Merge branch 'master' into firestore-test

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-benchmarks/load-generator-commons/src/main/java/rocks/theodolite/benchmarks/loadgenerator/HttpRecordSender.java

@@ -84,8 +84,9 @@ public class HttpRecordSender<T extends SpecificRecord> implements RecordSender<
     final String json = this.gson.toJson(message);
     final HttpRequest request = HttpRequest.newBuilder()
         .uri(this.uri)
-        .timeout(this.connectionTimeout)
         .POST(HttpRequest.BodyPublishers.ofString(json))
+        .header("Content-Type", "application/json")
+        .timeout(this.connectionTimeout)
         .build();
     final BodyHandler<Void> bodyHandler = BodyHandlers.discarding();
     // final BodyHandler<String> bodyHandler = BodyHandlers.ofString();

theodolite-benchmarks/uc2-beam/src/main/java/rocks/theodolite/benchmarks/uc2/beam/PipelineFactory.java

@@ -70,8 +70,10 @@ public class PipelineFactory extends AbstractPipelineFactory {
 
   @Override
   protected void registerCoders(final CoderRegistry registry) {
-    registry.registerCoderForClass(ActivePowerRecord.class,
-        AvroCoder.of(ActivePowerRecord.SCHEMA$));
+    registry.registerCoderForClass(
+        ActivePowerRecord.class,
+        // AvroCoder.of(ActivePowerRecord.SCHEMA$));
+        AvroCoder.of(ActivePowerRecord.class, false));
     registry.registerCoderForClass(StatsAggregation.class,
         SerializableCoder.of(StatsAggregation.class));
     registry.registerCoderForClass(StatsAccumulator.class,

theodolite-benchmarks/uc3-beam/src/main/java/rocks/theodolite/benchmarks/uc3/beam/PipelineFactory.java

@@ -95,7 +95,8 @@ public class PipelineFactory extends AbstractPipelineFactory {
   protected void registerCoders(final CoderRegistry registry) {
     registry.registerCoderForClass(
         ActivePowerRecord.class,
-        AvroCoder.of(ActivePowerRecord.SCHEMA$));
+        // AvroCoder.of(ActivePowerRecord.SCHEMA$));
+        AvroCoder.of(ActivePowerRecord.class, false));
     registry.registerCoderForClass(
         HourOfDayKey.class,
         new HourOfDayKeyCoder());

theodolite-benchmarks/uc4-beam/src/main/java/rocks/theodolite/benchmarks/uc4/beam/PipelineFactory.java

@@ -223,7 +223,8 @@ public class PipelineFactory extends AbstractPipelineFactory {
   protected void registerCoders(final CoderRegistry registry) {
     registry.registerCoderForClass(
         ActivePowerRecord.class,
-        AvroCoder.of(ActivePowerRecord.class));
+        // AvroCoder.of(ActivePowerRecord.SCHEMA$));
+        AvroCoder.of(ActivePowerRecord.class, false));
     registry.registerCoderForClass(
         AggregatedActivePowerRecord.class,
         new AggregatedActivePowerRecordCoder());

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,10 +45,11 @@ 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)

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

 package theodolite.execution.operator
 
+import io.fabric8.kubernetes.client.KubernetesClientException
 import io.fabric8.kubernetes.client.NamespacedKubernetesClient
 import io.fabric8.kubernetes.client.dsl.MixedOperation
 import io.fabric8.kubernetes.client.dsl.Resource
+import mu.KotlinLogging
 import theodolite.execution.Shutdown
 import theodolite.k8s.K8sContextFactory
 import theodolite.k8s.ResourceByLabelHandler
 import theodolite.model.crd.*
 
+private val logger = KotlinLogging.logger {}
+
 class ClusterSetup(
     private val executionCRDClient: MixedOperation<ExecutionCRD, BenchmarkExecutionList, Resource<ExecutionCRD>>,
     private val benchmarkCRDClient: MixedOperation<BenchmarkCRD, KubernetesBenchmarkList, Resource<BenchmarkCRD>>,
@@ -75,10 +79,15 @@ class ClusterSetup(
             labelName = "app.kubernetes.io/created-by",
             labelValue = "theodolite"
         )
+        try {
             resourceRemover.removeCR(
                 labelName = "app.kubernetes.io/created-by",
                 labelValue = "theodolite",
                 context = serviceMonitorContext
             )
+        } catch (e: KubernetesClientException) {
+            logger.warn { "Service monitors could not be cleaned up. It may be that service monitors are not registered by the Kubernetes API."}
+            logger.debug { "Error is: ${e.message}" }
+        }
     }
 }
\ No newline at end of file