diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 9634c635c37dd93bfb0bb0f6ef9641cbf0432cb2..c97acd1628e7816e70d7791d5702f4dcf4680c42 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -1,7 +1,7 @@ workflow: rules: - if: '$CI_PIPELINE_SOURCE == "merge_request_event"' - - if: '$CI_COMMIT_BRANCH && $CI_OPEN_MERGE_REQUESTS && $CI_PIPELINE_SOURCE == "push"' + - if: '$CI_COMMIT_BRANCH && $CI_OPEN_MERGE_REQUESTS && $CI_PIPELINE_SOURCE == "push" && $CI_COMMIT_REF_PROTECTED != "true"' when: never - when: always @@ -10,6 +10,7 @@ stages: - test - check - deploy + - smoketest default: tags: @@ -20,9 +21,9 @@ default: - exec-dind # see https://docs.gitlab.com/ee/ci/docker/using_docker_build.html#tls-enabled # for image usage and settings for building with TLS and docker in docker - image: docker:19.03.1 + image: docker:20.10.12 services: - - docker:19.03.1-dind + - docker:20.10.12-dind variables: DOCKER_TLS_CERTDIR: "/certs" @@ -45,6 +46,10 @@ default: fi - "[ $DOCKERFILE ] && KANIKO_DOCKERFILE=\"--dockerfile $DOCKERFILE\"" - /kaniko/executor --context `pwd`/$CONTEXT $KANIKO_DOCKERFILE $KANIKO_D + - echo "PUBLISHED_IMAGE_TAG=${CI_COMMIT_TAG-$CI_COMMIT_SHORT_SHA}" >> build.env + artifacts: + reports: + dotenv: build.env # Theodolite Docs @@ -372,7 +377,174 @@ deploy-uc4-load-generator: IMAGE_NAME: "theodolite-uc4-workload-generator" JAVA_PROJECT_NAME: "uc4-load-generator" JAVA_PROJECT_DEPS: "load-generator-commons" - + +deploy-http-bridge: + extends: .deploy-benchmarks + variables: + IMAGE_NAME: "theodolite-http-bridge" + JAVA_PROJECT_NAME: "http-bridge" + JAVA_PROJECT_DEPS: "load-generator-commons" + +.smoketest-benchmarks: + stage: smoketest + extends: + - .dind + image: ghcr.io/cau-se/theodolite-build-docker-compose-jq:20.10.12 + before_script: + - cd theodolite-benchmarks/docker-test + # variables: + # TEST_LOG_FILE: "test.log" + script: + - export THEODOLITE_TAG=$PUBLISHED_IMAGE_TAG + - ./smoketest-runner.sh ./$DOCKER_COMPOSE_DIR + # - cat test.log + after_script: + - cd ./$DOCKER_COMPOSE_DIR + - docker-compose down + rules: + - changes: + - theodolite-benchmarks/* + - theodolite-benchmarks/{$JAVA_PROJECT_DEPS}/**/* + if: "$CR_HOST && $CR_ORG && $CR_USER && $CR_PW && $DOCKER_COMPOSE_DIR && $JAVA_PROJECT_DEPS" + - if: "$CR_HOST && $CR_ORG && $CR_USER && $CR_PW && $DOCKER_COMPOSE_DIR && $JAVA_PROJECT_DEPS" + when: manual + allow_failure: true + +smoketest-uc1-kstreams: + extends: .smoketest-benchmarks + needs: + - deploy-uc1-kstreams + - deploy-uc1-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc1-kstreams" + JAVA_PROJECT_DEPS: "uc1-kstreams,kstreams-commons,uc1-load-generator,load-generator-commons" + +smoketest-uc1-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc1-flink + - deploy-uc1-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc1-flink" + JAVA_PROJECT_DEPS: "uc1-flink,flink-commons,uc1-load-generator,load-generator-commons" + +smoketest-uc1-beam-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc1-beam-flink + - deploy-uc1-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc1-beam-flink" + JAVA_PROJECT_DEPS: "uc1-beam-flink,uc1-beam,beam-commons,uc1-load-generator,load-generator-commons" + +smoketest-uc1-beam-samza: + extends: .smoketest-benchmarks + needs: + - deploy-uc1-beam-samza + - deploy-uc1-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc1-beam-samza" + JAVA_PROJECT_DEPS: "uc1-beam-samza,uc1-beam,beam-commons,uc1-load-generator,load-generator-commons" + +smoketest-uc2-kstreams: + extends: .smoketest-benchmarks + needs: + - deploy-uc2-kstreams + - deploy-uc2-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc2-kstreams" + JAVA_PROJECT_DEPS: "uc2-kstreams,kstreams-commons,uc2-load-generator,load-generator-commons" + +smoketest-uc2-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc2-flink + - deploy-uc2-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc2-flink" + JAVA_PROJECT_DEPS: "uc2-flink,flink-commons,uc2-load-generator,load-generator-commons" + +smoketest-uc2-beam-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc2-beam-flink + - deploy-uc2-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc2-beam-flink" + JAVA_PROJECT_DEPS: "uc2-beam-flink,uc2-beam,beam-commons,uc2-load-generator,load-generator-commons" + +smoketest-uc2-beam-samza: + extends: .smoketest-benchmarks + needs: + - deploy-uc2-beam-samza + - deploy-uc2-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc2-beam-samza" + JAVA_PROJECT_DEPS: "uc2-beam-samza,uc2-beam,beam-commons,uc2-load-generator,load-generator-commons" + +smoketest-uc3-kstreams: + extends: .smoketest-benchmarks + needs: + - deploy-uc3-kstreams + - deploy-uc3-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc3-kstreams" + JAVA_PROJECT_DEPS: "uc3-kstreams,kstreams-commons,uc3-load-generator,load-generator-commons" + +smoketest-uc3-beam-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc3-beam-flink + - deploy-uc3-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc3-beam-flink" + JAVA_PROJECT_DEPS: "uc3-beam-flink,uc3-beam,beam-commons,uc3-load-generator,load-generator-commons" + +smoketest-uc3-beam-samza: + extends: .smoketest-benchmarks + needs: + - deploy-uc3-beam-samza + - deploy-uc3-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc3-beam-samza" + JAVA_PROJECT_DEPS: "uc3-beam-samza,uc3-beam,beam-commons,uc3-load-generator,load-generator-commons" + +smoketest-uc4-kstreams: + extends: .smoketest-benchmarks + needs: + - deploy-uc4-kstreams + - deploy-uc4-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc4-kstreams" + JAVA_PROJECT_DEPS: "uc4-kstreams,kstreams-commons,uc4-load-generator,load-generator-commons" + +smoketest-uc4-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc4-flink + - deploy-uc4-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc4-flink" + JAVA_PROJECT_DEPS: "uc4-flink,flink-commons,uc4-load-generator,load-generator-commons" + +smoketest-uc4-beam-flink: + extends: .smoketest-benchmarks + needs: + - deploy-uc4-beam-flink + - deploy-uc4-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc4-beam-flink" + JAVA_PROJECT_DEPS: "uc4-beam-flink,uc4-beam,beam-commons,uc4-load-generator,load-generator-commons" + +smoketest-uc4-beam-samza: + extends: .smoketest-benchmarks + needs: + - deploy-uc4-beam-samza + - deploy-uc4-load-generator + variables: + DOCKER_COMPOSE_DIR: "uc4-beam-samza" + JAVA_PROJECT_DEPS: "uc4-beam-samza,uc4-beam,beam-commons,uc4-load-generator,load-generator-commons" + # Theodolite Framework diff --git a/CITATION.cff b/CITATION.cff index 04640de442f4458b09e11ce3d2939c850f594556..b6ca4542010b83e12206fbc0d9841683b43e1d57 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -8,7 +8,7 @@ authors: given-names: Wilhelm orcid: "https://orcid.org/0000-0001-6625-4335" title: Theodolite -version: "0.6.3" +version: "0.6.4" repository-code: "https://github.com/cau-se/theodolite" license: "Apache-2.0" doi: "10.1016/j.bdr.2021.100209" diff --git a/codemeta.json b/codemeta.json index 832b570681afb143978698fd47dad5d2835c700b..948a34628ec919e2492b61e3ac9997392dc5e030 100644 --- a/codemeta.json +++ b/codemeta.json @@ -8,7 +8,7 @@ "dateModified": "2022-01-24", "downloadUrl": "https://github.com/cau-se/theodolite/releases", "name": "Theodolite", - "version": "0.6.3", + "version": "0.6.4", "description": "Theodolite is a framework for benchmarking the horizontal and vertical scalability of cloud-native applications.", "developmentStatus": "active", "relatedLink": [ diff --git a/docs/api-reference/patchers.md b/docs/api-reference/patchers.md index 77f937e38f6a0ee4084cb0ad5b5838718eabff10..bea63ccd23decef5654f257221ce0358b4f68e45 100644 --- a/docs/api-reference/patchers.md +++ b/docs/api-reference/patchers.md @@ -53,6 +53,27 @@ Patchers can be seen as functions which take a value as input and modify a Kuber * **resource**: "uc1-kstreams-deployment.yaml" * **example value**: "random-scheduler" +* **LabelPatcher**: Changes the label of a Kubernetes Deployment or StatefulSet. The patched field is: `metadata.labels` + * **type**: "LabelPatcher" + * **resource**: "uc1-kstreams-deployment.yaml" + * **properties**: + * variableName: "app" + * **example value**: "theodolite-sut" + +* **MatchLabelPatcher**: Changes the match labels of a Kubernetes Deployment or StatefulSet. The patched field is: `spec.selector.matchLabels` + * **type**: "MatchLabelPatcher" + * **resource**: "uc1-kstreams-deployment.yaml" + * **properties**: + * variableName: "app" + * **example value**: "theodolite-sut" + +* **TemplateLabelPatcher**: Changes the template labels of a Kubernetes Deployment or StatefulSet. The patched field is: `spec.template.metadata.labels` + * **type**: "MatchLabelPatcher" + * **resource**: "uc1-kstreams-deployment.yaml" + * **properties**: + * variableName: "app" + * **example value**: "theodolite-sut" + * **ImagePatcher**: Changes the image of a Kubernetes resource. **Currently not fully implemented.** * **type**: "ImagePatcher" * **resource**: "uc1-kstreams-deployment.yaml" diff --git a/docs/index.yaml b/docs/index.yaml index 509844ab0bc371d29302f90f69e769cd52a8e11b..3e0de103a78f3529d314727ed59be3dcdc333fc9 100644 --- a/docs/index.yaml +++ b/docs/index.yaml @@ -1,6 +1,41 @@ apiVersion: v1 entries: theodolite: + - apiVersion: v2 + appVersion: 0.6.4 + created: "2022-02-16T16:09:11.967649304+01:00" + dependencies: + - condition: grafana.enabled + name: grafana + repository: https://grafana.github.io/helm-charts + version: 6.17.5 + - condition: kube-prometheus-stack.enabled + name: kube-prometheus-stack + repository: https://prometheus-community.github.io/helm-charts + version: 20.0.1 + - condition: cp-helm-charts.enabled + name: cp-helm-charts + repository: https://soerenhenning.github.io/cp-helm-charts + version: 0.6.0 + - condition: kafka-lag-exporter.enabled + name: kafka-lag-exporter + repository: https://seanglover.com/kafka-lag-exporter/repo + version: 0.6.7 + description: Theodolite is a framework for benchmarking the horizontal and vertical + scalability of cloud-native applications. + digest: 10156d9917233ffa297aab093532038667d25b2babb2b2058a0a32e1dccb0cca + home: https://www.theodolite.rocks + maintainers: + - email: soeren.henning@email.uni-kiel.de + name: Sören Henning + url: https://www.se.informatik.uni-kiel.de/en/team/soeren-henning-m-sc + name: theodolite + sources: + - https://github.com/cau-se/theodolite + type: application + urls: + - https://github.com/cau-se/theodolite/releases/download/v0.6.4/theodolite-0.6.4.tgz + version: 0.6.4 - apiVersion: v2 appVersion: 0.6.3 created: "2022-01-24T13:40:40.07330713+01:00" @@ -141,6 +176,41 @@ entries: urls: - https://github.com/cau-se/theodolite/releases/download/v0.6.0/theodolite-0.6.0.tgz version: 0.6.0 + - apiVersion: v2 + appVersion: 0.5.2 + created: "2022-02-16T15:43:43.534374597+01:00" + dependencies: + - condition: grafana.enabled + name: grafana + repository: https://grafana.github.io/helm-charts + version: 6.17.5 + - condition: kube-prometheus-stack.enabled + name: kube-prometheus-stack + repository: https://prometheus-community.github.io/helm-charts + version: 12.0.0 + - condition: cp-helm-charts.enabled + name: cp-helm-charts + repository: https://soerenhenning.github.io/cp-helm-charts + version: 0.6.0 + - condition: kafka-lag-exporter.enabled + name: kafka-lag-exporter + repository: https://seanglover.com/kafka-lag-exporter/repo + version: 0.6.6 + description: Theodolite is a framework for benchmarking the scalability stream + processing engines. + digest: 72df752883d2161fdfc0e96bb90fe11f9c0ed4f71013e588ec170f2cbb178e9c + home: https://cau-se.github.io/theodolite + maintainers: + - email: soeren.henning@email.uni-kiel.de + name: Sören Henning + url: https://www.se.informatik.uni-kiel.de/en/team/soeren-henning-m-sc + name: theodolite + sources: + - https://github.com/cau-se/theodolite + type: application + urls: + - https://github.com/cau-se/theodolite/releases/download/v0.5.2/theodolite-0.5.2.tgz + version: 0.5.2 - apiVersion: v2 appVersion: 0.5.1 created: "2021-11-12T16:15:01.629937292+01:00" @@ -246,4 +316,4 @@ entries: urls: - https://github.com/cau-se/theodolite/releases/download/v0.4.0/theodolite-0.4.0.tgz version: 0.4.0 -generated: "2022-01-24T13:40:40.036786105+01:00" +generated: "2022-02-16T16:09:11.93111234+01:00" diff --git a/docs/theodolite-benchmarks/load-generator.md b/docs/theodolite-benchmarks/load-generator.md index 6d42ea06d9cb008a9aeddcc8145a2868c8d916b1..17845c42d47e94a5b696dee1d774890de8d6fff1 100644 --- a/docs/theodolite-benchmarks/load-generator.md +++ b/docs/theodolite-benchmarks/load-generator.md @@ -17,7 +17,7 @@ For each benchmark, we provide a [load generator as OCI container image](https:/ You can simply run a load generator container, for example, for benchmark UC1 with: ```sh -docker run ghcr.io/cau-se/theodolite-uc1-workload-generator +docker run -it ghcr.io/cau-se/theodolite-uc1-workload-generator ``` ### Message format diff --git a/helm/Chart.yaml b/helm/Chart.yaml index 27451ad55ce75592db9dc7550b1f81dced3951bc..52fda2bbc83c722d70f6179c4ec53fb3775bd436 100644 --- a/helm/Chart.yaml +++ b/helm/Chart.yaml @@ -26,7 +26,7 @@ dependencies: condition: cp-helm-charts.enabled - name: kafka-lag-exporter version: 0.6.7 - repository: https://lightbend.github.io/kafka-lag-exporter/repo/ + repository: https://seanglover.com/kafka-lag-exporter/repo condition: kafka-lag-exporter.enabled version: 0.7.0-SNAPSHOT diff --git a/theodolite-benchmarks/beam-commons/build.gradle b/theodolite-benchmarks/beam-commons/build.gradle index a15b388a33ce53b0f1b8f2f0567996917b9f18d1..64ac2bb51ae1e6d741749a81e5c6c9e296d14d68 100644 --- a/theodolite-benchmarks/beam-commons/build.gradle +++ b/theodolite-benchmarks/beam-commons/build.gradle @@ -16,14 +16,14 @@ dependencies { implementation('org.industrial-devops:titan-ccp-common:0.1.0-SNAPSHOT') { changing = true } implementation('org.industrial-devops:titan-ccp-common-kafka:0.1.0-SNAPSHOT') { changing = true } - implementation group: 'org.apache.beam', name: 'beam-sdks-java-core', version: '2.22.0' - implementation('org.apache.beam:beam-sdks-java-io-kafka:2.22.0'){ + implementation group: 'org.apache.beam', name: 'beam-sdks-java-core', version: '2.35.0' + implementation('org.apache.beam:beam-sdks-java-io-kafka:2.35.0'){ exclude group: 'org.apache.kafka', module: 'kafka-clients' } implementation ('io.confluent:kafka-streams-avro-serde:5.3.2') implementation group: 'org.slf4j', name: 'slf4j-simple', version: '1.7.30' - + runtimeOnly 'org.slf4j:slf4j-api:1.7.32' runtimeOnly 'org.slf4j:slf4j-jdk14:1.7.32' diff --git a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.flink.gradle b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.flink.gradle index eb9bda1f84c4f20568fca1498462dff9082ea1fa..02bf925f8dc29456f9eeaddeef800d2edcf837f8 100644 --- a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.flink.gradle +++ b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.flink.gradle @@ -3,6 +3,6 @@ plugins { } dependencies { - implementation group: 'org.apache.beam', name: 'beam-runners-flink-1.12', version: '2.27.0' - implementation group: 'org.apache.flink', name: 'flink-statebackend-rocksdb_2.11', version: '1.12.0' + implementation group: 'org.apache.beam', name: 'beam-runners-flink-1.13', version: '2.35.0' + implementation group: 'org.apache.flink', name: 'flink-statebackend-rocksdb_2.11', version: '1.13.0' } \ No newline at end of file diff --git a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.gradle b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.gradle index 41d1ae4f2bdfa358aca3fca2b91ea2b57e4c3405..5849bd93221794d135f1c6cb3bcb62d2174724b5 100644 --- a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.gradle +++ b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.gradle @@ -18,13 +18,12 @@ repositories { } } -def apacheBeamVersion = '2.22.0' //'2.27.0' // '2.34.0' +def apacheBeamVersion = '2.35.0' dependencies { // These dependencies are used internally, and not exposed to consumers on their own compile classpath. implementation('org.industrial-devops:titan-ccp-common:0.1.0-SNAPSHOT') { changing = true } implementation('org.industrial-devops:titan-ccp-common-kafka:0.1.0-SNAPSHOT') { changing = true } - implementation 'com.google.code.gson:gson:2.8.2' implementation 'com.google.guava:guava:24.1-jre' implementation 'org.slf4j:slf4j-simple:1.7.25' implementation project(':beam-commons') diff --git a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.samza.gradle b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.samza.gradle index 73e916ccc867b9b3316776192f0dab56fa0710f0..44c59317472686cae88d6992382ae081c9b64ace 100644 --- a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.samza.gradle +++ b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.beam.samza.gradle @@ -3,7 +3,8 @@ plugins { } dependencies { - implementation('org.apache.beam:beam-runners-samza:2.22.0') { + implementation('org.apache.beam:beam-runners-samza:2.35.0') { exclude group: 'org.apache.samza', module: 'samza-yarn_2.11' } + implementation 'org.apache.samza:samza-kafka_2.11:1.5.0' } \ No newline at end of file diff --git a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.flink.gradle b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.flink.gradle index f5e93dd88d2234f8a9b0d6fea880f47d652dccfa..7671e602211b6d9e923a3b2a4c87f40fff84c6ec 100644 --- a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.flink.gradle +++ b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.flink.gradle @@ -20,7 +20,7 @@ shadowJar { tasks.distZip.enabled = false ext { - flinkVersion = '1.12.2' + flinkVersion = '1.13.5' scalaBinaryVersion = '2.12' } @@ -41,7 +41,6 @@ dependencies { implementation 'org.apache.kafka:kafka-clients:2.2.0' implementation 'com.google.guava:guava:30.1-jre' - implementation 'com.google.code.gson:gson:2.8.2' implementation 'org.slf4j:slf4j-simple:1.6.1' implementation project(':flink-commons') diff --git a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.kstreams.gradle b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.kstreams.gradle index da2d42176ac0ddc9a157f843e3268b37ac4397e2..bf533915a8fdf4a712754857702373264a30f80a 100644 --- a/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.kstreams.gradle +++ b/theodolite-benchmarks/buildSrc/src/main/groovy/theodolite.kstreams.gradle @@ -22,8 +22,7 @@ dependencies { // These dependencies are used internally, and not exposed to consumers on their own compile classpath. implementation('org.industrial-devops:titan-ccp-common:0.1.0-SNAPSHOT') { changing = true } implementation('org.industrial-devops:titan-ccp-common-kafka:0.1.0-SNAPSHOT') { changing = true } - implementation 'org.apache.kafka:kafka-streams:2.6.0' // enable TransformerSuppliers - implementation 'com.google.code.gson:gson:2.8.2' + implementation 'org.apache.kafka:kafka-streams:3.1.0' implementation 'com.google.guava:guava:24.1-jre' implementation 'org.slf4j:slf4j-simple:1.7.25' implementation project(':kstreams-commons') diff --git a/theodolite-benchmarks/config/spotbugs-exclude-filter.xml b/theodolite-benchmarks/config/spotbugs-exclude-filter.xml index 38cd4b6914b3f0cbdeac2a22bf29ce0d7cf487ee..4bc9e048fb20e27feb793d4c6c398345cd599cff 100644 --- a/theodolite-benchmarks/config/spotbugs-exclude-filter.xml +++ b/theodolite-benchmarks/config/spotbugs-exclude-filter.xml @@ -5,4 +5,7 @@ <Rank value="16" /> </Match> + <!-- Temporary disabled due to potential false positive reported in https://github.com/spotbugs/spotbugs/issues/1923. --> + <Bug code="NP" /> + </FindBugsFilter> \ No newline at end of file diff --git a/theodolite-benchmarks/docker-test/README.md b/theodolite-benchmarks/docker-test/README.md index fd1e9bf4730f897273be45a022ad2adeae1b7e6e..5d7ca3f4ac470202579f154fe8f066a246c84d23 100644 --- a/theodolite-benchmarks/docker-test/README.md +++ b/theodolite-benchmarks/docker-test/README.md @@ -36,3 +36,19 @@ the host, for example, from the IDE or Gradle. In such cases, the following adju You can now connect to Kafka from your host system with bootstrap server `localhost:19092` and contact the Schema Registry via `localhost:8081`. **Pay attention to the Kafka port, which is *19092* instead of the default one *9092*.** + +## Running Smoke Tests + +The `smoketest-runner.sh` script can be used to run a simple test for a specific Docker Compose file. You can call it with + +```sh +./smoketest-runner.sh <docker-compose-dir> +``` + +where `<docker-compose-dir>` is the directory of a Docker-Compose file, for example, `uc2-beam-samza`. The script exists with a zero exit code in case of success and a non-zero exit code otherwise. + +You can also run the set of all smoke test with: + +```sh +./smoketest-runner-all.sh +``` diff --git a/theodolite-benchmarks/docker-test/smoketest-runner-all.sh b/theodolite-benchmarks/docker-test/smoketest-runner-all.sh new file mode 100755 index 0000000000000000000000000000000000000000..0129a485d98a90d453b284408b755986f64208de --- /dev/null +++ b/theodolite-benchmarks/docker-test/smoketest-runner-all.sh @@ -0,0 +1,7 @@ +#!/bin/sh + +find . -name 'test.sh' -type f -exec dirname {} \; | + sort | + xargs -I %s sh -c "./smoketest-runner.sh %s 1>&2; echo $?" | + sort | + awk 'BEGIN {count[0]=0; count[1]=0} {count[$1!=0]++} END {print count[0] " tests successful, " count[1] " test failed."; exit count[1]}' diff --git a/theodolite-benchmarks/docker-test/smoketest-runner.sh b/theodolite-benchmarks/docker-test/smoketest-runner.sh new file mode 100755 index 0000000000000000000000000000000000000000..3167c90a567eed3cc2678c80c722dbd0b8684f1e --- /dev/null +++ b/theodolite-benchmarks/docker-test/smoketest-runner.sh @@ -0,0 +1,29 @@ +#!/bin/sh + +COMPOSE_FILE_PATH=$1 +echo "Run test for '$COMPOSE_FILE_PATH'." + +cd $COMPOSE_FILE_PATH +docker-compose pull -q +docker-compose up -d kafka zookeeper schema-registry +sleep 30s +docker-compose up -d +sleep 5s +docker-compose ps + +if test -f "./test.sh"; then + #timeout --foreground 3m ./test.sh + ./test.sh + RETURN=$? +else + RETURN=$? + echo "test.sh does not exists for '$COMPOSE_FILE_PATH'." +fi +if [ $RETURN -eq 0 ]; then + echo "Test for '$COMPOSE_FILE_PATH' has passed." +else + echo "Test for '$COMPOSE_FILE_PATH' has failed." +fi + +docker-compose down +exit $RETURN diff --git a/theodolite-benchmarks/docker-test/uc1-beam-flink/docker-compose.yml b/theodolite-benchmarks/docker-test/uc1-beam-flink/docker-compose.yml index 1b683b4ca65a2582aa6a4d68444c4bbef7895b73..69de6049aca50a6a8e10a9e709b38d322bc8cf61 100644 --- a/theodolite-benchmarks/docker-test/uc1-beam-flink/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc1-beam-flink/docker-compose.yml @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc1-workload-generator:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc1-beam-flink/test.sh b/theodolite-benchmarks/docker-test/uc1-beam-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..ebbecd1c5336c5dd907db11b8c8c45924e5924a8 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc1-beam-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +sleep 55s # to let the benchmark and produce some output +docker-compose logs --tail 100 benchmark-taskmanager | + sed -n "s/^.*Key:\s\(\S*\), Value:\s\(\S*\).*$/\2/p" | + tee /dev/stderr | + jq .identifier | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc1-beam-samza/docker-compose.yml b/theodolite-benchmarks/docker-test/uc1-beam-samza/docker-compose.yml index f5213799daa2d51eea53e794becdffc151a4da56..2212c3b539045114f31760d605ad928e237ed924 100644 --- a/theodolite-benchmarks/docker-test/uc1-beam-samza/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc1-beam-samza/docker-compose.yml @@ -22,18 +22,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc1-beam-samza:${THEODOLITE_TAG:-latest} scale: 1 diff --git a/theodolite-benchmarks/docker-test/uc1-beam-samza/test.sh b/theodolite-benchmarks/docker-test/uc1-beam-samza/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..ed17db3a44d5c4a8dacfbc956c2f36dd47503508 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc1-beam-samza/test.sh @@ -0,0 +1,13 @@ +#!/bin/sh + +sleep 55s # to let the benchmark and produce some output +docker-compose logs --tail 100 benchmark | + sed -n "s/^.*Key:\s\(\S*\), Value:\s\(\S*\).*$/\2/p" | + tee /dev/stderr | + jq .identifier | + sort | + uniq | + wc -l | + grep "\b10\b" + + diff --git a/theodolite-benchmarks/docker-test/uc1-flink-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc1-flink/docker-compose.yml similarity index 90% rename from theodolite-benchmarks/docker-test/uc1-flink-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc1-flink/docker-compose.yml index 6c661bb49cb4173357acc89c06783fe5e0a2ce49..55e359665de56cb03a5049c299761cb07690df30 100755 --- a/theodolite-benchmarks/docker-test/uc1-flink-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc1-flink/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc1-workload-generator:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc1-flink/test.sh b/theodolite-benchmarks/docker-test/uc1-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..7c7f11a94f42d56d91d383f27d58ad9a09a918e5 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc1-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +sleep 55s # to let the benchmark and produce some output +docker-compose logs --tail 100 benchmark-taskmanager | + sed -n "s/^.*Record:\s\(\S*\)$/\1/p" | + tee /dev/stderr | + jq .identifier | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc1-kstreams-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc1-kstreams/docker-compose.yml similarity index 83% rename from theodolite-benchmarks/docker-test/uc1-kstreams-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc1-kstreams/docker-compose.yml index 25c19e35e5dae1807ef46fb8ade4e888dff0c2d8..c85ce305c2f1383a77d4c405d52089ec1d2b02a6 100755 --- a/theodolite-benchmarks/docker-test/uc1-kstreams-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc1-kstreams/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,20 +20,25 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: - image: ghcr.io/cau-se/theodolite-uc1-kstreams-app:${THEODOLITE_TAG:-latest} + image: ghcr.io/cau-se/theodolite-uc1-kstreams-app:latest depends_on: - schema-registry - kafka diff --git a/theodolite-benchmarks/docker-test/uc1-kstreams/test.sh b/theodolite-benchmarks/docker-test/uc1-kstreams/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..da711e3fac1b1d664b1c8487687ceacdddce6efa --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc1-kstreams/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +sleep 55s # to let the benchmark and produce some output +docker-compose logs --tail 100 benchmark | + sed -n "s/^.*Record:\s\(\S*\)$/\1/p" | + tee /dev/stderr | + jq .identifier | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc2-beam-flink/docker-compose.yml b/theodolite-benchmarks/docker-test/uc2-beam-flink/docker-compose.yml index 8427161e43faa920e011973d76f32f9cc9d62f8c..79f6f040144d3c4da56d469095ad3ddce90026a2 100644 --- a/theodolite-benchmarks/docker-test/uc2-beam-flink/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc2-beam-flink/docker-compose.yml @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc2-workload-generator:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc2-beam-flink/test.sh b/theodolite-benchmarks/docker-test/uc2-beam-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..3c33834b0b21bc5dbe4e6a7c3ff947121bb2ce71 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc2-beam-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 20 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc2-beam-samza/docker-compose.yml b/theodolite-benchmarks/docker-test/uc2-beam-samza/docker-compose.yml index 820bb25dd370cf6c7410b20fbdbdb1d4281f47d3..cc6bc7a7112c35f11ce9cfd27d09aebe401c8c51 100644 --- a/theodolite-benchmarks/docker-test/uc2-beam-samza/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc2-beam-samza/docker-compose.yml @@ -22,18 +22,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc2-beam-samza:${THEODOLITE_TAG:-latest} scale: 1 diff --git a/theodolite-benchmarks/docker-test/uc2-beam-samza/test.sh b/theodolite-benchmarks/docker-test/uc2-beam-samza/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..3c33834b0b21bc5dbe4e6a7c3ff947121bb2ce71 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc2-beam-samza/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 20 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc2-flink-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc2-flink/docker-compose.yml similarity index 89% rename from theodolite-benchmarks/docker-test/uc2-flink-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc2-flink/docker-compose.yml index e51c544bfd64c6f540f5a84a5f617de745a1a6b6..1397c8ab0e7bb8d7ee67b7fa9c18143c404fd9ff 100755 --- a/theodolite-benchmarks/docker-test/uc2-flink-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc2-flink/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka - #ports: - # - "8081:8081" + restart: "on-failure" expose: - "8081" + #ports: + # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc2-workload-generator:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc2-flink/test.sh b/theodolite-benchmarks/docker-test/uc2-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..3c33834b0b21bc5dbe4e6a7c3ff947121bb2ce71 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc2-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 20 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc2-kstreams-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc2-kstreams/docker-compose.yml similarity index 86% rename from theodolite-benchmarks/docker-test/uc2-kstreams-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc2-kstreams/docker-compose.yml index 4cc4978d844a8b361af4061ab41d1adc4e7a1813..efdba90bef634bab76012316f67b0f9be9f79c77 100755 --- a/theodolite-benchmarks/docker-test/uc2-kstreams-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc2-kstreams/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka - #ports: - # - "8081:8081" + restart: "on-failure" expose: - "8081" + #ports: + # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc2-kstreams-app:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc2-kstreams/test.sh b/theodolite-benchmarks/docker-test/uc2-kstreams/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..3c33834b0b21bc5dbe4e6a7c3ff947121bb2ce71 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc2-kstreams/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 20 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc3-beam-flink/docker-compose.yml b/theodolite-benchmarks/docker-test/uc3-beam-flink/docker-compose.yml index b9e934b61f9742ad4e83601ff13707be0641c875..e68dc94d4185d789272279e8fa2e5d178ffdd14b 100644 --- a/theodolite-benchmarks/docker-test/uc3-beam-flink/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc3-beam-flink/docker-compose.yml @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc3-workload-generator:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc3-beam-flink/test.sh b/theodolite-benchmarks/docker-test/uc3-beam-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..0612838a3973a302c9acb3bbfa6b8d59ea2596c5 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc3-beam-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 600 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc3-beam-samza/docker-compose.yml b/theodolite-benchmarks/docker-test/uc3-beam-samza/docker-compose.yml index 3ae3a2faa7c0849b76ea838b09ba0254bdfea936..bf120f31dbfda2384b314ba4a90a25362f37b6c2 100644 --- a/theodolite-benchmarks/docker-test/uc3-beam-samza/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc3-beam-samza/docker-compose.yml @@ -22,18 +22,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc3-beam-samza:${THEODOLITE_TAG:-latest} scale: 1 diff --git a/theodolite-benchmarks/docker-test/uc3-beam-samza/test.sh b/theodolite-benchmarks/docker-test/uc3-beam-samza/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..0612838a3973a302c9acb3bbfa6b8d59ea2596c5 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc3-beam-samza/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 600 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc3-flink-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc3-flink/docker-compose.yml similarity index 89% rename from theodolite-benchmarks/docker-test/uc3-flink-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc3-flink/docker-compose.yml index 42c55950fbea022f00019699ae72678fbe88b8cb..0c83d8a25d19d4989f982d251893635ea1222cb0 100755 --- a/theodolite-benchmarks/docker-test/uc3-flink-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc3-flink/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka - #ports: - # - "8081:8081" + restart: "on-failure" expose: - "8081" + #ports: + # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc3-workload-generator:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc3-kstreams-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc3-kstreams/docker-compose.yml similarity index 85% rename from theodolite-benchmarks/docker-test/uc3-kstreams-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc3-kstreams/docker-compose.yml index f943de372b75bf64df0028679d56950fedfaec48..7da1fa3b420e2dfa0d6d357723583bc3a256823e 100755 --- a/theodolite-benchmarks/docker-test/uc3-kstreams-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc3-kstreams/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka - #ports: - # - "8081:8081" + restart: "on-failure" expose: - "8081" + #ports: + # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc3-kstreams-app:${THEODOLITE_TAG:-latest} depends_on: diff --git a/theodolite-benchmarks/docker-test/uc3-kstreams/test.sh b/theodolite-benchmarks/docker-test/uc3-kstreams/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..0612838a3973a302c9acb3bbfa6b8d59ea2596c5 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc3-kstreams/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=s -r http://schema-registry:8081 -f '%k:%s\n' -c 600 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b10\b" diff --git a/theodolite-benchmarks/docker-test/uc4-beam-flink/docker-compose.yml b/theodolite-benchmarks/docker-test/uc4-beam-flink/docker-compose.yml index b0d18dfe34c23847b3cc0b04d788fdc4b3552e68..2968b718eb06ade4d42bbe104e214d573881ac14 100644 --- a/theodolite-benchmarks/docker-test/uc4-beam-flink/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc4-beam-flink/docker-compose.yml @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1,configuration:3:1,aggregation-feedback:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc4-workload-generator:${THEODOLITE_TAG:-latest} depends_on: @@ -43,7 +48,7 @@ services: KAFKA_BOOTSTRAP_SERVERS: kafka:9092 SCHEMA_REGISTRY_URL: http://schema-registry:8081 NUM_SENSORS: 4 - NUM_NESTED_GROUPS: 4 + NUM_NESTED_GROUPS: 3 benchmark-jobmanager: image: ghcr.io/cau-se/theodolite-uc4-beam-flink:${THEODOLITE_TAG:-latest} #ports: diff --git a/theodolite-benchmarks/docker-test/uc4-beam-flink/test.sh b/theodolite-benchmarks/docker-test/uc4-beam-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..6a4c6dbf4f583e7598baefae8f48136bb2113630 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc4-beam-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=avro -r http://schema-registry:8081 -f '%k:%s\n' -c 2000 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b21\b" diff --git a/theodolite-benchmarks/docker-test/uc4-beam-samza/docker-compose.yml b/theodolite-benchmarks/docker-test/uc4-beam-samza/docker-compose.yml index b9e4142cb0be2701f7772cff5725b524aaffcbc7..d236af7d284ebb085c78110feb6001cb28d18290 100644 --- a/theodolite-benchmarks/docker-test/uc4-beam-samza/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc4-beam-samza/docker-compose.yml @@ -22,18 +22,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1,configuration:3:1,aggregation-feedback:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc4-beam-samza:${THEODOLITE_TAG:-latest} scale: 1 @@ -57,4 +62,4 @@ services: KAFKA_BOOTSTRAP_SERVERS: kafka:9092 SCHEMA_REGISTRY_URL: http://schema-registry:8081 NUM_SENSORS: 4 - NUM_NESTED_GROUPS: 4 + NUM_NESTED_GROUPS: 3 diff --git a/theodolite-benchmarks/docker-test/uc4-beam-samza/test.sh b/theodolite-benchmarks/docker-test/uc4-beam-samza/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..6a4c6dbf4f583e7598baefae8f48136bb2113630 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc4-beam-samza/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=avro -r http://schema-registry:8081 -f '%k:%s\n' -c 2000 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b21\b" diff --git a/theodolite-benchmarks/docker-test/uc4-flink-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc4-flink/docker-compose.yml similarity index 89% rename from theodolite-benchmarks/docker-test/uc4-flink-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc4-flink/docker-compose.yml index 5a5a5924ef13cc9691b267ce6169bdb85a1e267b..24e5acdf94d1ba3d5c2807d172ba054309a2ef4a 100755 --- a/theodolite-benchmarks/docker-test/uc4-flink-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc4-flink/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1,configuration:3:1,aggregation-feedback:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 load-generator: image: ghcr.io/cau-se/theodolite-uc4-workload-generator:${THEODOLITE_TAG:-latest} depends_on: @@ -43,7 +48,7 @@ services: KAFKA_BOOTSTRAP_SERVERS: kafka:9092 SCHEMA_REGISTRY_URL: http://schema-registry:8081 NUM_SENSORS: 4 - NUM_NESTED_GROUPS: 4 + NUM_NESTED_GROUPS: 3 benchmark-jobmanager: image: ghcr.io/cau-se/theodolite-uc4-flink:${THEODOLITE_TAG:-latest} #ports: diff --git a/theodolite-benchmarks/docker-test/uc4-flink/test.sh b/theodolite-benchmarks/docker-test/uc4-flink/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..0a478cf6641a1333f65281ae43cb525e32cb2510 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc4-flink/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=avro -r http://schema-registry:8081 -f '%k:%s\n' -c 500 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b21\b" diff --git a/theodolite-benchmarks/docker-test/uc4-kstreams-docker-compose/docker-compose.yml b/theodolite-benchmarks/docker-test/uc4-kstreams/docker-compose.yml similarity index 86% rename from theodolite-benchmarks/docker-test/uc4-kstreams-docker-compose/docker-compose.yml rename to theodolite-benchmarks/docker-test/uc4-kstreams/docker-compose.yml index 1818505787f662d57f1a96c8c79b03d61cf9da2a..6aaa02990841547edb6059e4e2fbf3b28b50985c 100755 --- a/theodolite-benchmarks/docker-test/uc4-kstreams-docker-compose/docker-compose.yml +++ b/theodolite-benchmarks/docker-test/uc4-kstreams/docker-compose.yml @@ -1,4 +1,4 @@ -version: '2' +version: '2.2' services: zookeeper: image: confluentinc/cp-zookeeper @@ -20,18 +20,23 @@ services: KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 30000 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" KAFKA_CREATE_TOPICS: "input:3:1,output:3:1,configuration:3:1,aggregation-feedback:3:1" + kcat: + image: edenhill/kcat:1.7.1 + entrypoint: "sh" + tty: true schema-registry: - image: confluentinc/cp-schema-registry:5.3.1 + image: confluentinc/cp-schema-registry:7.0.1 depends_on: - zookeeper - kafka + restart: "on-failure" expose: - "8081" #ports: # - 8081:8081 environment: SCHEMA_REGISTRY_HOST_NAME: schema-registry - SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181' + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092 benchmark: image: ghcr.io/cau-se/theodolite-uc4-kstreams-app:${THEODOLITE_TAG:-latest} depends_on: @@ -51,4 +56,4 @@ services: KAFKA_BOOTSTRAP_SERVERS: kafka:9092 SCHEMA_REGISTRY_URL: http://schema-registry:8081 NUM_SENSORS: 4 - NUM_NESTED_GROUPS: 4 \ No newline at end of file + NUM_NESTED_GROUPS: 3 \ No newline at end of file diff --git a/theodolite-benchmarks/docker-test/uc4-kstreams/test.sh b/theodolite-benchmarks/docker-test/uc4-kstreams/test.sh new file mode 100755 index 0000000000000000000000000000000000000000..9b9dee7dc78e7a587b9f2e5b778066e5bc099755 --- /dev/null +++ b/theodolite-benchmarks/docker-test/uc4-kstreams/test.sh @@ -0,0 +1,11 @@ +#!/bin/sh + +until docker-compose exec -T kcat kcat -L -b kafka:9092 -t output -J | jq -r '.topics[0].partitions | length' | grep "\b3\b"; do sleep 5s; done + +docker-compose exec -T kcat kcat -C -b kafka:9092 -t output -s key=s -s value=avro -r http://schema-registry:8081 -f '%k:%s\n' -c 32 | + tee /dev/stderr | + awk -F ':' '!/^%/ {print $1}' | + sort | + uniq | + wc -l | + grep "\b16\b" diff --git a/theodolite-benchmarks/flink-commons/build.gradle b/theodolite-benchmarks/flink-commons/build.gradle index a3a4a35752006bb10e15ff508ce0b37f70adc57d..bd07f1ca51ae6b781a92ae06ba8b2555fa83b873 100644 --- a/theodolite-benchmarks/flink-commons/build.gradle +++ b/theodolite-benchmarks/flink-commons/build.gradle @@ -3,7 +3,7 @@ plugins { } ext { - flinkVersion = '1.12.0' + flinkVersion = '1.13.5' scalaBinaryVersion = '2.12' } diff --git a/theodolite-benchmarks/http-bridge/.settings/org.eclipse.jdt.ui.prefs b/theodolite-benchmarks/http-bridge/.settings/org.eclipse.jdt.ui.prefs new file mode 100644 index 0000000000000000000000000000000000000000..a375cb792eeb842ecfd1f789fbf6a716df43e9c8 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/.settings/org.eclipse.jdt.ui.prefs @@ -0,0 +1,127 @@ +cleanup.add_default_serial_version_id=true +cleanup.add_generated_serial_version_id=false +cleanup.add_missing_annotations=true +cleanup.add_missing_deprecated_annotations=true +cleanup.add_missing_methods=false +cleanup.add_missing_nls_tags=false +cleanup.add_missing_override_annotations=true +cleanup.add_missing_override_annotations_interface_methods=true +cleanup.add_serial_version_id=false +cleanup.always_use_blocks=true +cleanup.always_use_parentheses_in_expressions=false +cleanup.always_use_this_for_non_static_field_access=true +cleanup.always_use_this_for_non_static_method_access=true +cleanup.convert_functional_interfaces=false +cleanup.convert_to_enhanced_for_loop=true +cleanup.correct_indentation=true +cleanup.format_source_code=true +cleanup.format_source_code_changes_only=false +cleanup.insert_inferred_type_arguments=false +cleanup.make_local_variable_final=true +cleanup.make_parameters_final=true +cleanup.make_private_fields_final=true +cleanup.make_type_abstract_if_missing_method=false +cleanup.make_variable_declarations_final=true +cleanup.never_use_blocks=false +cleanup.never_use_parentheses_in_expressions=true +cleanup.organize_imports=true +cleanup.qualify_static_field_accesses_with_declaring_class=false +cleanup.qualify_static_member_accesses_through_instances_with_declaring_class=true +cleanup.qualify_static_member_accesses_through_subtypes_with_declaring_class=true +cleanup.qualify_static_member_accesses_with_declaring_class=true +cleanup.qualify_static_method_accesses_with_declaring_class=false +cleanup.remove_private_constructors=true +cleanup.remove_redundant_modifiers=false +cleanup.remove_redundant_semicolons=true +cleanup.remove_redundant_type_arguments=true +cleanup.remove_trailing_whitespaces=true +cleanup.remove_trailing_whitespaces_all=true +cleanup.remove_trailing_whitespaces_ignore_empty=false +cleanup.remove_unnecessary_casts=true +cleanup.remove_unnecessary_nls_tags=true +cleanup.remove_unused_imports=true +cleanup.remove_unused_local_variables=false +cleanup.remove_unused_private_fields=true +cleanup.remove_unused_private_members=false +cleanup.remove_unused_private_methods=true +cleanup.remove_unused_private_types=true +cleanup.sort_members=false +cleanup.sort_members_all=false +cleanup.use_anonymous_class_creation=false +cleanup.use_blocks=true +cleanup.use_blocks_only_for_return_and_throw=false +cleanup.use_lambda=true +cleanup.use_parentheses_in_expressions=true +cleanup.use_this_for_non_static_field_access=true +cleanup.use_this_for_non_static_field_access_only_if_necessary=false +cleanup.use_this_for_non_static_method_access=true +cleanup.use_this_for_non_static_method_access_only_if_necessary=false +cleanup_profile=_CAU-SE-Style +cleanup_settings_version=2 +eclipse.preferences.version=1 +editor_save_participant_org.eclipse.jdt.ui.postsavelistener.cleanup=true +formatter_profile=_CAU-SE-Style +formatter_settings_version=21 +org.eclipse.jdt.ui.ignorelowercasenames=true +org.eclipse.jdt.ui.importorder=; +org.eclipse.jdt.ui.ondemandthreshold=99 +org.eclipse.jdt.ui.staticondemandthreshold=99 +sp_cleanup.add_default_serial_version_id=true +sp_cleanup.add_generated_serial_version_id=false +sp_cleanup.add_missing_annotations=true +sp_cleanup.add_missing_deprecated_annotations=true +sp_cleanup.add_missing_methods=false +sp_cleanup.add_missing_nls_tags=false +sp_cleanup.add_missing_override_annotations=true +sp_cleanup.add_missing_override_annotations_interface_methods=true +sp_cleanup.add_serial_version_id=false +sp_cleanup.always_use_blocks=true +sp_cleanup.always_use_parentheses_in_expressions=false +sp_cleanup.always_use_this_for_non_static_field_access=true +sp_cleanup.always_use_this_for_non_static_method_access=true +sp_cleanup.convert_functional_interfaces=false +sp_cleanup.convert_to_enhanced_for_loop=true +sp_cleanup.correct_indentation=true +sp_cleanup.format_source_code=true +sp_cleanup.format_source_code_changes_only=false +sp_cleanup.insert_inferred_type_arguments=false +sp_cleanup.make_local_variable_final=true +sp_cleanup.make_parameters_final=true +sp_cleanup.make_private_fields_final=true +sp_cleanup.make_type_abstract_if_missing_method=false +sp_cleanup.make_variable_declarations_final=true +sp_cleanup.never_use_blocks=false +sp_cleanup.never_use_parentheses_in_expressions=true +sp_cleanup.on_save_use_additional_actions=true +sp_cleanup.organize_imports=true +sp_cleanup.qualify_static_field_accesses_with_declaring_class=false +sp_cleanup.qualify_static_member_accesses_through_instances_with_declaring_class=true +sp_cleanup.qualify_static_member_accesses_through_subtypes_with_declaring_class=true +sp_cleanup.qualify_static_member_accesses_with_declaring_class=true +sp_cleanup.qualify_static_method_accesses_with_declaring_class=false +sp_cleanup.remove_private_constructors=true +sp_cleanup.remove_redundant_modifiers=false +sp_cleanup.remove_redundant_semicolons=true +sp_cleanup.remove_redundant_type_arguments=true +sp_cleanup.remove_trailing_whitespaces=true +sp_cleanup.remove_trailing_whitespaces_all=true +sp_cleanup.remove_trailing_whitespaces_ignore_empty=false +sp_cleanup.remove_unnecessary_casts=true +sp_cleanup.remove_unnecessary_nls_tags=true +sp_cleanup.remove_unused_imports=true +sp_cleanup.remove_unused_local_variables=false +sp_cleanup.remove_unused_private_fields=true +sp_cleanup.remove_unused_private_members=false +sp_cleanup.remove_unused_private_methods=true +sp_cleanup.remove_unused_private_types=true +sp_cleanup.sort_members=false +sp_cleanup.sort_members_all=false +sp_cleanup.use_anonymous_class_creation=false +sp_cleanup.use_blocks=true +sp_cleanup.use_blocks_only_for_return_and_throw=false +sp_cleanup.use_lambda=true +sp_cleanup.use_parentheses_in_expressions=true +sp_cleanup.use_this_for_non_static_field_access=true +sp_cleanup.use_this_for_non_static_field_access_only_if_necessary=false +sp_cleanup.use_this_for_non_static_method_access=true +sp_cleanup.use_this_for_non_static_method_access_only_if_necessary=false diff --git a/theodolite-benchmarks/http-bridge/.settings/qa.eclipse.plugin.checkstyle.prefs b/theodolite-benchmarks/http-bridge/.settings/qa.eclipse.plugin.checkstyle.prefs new file mode 100644 index 0000000000000000000000000000000000000000..87860c815222845c1d264d7d0ce498d3397f8280 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/.settings/qa.eclipse.plugin.checkstyle.prefs @@ -0,0 +1,4 @@ +configFilePath=../config/checkstyle.xml +customModulesJarPaths= +eclipse.preferences.version=1 +enabled=true diff --git a/theodolite-benchmarks/http-bridge/.settings/qa.eclipse.plugin.pmd.prefs b/theodolite-benchmarks/http-bridge/.settings/qa.eclipse.plugin.pmd.prefs new file mode 100644 index 0000000000000000000000000000000000000000..efbcb8c9e5d449194a48ca1ea42b7d807b573db9 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/.settings/qa.eclipse.plugin.pmd.prefs @@ -0,0 +1,4 @@ +customRulesJars= +eclipse.preferences.version=1 +enabled=true +ruleSetFilePath=../config/pmd.xml diff --git a/theodolite-benchmarks/http-bridge/Dockerfile b/theodolite-benchmarks/http-bridge/Dockerfile new file mode 100644 index 0000000000000000000000000000000000000000..b31dbcdef48f3b9eadf81a35c95e441c4b54955b --- /dev/null +++ b/theodolite-benchmarks/http-bridge/Dockerfile @@ -0,0 +1,6 @@ +FROM openjdk:11-slim + +ADD build/distributions/http-bridge.tar / + +CMD JAVA_OPTS="$JAVA_OPTS -Dorg.slf4j.simpleLogger.defaultLogLevel=$LOG_LEVEL" \ + /http-bridge/bin/http-bridge \ No newline at end of file diff --git a/theodolite-benchmarks/http-bridge/build.gradle b/theodolite-benchmarks/http-bridge/build.gradle new file mode 100644 index 0000000000000000000000000000000000000000..0377eefc76b456d8e0f94087b06d0c2689f977cf --- /dev/null +++ b/theodolite-benchmarks/http-bridge/build.gradle @@ -0,0 +1,31 @@ +plugins { + // common java conventions + id 'theodolite.java-conventions' + + // make executable + id 'application' +} + +tasks.distZip.enabled = false + +repositories { + mavenCentral() + maven { + url "https://oss.sonatype.org/content/repositories/snapshots/" + } + maven { + url 'https://packages.confluent.io/maven/' + } +} + +dependencies { + implementation('org.industrial-devops:titan-ccp-common:0.1.0-SNAPSHOT') { changing = true } + implementation('org.industrial-devops:titan-ccp-common-kafka:0.1.0-SNAPSHOT') { changing = true } + implementation project(':load-generator-commons') + + implementation 'io.javalin:javalin:4.3.0' + implementation 'com.google.code.gson:gson:2.8.2' + runtimeOnly 'org.slf4j:slf4j-simple:1.7.25' + + testImplementation 'junit:junit:4.12' +} diff --git a/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/Deserializer.java b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/Deserializer.java new file mode 100644 index 0000000000000000000000000000000000000000..f25c120d2165c4a1f747fdba32de43d4e4d157a6 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/Deserializer.java @@ -0,0 +1,13 @@ +package theodolite.commons.httpbridge; + +/** + * A class for converting objects to strings. + * + * @param <T> Type to be deserialized from. + */ +@FunctionalInterface +public interface Deserializer<T> { + + T deserialize(String json); + +} diff --git a/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/Endpoint.java b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/Endpoint.java new file mode 100644 index 0000000000000000000000000000000000000000..43850d80699a0db0b0fcebd76f625a17f8133f30 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/Endpoint.java @@ -0,0 +1,52 @@ +package theodolite.commons.httpbridge; + +import theodolite.commons.workloadgeneration.RecordSender; + +/** + * Class describing an endpoint of the {@link HttpBridge}, which converts JSON objects to Java + * objects and sends them using a {@link RecordSender}. + * + * @param <T> Type of objects this endpoint receives and converts. + */ +public class Endpoint<T> { + + private final String path; + + private final Deserializer<? extends T> recordDeserializer; + + private final RecordSender<? super T> recordSender; + + /** + * Create a new {@link Endpoint} at the given path. + */ + public Endpoint( + final String path, + final Deserializer<? extends T> recordDeserializer, + final RecordSender<? super T> recordSender) { + this.path = path; + this.recordDeserializer = recordDeserializer; + this.recordSender = recordSender; + } + + /** + * Create a new {@link Endpoint} at the given path with a {@link GsonDeserializer}. + */ + public Endpoint( + final String path, + final Class<T> recordType, + final RecordSender<? super T> recordSender) { + this.path = path; + this.recordDeserializer = new GsonDeserializer<>(recordType); + this.recordSender = recordSender; + } + + public String getPath() { + return this.path; + } + + public void convert(final String json) { + final T record = this.recordDeserializer.deserialize(json); + this.recordSender.send(record); + } + +} diff --git a/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/EnvVarHttpBridgeFactory.java b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/EnvVarHttpBridgeFactory.java new file mode 100644 index 0000000000000000000000000000000000000000..97956b8d47185c90efdc03393c03c8c44aea2335 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/EnvVarHttpBridgeFactory.java @@ -0,0 +1,59 @@ +package theodolite.commons.httpbridge; + +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import theodolite.commons.workloadgeneration.ConfigurationKeys; +import theodolite.commons.workloadgeneration.TitanKafkaSenderFactory; +import titan.ccp.model.records.ActivePowerRecord; + +class EnvVarHttpBridgeFactory { + + private static final String PORT_KEY = "PORT"; + private static final int PORT_DEFAULT = 8080; + + private static final String HOST_KEY = "HOST"; + private static final String HOST_DEFAULT = "0.0.0.0"; // NOPMD + + private static final String KAFKA_BOOTSTRAP_SERVERS_DEFAULT = "localhost:9092"; // NOPMD + private static final String KAFKA_TOPIC_DEFAULT = "input"; + private static final String SCHEMA_REGISTRY_URL_DEFAULT = "http://localhost:8081"; + + public HttpBridge create() { + final Endpoint<?> converter = new Endpoint<>( + "/", + ActivePowerRecord.class, + TitanKafkaSenderFactory.forKafkaConfig( + this.getKafkaBootstrapServer(), + this.getKafkaTopic(), + this.getSchemaRegistryUrl())); + return new HttpBridge(this.getHost(), this.getPort(), List.of(converter)); + } + + private String getHost() { + return Objects.requireNonNullElse(System.getenv(HOST_KEY), HOST_DEFAULT); + } + + private int getPort() { + return Optional.ofNullable(System.getenv(PORT_KEY)).map(Integer::parseInt).orElse(PORT_DEFAULT); + } + + private String getKafkaBootstrapServer() { + return Objects.requireNonNullElse( + System.getenv(ConfigurationKeys.KAFKA_BOOTSTRAP_SERVERS), + KAFKA_BOOTSTRAP_SERVERS_DEFAULT); + } + + private String getKafkaTopic() { + return Objects.requireNonNullElse( + System.getenv(ConfigurationKeys.KAFKA_INPUT_TOPIC), + KAFKA_TOPIC_DEFAULT); + } + + private String getSchemaRegistryUrl() { + return Objects.requireNonNullElse( + System.getenv(ConfigurationKeys.SCHEMA_REGISTRY_URL), + SCHEMA_REGISTRY_URL_DEFAULT); + } + +} diff --git a/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/GsonDeserializer.java b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/GsonDeserializer.java new file mode 100644 index 0000000000000000000000000000000000000000..42220617546527157d5463d6b9ce9208abc66d58 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/GsonDeserializer.java @@ -0,0 +1,29 @@ +package theodolite.commons.httpbridge; + +import com.google.gson.Gson; + +/** + * A {@link Deserializer} based on GSON. + * + * @param <T> Type to be serialized from. + */ +public class GsonDeserializer<T> implements Deserializer<T> { + + private final Gson gson; + private final Class<T> targetClass; + + public GsonDeserializer(final Class<T> targetClass) { + this(new Gson(), targetClass); + } + + public GsonDeserializer(final Gson gson, final Class<T> targetClass) { + this.gson = gson; + this.targetClass = targetClass; + } + + @Override + public T deserialize(final String json) { + return this.gson.fromJson(json, this.targetClass); + } + +} diff --git a/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/HttpBridge.java b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/HttpBridge.java new file mode 100644 index 0000000000000000000000000000000000000000..d36e191c8b0a591107de796f511aa853063dff73 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/HttpBridge.java @@ -0,0 +1,40 @@ +package theodolite.commons.httpbridge; + +import java.util.List; +import theodolite.commons.workloadgeneration.RecordSender; + +/** + * Class that creates a webserver with potentially multiple {@link Endpoint}s, which receives JSON + * objects at these endpoints, converts them to Java objects and send them using + * {@link RecordSender}s. + */ +public class HttpBridge { + + private final JavalinWebServer webServer; + + public HttpBridge(final String host, final int port, final List<Endpoint<?>> converters) { + this.webServer = new JavalinWebServer(converters, host, port); + } + + public void start() { + this.webServer.start(); + } + + public void stop() { + this.webServer.stop(); + } + + public void runAsStandalone() { + Runtime.getRuntime().addShutdownHook(new Thread(() -> this.stop())); + this.start(); + } + + public static HttpBridge fromEnvironment() { + return new EnvVarHttpBridgeFactory().create(); + } + + public static void main(final String[] args) { + HttpBridge.fromEnvironment().runAsStandalone(); + } + +} diff --git a/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/JavalinWebServer.java b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/JavalinWebServer.java new file mode 100644 index 0000000000000000000000000000000000000000..c23a17588d661fc5d1c6e9eb294d2d29fc165675 --- /dev/null +++ b/theodolite-benchmarks/http-bridge/src/main/java/theodolite/commons/httpbridge/JavalinWebServer.java @@ -0,0 +1,53 @@ +package theodolite.commons.httpbridge; + +import io.javalin.Javalin; +import java.util.Collection; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implementation of a webserver based on the Javalin framework. + */ +public class JavalinWebServer { + + private static final Logger LOGGER = LoggerFactory.getLogger(JavalinWebServer.class); + + private static final int HTTP_SUCCESS = 200; + + private final Javalin app = Javalin.create(); + + private final String host; + private final int port; + + /** + * Create a new instance, running on the specified host and port with the configured endpoints. + */ + public JavalinWebServer( + final Collection<Endpoint<?>> converters, + final String host, + final int port) { + this.host = host; + this.port = port; + this.configureRoutes(converters); + } + + private void configureRoutes(final Collection<Endpoint<?>> endpoints) { + for (final Endpoint<?> endpoint : endpoints) { + this.app.post(endpoint.getPath(), ctx -> { + final String record = ctx.body(); + LOGGER.debug("Received record at '{}': {}", ctx.path(), record); + endpoint.convert(record); + ctx.status(HTTP_SUCCESS); + }); + } + } + + public void start() { + this.app.start(this.host, this.port); + } + + public void stop() { + this.app.close(); + } + +} diff --git a/theodolite-benchmarks/kstreams-commons/build.gradle b/theodolite-benchmarks/kstreams-commons/build.gradle index 7683ffe39314ec375eda0ed4e139d618d44a7328..167a75327b251af20b1142fe42c82b3bbedfe62b 100644 --- a/theodolite-benchmarks/kstreams-commons/build.gradle +++ b/theodolite-benchmarks/kstreams-commons/build.gradle @@ -17,7 +17,7 @@ dependencies { // implementation 'org.slf4j:slf4j-simple:1.7.25' implementation('org.industrial-devops:titan-ccp-common:0.1.0-SNAPSHOT') { changing = true } implementation('org.industrial-devops:titan-ccp-common-kafka:0.1.0-SNAPSHOT') { changing = true } - implementation 'org.apache.kafka:kafka-streams:2.6.0' + implementation 'org.apache.kafka:kafka-streams:3.1.0' // Use JUnit test framework testImplementation 'junit:junit:4.12' diff --git a/theodolite-benchmarks/kstreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java b/theodolite-benchmarks/kstreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java index 89bd3147f0d3bb7a5fecc5d8c7d277bd294494ad..fe3cf484a81ee3561ad17b6b25d218cd011f2d5d 100644 --- a/theodolite-benchmarks/kstreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java +++ b/theodolite-benchmarks/kstreams-commons/src/main/java/theodolite/commons/kafkastreams/KafkaStreamsBuilder.java @@ -70,18 +70,15 @@ public abstract class KafkaStreamsBuilder { // optional configurations this.setOptionalProperty(propBuilder, StreamsConfig.ACCEPTABLE_RECOVERY_LAG_CONFIG, - this.config::getLong, - p -> p >= 0); + this.config::getLong, p -> p >= 0); this.setOptionalProperty(propBuilder, StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, this.config::getInt, p -> p > 0); this.setOptionalProperty(propBuilder, StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, - this.config::getInt, - p -> p >= 0); + this.config::getInt, p -> p >= 0); this.setOptionalProperty(propBuilder, StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, this.config::getInt, p -> p >= 0); this.setOptionalProperty(propBuilder, StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, - this.config::getLong, - p -> p >= 0); + this.config::getLong, p -> p >= 0); this.setOptionalProperty(propBuilder, StreamsConfig.MAX_WARMUP_REPLICAS_CONFIG, this.config::getInt, p -> p >= 1); this.setOptionalProperty(propBuilder, StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, @@ -89,22 +86,28 @@ public abstract class KafkaStreamsBuilder { this.setOptionalProperty(propBuilder, StreamsConfig.NUM_STREAM_THREADS_CONFIG, this.config::getInt, p -> p > 0); this.setOptionalProperty(propBuilder, StreamsConfig.POLL_MS_CONFIG, - this.config::getLong, - p -> p >= 0); + this.config::getLong, p -> p >= 0); this.setOptionalProperty(propBuilder, StreamsConfig.PROCESSING_GUARANTEE_CONFIG, - this.config::getString, p -> StreamsConfig.AT_LEAST_ONCE.equals(p) - || StreamsConfig.EXACTLY_ONCE.equals(p) || StreamsConfig.EXACTLY_ONCE_BETA.equals(p)); + this.config::getString, this::validateProcessingGuarantee); this.setOptionalProperty(propBuilder, StreamsConfig.REPLICATION_FACTOR_CONFIG, this.config::getInt, p -> p >= 0); - if (this.config.containsKey(StreamsConfig.TOPOLOGY_OPTIMIZATION) - && this.config.getBoolean(StreamsConfig.TOPOLOGY_OPTIMIZATION)) { - propBuilder.set(StreamsConfig.TOPOLOGY_OPTIMIZATION, StreamsConfig.OPTIMIZE); + if (this.config.containsKey(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG) + && this.config.getBoolean(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG)) { + propBuilder.set(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE); } return propBuilder.build(); } + @SuppressWarnings("deprecation") + private boolean validateProcessingGuarantee(final String processingGuarantee) { + return StreamsConfig.AT_LEAST_ONCE.equals(processingGuarantee) + // We continue support EXACTLY_ONCE to allow benchmarking it against v2 + || StreamsConfig.EXACTLY_ONCE.equals(processingGuarantee) + || StreamsConfig.EXACTLY_ONCE_V2.equals(processingGuarantee); + } + /** * Method to implement a {@link Topology} for a {@code KafkaStreams} application. * @@ -116,7 +119,7 @@ public abstract class KafkaStreamsBuilder { * Builds the {@link KafkaStreams} instance. */ public KafkaStreams build() { - // Create the Kafka streams instance. + // Create the Kafka Streams instance. final Properties properties = this.buildProperties(); return new KafkaStreams(this.buildTopology(properties), properties); } diff --git a/theodolite-benchmarks/settings.gradle b/theodolite-benchmarks/settings.gradle index ae4254e968a0bc09970752f95c6a40db86ae775c..776e7d8e4fe132839b6e27c70c368720415721ea 100644 --- a/theodolite-benchmarks/settings.gradle +++ b/theodolite-benchmarks/settings.gradle @@ -5,32 +5,33 @@ include 'kstreams-commons' include 'flink-commons' include 'beam-commons' -include 'uc1-beam' -include 'uc2-beam' -include 'uc3-beam' -include 'uc4-beam' - include 'uc1-load-generator' +include 'uc1-commons' include 'uc1-kstreams' include 'uc1-flink' +include 'uc1-beam' include 'uc1-beam-flink' include 'uc1-beam-samza' include 'uc2-load-generator' include 'uc2-kstreams' include 'uc2-flink' +include 'uc2-beam' include 'uc2-beam-flink' include 'uc2-beam-samza' include 'uc3-load-generator' include 'uc3-kstreams' include 'uc3-flink' +include 'uc3-beam' include 'uc3-beam-flink' include 'uc3-beam-samza' include 'uc4-load-generator' include 'uc4-kstreams' include 'uc4-flink' +include 'uc4-beam' include 'uc4-beam-flink' include 'uc4-beam-samza' +include 'http-bridge' diff --git a/theodolite-benchmarks/uc1-beam-samza/Dockerfile b/theodolite-benchmarks/uc1-beam-samza/Dockerfile index 9b729060532ea3a242ac3084ba0bebf88ca2e9b6..cf6ef6675464e3c9d37db492b39fd8a71ec60e63 100644 --- a/theodolite-benchmarks/uc1-beam-samza/Dockerfile +++ b/theodolite-benchmarks/uc1-beam-samza/Dockerfile @@ -5,5 +5,5 @@ ENV MAX_SOURCE_PARALLELISM=1024 ADD build/distributions/uc1-beam-samza.tar / ADD samza-standalone.properties / -CMD /uc1-beam-samza/bin/uc1-beam-samza --configFactory=org.apache.samza.config.factories.PropertiesConfigFactory --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" +CMD /uc1-beam-samza/bin/uc1-beam-samza --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" diff --git a/theodolite-benchmarks/uc1-beam/build.gradle b/theodolite-benchmarks/uc1-beam/build.gradle index 502e94fa737fb2ae1bab861407b27575cd8766ca..ae80dcd8c5b263892f4539098a9a29b25c133819 100644 --- a/theodolite-benchmarks/uc1-beam/build.gradle +++ b/theodolite-benchmarks/uc1-beam/build.gradle @@ -2,4 +2,6 @@ plugins { id 'theodolite.beam' } - +dependencies { + implementation project(':uc1-commons') +} diff --git a/theodolite-benchmarks/uc1-beam/src/main/java/application/ConverterAdapter.java b/theodolite-benchmarks/uc1-beam/src/main/java/application/ConverterAdapter.java new file mode 100644 index 0000000000000000000000000000000000000000..e368c3a06cde50ea8d49d84b038fb2ec5aa97d1a --- /dev/null +++ b/theodolite-benchmarks/uc1-beam/src/main/java/application/ConverterAdapter.java @@ -0,0 +1,40 @@ +package application; + +import org.apache.beam.sdk.transforms.SimpleFunction; +import org.apache.beam.sdk.values.TypeDescriptor; +import rocks.theodolite.benchmarks.uc1.commons.RecordConverter; +import titan.ccp.model.records.ActivePowerRecord; + +/** + * {@link SimpleFunction} which wraps a {@link RecordConverter} to be used with Beam. + * + * @param <T> type the {@link RecordConverter} is associated with. + */ +public class ConverterAdapter<T> extends SimpleFunction<ActivePowerRecord, T> { + + private static final long serialVersionUID = -5263671231838353747L; // NOPMD + + private final RecordConverter<T> recordConverter; + private final TypeDescriptor<T> type; + + /** + * Create a new {@link ConverterAdapter} with a given {@link RecordConverter} and the associated + * type. + */ + public ConverterAdapter(final RecordConverter<T> recordConverter, final Class<T> type) { + super(); + this.recordConverter = recordConverter; + this.type = TypeDescriptor.of(type); + } + + @Override + public T apply(final ActivePowerRecord record) { + return this.recordConverter.convert(record); + } + + @Override + public TypeDescriptor<T> getOutputTypeDescriptor() { + return this.type; + } + +} diff --git a/theodolite-benchmarks/uc1-beam/src/main/java/application/LogKeyValue.java b/theodolite-benchmarks/uc1-beam/src/main/java/application/LogKeyValue.java deleted file mode 100644 index 251523441e339cbaf58c7e3a1b30e97cc354df18..0000000000000000000000000000000000000000 --- a/theodolite-benchmarks/uc1-beam/src/main/java/application/LogKeyValue.java +++ /dev/null @@ -1,24 +0,0 @@ -package application; - -import org.apache.beam.sdk.transforms.DoFn; -import org.apache.beam.sdk.values.KV; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Logs all Key Value pairs. - */ -public class LogKeyValue extends DoFn<KV<String, String>, KV<String, String>> { - private static final long serialVersionUID = 4328743; - private static final Logger LOGGER = LoggerFactory.getLogger(LogKeyValue.class); - - /** - * Logs all key value pairs it processes. - */ - @ProcessElement - public void processElement(@Element final KV<String, String> kv, - final OutputReceiver<KV<String, String>> out) { - LOGGER.info("Key: {}, Value: {}", kv.getKey(), kv.getValue()); - out.output(kv); - } -} diff --git a/theodolite-benchmarks/uc1-beam/src/main/java/application/MapToGson.java b/theodolite-benchmarks/uc1-beam/src/main/java/application/MapToGson.java deleted file mode 100644 index 6b0c6bc4ddfe78c22028da5b8cf7dde7ed57fced..0000000000000000000000000000000000000000 --- a/theodolite-benchmarks/uc1-beam/src/main/java/application/MapToGson.java +++ /dev/null @@ -1,26 +0,0 @@ -package application; - -import com.google.gson.Gson; -import org.apache.beam.sdk.transforms.SimpleFunction; -import org.apache.beam.sdk.values.KV; -import titan.ccp.model.records.ActivePowerRecord; - -/** - * Converts a Map into a json String. - */ -public class MapToGson extends SimpleFunction<KV<String, ActivePowerRecord>, KV<String, String>> { - private static final long serialVersionUID = 7168356203579050214L; - private transient Gson gsonObj = new Gson(); - - @Override - public KV<String, String> apply( - final KV<String, ActivePowerRecord> kv) { - - if (this.gsonObj == null) { - this.gsonObj = new Gson(); - } - - final String gson = this.gsonObj.toJson(kv.getValue()); - return KV.of(kv.getKey(), gson); - } -} diff --git a/theodolite-benchmarks/uc1-beam/src/main/java/application/Uc1BeamPipeline.java b/theodolite-benchmarks/uc1-beam/src/main/java/application/Uc1BeamPipeline.java index eaff08ac78cd18ddfd47eb2949ca13340ecc27b8..dde2bc064af9b21e11b859fdbf3f6ba1374d17d7 100644 --- a/theodolite-benchmarks/uc1-beam/src/main/java/application/Uc1BeamPipeline.java +++ b/theodolite-benchmarks/uc1-beam/src/main/java/application/Uc1BeamPipeline.java @@ -6,7 +6,10 @@ import org.apache.beam.sdk.coders.CoderRegistry; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.Values; import org.apache.commons.configuration2.Configuration; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseAdapter; +import rocks.theodolite.benchmarks.uc1.commons.logger.LogWriterFactory; import theodolite.commons.beam.AbstractPipeline; import theodolite.commons.beam.kafka.KafkaActivePowerTimestampReader; import titan.ccp.model.records.ActivePowerRecord; @@ -22,6 +25,8 @@ import titan.ccp.model.records.ActivePowerRecord; */ public final class Uc1BeamPipeline extends AbstractPipeline { + private final DatabaseAdapter<String> databaseAdapter = LogWriterFactory.forJson(); + protected Uc1BeamPipeline(final PipelineOptions options, final Configuration config) { super(options, config); @@ -36,17 +41,14 @@ public final class Uc1BeamPipeline extends AbstractPipeline { final KafkaActivePowerTimestampReader kafka = new KafkaActivePowerTimestampReader(this.bootstrapServer, this.inputTopic, consumerConfig); - final LogKeyValue logKeyValue = new LogKeyValue(); - final MapToGson mapToGson = new MapToGson(); - // Apply pipeline transformations // Read from Kafka this.apply(kafka) - // Map to Gson - .apply(MapElements - .via(mapToGson)) - // Print to console - .apply(ParDo.of(logKeyValue)); + .apply(Values.create()) + .apply(MapElements.via(new ConverterAdapter<>( + this.databaseAdapter.getRecordConverter(), + String.class))) + .apply(ParDo.of(new WriterAdapter<>(this.databaseAdapter.getDatabaseWriter()))); } } diff --git a/theodolite-benchmarks/uc1-beam/src/main/java/application/WriterAdapter.java b/theodolite-benchmarks/uc1-beam/src/main/java/application/WriterAdapter.java new file mode 100644 index 0000000000000000000000000000000000000000..eb9a2670cd8e61ed103a277e9d26072dc926dbeb --- /dev/null +++ b/theodolite-benchmarks/uc1-beam/src/main/java/application/WriterAdapter.java @@ -0,0 +1,28 @@ +package application; + +import org.apache.beam.sdk.transforms.DoFn; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseAdapter; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseWriter; + +/** + * {@link DoFn} which wraps a {@link DatabaseAdapter} to be used with Beam. + * + * @param <T> type the {@link DatabaseWriter} is associated with. + */ +public class WriterAdapter<T> extends DoFn<T, Void> { + + private static final long serialVersionUID = -5263671231838353742L; // NOPMD + + private final DatabaseWriter<T> databaseWriter; + + public WriterAdapter(final DatabaseWriter<T> databaseWriter) { + super(); + this.databaseWriter = databaseWriter; + } + + @ProcessElement + public void processElement(@Element final T record, final OutputReceiver<Void> out) { + this.databaseWriter.write(record); + } + +} diff --git a/theodolite-benchmarks/uc1-commons/.settings/org.eclipse.jdt.ui.prefs b/theodolite-benchmarks/uc1-commons/.settings/org.eclipse.jdt.ui.prefs new file mode 100644 index 0000000000000000000000000000000000000000..713419c8d3d74d3bd7fd05c3e839367753fcdee0 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/.settings/org.eclipse.jdt.ui.prefs @@ -0,0 +1,127 @@ +cleanup.add_default_serial_version_id=true +cleanup.add_generated_serial_version_id=false +cleanup.add_missing_annotations=true +cleanup.add_missing_deprecated_annotations=true +cleanup.add_missing_methods=false +cleanup.add_missing_nls_tags=false +cleanup.add_missing_override_annotations=true +cleanup.add_missing_override_annotations_interface_methods=true +cleanup.add_serial_version_id=false +cleanup.always_use_blocks=true +cleanup.always_use_parentheses_in_expressions=false +cleanup.always_use_this_for_non_static_field_access=true +cleanup.always_use_this_for_non_static_method_access=true +cleanup.convert_functional_interfaces=false +cleanup.convert_to_enhanced_for_loop=true +cleanup.correct_indentation=true +cleanup.format_source_code=true +cleanup.format_source_code_changes_only=false +cleanup.insert_inferred_type_arguments=false +cleanup.make_local_variable_final=true +cleanup.make_parameters_final=true +cleanup.make_private_fields_final=true +cleanup.make_type_abstract_if_missing_method=false +cleanup.make_variable_declarations_final=true +cleanup.never_use_blocks=false +cleanup.never_use_parentheses_in_expressions=true +cleanup.organize_imports=true +cleanup.qualify_static_field_accesses_with_declaring_class=false +cleanup.qualify_static_member_accesses_through_instances_with_declaring_class=true +cleanup.qualify_static_member_accesses_through_subtypes_with_declaring_class=true +cleanup.qualify_static_member_accesses_with_declaring_class=true +cleanup.qualify_static_method_accesses_with_declaring_class=false +cleanup.remove_private_constructors=true +cleanup.remove_redundant_modifiers=false +cleanup.remove_redundant_semicolons=false +cleanup.remove_redundant_type_arguments=true +cleanup.remove_trailing_whitespaces=true +cleanup.remove_trailing_whitespaces_all=true +cleanup.remove_trailing_whitespaces_ignore_empty=false +cleanup.remove_unnecessary_casts=true +cleanup.remove_unnecessary_nls_tags=true +cleanup.remove_unused_imports=true +cleanup.remove_unused_local_variables=false +cleanup.remove_unused_private_fields=true +cleanup.remove_unused_private_members=false +cleanup.remove_unused_private_methods=true +cleanup.remove_unused_private_types=true +cleanup.sort_members=false +cleanup.sort_members_all=false +cleanup.use_anonymous_class_creation=false +cleanup.use_blocks=true +cleanup.use_blocks_only_for_return_and_throw=false +cleanup.use_lambda=true +cleanup.use_parentheses_in_expressions=true +cleanup.use_this_for_non_static_field_access=true +cleanup.use_this_for_non_static_field_access_only_if_necessary=false +cleanup.use_this_for_non_static_method_access=true +cleanup.use_this_for_non_static_method_access_only_if_necessary=false +cleanup_profile=_CAU-SE-Style +cleanup_settings_version=2 +eclipse.preferences.version=1 +editor_save_participant_org.eclipse.jdt.ui.postsavelistener.cleanup=true +formatter_profile=_CAU-SE-Style +formatter_settings_version=21 +org.eclipse.jdt.ui.ignorelowercasenames=true +org.eclipse.jdt.ui.importorder=; +org.eclipse.jdt.ui.ondemandthreshold=99 +org.eclipse.jdt.ui.staticondemandthreshold=99 +sp_cleanup.add_default_serial_version_id=true +sp_cleanup.add_generated_serial_version_id=false +sp_cleanup.add_missing_annotations=true +sp_cleanup.add_missing_deprecated_annotations=true +sp_cleanup.add_missing_methods=false +sp_cleanup.add_missing_nls_tags=false +sp_cleanup.add_missing_override_annotations=true +sp_cleanup.add_missing_override_annotations_interface_methods=true +sp_cleanup.add_serial_version_id=false +sp_cleanup.always_use_blocks=true +sp_cleanup.always_use_parentheses_in_expressions=false +sp_cleanup.always_use_this_for_non_static_field_access=true +sp_cleanup.always_use_this_for_non_static_method_access=true +sp_cleanup.convert_functional_interfaces=false +sp_cleanup.convert_to_enhanced_for_loop=true +sp_cleanup.correct_indentation=true +sp_cleanup.format_source_code=true +sp_cleanup.format_source_code_changes_only=false +sp_cleanup.insert_inferred_type_arguments=false +sp_cleanup.make_local_variable_final=true +sp_cleanup.make_parameters_final=true +sp_cleanup.make_private_fields_final=true +sp_cleanup.make_type_abstract_if_missing_method=false +sp_cleanup.make_variable_declarations_final=true +sp_cleanup.never_use_blocks=false +sp_cleanup.never_use_parentheses_in_expressions=true +sp_cleanup.on_save_use_additional_actions=true +sp_cleanup.organize_imports=true +sp_cleanup.qualify_static_field_accesses_with_declaring_class=false +sp_cleanup.qualify_static_member_accesses_through_instances_with_declaring_class=true +sp_cleanup.qualify_static_member_accesses_through_subtypes_with_declaring_class=true +sp_cleanup.qualify_static_member_accesses_with_declaring_class=true +sp_cleanup.qualify_static_method_accesses_with_declaring_class=false +sp_cleanup.remove_private_constructors=true +sp_cleanup.remove_redundant_modifiers=false +sp_cleanup.remove_redundant_semicolons=false +sp_cleanup.remove_redundant_type_arguments=true +sp_cleanup.remove_trailing_whitespaces=true +sp_cleanup.remove_trailing_whitespaces_all=true +sp_cleanup.remove_trailing_whitespaces_ignore_empty=false +sp_cleanup.remove_unnecessary_casts=true +sp_cleanup.remove_unnecessary_nls_tags=true +sp_cleanup.remove_unused_imports=true +sp_cleanup.remove_unused_local_variables=false +sp_cleanup.remove_unused_private_fields=true +sp_cleanup.remove_unused_private_members=false +sp_cleanup.remove_unused_private_methods=true +sp_cleanup.remove_unused_private_types=true +sp_cleanup.sort_members=false +sp_cleanup.sort_members_all=false +sp_cleanup.use_anonymous_class_creation=false +sp_cleanup.use_blocks=true +sp_cleanup.use_blocks_only_for_return_and_throw=false +sp_cleanup.use_lambda=true +sp_cleanup.use_parentheses_in_expressions=true +sp_cleanup.use_this_for_non_static_field_access=true +sp_cleanup.use_this_for_non_static_field_access_only_if_necessary=false +sp_cleanup.use_this_for_non_static_method_access=true +sp_cleanup.use_this_for_non_static_method_access_only_if_necessary=false diff --git a/theodolite-benchmarks/uc1-commons/.settings/qa.eclipse.plugin.checkstyle.prefs b/theodolite-benchmarks/uc1-commons/.settings/qa.eclipse.plugin.checkstyle.prefs new file mode 100644 index 0000000000000000000000000000000000000000..87860c815222845c1d264d7d0ce498d3397f8280 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/.settings/qa.eclipse.plugin.checkstyle.prefs @@ -0,0 +1,4 @@ +configFilePath=../config/checkstyle.xml +customModulesJarPaths= +eclipse.preferences.version=1 +enabled=true diff --git a/theodolite-benchmarks/uc1-commons/.settings/qa.eclipse.plugin.pmd.prefs b/theodolite-benchmarks/uc1-commons/.settings/qa.eclipse.plugin.pmd.prefs new file mode 100644 index 0000000000000000000000000000000000000000..efbcb8c9e5d449194a48ca1ea42b7d807b573db9 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/.settings/qa.eclipse.plugin.pmd.prefs @@ -0,0 +1,4 @@ +customRulesJars= +eclipse.preferences.version=1 +enabled=true +ruleSetFilePath=../config/pmd.xml diff --git a/theodolite-benchmarks/uc1-commons/build.gradle b/theodolite-benchmarks/uc1-commons/build.gradle new file mode 100644 index 0000000000000000000000000000000000000000..cd95e28ed53cc8e33a416564e613574faf1fc6cb --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/build.gradle @@ -0,0 +1,22 @@ +plugins { + id 'theodolite.java-commons' +} + +repositories { + mavenCentral() + maven { + url "https://oss.sonatype.org/content/repositories/snapshots/" + } + maven { + url 'https://packages.confluent.io/maven/' + } +} + +dependencies { + implementation('org.industrial-devops:titan-ccp-common:0.1.0-SNAPSHOT') { changing = true } + implementation('org.industrial-devops:titan-ccp-common-kafka:0.1.0-SNAPSHOT') { changing = true } + + implementation 'com.google.code.gson:gson:2.8.9' + + testImplementation 'junit:junit:4.12' +} diff --git a/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/DatabaseAdapter.java b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/DatabaseAdapter.java new file mode 100644 index 0000000000000000000000000000000000000000..a1cb1ade0dc76b168cf9ee54f64d5ac88d6b3a98 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/DatabaseAdapter.java @@ -0,0 +1,46 @@ +package rocks.theodolite.benchmarks.uc1.commons; + +import java.util.Objects; +import titan.ccp.model.records.ActivePowerRecord; + +/** + * A database adapter consisting of a {@link RecordConverter} and a {@link DatabaseWriter}. + * + * @param <T> intermediate data type written to the database. + */ +public final class DatabaseAdapter<T> { + + private final RecordConverter<T> recordConverter; + + private final DatabaseWriter<T> databaseWriter; + + private DatabaseAdapter(final RecordConverter<T> recordConverter, + final DatabaseWriter<T> databaseWriter) { + this.recordConverter = recordConverter; + this.databaseWriter = databaseWriter; + } + + public RecordConverter<T> getRecordConverter() { + return this.recordConverter; + } + + public DatabaseWriter<T> getDatabaseWriter() { + return this.databaseWriter; + } + + /** + * Create a new {@link DatabaseAdapter}. + * + * @param <T> intermediate data type written to the database. + * @param recordConverter RecordConverter for converting {@link ActivePowerRecord}s to {@code T} + * @param databaseWriter DatabaseWriter for writing converted records to the database. + * @return the {@link DatabaseAdapter}. + */ + public static <T> DatabaseAdapter<T> from(final RecordConverter<T> recordConverter, + final DatabaseWriter<T> databaseWriter) { + Objects.requireNonNull(recordConverter); + Objects.requireNonNull(databaseWriter); + return new DatabaseAdapter<>(recordConverter, databaseWriter); + } + +} diff --git a/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/DatabaseWriter.java b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/DatabaseWriter.java new file mode 100644 index 0000000000000000000000000000000000000000..1beb269e4b75252ac72f7c30c4a26f7a11de4fb6 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/DatabaseWriter.java @@ -0,0 +1,13 @@ +package rocks.theodolite.benchmarks.uc1.commons; + +/** + * Writes an object to a database. + * + * @param <T> Type expected by the database. + */ +@FunctionalInterface +public interface DatabaseWriter<T> { + + void write(T record); + +} diff --git a/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/RecordConverter.java b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/RecordConverter.java new file mode 100644 index 0000000000000000000000000000000000000000..105f19e0e920e3516f7277cd7804dae210a7d0b1 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/RecordConverter.java @@ -0,0 +1,15 @@ +package rocks.theodolite.benchmarks.uc1.commons; + +import titan.ccp.model.records.ActivePowerRecord; + +/** + * Converts an {@link ActivePowerRecord} to the type required by a database. + * + * @param <T> Type required by the database. + */ +@FunctionalInterface +public interface RecordConverter<T> { + + T convert(ActivePowerRecord record); + +} diff --git a/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/JsonConverter.java b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/JsonConverter.java new file mode 100644 index 0000000000000000000000000000000000000000..f9974affb7bf57fc63e9bfe8ba92fd056da9a97b --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/JsonConverter.java @@ -0,0 +1,22 @@ +package rocks.theodolite.benchmarks.uc1.commons.logger; + +import com.google.gson.Gson; +import java.io.Serializable; +import rocks.theodolite.benchmarks.uc1.commons.RecordConverter; +import titan.ccp.model.records.ActivePowerRecord; + +/** + * {@link RecordConverter} that converts {@link ActivePowerRecord}s to JSON strings. + */ +public class JsonConverter implements RecordConverter<String>, Serializable { + + private static final long serialVersionUID = -5263671231838353748L; // NOPMD + + private static final Gson GSON = new Gson(); + + @Override + public String convert(final ActivePowerRecord activePowerRecord) { + return GSON.toJson(activePowerRecord); + } + +} diff --git a/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/LogWriter.java b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/LogWriter.java new file mode 100644 index 0000000000000000000000000000000000000000..d606a6dffd01257b308bf2afebc3088b52793ccf --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/LogWriter.java @@ -0,0 +1,22 @@ +package rocks.theodolite.benchmarks.uc1.commons.logger; + +import java.io.Serializable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseWriter; + +/** + * Writes string records to a {@link Logger}. + */ +public class LogWriter implements DatabaseWriter<String>, Serializable { + + private static final long serialVersionUID = -5263671231838353749L; // NOPMD + + private static final Logger LOGGER = LoggerFactory.getLogger(LogWriter.class); + + @Override + public void write(final String string) { + LOGGER.info("Record: {}", string); + } + +} diff --git a/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/LogWriterFactory.java b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/LogWriterFactory.java new file mode 100644 index 0000000000000000000000000000000000000000..305ed933ba3e0d885de9c65aacc6ace8a0884621 --- /dev/null +++ b/theodolite-benchmarks/uc1-commons/src/main/java/rocks/theodolite/benchmarks/uc1/commons/logger/LogWriterFactory.java @@ -0,0 +1,18 @@ +package rocks.theodolite.benchmarks.uc1.commons.logger; + +import org.slf4j.Logger; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseAdapter; + +/** + * Provides factory methods for creating a dummy {@link DatabaseAdapter} writing records as logs + * using a SLF4J {@link Logger}. + */ +public final class LogWriterFactory { + + private LogWriterFactory() {} + + public static DatabaseAdapter<String> forJson() { + return DatabaseAdapter.from(new JsonConverter(), new LogWriter()); + } + +} diff --git a/theodolite-benchmarks/uc1-flink/Dockerfile b/theodolite-benchmarks/uc1-flink/Dockerfile index 795b9e343a03cf0209e1625f5cbc3d45dcb77cda..fdaecd661e86275d670423351664e54221ce33cc 100644 --- a/theodolite-benchmarks/uc1-flink/Dockerfile +++ b/theodolite-benchmarks/uc1-flink/Dockerfile @@ -1,3 +1,3 @@ -FROM flink:1.12-scala_2.12-java11 +FROM flink:1.13-java11 ADD build/libs/uc1-flink-all.jar /opt/flink/usrlib/artifacts/uc1-flink-all.jar diff --git a/theodolite-benchmarks/uc1-flink/build.gradle b/theodolite-benchmarks/uc1-flink/build.gradle index 8a2a359c4840e67581f7bc24f1544ff519f82525..681effe9a347f0fa9f26d6a2caf0668ade09d6c2 100644 --- a/theodolite-benchmarks/uc1-flink/build.gradle +++ b/theodolite-benchmarks/uc1-flink/build.gradle @@ -2,4 +2,8 @@ plugins { id 'theodolite.flink' } +dependencies { + implementation project(':uc1-commons') +} + mainClassName = "theodolite.uc1.application.HistoryServiceFlinkJob" diff --git a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/ConverterAdapter.java b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/ConverterAdapter.java new file mode 100644 index 0000000000000000000000000000000000000000..af0a0b1cf5a25d22c1fdc5e7adb7467be03f9b9f --- /dev/null +++ b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/ConverterAdapter.java @@ -0,0 +1,27 @@ +package theodolite.uc1.application; + +import org.apache.flink.api.common.functions.MapFunction; +import rocks.theodolite.benchmarks.uc1.commons.RecordConverter; +import titan.ccp.model.records.ActivePowerRecord; + +/** + * {@link MapFunction} which wraps a {@link RecordConverter} to be used with Flink. + * + * @param <T> type the {@link RecordConverter} is associated with. + */ +public class ConverterAdapter<T> implements MapFunction<ActivePowerRecord, T> { + + private static final long serialVersionUID = -5263671231838353747L; // NOPMD + + private final RecordConverter<T> recordConverter; + + public ConverterAdapter(final RecordConverter<T> recordConverter) { + this.recordConverter = recordConverter; + } + + @Override + public T map(final ActivePowerRecord record) throws Exception { + return this.recordConverter.convert(record); + } + +} diff --git a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/GsonMapper.java b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/GsonMapper.java deleted file mode 100644 index 831db7fe63be6529e6b7ba299dca92b138ff7d13..0000000000000000000000000000000000000000 --- a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/GsonMapper.java +++ /dev/null @@ -1,22 +0,0 @@ -package theodolite.uc1.application; - -import com.google.gson.Gson; -import org.apache.flink.api.common.functions.MapFunction; -import titan.ccp.model.records.ActivePowerRecord; - -/** - * {@link MapFunction} which maps {@link ActivePowerRecord}s to their representation as JSON - * strings. - */ -public class GsonMapper implements MapFunction<ActivePowerRecord, String> { - - private static final long serialVersionUID = -5263671231838353747L; // NOPMD - - private static final Gson GSON = new Gson(); - - @Override - public String map(final ActivePowerRecord value) throws Exception { - return GSON.toJson(value); - } - -} diff --git a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java index 0cb132e526486e71409736b843dd25bdfa52da4a..41131152734f68dd34489461b1ad31d94a970eac 100644 --- a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java +++ b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java @@ -7,6 +7,8 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseAdapter; +import rocks.theodolite.benchmarks.uc1.commons.logger.LogWriterFactory; import theodolite.commons.flink.KafkaConnectorFactory; import titan.ccp.common.configuration.ServiceConfigurations; import titan.ccp.model.records.ActivePowerRecord; @@ -22,6 +24,8 @@ public final class HistoryServiceFlinkJob { private final StreamExecutionEnvironment env; private final String applicationId; + private final DatabaseAdapter<String> databaseAdapter = LogWriterFactory.forJson(); + /** * Create a new instance of the {@link HistoryServiceFlinkJob}. */ @@ -69,9 +73,10 @@ public final class HistoryServiceFlinkJob { stream // .rebalance() - .map(new GsonMapper()) - .flatMap((record, c) -> LOGGER.info("Record: {}", record)) - .returns(Types.GENERIC(Object.class)); // Will never be used + .map(new ConverterAdapter<>(this.databaseAdapter.getRecordConverter())) + .returns(Types.STRING) + .flatMap(new WriterAdapter<>(this.databaseAdapter.getDatabaseWriter())) + .returns(Types.VOID); // Will never be used } /** diff --git a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/WriterAdapter.java b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/WriterAdapter.java new file mode 100644 index 0000000000000000000000000000000000000000..b2f375ec5f5a66141a2551015fb9fbd013ab9c16 --- /dev/null +++ b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/WriterAdapter.java @@ -0,0 +1,28 @@ +package theodolite.uc1.application; + +import org.apache.flink.api.common.functions.FlatMapFunction; +import org.apache.flink.util.Collector; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseAdapter; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseWriter; + +/** + * {@link FlatMapFunction} which wraps a {@link DatabaseAdapter} to be used with Flink. + * + * @param <T> type the {@link DatabaseWriter} is associated with. + */ +public class WriterAdapter<T> implements FlatMapFunction<T, Void> { + + private static final long serialVersionUID = -5263671231838353747L; // NOPMD + + private final DatabaseWriter<T> databaseWriter; + + public WriterAdapter(final DatabaseWriter<T> databaseWriter) { + this.databaseWriter = databaseWriter; + } + + @Override + public void flatMap(final T value, final Collector<Void> out) throws Exception { + this.databaseWriter.write(value); + } + +} diff --git a/theodolite-benchmarks/uc1-kstreams/build.gradle b/theodolite-benchmarks/uc1-kstreams/build.gradle index 74cfb450ec80759f60582c25ab844e3398d5bf02..1460a99a2aad7767b84259494c4c231344862545 100644 --- a/theodolite-benchmarks/uc1-kstreams/build.gradle +++ b/theodolite-benchmarks/uc1-kstreams/build.gradle @@ -2,4 +2,8 @@ plugins { id 'theodolite.kstreams' } +dependencies { + implementation project(':uc1-commons') +} + mainClassName = "theodolite.uc1.application.HistoryService" diff --git a/theodolite-benchmarks/uc1-kstreams/src/main/java/theodolite/uc1/streamprocessing/TopologyBuilder.java b/theodolite-benchmarks/uc1-kstreams/src/main/java/theodolite/uc1/streamprocessing/TopologyBuilder.java index 427a838f45f6807ede00dcb68ebf8c5580f28ce6..64d6d08c30c1a015c668e744fe164bda3f493aa4 100644 --- a/theodolite-benchmarks/uc1-kstreams/src/main/java/theodolite/uc1/streamprocessing/TopologyBuilder.java +++ b/theodolite-benchmarks/uc1-kstreams/src/main/java/theodolite/uc1/streamprocessing/TopologyBuilder.java @@ -1,13 +1,12 @@ package theodolite.uc1.streamprocessing; -import com.google.gson.Gson; import java.util.Properties; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.kstream.Consumed; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import rocks.theodolite.benchmarks.uc1.commons.DatabaseAdapter; +import rocks.theodolite.benchmarks.uc1.commons.logger.LogWriterFactory; import titan.ccp.common.kafka.avro.SchemaRegistryAvroSerdeFactory; import titan.ccp.model.records.ActivePowerRecord; @@ -16,12 +15,11 @@ import titan.ccp.model.records.ActivePowerRecord; */ public class TopologyBuilder { - private static final Logger LOGGER = LoggerFactory.getLogger(TopologyBuilder.class); - private static final Gson GSON = new Gson(); - private final String inputTopic; private final SchemaRegistryAvroSerdeFactory srAvroSerdeFactory; + private final DatabaseAdapter<String> databaseAdapter = LogWriterFactory.forJson(); + private final StreamsBuilder builder = new StreamsBuilder(); @@ -42,8 +40,8 @@ public class TopologyBuilder { .stream(this.inputTopic, Consumed.with( Serdes.String(), this.srAvroSerdeFactory.<ActivePowerRecord>forValues())) - .mapValues(v -> GSON.toJson(v)) - .foreach((k, record) -> LOGGER.info("Record: {}", record)); + .mapValues(this.databaseAdapter.getRecordConverter()::convert) + .foreach((k, record) -> this.databaseAdapter.getDatabaseWriter().write(record)); return this.builder.build(properties); } diff --git a/theodolite-benchmarks/uc2-beam-samza/Dockerfile b/theodolite-benchmarks/uc2-beam-samza/Dockerfile index 22855cea279819cacbf6eee253c30c60409fdba3..ae762791c40fc6981ce7e5fd08bea860ed9208ec 100644 --- a/theodolite-benchmarks/uc2-beam-samza/Dockerfile +++ b/theodolite-benchmarks/uc2-beam-samza/Dockerfile @@ -5,4 +5,4 @@ ENV MAX_SOURCE_PARALLELISM=1024 ADD build/distributions/uc2-beam-samza.tar / ADD samza-standalone.properties / -CMD /uc2-beam-samza/bin/uc2-beam-samza --configFactory=org.apache.samza.config.factories.PropertiesConfigFactory --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" +CMD /uc2-beam-samza/bin/uc2-beam-samza --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" diff --git a/theodolite-benchmarks/uc2-flink/Dockerfile b/theodolite-benchmarks/uc2-flink/Dockerfile index 537ab28e2d4e5fb8edfc2760142acc33cc49b91d..01a85c57b00ea8bb4af8faa697708fd2b650de50 100644 --- a/theodolite-benchmarks/uc2-flink/Dockerfile +++ b/theodolite-benchmarks/uc2-flink/Dockerfile @@ -1,3 +1,3 @@ -FROM flink:1.12-scala_2.12-java11 +FROM flink:1.13-java11 ADD build/libs/uc2-flink-all.jar /opt/flink/usrlib/artifacts/uc2-flink-all.jar \ No newline at end of file diff --git a/theodolite-benchmarks/uc2-flink/src/main/java/theodolite/uc2/application/HistoryServiceFlinkJob.java b/theodolite-benchmarks/uc2-flink/src/main/java/theodolite/uc2/application/HistoryServiceFlinkJob.java index d156d895d86bb01a31f96e08764df8b8df743c4d..7a97099c71a18449b7cc3f0413632b52fd5b69f5 100644 --- a/theodolite-benchmarks/uc2-flink/src/main/java/theodolite/uc2/application/HistoryServiceFlinkJob.java +++ b/theodolite-benchmarks/uc2-flink/src/main/java/theodolite/uc2/application/HistoryServiceFlinkJob.java @@ -5,7 +5,6 @@ import org.apache.commons.configuration2.Configuration; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.runtime.state.StateBackend; -import org.apache.flink.streaming.api.TimeCharacteristic; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; @@ -48,8 +47,6 @@ public final class HistoryServiceFlinkJob { } private void configureEnv() { - this.env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); - final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true); final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS); if (checkpointing) { diff --git a/theodolite-benchmarks/uc2-kstreams/src/main/java/theodolite/uc2/streamprocessing/TopologyBuilder.java b/theodolite-benchmarks/uc2-kstreams/src/main/java/theodolite/uc2/streamprocessing/TopologyBuilder.java index eda7c495a2cff6d58b62a8a6a74ea8e1b2d89aca..21dcf14a9322ce5a6381f96f22f5fadb85cc78f0 100644 --- a/theodolite-benchmarks/uc2-kstreams/src/main/java/theodolite/uc2/streamprocessing/TopologyBuilder.java +++ b/theodolite-benchmarks/uc2-kstreams/src/main/java/theodolite/uc2/streamprocessing/TopologyBuilder.java @@ -53,7 +53,7 @@ public class TopologyBuilder { Consumed.with(Serdes.String(), this.srAvroSerdeFactory.<ActivePowerRecord>forValues())) .groupByKey() - .windowedBy(TimeWindows.of(this.duration)) + .windowedBy(TimeWindows.ofSizeWithNoGrace(this.duration)) // .aggregate( // () -> 0.0, // (key, activePowerRecord, agg) -> agg + activePowerRecord.getValueInW(), diff --git a/theodolite-benchmarks/uc3-beam-samza/Dockerfile b/theodolite-benchmarks/uc3-beam-samza/Dockerfile index d3e860bd72c54121d616bc5562d519e6e1e21dec..54979b8e1fa8aa9ac7d073302301bd10cbff5f34 100644 --- a/theodolite-benchmarks/uc3-beam-samza/Dockerfile +++ b/theodolite-benchmarks/uc3-beam-samza/Dockerfile @@ -5,4 +5,4 @@ ENV MAX_SOURCE_PARALLELISM=1024 ADD build/distributions/uc3-beam-samza.tar / ADD samza-standalone.properties / -CMD /uc3-beam-samza/bin/uc3-beam-samza --configFactory=org.apache.samza.config.factories.PropertiesConfigFactory --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" +CMD /uc3-beam-samza/bin/uc3-beam-samza --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" diff --git a/theodolite-benchmarks/uc3-flink/Dockerfile b/theodolite-benchmarks/uc3-flink/Dockerfile index 34c6da692cb30b738adf47b9d4ca893e72f330e4..cef05c0296f55f0cf7391dd35dd1806ec0efa287 100644 --- a/theodolite-benchmarks/uc3-flink/Dockerfile +++ b/theodolite-benchmarks/uc3-flink/Dockerfile @@ -1,3 +1,3 @@ -FROM flink:1.12-scala_2.12-java11 +FROM flink:1.13-java11 ADD build/libs/uc3-flink-all.jar /opt/flink/usrlib/artifacts/uc3-flink-all.jar \ No newline at end of file diff --git a/theodolite-benchmarks/uc3-flink/src/main/java/theodolite/uc3/application/HistoryServiceFlinkJob.java b/theodolite-benchmarks/uc3-flink/src/main/java/theodolite/uc3/application/HistoryServiceFlinkJob.java index 091b25674a2a31671ca68bd2076c694da9533d77..621146864fc84d032a7dc3c65fa253df9b940b2d 100644 --- a/theodolite-benchmarks/uc3-flink/src/main/java/theodolite/uc3/application/HistoryServiceFlinkJob.java +++ b/theodolite-benchmarks/uc3-flink/src/main/java/theodolite/uc3/application/HistoryServiceFlinkJob.java @@ -9,7 +9,6 @@ import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.runtime.state.StateBackend; -import org.apache.flink.streaming.api.TimeCharacteristic; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; @@ -55,8 +54,6 @@ public final class HistoryServiceFlinkJob { } private void configureEnv() { - this.env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); - final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true); final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS); if (checkpointing) { diff --git a/theodolite-benchmarks/uc3-kstreams/src/main/java/theodolite/uc3/streamprocessing/TopologyBuilder.java b/theodolite-benchmarks/uc3-kstreams/src/main/java/theodolite/uc3/streamprocessing/TopologyBuilder.java index 1e976c07158720b3681d89413a5f277b1395f32d..4c63e21f3d9f1af6c9ef0363d7d01939faae9aef 100644 --- a/theodolite-benchmarks/uc3-kstreams/src/main/java/theodolite/uc3/streamprocessing/TopologyBuilder.java +++ b/theodolite-benchmarks/uc3-kstreams/src/main/java/theodolite/uc3/streamprocessing/TopologyBuilder.java @@ -60,17 +60,18 @@ public class TopologyBuilder { final Serde<HourOfDayKey> keySerde = HourOfDayKeySerde.create(); this.builder - .stream(this.inputTopic, - Consumed.with(Serdes.String(), - this.srAvroSerdeFactory.<ActivePowerRecord>forValues())) + .stream(this.inputTopic, Consumed.with( + Serdes.String(), + this.srAvroSerdeFactory.<ActivePowerRecord>forValues())) .selectKey((key, value) -> { final Instant instant = Instant.ofEpochMilli(value.getTimestamp()); final LocalDateTime dateTime = LocalDateTime.ofInstant(instant, this.zone); return keyFactory.createKey(value.getIdentifier(), dateTime); }) - .groupByKey( - Grouped.with(keySerde, this.srAvroSerdeFactory.forValues())) - .windowedBy(TimeWindows.of(this.aggregtionDuration).advanceBy(this.aggregationAdvance)) + .groupByKey(Grouped.with(keySerde, this.srAvroSerdeFactory.forValues())) + .windowedBy(TimeWindows + .ofSizeWithNoGrace(this.aggregtionDuration) + .advanceBy(this.aggregationAdvance)) .aggregate( () -> Stats.of(), (k, record, stats) -> StatsFactory.accumulate(stats, record.getValueInW()), diff --git a/theodolite-benchmarks/uc4-beam-samza/Dockerfile b/theodolite-benchmarks/uc4-beam-samza/Dockerfile index ffb82ee09e99384d4914a0f86b6d9214fc161381..b59585f748b95cf62e59be01c8fa9dc0d919f43a 100644 --- a/theodolite-benchmarks/uc4-beam-samza/Dockerfile +++ b/theodolite-benchmarks/uc4-beam-samza/Dockerfile @@ -5,4 +5,4 @@ ENV MAX_SOURCE_PARALLELISM=1024 ADD build/distributions/uc4-beam-samza.tar / ADD samza-standalone.properties / -CMD /uc4-beam-samza/bin/uc4-beam-samza --configFactory=org.apache.samza.config.factories.PropertiesConfigFactory --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" +CMD /uc4-beam-samza/bin/uc4-beam-samza --configFilePath=samza-standalone.properties --samzaExecutionEnvironment=STANDALONE --maxSourceParallelism=$MAX_SOURCE_PARALLELISM --enableMetrics=false --configOverride="{\"job.coordinator.zk.connect\":\"$SAMZA_JOB_COORDINATOR_ZK_CONNECT\"}" diff --git a/theodolite-benchmarks/uc4-flink/Dockerfile b/theodolite-benchmarks/uc4-flink/Dockerfile index 4f51f379e5da436104bb8c914e3233d6ecb4ec1f..1fe58e9930db0ef30fa10d12519dd68ccdb2e97e 100644 --- a/theodolite-benchmarks/uc4-flink/Dockerfile +++ b/theodolite-benchmarks/uc4-flink/Dockerfile @@ -1,3 +1,3 @@ -FROM flink:1.12-scala_2.12-java11 +FROM flink:1.13-java11 ADD build/libs/uc4-flink-all.jar /opt/flink/usrlib/artifacts/uc4-flink-all.jar \ No newline at end of file diff --git a/theodolite-benchmarks/uc4-flink/src/main/java/theodolite/uc4/application/AggregationServiceFlinkJob.java b/theodolite-benchmarks/uc4-flink/src/main/java/theodolite/uc4/application/AggregationServiceFlinkJob.java index 3e2878a893057024de00333492462f5029eb6d77..48eadff101905b746ab239f6030decb728c12475 100644 --- a/theodolite-benchmarks/uc4-flink/src/main/java/theodolite/uc4/application/AggregationServiceFlinkJob.java +++ b/theodolite-benchmarks/uc4-flink/src/main/java/theodolite/uc4/application/AggregationServiceFlinkJob.java @@ -9,12 +9,12 @@ import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.runtime.state.StateBackend; -import org.apache.flink.streaming.api.TimeCharacteristic; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; +import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import org.apache.kafka.common.serialization.Serdes; import org.slf4j.Logger; @@ -68,8 +68,6 @@ public final class AggregationServiceFlinkJob { } private void configureEnv() { - this.env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); - final boolean checkpointing = this.config.getBoolean(ConfigurationKeys.CHECKPOINTING, true); final int commitIntervalMs = this.config.getInt(ConfigurationKeys.COMMIT_INTERVAL_MS); if (checkpointing) { @@ -103,9 +101,11 @@ public final class AggregationServiceFlinkJob { this.env.getConfig().registerTypeWithKryoSerializer(Set.of(1, 2, 3, 4).getClass(), // NOCS new ImmutableSetSerializer()); - this.env.getConfig().getRegisteredTypesWithKryoSerializers() - .forEach((c, s) -> LOGGER.info("Class " + c.getName() + " registered with serializer " - + s.getSerializer().getClass().getName())); + this.env + .getConfig() + .getRegisteredTypesWithKryoSerializers() + .forEach((c, s) -> LOGGER.info("Class '{}' registered with serializer '{}'.", c.getName(), + s.getSerializer().getClass().getName())); } private void buildPipeline() { @@ -134,12 +134,13 @@ public final class AggregationServiceFlinkJob { final FlinkKafkaConsumer<AggregatedActivePowerRecord> kafkaOutputSource = kafkaConnector.createConsumer(outputTopic, AggregatedActivePowerRecord.class); - final FlinkKafkaConsumer<Tuple2<Event, String>> kafkaConfigSource = + final FlinkKafkaConsumerBase<Tuple2<Event, String>> kafkaConfigSource = kafkaConnector.createConsumer( configurationTopic, EventSerde::serde, Serdes::String, - TupleType.of(TypeInformation.of(Event.class), Types.STRING)); + TupleType.of(TypeInformation.of(Event.class), Types.STRING)) + .setStartFromEarliest(); // Sink to output topic with SensorId, AggregatedActivePowerRecord final FlinkKafkaProducer<Tuple2<String, AggregatedActivePowerRecord>> kafkaAggregationSink = diff --git a/theodolite-benchmarks/uc4-kstreams/src/main/java/theodolite/uc4/streamprocessing/TopologyBuilder.java b/theodolite-benchmarks/uc4-kstreams/src/main/java/theodolite/uc4/streamprocessing/TopologyBuilder.java index 623870313cd341d0594fee38d2fd0ae297abbeae..712b20cb63c2d9f6b77321eaf18eafe4b16854d2 100644 --- a/theodolite-benchmarks/uc4-kstreams/src/main/java/theodolite/uc4/streamprocessing/TopologyBuilder.java +++ b/theodolite-benchmarks/uc4-kstreams/src/main/java/theodolite/uc4/streamprocessing/TopologyBuilder.java @@ -146,7 +146,7 @@ public class TopologyBuilder { .groupByKey(Grouped.with( SensorParentKeySerde.serde(), this.srAvroSerdeFactory.forValues())) - .windowedBy(TimeWindows.of(this.emitPeriod).grace(this.gracePeriod)) + .windowedBy(TimeWindows.ofSizeAndGrace(this.emitPeriod, this.gracePeriod)) .reduce( // TODO Configurable window aggregation function (oldVal, newVal) -> newVal.getTimestamp() >= oldVal.getTimestamp() ? newVal : oldVal, diff --git a/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/AbstractK8sLoader.kt b/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/AbstractK8sLoader.kt index 871b8cf43907fcb8b0b5ea501c6b47f82e56ff69..36cfef9ce912886a638c200b502923dfe03ef5d0 100644 --- a/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/AbstractK8sLoader.kt +++ b/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/AbstractK8sLoader.kt @@ -13,6 +13,7 @@ abstract class AbstractK8sLoader: K8sResourceLoader { "Deployment" -> loadDeployment(resourceString) "Service" -> loadService(resourceString) "ServiceMonitor" -> loadServiceMonitor(resourceString) + "PodMonitor" -> loadPodMonitor(resourceString) "ConfigMap" -> loadConfigmap(resourceString) "StatefulSet" -> loadStatefulSet(resourceString) "Execution" -> loadExecution(resourceString) @@ -51,6 +52,16 @@ abstract class AbstractK8sLoader: K8sResourceLoader { return loadCustomResourceWrapper(resource, context) } + override fun loadPodMonitor(resource: String): KubernetesResource { + val context = K8sContextFactory().create( + api = "v1", + scope = "Namespaced", + group = "monitoring.coreos.com", + plural = "podmonitors" + ) + return loadCustomResourceWrapper(resource, context) + } + override fun loadExecution(resource: String): KubernetesResource { val context = K8sContextFactory().create( api = "v1", diff --git a/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/K8sResourceLoader.kt b/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/K8sResourceLoader.kt index c123ab2958132cb43ad188136f738b561e91310b..1487b64bf4f7fbcc735539a429be9237d41205bc 100644 --- a/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/K8sResourceLoader.kt +++ b/theodolite/src/main/kotlin/theodolite/k8s/resourceLoader/K8sResourceLoader.kt @@ -11,5 +11,6 @@ interface K8sResourceLoader { fun loadBenchmark(resource: String): KubernetesResource fun loadConfigmap(resource: String): KubernetesResource fun loadServiceMonitor(resource: String): KubernetesResource + fun loadPodMonitor(resource: String): KubernetesResource fun loadCustomResourceWrapper(resource: String, context: CustomResourceDefinitionContext): KubernetesResource } \ No newline at end of file diff --git a/theodolite/src/main/kotlin/theodolite/patcher/MatchLabelPatcher.kt b/theodolite/src/main/kotlin/theodolite/patcher/MatchLabelPatcher.kt new file mode 100644 index 0000000000000000000000000000000000000000..30ff73b5da3b551119ad085adbc982180e4fc066 --- /dev/null +++ b/theodolite/src/main/kotlin/theodolite/patcher/MatchLabelPatcher.kt @@ -0,0 +1,34 @@ +package theodolite.patcher + +import io.fabric8.kubernetes.api.model.KubernetesResource +import io.fabric8.kubernetes.api.model.apps.Deployment +import io.fabric8.kubernetes.api.model.apps.StatefulSet + +/** + * This patcher is able to set the `spec.selector.matchLabels` for a `Deployment` or `StatefulSet` Kubernetes resource. + * + * @property k8sResource The Kubernetes manifests to patch + * @property variableName The matchLabel which should be set + */ +class MatchLabelPatcher(private val k8sResource: KubernetesResource, val variableName: String) : + AbstractPatcher(k8sResource) { + + override fun <String> patch(labelValue: String) { + if (labelValue is kotlin.String) { + when (k8sResource) { + is Deployment -> { + if (k8sResource.spec.selector.matchLabels == null) { + k8sResource.spec.selector.matchLabels = mutableMapOf() + } + k8sResource.spec.selector.matchLabels[this.variableName] = labelValue + } + is StatefulSet -> { + if (k8sResource.spec.selector.matchLabels == null) { + k8sResource.spec.selector.matchLabels = mutableMapOf() + } + k8sResource.spec.selector.matchLabels[this.variableName] = labelValue + } + } + } + } +} \ No newline at end of file diff --git a/theodolite/src/main/kotlin/theodolite/patcher/PatcherFactory.kt b/theodolite/src/main/kotlin/theodolite/patcher/PatcherFactory.kt index 88b3e19e999a889cdcb8345ca7c90c37a6e6d275..e92de4dba7de298c9df76600f2c6785f5878103e 100644 --- a/theodolite/src/main/kotlin/theodolite/patcher/PatcherFactory.kt +++ b/theodolite/src/main/kotlin/theodolite/patcher/PatcherFactory.kt @@ -79,6 +79,14 @@ class PatcherFactory { k8sResource = resource, variableName = patcherDefinition.properties["variableName"]!! ) + "MatchLabelPatcher" -> MatchLabelPatcher( + k8sResource = resource, + variableName = patcherDefinition.properties["variableName"]!! + ) + "TemplateLabelPatcher" -> TemplateLabelPatcher( + k8sResource = resource, + variableName = patcherDefinition.properties["variableName"]!! + ) "ImagePatcher" -> ImagePatcher( k8sResource = resource, container = patcherDefinition.properties["container"]!! diff --git a/theodolite/src/main/kotlin/theodolite/patcher/TemplateLabelPatcher.kt b/theodolite/src/main/kotlin/theodolite/patcher/TemplateLabelPatcher.kt new file mode 100644 index 0000000000000000000000000000000000000000..a524e5c40f90ccf98dc95003cc33dcfceb6f8598 --- /dev/null +++ b/theodolite/src/main/kotlin/theodolite/patcher/TemplateLabelPatcher.kt @@ -0,0 +1,34 @@ +package theodolite.patcher + +import io.fabric8.kubernetes.api.model.KubernetesResource +import io.fabric8.kubernetes.api.model.apps.Deployment +import io.fabric8.kubernetes.api.model.apps.StatefulSet + +/** + * This patcher is able to set the field `spec.template.metadata.labels` for a `Deployment` or `StatefulSet` Kubernetes resource. + * + * @property k8sResource The Kubernetes manifests to patch + * @property variableName The label which should be set + */ +class TemplateLabelPatcher(private val k8sResource: KubernetesResource, val variableName: String) : + AbstractPatcher(k8sResource) { + + override fun <String> patch(labelValue: String) { + if (labelValue is kotlin.String) { + when (k8sResource) { + is Deployment -> { + if (k8sResource.spec.template.metadata.labels == null) { + k8sResource.spec.template.metadata.labels = mutableMapOf() + } + k8sResource.spec.template.metadata.labels[this.variableName] = labelValue + } + is StatefulSet -> { + if (k8sResource.spec.template.metadata.labels == null) { + k8sResource.spec.template.metadata.labels = mutableMapOf() + } + k8sResource.spec.template.metadata.labels[this.variableName] = labelValue + } + } + } + } +} \ No newline at end of file