Merge branch 'master' of git.se.informatik.uni-kiel.de:she/theodolite into make-sm-cleanup-optional

helm/templates/kafka/kafka-cluster.yaml

@@ -39,6 +39,11 @@ spec:
               nodeSelectorTerms:
                 {{- toYaml . | nindent 16 }}
     {{- end}}
+    {{- with .Values.strimzi.kafka.resources}}
+    resources:
+      {{- toYaml . | nindent 6 }}
+    {{- end}}
+
 
   zookeeper:
     {{- with .Values.strimzi.zookeeper.replicas }}

helm/values.yaml

@@ -174,6 +174,7 @@ strimzi:
       "-Xmx": "512M"
       "-Xms": "512M"
     nodeSelectorTerms: []
+    resources: {}
     
   zookeeper:
     replicas: 3

theodolite/src/main/kotlin/theodolite/benchmark/Action.kt

@@ -38,7 +38,7 @@ class ActionSelector {
 @JsonDeserialize
 @RegisterForReflection
 class PodSelector {
-    lateinit var matchLabels: MutableMap<String, String>
+    lateinit var matchLabels: Map<String, String>
 }
 @JsonDeserialize
 @RegisterForReflection

theodolite/src/main/kotlin/theodolite/benchmark/ActionCommand.kt

@@ -33,7 +33,7 @@ class ActionCommand(val client: NamespacedKubernetesClient) {
      * @return the exit code of this executed command
      */
     fun exec(
-        matchLabels: MutableMap<String, String>,
+        matchLabels: Map<String, String>,
         command: Array<String>,
         timeout: Long = Configuration.TIMEOUT_SECONDS,
         container: String = ""
@@ -58,7 +58,7 @@ class ActionCommand(val client: NamespacedKubernetesClient) {
 
             val latchTerminationStatus = execLatch.await(timeout, TimeUnit.SECONDS)
             if (!latchTerminationStatus) {
-                throw ActionCommandFailedException("Latch could not terminate within specified time")
+                throw ActionCommandFailedException("Timeout while running action command")
             }
             execWatch.close()
         } catch (e: Exception) {
@@ -112,7 +112,7 @@ class ActionCommand(val client: NamespacedKubernetesClient) {
      * it can take a while until the status is ready and the pod can be selected.
      * @return the name of the pod or throws [ActionCommandFailedException]
      */
-    fun getPodName(matchLabels: MutableMap<String, String>, tries: Int): String {
+    fun getPodName(matchLabels: Map<String, String>, tries: Int): String {
         for (i in 1..tries) {
 
             try {
@@ -125,7 +125,7 @@ class ActionCommand(val client: NamespacedKubernetesClient) {
         throw ActionCommandFailedException("Couldn't find any pod that matches the specified labels.")
     }
 
-    private fun getPodName(matchLabels: MutableMap<String, String>): String {
+    private fun getPodName(matchLabels: Map<String, String>): String {
         return try {
             val podNames = this.client
                 .pods()

theodolite/src/main/kotlin/theodolite/evaluation/AnalysisExecutor.kt

@@ -38,7 +38,7 @@ class AnalysisExecutor(
 
         try {
             val ioHandler = IOHandler()
-            val resultsFolder: String = ioHandler.getResultFolderURL()
+            val resultsFolder = ioHandler.getResultFolderURL()
             val fileURL = "${resultsFolder}exp${executionId}_${load.get()}_${res.get()}_${slo.sloType.toSlug()}"
 
             val prometheusData = executionIntervals

theodolite/src/main/kotlin/theodolite/evaluation/MetricFetcher.kt

@@ -45,15 +45,16 @@ class MetricFetcher(private val prometheusURL: String, private val offset: Durat
         )
 
         while (counter < RETRIES) {
+            logger.info { "Request collected metrics from Prometheus for interval [$offsetStart,$offsetEnd]." }
             val response = get("$prometheusURL/api/v1/query_range", params = parameter, timeout = TIMEOUT)
             if (response.statusCode != 200) {
                 val message = response.jsonObject.toString()
-                logger.warn { "Could not connect to Prometheus: $message. Retrying now." }
+                logger.warn { "Could not connect to Prometheus: $message. Retry $counter/$RETRIES." }
                 counter++
             } else {
                 val values = parseValues(response)
                 if (values.data?.result.isNullOrEmpty()) {
-                    throw NoSuchFieldException("Empty query result: $values between $start and $end for query $query.")
+                    throw NoSuchFieldException("Empty query result: $values between for query '$query' in interval [$offsetStart,$offsetEnd] .")
                 }
                 return parseValues(response)
             }

theodolite/src/main/kotlin/theodolite/evaluation/SloConfigHandler.kt

@@ -4,6 +4,7 @@ import theodolite.benchmark.BenchmarkExecution
 import theodolite.util.InvalidPatcherConfigurationException
 import javax.enterprise.context.ApplicationScoped
 
+private const val DEFAULT_CONSUMER_LAG_METRIC_BASE = "kafka_consumergroup_lag"
 private const val DEFAULT_CONSUMER_LAG_QUERY = "sum by(consumergroup) (kafka_consumergroup_lag >= 0)"
 private const val DEFAULT_DROPPED_RECORDS_QUERY = "sum by(job) (kafka_streams_stream_task_metrics_dropped_records_total>=0)"
 
@@ -13,9 +14,14 @@ class SloConfigHandler {
         fun getQueryString(slo: BenchmarkExecution.Slo): String {
             return when (slo.sloType.lowercase()) {
                 SloTypes.GENERIC.value -> slo.properties["promQLQuery"] ?: throw IllegalArgumentException("promQLQuery expected")
+                SloTypes.LAG_TREND.value, SloTypes.LAG_TREND_RATIO.value -> slo.properties["promQLQuery"] ?:
+                    (slo.properties["consumerGroup"]?.let { "{consumergroup='$it'}" } ?: "").let {
+                        "sum by(consumergroup) ($DEFAULT_CONSUMER_LAG_METRIC_BASE$it >= 0)"
+                    }
+                SloTypes.DROPPED_RECORDS.value, SloTypes.DROPPED_RECORDS_RATIO.value -> slo.properties["promQLQuery"] ?: DEFAULT_DROPPED_RECORDS_QUERY
                 SloTypes.LAG_TREND.value, SloTypes.LAG_TREND_RATIO.value -> slo.properties["promQLQuery"] ?: DEFAULT_CONSUMER_LAG_QUERY
                 SloTypes.DROPPED_RECORDS.value, SloTypes.DROPPED_RECORDS_RATIO.value -> slo.properties["promQLQuery"] ?: DEFAULT_DROPPED_RECORDS_QUERY
-                else -> throw  InvalidPatcherConfigurationException("Could not find Prometheus query string for slo type $slo.sloType")
+                else -> throw  InvalidPatcherConfigurationException("Could not find Prometheus query string for slo type ${slo.sloType}")
             }
         }
     }

theodolite/src/main/kotlin/theodolite/execution/operator/BenchmarkStateChecker.kt

@@ -190,7 +190,7 @@ class BenchmarkStateChecker(
     }
 }
 
-private fun <K, V> MutableMap<K, V>.containsMatchLabels(matchLabels: MutableMap<V, V>): Boolean {
+private fun <K, V> Map<K, V>.containsMatchLabels(matchLabels: Map<V, V>): Boolean {
     for (kv in matchLabels) {
         if (kv.value != this[kv.key as K]) {
             return false

theodolite/src/test/kotlin/theodolite/benchmark/ActionCommandTest.kt

@@ -102,7 +102,7 @@ class ActionCommandTest {
         val action = Action()
         action.selector = ActionSelector()
         action.selector.pod = PodSelector()
-        action.selector.pod.matchLabels = mutableMapOf("app" to "pod")
+        action.selector.pod.matchLabels = mapOf("app" to "pod")
         action.exec = Command()
         action.exec.command = arrayOf("ls")
         action.exec.timeoutSeconds = 10L
@@ -118,7 +118,7 @@ class ActionCommandTest {
         val action = Action()
         action.selector = ActionSelector()
         action.selector.pod = PodSelector()
-        action.selector.pod.matchLabels = mutableMapOf("app" to "pod")
+        action.selector.pod.matchLabels = mapOf("app" to "pod")
         action.exec = Command()
         action.exec.command = arrayOf("error-command")
         action.exec.timeoutSeconds = 10L