Create kubernetes resources and restructure for test preparation

theodolite-quarkus/src/main/kotlin/theodolite/execution/KafkaBenchmarkExecutor.kt

@@ -13,7 +13,7 @@ class KafkaBenchmarkExecutor(benchmark: Benchmark, results: Results, executionDu
     override fun runExperiment(load: LoadDimension, res: Resource): Boolean {
         benchmark.start(load, res)
         this.waitAndLog()
-        benchmark.stop()
+        benchmark.clearClusterEnvironment()
         // todo evaluate
         val result = false // if success else false
         this.results.setResult(Pair(load, res), result)

theodolite-quarkus/src/main/kotlin/theodolite/execution/TheodoliteExecutor.kt

 package theodolite.execution
 
+import theodolite.k8s.UC1Benchmark
 import theodolite.strategies.restriction.LowerBoundRestriction
 import theodolite.strategies.searchstrategy.CompositeStrategy
 import theodolite.strategies.searchstrategy.LinearSearch
@@ -8,7 +9,18 @@ import java.time.Duration
 
 class TheodoliteExecutor() {
     private fun loadConfig(): Config {
-        val benchmark: KafkaBenchmark = KafkaBenchmark(emptyMap())
+        val benchmark: UC1Benchmark = UC1Benchmark(
+            UC1Benchmark.UC1BenchmarkConfig(
+                zookeeperConnectionString = "my-confluent-cp-zookeeper:2181",
+                kafkaIPConnectionString = "my-confluent-cp-kafka:9092",
+                schemaRegistryConnectionString = "http://my-confluent-cp-schema-registry:8081",
+                kafkaPartition = 40,
+                kafkaReplication = 3,
+                kafkaTopics = listOf("input", "output"),
+                ucDeploymentPath = "",
+                ucServicePath = "",
+                wgDeploymentPath = ""
+        ))
         val results: Results = Results()
         val executionDuration = Duration.ofSeconds(60*5 )
         val executor: BenchmarkExecutor = KafkaBenchmarkExecutor(benchmark, results, executionDuration)

theodolite-quarkus/src/main/kotlin/theodolite/k8s/UC1Benchmark.kt

@@ -3,14 +3,10 @@ package theodolite.k8s
 import io.fabric8.kubernetes.api.model.Service
 import io.fabric8.kubernetes.api.model.apps.Deployment
 import io.fabric8.kubernetes.client.DefaultKubernetesClient
-import io.fabric8.kubernetes.client.KubernetesClient
 import io.fabric8.kubernetes.client.NamespacedKubernetesClient
-import org.apache.kafka.common.internals.Topic
 import theodolite.util.Benchmark
 import theodolite.util.LoadDimension
 import theodolite.util.Resource
-import theodolite.k8s.WorkloadGeneratorStateCleaner
-import java.io.FileNotFoundException
 
 class UC1Benchmark(config: UC1BenchmarkConfig) : Benchmark(config) {
     val workloadGeneratorStateCleaner: WorkloadGeneratorStateCleaner
@@ -31,49 +27,49 @@ class UC1Benchmark(config: UC1BenchmarkConfig) : Benchmark(config) {
         this.yamlLoader = YamlLoader(this.kubernetesClient)
         this.deploymentManager = DeploymentManager(this.kubernetesClient)
         this.serviceManager = ServiceManager(this.kubernetesClient)
-        ucDeployment = this.yamlLoader.loadDeployment(this.config.ucDeploymentPath)!!
-        ucService = this.yamlLoader.loadService(this.config.ucServicePath)!!
+        ucDeployment = this.yamlLoader.loadDeployment(this.config.ucDeploymentPath)
+        ucService = this.yamlLoader.loadService(this.config.ucServicePath)
 
 
 
     }
 
-    override fun start(load: LoadDimension, resources: Resource) {
-        // TODO("extract code to dedicated functions. And, we should create a abstration layer to create the benchmark core, which are identically by all benchmarks")
+    override fun clearClusterEnvironment() {
+        this.workloadGeneratorStateCleaner.deleteAll()
+        this.topicManager.deleteTopics(this.config.kafkaTopics)
+        this.deploymentManager.delete(this.ucDeployment)
+        this.serviceManager.delete(this.ucService)
+    }
+
+    override fun initializeClusterEnvironment() {
         this.workloadGeneratorStateCleaner.deleteAll()
         this.topicManager.deleteTopics(this.config.kafkaTopics)
         this.topicManager.createTopics(this.config.kafkaTopics, this.config.kafkaPartition, this.config.kafkaReplication)
+    }
 
+    override fun startSUT(resources: Resource) {
         // set environment variables
         val environmentVariables: MutableMap<String, String> = mutableMapOf()
         environmentVariables.put("KAFKA_BOOTSTRAP_SERVER", this.config.kafkaIPConnectionString)
-        // environmentVariables.put("replicas", this.config.deploymentReplicas) TODO("add possibility to set replicas")
-       environmentVariables.put("SCHEMA_REGISTRY_URL", this.config.schemaRegistryConnectionString)
+        environmentVariables.put("SCHEMA_REGISTRY_URL", this.config.schemaRegistryConnectionString)
+
+        // setup deployment
+        this.deploymentManager.setReplica(ucDeployment, resources.get())
         this.deploymentManager.setWorkloadEnv(ucDeployment,"uc-application", environmentVariables)
 
         // create kubernetes resources
         this.deploymentManager.deploy(ucDeployment)
         this.serviceManager.deploy(ucService)
-
-        this.startWorkloadGenerator("uc1", load, "uc1")
-
-    }
-
-    override fun stop() {
-        this.workloadGeneratorStateCleaner.deleteAll()
-        this.topicManager.deleteTopics(this.config.kafkaTopics)
-        this.deploymentManager.delete(this.ucDeployment)
-        this.serviceManager.delete(this.ucService)
     }
 
-    override fun startWorkloadGenerator(wg: String, load: LoadDimension, ucId: String) {
+    override fun startWorkloadGenerator(load: LoadDimension) {
         val wgDeployment = this.yamlLoader.loadDeployment(this.config.wgDeploymentPath)
         val environmentVariables: MutableMap<String, String> = mutableMapOf()
         environmentVariables.put("NUM_SENSORS", load.get().toString())
         // TODO ("calculate number of required instances")
         val requiredInstances: Int = 1
         environmentVariables.put("NUM_INSTANCES", requiredInstances.toString())
-        wgDeployment?.let { this.deploymentManager.setWorkloadEnv(it, "workload-generator", environmentVariables) }
+        this.deploymentManager.setWorkloadEnv(wgDeployment, "workload-generator", environmentVariables)
     }
 
     data class UC1BenchmarkConfig(
@@ -84,7 +80,6 @@ class UC1Benchmark(config: UC1BenchmarkConfig) : Benchmark(config) {
         val kafkaReplication: Short,
         val kafkaPartition: Int,
         val ucDeploymentPath: String,
-        val ucDeploymentReplicas: String,
         val ucServicePath: String,
         val wgDeploymentPath: String
         ) {}

theodolite-quarkus/src/main/kotlin/theodolite/util/Benchmark.kt

@@ -4,10 +4,17 @@ import theodolite.k8s.UC1Benchmark
 
 // todo: needs cluster and resource config
 abstract class Benchmark(val config: UC1Benchmark.UC1BenchmarkConfig) {
-    abstract fun start(load: LoadDimension, resources: Resource);
+    fun start(load: LoadDimension, resources: Resource) {
+        this.initializeClusterEnvironment()
+        this.startSUT(resources)
+        this.startWorkloadGenerator(load)
+    }
 
-    abstract fun stop();
+    abstract fun initializeClusterEnvironment();
+    abstract fun clearClusterEnvironment();
 
-    abstract fun startWorkloadGenerator(wg: String, load: LoadDimension, ucId: String);
+    abstract fun startSUT(resources: Resource);
+
+    abstract fun startWorkloadGenerator(load: LoadDimension);
 
 }
\ No newline at end of file

theodolite-quarkus/src/main/kotlin/theodolite/util/KafkaBenchmark.kt

-package theodolite.util
-
-class KafkaBenchmark(config: Map<String, Any>): Benchmark(config) {
-
-    override fun start(load: LoadDimension, resources: Resource) {
-        TODO("Not yet implemented")
-    }
-
-    override fun stop() {
-        TODO("Not yet implemented")
-    }
-
-    override fun startWorkloadGenerator(wg: String, load: LoadDimension, ucId: String) {
-        TODO("Not yet implemented")
-    }
-}
\ No newline at end of file

theodolite-quarkus/src/main/kotlin/theodolite/util/TestBenchmark.kt

@@ -5,7 +5,7 @@ class TestBenchmark: Benchmark(config = emptyMap()) {
         TODO("Not yet implemented")
     }
 
-    override fun stop() {
+    override fun clearClusterEnvironment() {
         TODO("Not yet implemented")
     }