Merge branch 'theodolite-kotlin' into store-results-in-volume

theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java

@@ -47,7 +47,7 @@ public final class HistoryServiceFlinkJob {
     // Parallelism
     final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
     if (parallelism != null) {
-      LOGGER.error("Set parallelism: {}.", parallelism);
+      LOGGER.info("Set parallelism: {}.", parallelism);
       this.env.setParallelism(parallelism);
     }
 
@@ -68,7 +68,7 @@ public final class HistoryServiceFlinkJob {
     final DataStream<ActivePowerRecord> stream = this.env.addSource(kafkaConsumer);
 
     stream
-        .rebalance()
+        // .rebalance()
         .map(new GsonMapper())
         .flatMap((record, c) -> LOGGER.info("Record: {}", record))
         .returns(Types.GENERIC(Object.class)); // Will never be used

theodolite-benchmarks/uc2-flink/src/main/java/theodolite/uc2/application/HistoryServiceFlinkJob.java

@@ -59,7 +59,7 @@ public final class HistoryServiceFlinkJob {
     // Parallelism
     final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
     if (parallelism != null) {
-      LOGGER.error("Set parallelism: {}.", parallelism);
+      LOGGER.info("Set parallelism: {}.", parallelism);
       this.env.setParallelism(parallelism);
     }
 
@@ -83,7 +83,9 @@ public final class HistoryServiceFlinkJob {
     final String schemaRegistryUrl = this.config.getString(ConfigurationKeys.SCHEMA_REGISTRY_URL);
     final String inputTopic = this.config.getString(ConfigurationKeys.KAFKA_INPUT_TOPIC);
     final String outputTopic = this.config.getString(ConfigurationKeys.KAFKA_OUTPUT_TOPIC);
-    final int windowDuration = this.config.getInt(ConfigurationKeys.KAFKA_WINDOW_DURATION_MINUTES);
+    final int windowDurationMinutes =
+        this.config.getInt(ConfigurationKeys.KAFKA_WINDOW_DURATION_MINUTES);
+    final Time windowDuration = Time.minutes(windowDurationMinutes);
     final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true);
 
     final KafkaConnectorFactory kafkaConnector = new KafkaConnectorFactory(
@@ -100,9 +102,9 @@ public final class HistoryServiceFlinkJob {
 
     this.env
         .addSource(kafkaSource).name("[Kafka Consumer] Topic: " + inputTopic)
-        .rebalance()
+        // .rebalance()
         .keyBy(ActivePowerRecord::getIdentifier)
-        .window(TumblingEventTimeWindows.of(Time.minutes(windowDuration)))
+        .window(TumblingEventTimeWindows.of(windowDuration))
         .aggregate(new StatsAggregateFunction(), new StatsProcessWindowFunction())
         .map(t -> {
           final String key = t.f0;

theodolite-benchmarks/uc3-flink/src/main/java/theodolite/uc3/application/HistoryServiceFlinkJob.java

@@ -117,9 +117,8 @@ public final class HistoryServiceFlinkJob {
     // Streaming topology
     final StatsKeyFactory<HourOfDayKey> keyFactory = new HourOfDayKeyFactory();
     this.env
-        .addSource(kafkaSource)
-        .name("[Kafka Consumer] Topic: " + inputTopic)
-        .rebalance()
+        .addSource(kafkaSource).name("[Kafka Consumer] Topic: " + inputTopic)
+        // .rebalance()
         .keyBy((KeySelector<ActivePowerRecord, HourOfDayKey>) record -> {
           final Instant instant = Instant.ofEpochMilli(record.getTimestamp());
           final LocalDateTime dateTime = LocalDateTime.ofInstant(instant, timeZone);

theodolite-benchmarks/uc4-flink/src/main/java/theodolite/uc4/application/AggregationServiceFlinkJob.java

@@ -79,7 +79,7 @@ public final class AggregationServiceFlinkJob {
     // Parallelism
     final Integer parallelism = this.config.getInteger(ConfigurationKeys.PARALLELISM, null);
     if (parallelism != null) {
-      LOGGER.error("Set parallelism: {}.", parallelism);
+      LOGGER.info("Set parallelism: {}.", parallelism);
       this.env.setParallelism(parallelism);
     }
 
@@ -152,7 +152,7 @@ public final class AggregationServiceFlinkJob {
     // Build input stream
     final DataStream<ActivePowerRecord> inputStream = this.env.addSource(kafkaInputSource)
         .name("[Kafka Consumer] Topic: " + inputTopic)// NOCS
-        .rebalance()
+        // .rebalance()
         .map(r -> r)
         .name("[Map] Rebalance Forward");
 
@@ -160,7 +160,7 @@ public final class AggregationServiceFlinkJob {
     final DataStream<ActivePowerRecord> aggregationsInputStream =
         this.env.addSource(kafkaOutputSource)
             .name("[Kafka Consumer] Topic: " + outputTopic) // NOCS
-            .rebalance()
+            // .rebalance()
             .map(r -> new ActivePowerRecord(r.getIdentifier(), r.getTimestamp(), r.getSumInW()))
             .name("[Map] AggregatedActivePowerRecord -> ActivePowerRecord");
 

theodolite-quarkus/config/thedolite-operator.yaml

-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: theodolite
-spec:
-  selector:
-    matchLabels:
-      app: theodolite
-  replicas: 1
-  template:
-    metadata:
-      labels:
-        app: theodolite
-    spec:
-      terminationGracePeriodSeconds: 0
-      serviceAccountName: theodolite
-      containers:
-        - name: thedolite
-          image: ghcr.io/cau-se/theodolite:theodolite-kotlin-latest
-          env:
-            - name: KUBECONFIG
-              value: "~/.kube/config"
-            - name: NAMESPACE
-              value: "default"
\ No newline at end of file

theodolite-quarkus/src/main/resources/operator/example-benchmark-k8s-resource.yaml → theodolite-quarkus/examples/operator/example-benchmark.yaml

@@ -3,15 +3,14 @@ kind: benchmark
 metadata:
   name: uc1-kstreams
 spec:
-  name: test
   appResource:
     - "uc1-kstreams-deployment.yaml"
     - "aggregation-service.yaml"
     - "jmx-configmap.yaml"
+    - "uc1-service-monitor.yaml"
   loadGenResource:
     - "uc1-load-generator-deployment.yaml"
     - "uc1-load-generator-service.yaml"
-    - "uc1-load-generator-service.yaml"
   resourceTypes:
     - typeName: "Instances"
       patchers:
@@ -23,14 +22,14 @@ spec:
         - type: "EnvVarPatcher"
           resource: "uc1-load-generator-deployment.yaml"
           properties:
-            container: "workload-generator"
             variableName: "NUM_SENSORS"
+            container: "workload-generator"
         - type: "NumSensorsLoadGeneratorReplicaPatcher"
           resource: "uc1-load-generator-deployment.yaml"
           properties:
             loadGenMaxRecords: "15000"
   kafkaConfig:
-    bootstrapServer: "localhost:31290"
+    bootstrapServer: "theodolite-cp-kafka:9092"
     topics:
       - name: "input"
         numPartitions: 40

theodolite-quarkus/examples/operator/example-execution.yaml

+apiVersion: theodolite.com/v1
+kind: execution
+metadata:
+  name: theodolite-example-execution
+spec:
+  benchmark: "uc1-kstreams"
+  load:
+    loadType: "NumSensors"
+    loadValues: [25000, 50000, 75000, 100000, 125000, 150000]
+  resources:
+    resourceType: "Instances"
+    resourceValues: [1, 2, 3, 4, 5]
+  slos:
+    - sloType: "lag trend"
+      threshold: 2000
+      prometheusUrl: "http://prometheus-operated:9090"
+      externalSloUrl: "http://localhost:80/evaluate-slope"
+      offset: 0
+      warmup: 60 # in seconds
+  execution:
+    strategy: "LinearSearch"
+    duration: 300 # in seconds
+    repetitions: 1
+    loadGenerationDelay: 30 # in seconds
+    restrictions:
+      - "LowerBound"
+  configOverrides:
+  # - patcher:
+  #     type: "NodeSelectorPatcher"
+  #     resource: "uc1-load-generator-deployment.yaml"
+  #     properties:
+  #       variableName: "env"
+  #     value: "prod"
+  # - patcher:
+  #     type: "NodeSelectorPatcher"
+  #     resource: "uc1-kstreams-deployment.yaml"
+  #     properties:
+  #       variableName: "env"
+  #   value: "prod"
+  # - patcher:
+  #     type: "ResourceLimitPatcher"
+  #     resource: "uc1-kstreams-deployment.yaml"
+  #     properties:
+  #       container: "uc-application"
+  #       limitedResource: "cpu"
+  #   value: "1000m"
+  # - patcher:
+  #     type: "ResourceLimitPatcher"
+  #     resource: "uc1-kstreams-deployment.yaml"
+  #     properties:
+  #       container: "uc-application"
+  #       limitedResource: "memory"
+  #   value: "2Gi"
+  #  - patcher:
+  #      type: "SchedulerNamePatcher"
+  #      resource: "uc1-kstreams-deployment.yaml"
+  #    value: "random-scheduler"

theodolite-quarkus/examples/standalone/example-execution.yaml

+name: example-execution
+benchmark: "uc1-kstreams"
+load:
+  loadType: "NumSensors"
+  loadValues: [25000, 50000, 75000, 100000, 125000, 150000]
+resources:
+  resourceType: "Instances"
+  resourceValues: [1, 2, 3, 4, 5]
+slos:
+  - sloType: "lag trend"
+    threshold: 2000
+    prometheusUrl: "http://prometheus-operated:9090"
+    externalSloUrl: "http://localhost:80/evaluate-slope"
+    offset: 0
+    warmup: 60 # in seconds
+execution:
+  strategy: "LinearSearch"
+  duration: 300 # in seconds
+  repetitions: 1
+  loadGenerationDelay: 30 # in seconds, optional field, default is 0 seconds
+  restrictions:
+    - "LowerBound"
+configOverrides: []
\ No newline at end of file