Merge branch '109-implement-kotlin-prototype' into 'theodolite-kotlin'

Resolve "Implement Quarkus/Kotlin protype"

See merge request !78

analysis/README.md

@@ -9,7 +9,7 @@ benchmark execution results and plotting. The following notebooks are provided:
 For legacy reasons, we also provide the following notebooks, which, however, are not documented:
 
 * [scalability-graph.ipynb](scalability-graph.ipynb): Creates a scalability graph for a certain benchmark execution.
-* [scalability-graph-final.ipynb](scalability-graph-final.ipynb): Combines the scalability graphs of multiple benchmarks executions (e.g. for comparing different configuration).
+* [scalability-graph-plotter.ipynb](scalability-graph-plotter.ipynb): Combines the scalability graphs of multiple benchmarks executions (e.g. for comparing different configuration).
 * [lag-trend-graph.ipynb](lag-trend-graph.ipynb): Visualizes the consumer lag evaluation over time along with the computed trend.
 
 ## Usage

analysis/demand-metric-plot.ipynb

@@ -34,7 +34,7 @@
   },
   {
    "source": [
-    "We need to specify the directory, where the demand CSV files can be found, and a dictionary that maps a system description (e.g. its name) to the corresponding CSV file (prefix). "
+    "We need to specify the directory, where the demand CSV files can be found, and a dictionary that maps a system description (e.g. its name) to the corresponding CSV file (prefix). To use Unicode narrow non-breaking spaces in the description format it as `u\"1000\\u202FmCPU\"`."
    ],
    "cell_type": "markdown",
    "metadata": {}

analysis/demand-metric.ipynb

@@ -4,7 +4,7 @@
    "source": [
     "# Theodolite Analysis - Demand Metric\n",
     "\n",
-    "This notebook allows applies Theodolite's *demand* metric to describe scalability of a SUT based on Theodolite measurement data.\n",
+    "This notebook applies Theodolite's *demand* metric to describe scalability of a SUT based on Theodolite measurement data.\n",
     "\n",
     "Theodolite's *demand* metric is a function, mapping load intensities to the minimum required resources (e.g., instances) that are required to process this load. With this notebook, the *demand* metric function is approximated by a map of tested load intensities to their minimum required resources.\n",
     "\n",

docs/README.md

+---
+title: Theodolite
+nav_order: 1
+permalink: /
+---
+
+# Theodolite
+
+> A theodolite is a precision optical instrument for measuring angles between designated visible points in the horizontal and vertical planes.  -- <cite>[Wikipedia](https://en.wikipedia.org/wiki/Theodolite)</cite>
+
+Theodolite is a framework for benchmarking the horizontal and vertical scalability of stream processing engines. It consists of three modules:
+
+## Theodolite Benchmarks
+
+Theodolite contains 4 application benchmarks, which are based on typical use cases for stream processing within microservices. For each benchmark, a corresponding workload generator is provided. Currently, this repository provides benchmark implementations for Kafka Streams.
+
+
+## Theodolite Execution Framework
+
+Theodolite aims to benchmark scalability of stream processing engines for real use cases. Microservices that apply stream processing techniques are usually deployed in elastic cloud environments. Hence, Theodolite's cloud-native benchmarking framework deploys as components in a cloud environment, orchestrated by Kubernetes. More information on how to execute scalability benchmarks can be found in [Thedolite execution framework](execution).
+
+
+## Theodolite Analysis Tools
+
+Theodolite's benchmarking method create a *scalability graph* allowing to draw conclusions about the scalability of a stream processing engine or its deployment. A scalability graph shows how resource demand evolves with an increasing workload. Theodolite provides Jupyter notebooks for creating such scalability graphs based on benchmarking results from the execution framework. More information can be found in [Theodolite analysis tool](analysis).

docs/_config.yml

+title: "Theodolite"
+remote_theme: pmarsceill/just-the-docs
+#color_scheme: "dark"
+aux_links:
+    "Theodolite on GitHub":
+      - "//github.com/cau-se/theodolite"
\ No newline at end of file

docs/release-process.md

+---
+title: Release Process
+has_children: false
+nav_order: 2
+---
+
 # Release Process
 
 We assume that we are creating the release `v0.1.1`. Please make sure to adjust

execution/infrastructure/kafka/values.yaml

@@ -48,14 +48,15 @@ cp-kafka:
   #   cpu: 100m
   #   memory: 128Mi
   configurationOverrides:
-    #"offsets.topic.replication.factor": "3"
+    # offsets.topic.replication.factor: "3"
     "message.max.bytes": "134217728" # 128 MB
     "replica.fetch.max.bytes": "134217728" # 128 MB
     # "default.replication.factor": 3
     # "min.insync.replicas": 2
     "auto.create.topics.enable": false
     "log.retention.ms": "10000" # 10s
-    #"log.retention.ms": "86400000" # 24h
+    # "log.retention.ms": "86400000" # 24h
+    # "group.initial.rebalance.delay.ms": "30000" # 30s
     "metrics.sample.window.ms": "5000" #5s
 
 ## ------------------------------------------------------

execution/infrastructure/kafka/values_kafka_nodeport.yaml

+## ------------------------------------------------------
+## Zookeeper
+## ------------------------------------------------------
+cp-zookeeper:
+  enabled: true
+  servers: 1
+  image: confluentinc/cp-zookeeper
+  imageTag: 5.4.0
+  ## Optionally specify an array of imagePullSecrets. Secrets must be manually created in the namespace.
+  ## https://kubernetes.io/docs/concepts/containers/images/#specifying-imagepullsecrets-on-a-pod
+  imagePullSecrets:
+  #  - name: "regcred"
+  heapOptions: "-Xms512M -Xmx512M"
+  persistence:
+    enabled: false
+  resources: {}
+  ## If you do want to specify resources, uncomment the following lines, adjust them as necessary,
+  ## and remove the curly braces after 'resources:'
+  #  limits:
+  #   cpu: 100m
+  #   memory: 128Mi
+  #  requests:
+  #   cpu: 100m
+  #   memory: 128Mi
+
+## ------------------------------------------------------
+## Kafka
+## ------------------------------------------------------
+cp-kafka:
+  enabled: true
+  brokers: 1
+  image: confluentinc/cp-enterprise-kafka
+  imageTag: 5.4.0
+  ## Optionally specify an array of imagePullSecrets. Secrets must be manually created in the namespace.
+  ## https://kubernetes.io/docs/concepts/containers/images/#specifying-imagepullsecrets-on-a-pod
+  imagePullSecrets:
+  #  - name: "regcred"
+  heapOptions: "-Xms512M -Xmx512M"
+  persistence:
+    enabled: false
+  resources: {}
+  ## If you do want to specify resources, uncomment the following lines, adjust them as necessary,
+  ## and remove the curly braces after 'resources:'
+  #  limits:
+  #   cpu: 100m
+  #   memory: 128Mi
+  #  requests:
+  #   cpu: 100m
+  #   memory: 128Mi
+  configurationOverrides:
+    offsets.topic.replication.factor: "1"
+    "message.max.bytes": "134217728" # 128 MB
+    "replica.fetch.max.bytes": "134217728" # 128 MB
+    # "default.replication.factor": 3
+    # "min.insync.replicas": 2
+    "auto.create.topics.enable": false
+    "log.retention.ms": "10000" # 10s
+    #"log.retention.ms": "86400000" # 24h
+    "metrics.sample.window.ms": "5000" #5s
+  
+  # access kafka from outside
+  nodeport:
+    enabled: true
+
+## ------------------------------------------------------
+## Schema Registry
+## ------------------------------------------------------
+cp-schema-registry:
+  enabled: true
+  image: confluentinc/cp-schema-registry
+  imageTag: 5.4.0
+  ## Optionally specify an array of imagePullSecrets. Secrets must be manually created in the namespace.
+  ## https://kubernetes.io/docs/concepts/containers/images/#specifying-imagepullsecrets-on-a-pod
+  imagePullSecrets:
+  #  - name: "regcred"
+  heapOptions: "-Xms512M -Xmx512M"
+  resources: {}
+  ## If you do want to specify resources, uncomment the following lines, adjust them as necessary,
+  ## and remove the curly braces after 'resources:'
+  #  limits:
+  #   cpu: 100m
+  #   memory: 128Mi
+  #  requests:
+  #   cpu: 100m
+  #   memory: 128Mi
+
+cp-kafka-rest:
+  enabled: false
+
+cp-kafka-connect:
+  enabled: false
+
+cp-ksql-server:
+  enabled: false
+
+cp-control-center:
+  enabled: false

theodolite-quarkus/build.gradle

@@ -20,9 +20,11 @@ dependencies {
     testImplementation 'io.quarkus:quarkus-junit5'
     testImplementation 'io.rest-assured:rest-assured'
 
-
+    implementation 'org.slf4j:slf4j-simple:1.7.29'
+    implementation 'io.github.microutils:kotlin-logging:1.12.0'
     implementation 'io.fabric8:kubernetes-client:5.0.0-alpha-2'
-    //implementation 'com.fkorotkov:kubernetes-dsl:2.8.1'
+    compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.7.0'
+    compile group: 'org.apache.zookeeper', name: 'zookeeper', version: '3.6.2'
 }
 
 group 'theodolite'
@@ -47,9 +49,8 @@ compileKotlin {
 compileTestKotlin {
     kotlinOptions.jvmTarget = JavaVersion.VERSION_11
 }
-
 detekt {
     failFast = true // fail build on any finding
     buildUponDefaultConfig = true
     ignoreFailures = true
-}
+}
\ No newline at end of file

theodolite-quarkus/src/main/kotlin/theodolite/GreetingResource.kt

-package theodolite
-
-import javax.ws.rs.GET
-import javax.ws.rs.Path
-import javax.ws.rs.Produces
-import javax.ws.rs.core.MediaType
-
-@Path("/hello-resteasy")
-class GreetingResource {
-
-    @GET
-    @Produces(MediaType.TEXT_PLAIN)
-    fun hello() = "Hello RESTEasy"
-}
\ No newline at end of file

theodolite-quarkus/src/main/kotlin/theodolite/Main.kt

 package theodolite
+
 import io.quarkus.runtime.annotations.QuarkusMain
+import theodolite.execution.TheodoliteExecutor
+import mu.KotlinLogging
 
+private val logger = KotlinLogging.logger {}
 
 @QuarkusMain
 object Main {
     @JvmStatic
     fun main(args: Array<String>) {
-        println("Running main method")
-
-
-        //Quarkus.run()
+        val theodolite = TheodoliteExecutor()
+        theodolite.run()
+        logger.info("Application started")
     }
 }

theodolite-quarkus/src/main/kotlin/theodolite/evaluation/SLOChecker.kt

+package theodolite.evaluation
+
+interface SLOChecker {
+}
\ No newline at end of file

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

+package theodolite.execution
+
+import mu.KotlinLogging
+import theodolite.util.AbstractBenchmark
+import theodolite.util.LoadDimension
+import theodolite.util.Resource
+import theodolite.util.Results
+import java.time.Duration
+
+private val logger = KotlinLogging.logger {}
+
+/**
+ * The Benchmark Executor runs a single experiment.
+ *
+ * @property benchmark
+ * @property results
+ * @property executionDuration
+ * @constructor Create empty Benchmark executor
+ */
+abstract class BenchmarkExecutor(val benchmark: AbstractBenchmark, val results: Results, val executionDuration: Duration) {
+    /**
+     * Run a experiment for the given parametrization, evaluate the experiment and save the result.
+     *
+     * @param load load to be tested.
+     * @param res resources to be tested.
+     * @return True, if the number of resources are suitable for the given load, false otherwise.
+     */
+    abstract fun runExperiment(load: LoadDimension, res: Resource): Boolean
+
+    /**
+     * Wait while the benchmark is running and log the number of minutes executed every 1 minute.
+     *
+     */
+    fun waitAndLog() {
+        for (i in 1.rangeTo(executionDuration.toMinutes())) {
+            Thread.sleep(Duration.ofMinutes(1).toMillis())
+            logger.info { "Executed: $i minutes" }
+        }
+    }
+}

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

+package theodolite.execution
+
+import theodolite.util.AbstractBenchmark
+import theodolite.util.LoadDimension
+import theodolite.util.Resource
+import theodolite.util.Results
+import java.time.Duration
+
+class BenchmarkExecutorImpl(benchmark: AbstractBenchmark, results: Results, executionDuration: Duration) : BenchmarkExecutor(benchmark, results, executionDuration) {
+    override fun runExperiment(load: LoadDimension, res: Resource): Boolean {
+        benchmark.start(load, res)
+        this.waitAndLog()
+        benchmark.clearClusterEnvironment()
+        // todo evaluate
+        val result = false // if success else false
+        this.results.setResult(Pair(load, res), result)
+        return result;
+    }
+}
\ No newline at end of file

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

+package theodolite.execution
+
+import theodolite.util.AbstractBenchmark
+import theodolite.util.LoadDimension
+import theodolite.util.Resource
+import theodolite.util.Results
+import java.time.Duration
+
+class TestBenchmarkExecutorImpl(private val mockResults: Array<Array<Boolean>>, benchmark: AbstractBenchmark, results: Results):
+    BenchmarkExecutor(benchmark, results, executionDuration = Duration.ofSeconds(1)) {
+
+    override fun runExperiment(load: LoadDimension, res: Resource): Boolean {
+        val result = this.mockResults[load.get()][res.get()]
+
+        this.results.setResult(Pair(load, res), result)
+        return result;
+    }
+}
\ No newline at end of file

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

+package theodolite.execution
+
+import mu.KotlinLogging
+import theodolite.k8s.UC1Benchmark
+import theodolite.strategies.restriction.LowerBoundRestriction
+import theodolite.strategies.searchstrategy.CompositeStrategy
+import theodolite.strategies.searchstrategy.LinearSearch
+import theodolite.util.*
+import java.nio.file.Paths
+import java.time.Duration
+
+private val logger = KotlinLogging.logger {}
+
+class TheodoliteExecutor() {
+    val projectDirAbsolutePath = Paths.get("").toAbsolutePath().toString()
+    val resourcesPath = Paths.get(projectDirAbsolutePath, "./../../../resources/main/yaml/")
+    private fun loadConfig(): Config {
+        logger.info { resourcesPath }
+        val benchmark: UC1Benchmark = UC1Benchmark(
+            AbstractBenchmark.Config(
+                clusterZookeeperConnectionString = "my-confluent-cp-zookeeper:2181",
+                clusterKafkaConnectionString = "my-confluent-cp-kafka:9092",
+                externalZookeeperConnectionString = "localhost:2181",
+                externalKafkaConnectionString = "localhost:9092",
+                schemaRegistryConnectionString = "http://my-confluent-cp-schema-registry:8081",
+                kafkaPartition = 40,
+                kafkaReplication = 1,
+                kafkaTopics = listOf("input", "output"),
+                // TODO("handle path in a more nice way (not absolut)")
+                ucDeploymentPath = "$resourcesPath/aggregation-deployment.yaml",
+                ucServicePath = "$resourcesPath/aggregation-service.yaml",
+                wgDeploymentPath = "$resourcesPath/workloadGenerator.yaml",
+                configMapPath = "$resourcesPath/jmx-configmap.yaml",
+                ucImageURL = "ghcr.io/cau-se/theodolite-uc1-kstreams-app:latest",
+                wgImageURL = "ghcr.io/cau-se/theodolite-uc1-workload-generator:theodolite-kotlin-latest"
+            )
+        )
+        val results: Results = Results()
+
+        val executionDuration = Duration.ofSeconds(60 * 5)
+
+        val executor: BenchmarkExecutor = BenchmarkExecutorImpl(benchmark, results, executionDuration)
+
+        val restrictionStrategy = LowerBoundRestriction(results)
+        val searchStrategy = LinearSearch(executor)
+
+        return Config(
+            loads = listOf(5000, 10000).map { number -> LoadDimension(number) },
+            resources = (1..6).map { number -> Resource(number) },
+            compositeStrategy = CompositeStrategy(
+                executor,
+                searchStrategy,
+                restrictionStrategies = setOf(restrictionStrategy)
+            ),
+            executionDuration = executionDuration
+        )
+    }
+
+    fun run() {
+        // read or get benchmark config
+        val config = this.loadConfig()
+
+        // execute benchmarks for each load
+        for (load in config.loads) {
+            config.compositeStrategy.findSuitableResource(load, config.resources)
+        }
+    }
+}

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

+package theodolite.k8s
+
+import io.fabric8.kubernetes.api.model.ConfigMap
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient
+
+class ConfigMapManager(private val client: NamespacedKubernetesClient) {
+
+    fun deploy(configMap: ConfigMap) {
+        this.client.configMaps().createOrReplace(configMap)
+    }
+
+    fun delete(configMap: ConfigMap) {
+        this.client.configMaps().delete(configMap)
+    }
+}

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

+package theodolite.k8s
+
+import io.fabric8.kubernetes.api.model.Container
+import io.fabric8.kubernetes.api.model.EnvVar
+import io.fabric8.kubernetes.api.model.EnvVarSource
+import io.fabric8.kubernetes.api.model.Quantity
+import io.fabric8.kubernetes.api.model.apps.Deployment
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient
+import mu.KotlinLogging
+
+private val logger = KotlinLogging.logger {}
+
+class DeploymentManager(private val client: NamespacedKubernetesClient) {
+
+    /**
+     * Sets the ContainerEvironmentVariables, creates new if variable don t exist.
+     * @param container - The Container
+     * @param map - Map of k=Name,v =Value of EnviromentVariables
+     */
+    private fun setContainerEnv(container: Container, map: Map<String, String>) {
+        map.forEach { k, v ->
+            // filter for mathing name and set value
+            val x = container.env.filter { envVar -> envVar.name == k }
+
+            if (x.isEmpty()) {
+                val newVar = EnvVar(k, v, EnvVarSource())
+                container.env.add(newVar)
+            } else {
+                x.forEach {
+                    it.value = v
+                }
+            }
+        }
+    }
+
+    /**
+     * Set the environment Variable for a container
+     */
+    fun setWorkloadEnv(workloadDeployment: Deployment, containerName: String, map: Map<String, String>) {
+        workloadDeployment.spec.template.spec.containers.filter { it.name == containerName }
+            .forEach { it: Container ->
+                setContainerEnv(it, map)
+            }
+    }
+
+    /**
+     *  Change the RessourceLimit of a container (Usally SUT)
+     */
+    fun changeRessourceLimits(deployment: Deployment, ressource: String, containerName: String, limit: String) {
+        deployment.spec.template.spec.containers.filter { it.name == containerName }.forEach {
+            it.resources.limits.replace(ressource, Quantity(limit))
+        }
+    }
+
+    /**
+     * Change the image name of a container (SUT and the Worklaodgenerators)
+     */
+    fun setImageName(deployment: Deployment, containerName: String, image: String) {
+        deployment.spec.template.spec.containers.filter { it.name == containerName }.forEach {
+            it.image = image
+        }
+    }
+
+    /**
+     * Change the image name of a container (SUT and the Worklaodgenerators)
+     */
+    fun setReplica(deployment: Deployment, replicas: Int) {
+        deployment.spec.setReplicas(replicas)
+    }
+
+    // TODO potential add exception handling
+    fun deploy(deployment: Deployment) {
+        this.client.apps().deployments().createOrReplace(deployment)
+    }
+
+    // TODO potential add exception handling
+    fun delete(deployment: Deployment) {
+        this.client.apps().deployments().delete(deployment)
+    }
+}

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

+package theodolite.k8s
+
+import io.fabric8.kubernetes.api.model.Service
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient
+
+class ServiceManager(private val client: NamespacedKubernetesClient) {
+
+    fun changeServiceName(service: Service, newName: String) {
+
+        service.metadata.apply {
+            name = newName
+        }
+    }
+
+    fun deploy(service: Service) {
+        client.services().createOrReplace(service)
+    }
+
+    fun delete(service: Service) {
+        client.services().delete(service)
+    }
+}