diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/Benchmark.kt b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/Benchmark.kt
index 05d021b1bcfb77fa8ffeb0522510d49e39ef501c..d57a28e8bbcf4dc101e4814ecaa0d52fe28c08a9 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/Benchmark.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/Benchmark.kt
@@ -1,5 +1,8 @@
package theodolite.benchmark
+import io.fabric8.kubernetes.api.model.KubernetesResource
+import io.fabric8.kubernetes.api.model.Namespaced
+import io.fabric8.kubernetes.client.CustomResource
import io.quarkus.runtime.annotations.RegisterForReflection
import theodolite.util.ConfigurationOverride
import theodolite.util.LoadDimension
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecution.kt b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecution.kt
index 38d0f0389ce92a8720df05e892d11cf4f1ac480a..62ab75898d16ff2732ab6aa5c254ec8f87fb7266 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecution.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecution.kt
@@ -2,8 +2,6 @@ package theodolite.benchmark
import com.fasterxml.jackson.databind.annotation.JsonDeserialize
import io.fabric8.kubernetes.api.model.KubernetesResource
-import io.fabric8.kubernetes.api.model.Namespaced
-import io.fabric8.kubernetes.client.CustomResource
import io.quarkus.runtime.annotations.RegisterForReflection
import theodolite.util.ConfigurationOverride
import kotlin.properties.Delegates
@@ -26,7 +24,7 @@ import kotlin.properties.Delegates
*/
@JsonDeserialize
@RegisterForReflection
-class BenchmarkExecution : CustomResource(), Namespaced {
+class BenchmarkExecution : KubernetesResource {
var executionId: Int = 0
lateinit var name: String
lateinit var benchmark: String
@@ -34,7 +32,7 @@ class BenchmarkExecution : CustomResource(), Namespaced {
lateinit var resources: ResourceDefinition
lateinit var slos: List<Slo>
lateinit var execution: Execution
- lateinit var configOverrides: List<ConfigurationOverride?>
+ lateinit var configOverrides: MutableList<ConfigurationOverride?>
/**
* This execution encapsulates the [strategy], the [duration], the [repetitions], and the [restrictions]
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecutionList.kt b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecutionList.kt
deleted file mode 100644
index 50e8967f20aebad880ebd218136749af8e3ea6ee..0000000000000000000000000000000000000000
--- a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/BenchmarkExecutionList.kt
+++ /dev/null
@@ -1,5 +0,0 @@
-package theodolite.benchmark
-
-import io.fabric8.kubernetes.client.CustomResourceList
-
-class BenchmarkExecutionList : CustomResourceList<BenchmarkExecution>()
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmark.kt b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmark.kt
index c89e9e85323d6e51c104b1d34ff1ef9d8d4d60cd..aa9c36ad912437e3b104dccf6ff1f4dea5905946 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmark.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmark.kt
@@ -1,5 +1,6 @@
package theodolite.benchmark
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
import io.fabric8.kubernetes.api.model.KubernetesResource
import io.fabric8.kubernetes.api.model.Namespaced
import io.fabric8.kubernetes.client.CustomResource
@@ -30,16 +31,18 @@ private var DEFAULT_NAMESPACE = "default"
* for the deserializing in the [theodolite.execution.operator.TheodoliteOperator].
* @constructor construct an empty Benchmark.
*/
+@JsonDeserialize
@RegisterForReflection
-class KubernetesBenchmark : Benchmark, CustomResource(), Namespaced {
+class KubernetesBenchmark: KubernetesResource, Benchmark{
lateinit var name: String
lateinit var appResource: List<String>
lateinit var loadGenResource: List<String>
lateinit var resourceTypes: List<TypeName>
lateinit var loadTypes: List<TypeName>
lateinit var kafkaConfig: KafkaConfig
- private val namespace = System.getenv("NAMESPACE") ?: DEFAULT_NAMESPACE
- var path = System.getenv("THEODOLITE_APP_RESOURCES") ?: "./config"
+ var namespace = System.getenv("NAMESPACE") ?: DEFAULT_NAMESPACE
+ var path = System.getenv("THEODOLITE_APP_RESOURCES") ?: "./config"
+
/**
* Loads [KubernetesResource]s.
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmarkList.kt b/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmarkList.kt
deleted file mode 100644
index 0930875e96146fda58301478bda68b00c229e99f..0000000000000000000000000000000000000000
--- a/theodolite-quarkus/src/main/kotlin/theodolite/benchmark/KubernetesBenchmarkList.kt
+++ /dev/null
@@ -1,5 +0,0 @@
-package theodolite.benchmark
-
-import io.fabric8.kubernetes.client.CustomResourceList
-
-class KubernetesBenchmarkList : CustomResourceList<KubernetesBenchmark>()
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/TheodoliteExecutor.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/TheodoliteExecutor.kt
index ded4943c14ff8250f3dcad5279ff670af5fd7fd3..1fecc4664b02e51b581d84d6cbc19c9ddda82c9a 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/execution/TheodoliteExecutor.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/TheodoliteExecutor.kt
@@ -13,10 +13,6 @@ import theodolite.util.Resource
import theodolite.util.Results
import java.io.File
import java.io.PrintWriter
-import java.lang.IllegalArgumentException
-import java.lang.Thread.sleep
-import java.nio.file.Files
-import java.nio.file.Path
import java.time.Duration
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/AbstractStateHandler.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/AbstractStateHandler.kt
new file mode 100644
index 0000000000000000000000000000000000000000..75cbcad051e2055f25d876e66e0fffcdc249c4f5
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/AbstractStateHandler.kt
@@ -0,0 +1,55 @@
+package theodolite.execution.operator
+
+import io.fabric8.kubernetes.api.model.Namespaced
+import io.fabric8.kubernetes.client.CustomResource
+import io.fabric8.kubernetes.client.CustomResourceDoneable
+import io.fabric8.kubernetes.client.CustomResourceList
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.dsl.MixedOperation
+import io.fabric8.kubernetes.client.dsl.Resource
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext
+import java.lang.Thread.sleep
+
+abstract class AbstractStateHandler<T,L,D>(
+ private val context: CustomResourceDefinitionContext,
+ private val client: KubernetesClient,
+ private val crd: Class<T>,
+ private val crdList: Class<L>,
+ private val donableCRD: Class<D>
+ ): StateHandler where T : CustomResource, T: Namespaced, L: CustomResourceList<T>, D: CustomResourceDoneable<T> {
+
+ private val crdClient: MixedOperation<T, L, D, Resource<T, D>> =
+ this.client.customResources(this.context, this.crd, this.crdList, this.donableCRD)
+
+ @Synchronized
+ override fun setState(resourceName: String, f: (CustomResource) -> CustomResource?) {
+ this.crdClient
+ .inNamespace(this.client.namespace)
+ .list().items
+ .filter { item -> item.metadata.name == resourceName }
+ .map { customResource -> f(customResource) }
+ .forEach { this.crdClient.updateStatus(it as T) }
+ }
+
+ @Synchronized
+ override fun getState(resourceName: String, f: (CustomResource) -> String?): String? {
+ return this.crdClient
+ .inNamespace(this.client.namespace)
+ .list().items
+ .filter { item -> item.metadata.name == resourceName }
+ .map { customResource -> f(customResource) }
+ .firstOrNull()
+ }
+
+ @Synchronized
+ override fun blockUntilStateIsSet(resourceName: String, desiredStatusString: String, f: (CustomResource) -> String?, maxTries: Int): Boolean {
+ for (i in 0.rangeTo(maxTries)) {
+ val currentStatus = getState(resourceName, f)
+ if(currentStatus == desiredStatusString) {
+ return true
+ }
+ sleep(50)
+ }
+ return false
+ }
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/BenchmarkEventHandler.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/BenchmarkEventHandler.kt
deleted file mode 100644
index 7c8188e3c342cfc1b22fefdd3ca91e7dbce85905..0000000000000000000000000000000000000000
--- a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/BenchmarkEventHandler.kt
+++ /dev/null
@@ -1,67 +0,0 @@
-package theodolite.execution.operator
-
-import io.fabric8.kubernetes.client.informers.ResourceEventHandler
-import mu.KotlinLogging
-import theodolite.benchmark.KubernetesBenchmark
-
-private val logger = KotlinLogging.logger {}
-
-/**
- * Handles adding, updating and deleting KubernetesBenchmarks.
- *
- * @param controller The TheodoliteController that handles the application state
- *
- * @see TheodoliteController
- * @see KubernetesBenchmark
- */
-class BenchmarkEventHandler(private val controller: TheodoliteController) : ResourceEventHandler<KubernetesBenchmark> {
-
- /**
- * Add a KubernetesBenchmark.
- *
- * @param benchmark the KubernetesBenchmark to add
- *
- * @see KubernetesBenchmark
- */
- override fun onAdd(benchmark: KubernetesBenchmark) {
- benchmark.name = benchmark.metadata.name
- logger.info { "Add new benchmark ${benchmark.name}." }
- this.controller.benchmarks[benchmark.name] = benchmark
- }
-
- /**
- * Update a KubernetesBenchmark.
- *
- * @param oldBenchmark the KubernetesBenchmark to update
- * @param newBenchmark the updated KubernetesBenchmark
- *
- * @see KubernetesBenchmark
- */
- override fun onUpdate(oldBenchmark: KubernetesBenchmark, newBenchmark: KubernetesBenchmark) {
- logger.info { "Update benchmark ${newBenchmark.metadata.name}." }
- newBenchmark.name = newBenchmark.metadata.name
- if (this.controller.isInitialized() && this.controller.executor.getBenchmark().name == oldBenchmark.metadata.name) {
- this.controller.isUpdated.set(true)
- this.controller.executor.executor.run.compareAndSet(true, false)
- } else {
- onAdd(newBenchmark)
- }
- }
-
- /**
- * Delete a KubernetesBenchmark.
- *
- * @param benchmark the KubernetesBenchmark to delete
- *
- * @see KubernetesBenchmark
- */
- override fun onDelete(benchmark: KubernetesBenchmark, b: Boolean) {
- logger.info { "Delete benchmark ${benchmark.metadata.name}." }
- this.controller.benchmarks.remove(benchmark.metadata.name)
- if (this.controller.isInitialized() && this.controller.executor.getBenchmark().name == benchmark.metadata.name) {
- this.controller.isUpdated.set(true)
- this.controller.executor.executor.run.compareAndSet(true, false)
- logger.info { "Current benchmark stopped." }
- }
- }
-}
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ClusterSetup.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ClusterSetup.kt
new file mode 100644
index 0000000000000000000000000000000000000000..f3b4e9131a7184962310fe107586757cfb983097
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ClusterSetup.kt
@@ -0,0 +1,76 @@
+package theodolite.execution.operator
+
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient
+import io.fabric8.kubernetes.client.dsl.MixedOperation
+import io.fabric8.kubernetes.client.dsl.Resource
+import mu.KotlinLogging
+import org.json.JSONObject
+import theodolite.execution.Shutdown
+import theodolite.k8s.K8sContextFactory
+import theodolite.model.crd.*
+
+private val logger = KotlinLogging.logger {}
+
+class ClusterSetup(
+ private val executionCRDClient: MixedOperation<ExecutionCRD, BenchmarkExecutionList, DoneableExecution, Resource<ExecutionCRD, DoneableExecution>>,
+ private val benchmarkCRDClient: MixedOperation<BenchmarkCRD, KubernetesBenchmarkList, DoneableBenchmark, Resource<BenchmarkCRD, DoneableBenchmark>>,
+ private val client: NamespacedKubernetesClient
+
+ ) {
+ private val serviceMonitorContext = K8sContextFactory().create(
+ api = "v1",
+ scope = "Namespaced",
+ group = "monitoring.coreos.com",
+ plural = "servicemonitors"
+ )
+
+ fun clearClusterState(){
+ stopRunningExecution()
+ clearByLabel()
+ }
+
+ private fun stopRunningExecution() {
+ executionCRDClient
+ .inNamespace(client.namespace)
+ .list()
+ .items
+ .asSequence()
+ .filter { it.status.executionState == STATES.Running.value }
+ .forEach { execution ->
+ val benchmark = benchmarkCRDClient
+ .inNamespace(client.namespace)
+ .list()
+ .items
+ .firstOrNull { it.metadata.name == execution.spec.benchmark }
+
+ if (benchmark != null) {
+ execution.spec.name = execution.metadata.name
+ benchmark.spec.name = benchmark.metadata.name
+ Shutdown(execution.spec, benchmark.spec).start()
+ } else {
+ logger.error {
+ "Execution with state ${STATES.Running.value} was found, but no corresponding benchmark. " +
+ "Could not initialize cluster." }
+ }
+
+
+ }
+ }
+
+ private fun clearByLabel() {
+ this.client.services().withLabel("app.kubernetes.io/created-by=theodolite").delete()
+ this.client.apps().deployments().withLabel("app.kubernetes.io/created-by=theodolite").delete()
+ this.client.apps().statefulSets().withLabel("app.kubernetes.io/created-by=theodolite").delete()
+ this.client.configMaps().withLabel("app.kubernetes.io/created-by=theodolite").delete()
+
+ val serviceMonitors = JSONObject(
+ this.client.customResource(serviceMonitorContext)
+ .list(client.namespace, mapOf(Pair("app.kubernetes.io/created-by", "theodolite")))
+ )
+ .getJSONArray("items")
+
+ (0 until serviceMonitors.length())
+ .map { serviceMonitors.getJSONObject(it).getJSONObject("metadata").getString("name") }
+ .forEach { this.client.customResource(serviceMonitorContext).delete(client.namespace, it) }
+ }
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionEventHandler.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionEventHandler.kt
index ea62b7b895fce772a1f89019ea4aaac0f3957dc1..2469d0f7c016447b8c49135b907135beec74af63 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionEventHandler.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionEventHandler.kt
@@ -1,8 +1,11 @@
package theodolite.execution.operator
+import com.google.gson.Gson
+import com.google.gson.GsonBuilder
import io.fabric8.kubernetes.client.informers.ResourceEventHandler
import mu.KotlinLogging
import theodolite.benchmark.BenchmarkExecution
+import theodolite.model.crd.*
private val logger = KotlinLogging.logger {}
@@ -14,17 +17,30 @@ private val logger = KotlinLogging.logger {}
* @see TheodoliteController
* @see BenchmarkExecution
*/
-class ExecutionHandler(private val controller: TheodoliteController) : ResourceEventHandler<BenchmarkExecution> {
+class ExecutionHandler(
+ private val controller: TheodoliteController,
+ private val stateHandler: ExecutionStateHandler
+) : ResourceEventHandler<ExecutionCRD> {
+ private val gson: Gson = GsonBuilder().enableComplexMapKeySerialization().create()
/**
* Add an execution to the end of the queue of the TheodoliteController.
*
- * @param execution the execution to add
+ * @param ExecutionCRD the execution to add
*/
- override fun onAdd(execution: BenchmarkExecution) {
- execution.name = execution.metadata.name
- logger.info { "Add new execution ${execution.metadata.name} to queue." }
- this.controller.executionsQueue.add(execution)
+ @Synchronized
+ override fun onAdd(execution: ExecutionCRD) {
+ logger.info { "Add execution ${execution.metadata.name}" }
+ execution.spec.name = execution.metadata.name
+ when (this.stateHandler.getExecutionState(execution.metadata.name)) {
+ null -> this.stateHandler.setExecutionState(execution.spec.name, STATES.Pending)
+ STATES.Running -> {
+ this.stateHandler.setExecutionState(execution.spec.name, STATES.Restart)
+ if(this.controller.isExecutionRunning(execution.spec.name)){
+ this.controller.stop(restart=true)
+ }
+ }
+ }
}
/**
@@ -32,40 +48,39 @@ class ExecutionHandler(private val controller: TheodoliteController) : ResourceE
* added to the beginning of the queue of the TheodoliteController.
* Otherwise, it is just added to the beginning of the queue.
*
- * @param oldExecution the old execution
- * @param newExecution the new execution
+ * @param oldExecutionCRD the old execution
+ * @param newExecutionCRD the new execution
*/
- override fun onUpdate(oldExecution: BenchmarkExecution, newExecution: BenchmarkExecution) {
- logger.info { "Add updated execution to queue." }
- newExecution.name = newExecution.metadata.name
- try {
- this.controller.executionsQueue.removeIf { e -> e.name == newExecution.metadata.name }
- } catch (e: NullPointerException) {
- logger.warn { "No execution found for deletion" }
- }
- this.controller.executionsQueue.addFirst(newExecution)
- if (this.controller.isInitialized() && this.controller.executor.getExecution().name == newExecution.metadata.name) {
- this.controller.isUpdated.set(true)
- this.controller.executor.executor.run.compareAndSet(true, false)
+ @Synchronized
+ override fun onUpdate(oldExecution: ExecutionCRD, newExecution: ExecutionCRD) {
+ logger.info { "Receive update event for execution ${oldExecution.metadata.name}" }
+ newExecution.spec.name = newExecution.metadata.name
+ oldExecution.spec.name = oldExecution.metadata.name
+ if(gson.toJson(oldExecution.spec) != gson.toJson(newExecution.spec)) {
+ when(this.stateHandler.getExecutionState(newExecution.metadata.name)) {
+ STATES.Running -> {
+ this.stateHandler.setExecutionState(newExecution.spec.name, STATES.Restart)
+ if (this.controller.isExecutionRunning(newExecution.spec.name)){
+ this.controller.stop(restart=true)
+ }
+ }
+ STATES.Restart -> {} // should this set to pending?
+ else -> this.stateHandler.setExecutionState(newExecution.spec.name, STATES.Pending)
+ }
+ }
}
- }
/**
* Delete an execution from the queue of the TheodoliteController.
*
- * @param execution the execution to delete
+ * @param ExecutionCRD the execution to delete
*/
- override fun onDelete(execution: BenchmarkExecution, b: Boolean) {
- try {
- this.controller.executionsQueue.removeIf { e -> e.name == execution.metadata.name }
- logger.info { "Delete execution ${execution.metadata.name} from queue." }
- } catch (e: NullPointerException) {
- logger.warn { "No execution found for deletion" }
- }
- if (this.controller.isInitialized() && this.controller.executor.getExecution().name == execution.metadata.name) {
- this.controller.isUpdated.set(true)
- this.controller.executor.executor.run.compareAndSet(true, false)
- logger.info { "Current benchmark stopped." }
+ @Synchronized
+ override fun onDelete(execution: ExecutionCRD, b: Boolean) {
+ logger.info { "Delete execution ${execution.metadata.name}" }
+ if(execution.status.executionState == STATES.Running.value
+ && this.controller.isExecutionRunning(execution.spec.name)) {
+ this.controller.stop()
}
}
}
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionStateHandler.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionStateHandler.kt
new file mode 100644
index 0000000000000000000000000000000000000000..0e99e92360ef08c389ada40ea30728f118a05adc
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/ExecutionStateHandler.kt
@@ -0,0 +1,82 @@
+package theodolite.execution.operator
+
+import io.fabric8.kubernetes.client.CustomResource
+import io.fabric8.kubernetes.client.KubernetesClient
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext
+import theodolite.model.crd.BenchmarkExecutionList
+import theodolite.model.crd.ExecutionCRD
+import theodolite.model.crd.STATES
+import java.lang.Thread.sleep
+import java.time.Duration
+import java.time.Instant
+import java.util.concurrent.atomic.AtomicBoolean
+
+class ExecutionStateHandler(context: CustomResourceDefinitionContext, val client: KubernetesClient):
+ AbstractStateHandler<ExecutionCRD, BenchmarkExecutionList, DoneableExecution>(
+ context = context,
+ client = client,
+ crd = ExecutionCRD::class.java,
+ crdList = BenchmarkExecutionList::class.java,
+ donableCRD = DoneableExecution::class.java) {
+
+ private var runExecutionDurationTimer: AtomicBoolean = AtomicBoolean(false)
+
+ private fun getExecutionLambda() = { cr: CustomResource ->
+ var execState = ""
+ if (cr is ExecutionCRD) { execState = cr.status.executionState }
+ execState
+ }
+
+ private fun getDurationLambda() = { cr: CustomResource ->
+ var execState = ""
+ if (cr is ExecutionCRD) { execState = cr.status.executionState }
+ execState
+ }
+
+ fun setExecutionState(resourceName: String, status: STATES): Boolean {
+ setState(resourceName) {cr -> if(cr is ExecutionCRD) cr.status.executionState = status.value; cr}
+ return blockUntilStateIsSet(resourceName, status.value, getExecutionLambda())
+ }
+
+ fun getExecutionState(resourceName: String) : STATES? {
+ val status = this.getState(resourceName, getExecutionLambda())
+ return STATES.values().firstOrNull { it.value == status }
+ }
+
+ fun setDurationState(resourceName: String, duration: Duration) {
+ setState(resourceName) { cr -> if (cr is ExecutionCRD) cr.status.executionDuration = durationToK8sString(duration); cr }
+ blockUntilStateIsSet(resourceName, durationToK8sString(duration), getDurationLambda())
+ }
+
+ fun getDurationState(resourceName: String): String? {
+ return this.getState(resourceName, getDurationLambda())
+ }
+
+ private fun durationToK8sString(duration: Duration): String {
+ val sec = duration.seconds
+ return when {
+ sec <= 120 -> "${sec}s" // max 120s
+ sec < 60 * 99 -> "${duration.toMinutes()}m" // max 99m
+ sec < 60 * 60 * 99 -> "${duration.toHours()}h" // max 99h
+ else -> "${duration.toDays()}d + ${duration.minusDays(duration.toDays()).toHours()}h"
+ }
+ }
+
+ fun startDurationStateTimer(resourceName: String) {
+ this.runExecutionDurationTimer.set(true)
+ val startTime = Instant.now().toEpochMilli()
+ Thread {
+ while (this.runExecutionDurationTimer.get()) {
+ val duration = Duration.ofMillis(Instant.now().minusMillis(startTime).toEpochMilli())
+ setDurationState(resourceName, duration)
+ sleep(100 * 1)
+ }
+ }.start()
+ }
+
+ @Synchronized
+ fun stopDurationStateTimer() {
+ this.runExecutionDurationTimer.set(false)
+ sleep(100 * 2)
+ }
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/StateHandler.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/StateHandler.kt
new file mode 100644
index 0000000000000000000000000000000000000000..0fbd97e5cca4a9be220eb0b0c89ea0af129a7860
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/StateHandler.kt
@@ -0,0 +1,15 @@
+package theodolite.execution.operator
+
+import io.fabric8.kubernetes.client.CustomResource
+private const val MAX_TRIES: Int = 5
+
+interface StateHandler {
+ fun setState(resourceName: String, f: (CustomResource) -> CustomResource?)
+ fun getState(resourceName: String, f: (CustomResource) -> String?): String?
+ fun blockUntilStateIsSet(
+ resourceName: String,
+ desiredStatusString: String,
+ f: (CustomResource) -> String?,
+ maxTries: Int = MAX_TRIES): Boolean
+
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteController.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteController.kt
index 0e889e0393e17fd65f4d8f12c5b95a3dbed7f593..3d76a8620c0112a15d94f0227efcdd5ef9afce1d 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteController.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteController.kt
@@ -1,122 +1,194 @@
package theodolite.execution.operator
import io.fabric8.kubernetes.client.NamespacedKubernetesClient
-import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext
+import io.fabric8.kubernetes.client.dsl.MixedOperation
+import io.fabric8.kubernetes.client.dsl.Resource
import mu.KotlinLogging
import theodolite.benchmark.BenchmarkExecution
import theodolite.benchmark.KubernetesBenchmark
import theodolite.execution.TheodoliteExecutor
+import theodolite.model.crd.*
+import theodolite.util.ConfigurationOverride
+import theodolite.util.PatcherDefinition
import java.lang.Thread.sleep
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.ConcurrentLinkedDeque
-import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.atomic.AtomicInteger
private val logger = KotlinLogging.logger {}
/**
* The controller implementation for Theodolite.
*
- * Maintains a Dequeue, based on ConcurrentLinkedDequeue, of executions to be executed for a benchmark.
- *
- * @param client The NamespacedKubernetesClient
- * @param executionContext The CustomResourceDefinitionContext
- *
* @see NamespacedKubernetesClient
* @see CustomResourceDefinitionContext
* @see BenchmarkExecution
* @see KubernetesBenchmark
* @see ConcurrentLinkedDeque
*/
+
class TheodoliteController(
- val client: NamespacedKubernetesClient,
- val executionContext: CustomResourceDefinitionContext,
- val path: String
+ private val namespace: String,
+ val path: String,
+ private val executionCRDClient: MixedOperation<ExecutionCRD, BenchmarkExecutionList, DoneableExecution, Resource<ExecutionCRD, DoneableExecution>>,
+ private val benchmarkCRDClient: MixedOperation<BenchmarkCRD, KubernetesBenchmarkList, DoneableBenchmark, Resource<BenchmarkCRD, DoneableBenchmark>>,
+ private val executionStateHandler: ExecutionStateHandler
) {
lateinit var executor: TheodoliteExecutor
- val executionsQueue: ConcurrentLinkedDeque<BenchmarkExecution> = ConcurrentLinkedDeque()
- val benchmarks: ConcurrentHashMap<String, KubernetesBenchmark> = ConcurrentHashMap()
- var isUpdated = AtomicBoolean(false)
- var executionID = AtomicInteger(0)
-
/**
+ *
* Runs the TheodoliteController forever.
*/
fun run() {
+ sleep(5000) // wait until all states are correctly set
while (true) {
- try {
- reconcile()
- logger.info { "Theodolite is waiting for new matching benchmark and execution." }
- logger.info { "Currently available executions: " }
- executionsQueue.forEach {
- logger.info { "${it.name} : waiting for : ${it.benchmark}" }
- }
- logger.info { "Currently available benchmarks: " }
- benchmarks.forEach {
- logger.info { it.key }
- }
- sleep(2000)
- } catch (e: InterruptedException) {
- logger.error { "Execution interrupted with error: $e." }
- }
+ reconcile()
+ sleep(2000)
}
}
- /**
- * Ensures that the application state corresponds to the defined KubernetesBenchmarks and BenchmarkExecutions.
- *
- * @see KubernetesBenchmark
- * @see BenchmarkExecution
- */
- @Synchronized
private fun reconcile() {
- while (executionsQueue.isNotEmpty()) {
- val execution = executionsQueue.peek()
- val benchmark = benchmarks[execution.benchmark]
-
- if (benchmark == null) {
- logger.debug { "No benchmark found for execution ${execution.name}." }
- sleep(1000)
+ do {
+ val execution = getNextExecution()
+ if (execution != null) {
+ val benchmark = getBenchmarks()
+ .firstOrNull { it.name == execution.benchmark }
+ if (benchmark != null) {
+ runExecution(execution, benchmark)
+ }
} else {
- runExecution(execution, benchmark)
+ logger.info { "Could not find executable execution." }
}
- }
+ } while (execution != null)
}
/**
* Execute a benchmark with a defined KubernetesBenchmark and BenchmarkExecution
*
- * @see KubernetesBenchmark
* @see BenchmarkExecution
*/
- @Synchronized
- fun runExecution(execution: BenchmarkExecution, benchmark: KubernetesBenchmark) {
- execution.executionId = executionID.getAndSet(executionID.get() + 1)
- isUpdated.set(false)
- benchmark.path = path
- logger.info { "Start execution ${execution.name} with benchmark ${benchmark.name}." }
- executor = TheodoliteExecutor(config = execution, kubernetesBenchmark = benchmark)
- executor.run()
+ private fun runExecution(execution: BenchmarkExecution, benchmark: KubernetesBenchmark) {
+ setAdditionalLabels(execution.name,
+ "deployed-for-execution",
+ benchmark.appResource + benchmark.loadGenResource,
+ execution)
+ setAdditionalLabels(benchmark.name,
+ "deployed-for-benchmark",
+ benchmark.appResource + benchmark.loadGenResource,
+ execution)
+ setAdditionalLabels("theodolite",
+ "app.kubernetes.io/created-by",
+ benchmark.appResource + benchmark.loadGenResource,
+ execution)
+
+ executionStateHandler.setExecutionState(execution.name, STATES.Running)
+ executionStateHandler.startDurationStateTimer(execution.name)
try {
- if (!isUpdated.get()) {
- this.executionsQueue.removeIf { e -> e.name == execution.name }
- client.customResource(executionContext).delete(client.namespace, execution.metadata.name)
+ executor = TheodoliteExecutor(execution, benchmark)
+ executor.run()
+ when (executionStateHandler.getExecutionState(execution.name)) {
+ STATES.Restart -> runExecution(execution, benchmark)
+ STATES.Running -> {
+ executionStateHandler.setExecutionState(execution.name, STATES.Finished)
+ logger.info { "Execution of ${execution.name} is finally stopped." }
+ }
}
} catch (e: Exception) {
- logger.warn { "Deletion skipped." }
+ logger.error { "Failure while executing execution ${execution.name} with benchmark ${benchmark.name}." }
+ logger.error { "Problem is: $e" }
+ executionStateHandler.setExecutionState(execution.name, STATES.Failure)
}
+ executionStateHandler.stopDurationStateTimer()
+ }
- logger.info { "Execution of ${execution.name} is finally stopped." }
+ @Synchronized
+ fun stop(restart: Boolean = false) {
+ if (!::executor.isInitialized) return
+ if (restart) {
+ executionStateHandler.setExecutionState(this.executor.getExecution().name, STATES.Restart)
+ } else {
+ executionStateHandler.setExecutionState(this.executor.getExecution().name, STATES.Interrupted)
+ logger.warn { "Execution ${executor.getExecution().name} unexpected interrupted" }
+ }
+ this.executor.executor.run.set(false)
+ }
+
+ /**
+ * @return all available [BenchmarkCRD]s
+ */
+ private fun getBenchmarks(): List<KubernetesBenchmark> {
+ return this.benchmarkCRDClient
+ .inNamespace(namespace)
+ .list()
+ .items
+ .map { it.spec.name = it.metadata.name; it }
+ .map { it.spec.path = path; it }
+ .map { it.spec }
}
/**
- * @return true if the TheodoliteExecutor of this controller is initialized. Else returns false.
+ * Get the [BenchmarkExecution] for the next run. Which [BenchmarkExecution]
+ * is selected for the next execution depends on three points:
*
- * @see TheodoliteExecutor
+ * 1. Only executions are considered for which a matching benchmark is available on the cluster
+ * 2. The Status of the execution must be [STATES.Pending] or [STATES.Restart]
+ * 3. Of the remaining [BenchmarkCRD], those with status [STATES.Restart] are preferred,
+ * then, if there is more than one, the oldest execution is chosen.
+ *
+ * @return the next execution or null
*/
- @Synchronized
- fun isInitialized(): Boolean {
- return ::executor.isInitialized
+ private fun getNextExecution(): BenchmarkExecution? {
+ val availableBenchmarkNames = getBenchmarks()
+ .map { it.name }
+
+ return executionCRDClient
+ .inNamespace(namespace)
+ .list()
+ .items
+ .asSequence()
+ .map { it.spec.name = it.metadata.name; it }
+ .filter {
+ it.status.executionState == STATES.Pending.value ||
+ it.status.executionState == STATES.Restart.value
+ }
+ .filter { availableBenchmarkNames.contains(it.spec.benchmark) }
+ .sortedWith(stateComparator().thenBy { it.metadata.creationTimestamp })
+ .map { it.spec }
+ .firstOrNull()
+ }
+
+ /**
+ * Simple comparator which can be used to order a list of [ExecutionCRD] such that executions with
+ * status [STATES.Restart] are before all other executions.
+ */
+ private fun stateComparator() = Comparator<ExecutionCRD> { a, b ->
+ when {
+ (a == null && b == null) -> 0
+ (a.status.executionState == STATES.Restart.value) -> -1
+ else -> 1
+ }
+ }
+
+ fun isExecutionRunning(executionName: String): Boolean {
+ return this.executor.getExecution().name == executionName
+ }
+
+ private fun setAdditionalLabels(
+ labelValue: String,
+ labelName: String,
+ resources: List<String>,
+ execution: BenchmarkExecution
+ ) {
+ val additionalConfigOverrides = mutableListOf<ConfigurationOverride>()
+ resources.forEach {
+ run {
+ val configurationOverride = ConfigurationOverride()
+ configurationOverride.patcher = PatcherDefinition()
+ configurationOverride.patcher.type = "LabelPatcher"
+ configurationOverride.patcher.variableName = labelName
+ configurationOverride.patcher.resource = it
+ configurationOverride.value = labelValue
+ additionalConfigOverrides.add(configurationOverride)
+ }
+ }
+ execution.configOverrides.addAll(additionalConfigOverrides)
}
-}
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteOperator.kt b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteOperator.kt
index 5e15a4a80e67f47a42d605d4af39102927139331..49fbf54eecf2a178ae8a0b38bedee1c597a80e1f 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteOperator.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/execution/operator/TheodoliteOperator.kt
@@ -1,13 +1,12 @@
package theodolite.execution.operator
import io.fabric8.kubernetes.client.DefaultKubernetesClient
+import io.fabric8.kubernetes.client.dsl.MixedOperation
+import io.fabric8.kubernetes.client.dsl.Resource
import io.fabric8.kubernetes.internal.KubernetesDeserializer
import mu.KotlinLogging
-import theodolite.benchmark.BenchmarkExecution
-import theodolite.benchmark.BenchmarkExecutionList
-import theodolite.benchmark.KubernetesBenchmark
-import theodolite.benchmark.KubernetesBenchmarkList
import theodolite.k8s.K8sContextFactory
+import theodolite.model.crd.*
private const val DEFAULT_NAMESPACE = "default"
@@ -16,7 +15,7 @@ private const val EXECUTION_SINGULAR = "execution"
private const val EXECUTION_PLURAL = "executions"
private const val BENCHMARK_SINGULAR = "benchmark"
private const val BENCHMARK_PLURAL = "benchmarks"
-private const val API_VERSION = "v1alpha1"
+private const val API_VERSION = "v1"
private const val RESYNC_PERIOD = 10 * 60 * 1000.toLong()
private const val GROUP = "theodolite.com"
private val logger = KotlinLogging.logger {}
@@ -33,42 +32,82 @@ class TheodoliteOperator {
* Start the operator.
*/
fun start() {
+ // FIXME("Remove all benchmark state handling")
logger.info { "Using $namespace as namespace." }
val client = DefaultKubernetesClient().inNamespace(namespace)
+ client.use {
+ KubernetesDeserializer.registerCustomKind(
+ "$GROUP/$API_VERSION",
+ EXECUTION_SINGULAR,
+ ExecutionCRD::class.java
+ )
- KubernetesDeserializer.registerCustomKind(
- "$GROUP/$API_VERSION",
- EXECUTION_SINGULAR,
- BenchmarkExecution::class.java
- )
-
- KubernetesDeserializer.registerCustomKind(
- "$GROUP/$API_VERSION",
- BENCHMARK_SINGULAR,
- KubernetesBenchmark::class.java
- )
-
- val contextFactory = K8sContextFactory()
- val executionContext = contextFactory.create(API_VERSION, SCOPE, GROUP, EXECUTION_PLURAL)
- val benchmarkContext = contextFactory.create(API_VERSION, SCOPE, GROUP, BENCHMARK_PLURAL)
-
- val appResource = System.getenv("THEODOLITE_APP_RESOURCES") ?: "./config"
- val controller = TheodoliteController(client = client, executionContext = executionContext, path = appResource)
-
- val informerFactory = client.informers()
- val informerExecution = informerFactory.sharedIndexInformerForCustomResource(
- executionContext, BenchmarkExecution::class.java,
- BenchmarkExecutionList::class.java, RESYNC_PERIOD
- )
- val informerBenchmark = informerFactory.sharedIndexInformerForCustomResource(
- benchmarkContext, KubernetesBenchmark::class.java,
- KubernetesBenchmarkList::class.java, RESYNC_PERIOD
- )
-
- informerExecution.addEventHandler(ExecutionHandler(controller))
- informerBenchmark.addEventHandler(BenchmarkEventHandler(controller))
- informerFactory.startAllRegisteredInformers()
-
- controller.run()
+ KubernetesDeserializer.registerCustomKind(
+ "$GROUP/$API_VERSION",
+ BENCHMARK_SINGULAR,
+ BenchmarkCRD::class.java
+ )
+
+ val contextFactory = K8sContextFactory()
+ val executionContext = contextFactory.create(API_VERSION, SCOPE, GROUP, EXECUTION_PLURAL)
+ val benchmarkContext = contextFactory.create(API_VERSION, SCOPE, GROUP, BENCHMARK_PLURAL)
+
+ val executionCRDClient: MixedOperation<
+ ExecutionCRD,
+ BenchmarkExecutionList,
+ DoneableExecution,
+ Resource<ExecutionCRD, DoneableExecution>>
+ = client.customResources(
+ executionContext,
+ ExecutionCRD::class.java,
+ BenchmarkExecutionList::class.java,
+ DoneableExecution::class.java)
+
+ val benchmarkCRDClient: MixedOperation<
+ BenchmarkCRD,
+ KubernetesBenchmarkList,
+ DoneableBenchmark,
+ Resource<BenchmarkCRD, DoneableBenchmark>>
+ = client.customResources(
+ benchmarkContext,
+ BenchmarkCRD::class.java,
+ KubernetesBenchmarkList::class.java,
+ DoneableBenchmark::class.java)
+
+ val executionStateHandler = ExecutionStateHandler(
+ context = executionContext,
+ client = client)
+
+ val appResource = System.getenv("THEODOLITE_APP_RESOURCES") ?: "./config"
+ val controller =
+ TheodoliteController(
+ namespace = client.namespace,
+ path = appResource,
+ benchmarkCRDClient = benchmarkCRDClient,
+ executionCRDClient = executionCRDClient,
+ executionStateHandler = executionStateHandler)
+
+ val informerFactory = client.informers()
+ val informerExecution = informerFactory.sharedIndexInformerForCustomResource(
+ executionContext,
+ ExecutionCRD::class.java,
+ BenchmarkExecutionList::class.java,
+ RESYNC_PERIOD
+ )
+
+ informerExecution.addEventHandler(ExecutionHandler(
+ controller = controller,
+ stateHandler = executionStateHandler))
+
+ ClusterSetup(
+ executionCRDClient = executionCRDClient,
+ benchmarkCRDClient = benchmarkCRDClient,
+ client = client
+ ).clearClusterState()
+
+ informerFactory.startAllRegisteredInformers()
+ controller.run()
+
+ }
}
}
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/K8sManager.kt b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/K8sManager.kt
index c58225bf855c70a5a7057132617418a89b0816a8..af9e780d505d7392eff535ee9d3be11780561ce4 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/K8sManager.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/K8sManager.kt
@@ -63,6 +63,12 @@ class K8sManager(private val client: NamespacedKubernetesClient) {
}
}
+ fun setLabel(resource: KubernetesResource) {
+ when(resource) {
+ is Deployment -> this.client
+ }
+ }
+
private fun blockUntilPodsDeleted(podLabel: String) {
while (!this.client.pods().withLabel(podLabel).list().items.isNullOrEmpty()) {
logger.info { "Wait for pods with label '$podLabel' to be deleted." }
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/ServiceMonitorWrapper.kt b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/ServiceMonitorWrapper.kt
index 4950cee225e103ff095def91de64471ec1894a79..56452d74968db0fd4c939f44f3ed8a7abbe7b928 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/ServiceMonitorWrapper.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/ServiceMonitorWrapper.kt
@@ -53,4 +53,9 @@ class ServiceMonitorWrapper(private val serviceMonitor: Map<String, String>) : C
val smAsMap = this.serviceMonitor["metadata"]!! as Map<String, String>
return smAsMap["name"]!!
}
+
+ fun getLabels(): Map<String, String>{
+ val smAsMap = this.serviceMonitor["metadata"]!! as Map<String, String>
+ return smAsMap["labels"]!! as Map<String, String>
+ }
}
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/BenchmarkCRD.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/BenchmarkCRD.kt
new file mode 100644
index 0000000000000000000000000000000000000000..326aa10a21bebd913eaafcb8315188288ae97ff1
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/BenchmarkCRD.kt
@@ -0,0 +1,11 @@
+package theodolite.model.crd
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
+import io.fabric8.kubernetes.api.model.Namespaced
+import io.fabric8.kubernetes.client.CustomResource
+import theodolite.benchmark.KubernetesBenchmark
+
+@JsonDeserialize
+class BenchmarkCRD(
+ var spec: KubernetesBenchmark = KubernetesBenchmark()
+) : CustomResource(), Namespaced
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/BenchmarkExecutionList.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/BenchmarkExecutionList.kt
new file mode 100644
index 0000000000000000000000000000000000000000..2b2dcc07f9c37f1712109e3d092f2db0c139e1c8
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/BenchmarkExecutionList.kt
@@ -0,0 +1,5 @@
+package theodolite.model.crd
+
+import io.fabric8.kubernetes.client.CustomResourceList
+
+class BenchmarkExecutionList : CustomResourceList<ExecutionCRD>()
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/DoneableBenchmark.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/DoneableBenchmark.kt
new file mode 100644
index 0000000000000000000000000000000000000000..e00e8268b2ec8eba17b3706feb3940eded1b2b0c
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/DoneableBenchmark.kt
@@ -0,0 +1,7 @@
+package theodolite.model.crd
+
+import io.fabric8.kubernetes.api.builder.Function
+import io.fabric8.kubernetes.client.CustomResourceDoneable
+
+class DoneableBenchmark(resource: BenchmarkCRD, function: Function<BenchmarkCRD, BenchmarkCRD>) :
+ CustomResourceDoneable<BenchmarkCRD>(resource, function)
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/DoneableExecution.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/DoneableExecution.kt
new file mode 100644
index 0000000000000000000000000000000000000000..be07725b405c29a0d9000b6e6ec455536ad111fb
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/DoneableExecution.kt
@@ -0,0 +1,8 @@
+package theodolite.execution.operator
+
+import io.fabric8.kubernetes.client.CustomResourceDoneable
+import io.fabric8.kubernetes.api.builder.Function
+import theodolite.model.crd.ExecutionCRD
+
+class DoneableExecution(resource: ExecutionCRD, function: Function<ExecutionCRD, ExecutionCRD>) :
+ CustomResourceDoneable<ExecutionCRD>(resource, function)
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/ExecutionCRD.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/ExecutionCRD.kt
new file mode 100644
index 0000000000000000000000000000000000000000..79a387cee250d3abf0fdb576a5f0f33424596792
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/ExecutionCRD.kt
@@ -0,0 +1,13 @@
+package theodolite.model.crd
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
+import io.fabric8.kubernetes.api.model.KubernetesResource
+import io.fabric8.kubernetes.api.model.Namespaced
+import io.fabric8.kubernetes.client.CustomResource
+import theodolite.benchmark.BenchmarkExecution
+
+@JsonDeserialize
+class ExecutionCRD(
+ var spec: BenchmarkExecution = BenchmarkExecution(),
+ var status: ExecutionStatus = ExecutionStatus()
+ ) : CustomResource(), Namespaced
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/ExecutionStatus.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/ExecutionStatus.kt
new file mode 100644
index 0000000000000000000000000000000000000000..43e9035b3120eb22304576f2006092eec376b6d2
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/ExecutionStatus.kt
@@ -0,0 +1,13 @@
+package theodolite.model.crd
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize
+import io.fabric8.kubernetes.api.model.KubernetesResource
+import io.fabric8.kubernetes.api.model.Namespaced
+import io.fabric8.kubernetes.client.CustomResource
+
+@JsonDeserialize
+class ExecutionStatus(): KubernetesResource, CustomResource(), Namespaced {
+ var executionState: String = ""
+ var executionDuration: String = "-"
+
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/KubernetesBenchmarkList.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/KubernetesBenchmarkList.kt
new file mode 100644
index 0000000000000000000000000000000000000000..8ad0a493d948bf5f78741052100766dcf6e316ec
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/KubernetesBenchmarkList.kt
@@ -0,0 +1,5 @@
+package theodolite.model.crd
+
+import io.fabric8.kubernetes.client.CustomResourceList
+
+class KubernetesBenchmarkList : CustomResourceList<BenchmarkCRD>()
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/STATES.kt b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/STATES.kt
new file mode 100644
index 0000000000000000000000000000000000000000..3fcd9751c3ba755121b4b467b45e2cb3f4f13bee
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/model/crd/STATES.kt
@@ -0,0 +1,11 @@
+package theodolite.model.crd
+
+enum class STATES(val value: String) {
+ Running("RUNNING"),
+ Pending("PENDING"),
+ Failure("FAILURE"),
+ Finished("FINISHED"),
+ Restart("RESTART"),
+ Interrupted("INTERRUPTED"),
+ NoState("NoState")
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/patcher/LabelPatcher.kt b/theodolite-quarkus/src/main/kotlin/theodolite/patcher/LabelPatcher.kt
new file mode 100644
index 0000000000000000000000000000000000000000..d9feff00726c8c73483118276eeae7b7975d8c8e
--- /dev/null
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/patcher/LabelPatcher.kt
@@ -0,0 +1,50 @@
+package theodolite.patcher
+
+import io.fabric8.kubernetes.api.model.ConfigMap
+import io.fabric8.kubernetes.api.model.KubernetesResource
+import io.fabric8.kubernetes.api.model.Service
+import io.fabric8.kubernetes.api.model.apps.Deployment
+import io.fabric8.kubernetes.api.model.apps.StatefulSet
+import io.fabric8.kubernetes.client.CustomResource
+
+class LabelPatcher(private val k8sResource: KubernetesResource, val variableName: String) :
+ AbstractPatcher(k8sResource, variableName) {
+
+ override fun <String> patch(labelValue: String) {
+ println("call patcher for resource $k8sResource !")
+ if(labelValue is kotlin.String){
+ when(k8sResource){
+ is Deployment -> {
+ if (k8sResource.metadata.labels == null){
+ k8sResource.metadata.labels = mutableMapOf()
+ }
+ k8sResource.metadata.labels[this.variableName] = labelValue
+ }
+ is StatefulSet -> {
+ if (k8sResource.metadata.labels == null){
+ k8sResource.metadata.labels = mutableMapOf()
+ }
+ k8sResource.metadata.labels[this.variableName] = labelValue
+ }
+ is Service -> {
+ if (k8sResource.metadata.labels == null){
+ k8sResource.metadata.labels = mutableMapOf()
+ }
+ k8sResource.metadata.labels[this.variableName] = labelValue
+ }
+ is ConfigMap -> {
+ if (k8sResource.metadata.labels == null){
+ k8sResource.metadata.labels = mutableMapOf()
+ }
+ k8sResource.metadata.labels[this.variableName] = labelValue
+ }
+ is CustomResource -> {
+ if (k8sResource.metadata.labels == null){
+ k8sResource.metadata.labels = mutableMapOf()
+ }
+ k8sResource.metadata.labels[this.variableName] = labelValue
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/patcher/PatcherFactory.kt b/theodolite-quarkus/src/main/kotlin/theodolite/patcher/PatcherFactory.kt
index 2ee1f6c7b46322cb0f8de03c37aabe64ccf0ba5a..9ca6570ff56a673ffde144b68d3f3d9c90913ef9 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/patcher/PatcherFactory.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/patcher/PatcherFactory.kt
@@ -45,6 +45,7 @@ class PatcherFactory {
patcherDefinition.variableName
)
"SchedulerNamePatcher" -> SchedulerNamePatcher(resource)
+ "LabelPatcher" -> LabelPatcher(resource, patcherDefinition.variableName)
else -> throw IllegalArgumentException("Patcher type ${patcherDefinition.type} not found")
}
}
diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/util/KafkaConfig.kt b/theodolite-quarkus/src/main/kotlin/theodolite/util/KafkaConfig.kt
index 398ff90bed8f48683321e2375458b3a065c39463..4e72ccb0d86749a6538c26556241ac114ef8d9a4 100644
--- a/theodolite-quarkus/src/main/kotlin/theodolite/util/KafkaConfig.kt
+++ b/theodolite-quarkus/src/main/kotlin/theodolite/util/KafkaConfig.kt
@@ -28,6 +28,7 @@ class KafkaConfig {
* Wrapper for a topic definition.
*/
@RegisterForReflection
+ @JsonDeserialize
class TopicWrapper {
/**
* The topic name