Merge pull request #60 from stefanDeveloper/update-benchmark-tests

Update benchmark tests

Author: stefanDeveloper

docker/benchmark_tests/run_test.py

@@ -210,39 +210,96 @@ def __init__(
             self.interval_lengths.append(normal_rate_interval_length)
 
 
-class LongTermTest(ScalabilityTest):
-    """Starts with a low rate and increases the rate in fixed intervals."""
+class LongTermTest:
+    """Keeps a consistent rate for a long time."""
 
-    def __init__(self, full_length: float | int, msg_per_sec: float | int):
-        super().__init__()
+    def __init__(self, full_length_in_min: float | int, msg_per_sec: float | int):
+        self.dataset_generator = DatasetGenerator()
+        self.kafka_producer = SimpleKafkaProduceHandler()
+
+        self.msg_per_sec = msg_per_sec
+        self.full_length_in_min = full_length_in_min
+
+    def execute(self):
+        """Executes the test with the configured parameters."""
+        start_timestamp = datetime.datetime.now()
+        logger.warning(
+            f"Start {self.full_length_in_min} minute-test with "
+            f"rate {self.msg_per_sec} msg/sec at: {start_timestamp}"
+        )
+
+        cur_index = 0
+        while datetime.datetime.now() - start_timestamp < datetime.timedelta(
+            minutes=self.full_length_in_min
+        ):
+            try:
+                self.kafka_producer.produce(
+                    PRODUCE_TO_TOPIC,
+                    self.dataset_generator.generate_random_logline(),
+                )
+                logger.info(
+                    f"Sent message {cur_index + 1} at: {datetime.datetime.now()}"
+                )
+                cur_index += 1
+            except KafkaError:
+                logger.warning(KafkaError)
+            time.sleep(1.0 / self.msg_per_sec)
+
+        logger.warning(
+            f"Stop at: {datetime.datetime.now()}, sent {cur_index} messages in the "
+            f"past {(datetime.datetime.now() - start_timestamp).total_seconds() / 60} minutes."
+        )
 
-        self.msg_per_sec_in_intervals = [msg_per_sec]
-        self.interval_lengths = [full_length]
 
+class MaximumThroughputTest(LongTermTest):
+    """Keeps a consistent rate that is too high to be handled."""
 
-def main():
+    def __init__(self, length_in_min: float | int, msg_per_sec: int = 10000):
+        super().__init__(full_length_in_min=length_in_min, msg_per_sec=msg_per_sec)
+
+
+def main(test_type_nr):
     """Creates the test instance and executes the test."""
-    # ramp_up_test = RampUpTest(
-    #     msg_per_sec_in_intervals=[1, 10, 50, 100, 150],
-    #     interval_length_in_sec=[10, 5, 4, 4, 2],
-    # )
-    # ramp_up_test.execute()
-
-    burst_test = BurstTest(
-        normal_rate_msg_per_sec=20,
-        burst_rate_msg_per_sec=10000,
-        normal_rate_interval_length=10,
-        burst_rate_interval_length=2,
-        number_of_intervals=3,
-    )
-    burst_test.execute()
-
-    # long_term_test = LongTermTest(
-    #     full_length=10.4,
-    #     msg_per_sec=15,
-    # )
-    # long_term_test.execute()
+    match test_type_nr:
+        case 1:
+            ramp_up_test = RampUpTest(
+                msg_per_sec_in_intervals=[1, 10, 50, 100, 150, 200],
+                interval_length_in_sec=[30, 30, 30, 30, 30, 30],
+            )
+            ramp_up_test.execute()
+
+        case 2:
+            burst_test = BurstTest(
+                normal_rate_msg_per_sec=20,
+                burst_rate_msg_per_sec=10000,
+                normal_rate_interval_length=10,
+                burst_rate_interval_length=2,
+                number_of_intervals=3,
+            )
+            burst_test.execute()
+
+        case 3:
+            maximum_throughput_test = MaximumThroughputTest(
+                length_in_min=1,
+            )
+            maximum_throughput_test.execute()
+
+        case 4:
+            long_term_test = LongTermTest(
+                full_length_in_min=10,
+                msg_per_sec=15,
+            )
+            long_term_test.execute()
+
+        case _:
+            pass
 
 
 if __name__ == "__main__":
-    main()
+    """
+    1 - Ramp-up test
+    2 - Burst test
+    3 - Maximum throughput test
+    4 - Long-term test
+    """
+    main(1)

docker/docker-compose-swarm/docker-compose.swarm-kafka.yml

@@ -0,0 +1,111 @@
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.3.2
+    networks:
+      - heidgaf
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_SERVER_ID: 1
+    healthcheck:
+      test: [ "CMD-SHELL", "nc -z localhost 2181" ]
+      interval: 10s
+      timeout: 5s
+      retries: 3
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-1 ]
+      restart_policy:
+        condition: on-failure
+
+  kafka1:
+    image: confluentinc/cp-kafka:7.3.2
+    networks:
+      - heidgaf
+    ports:
+      - "8097:8097"
+      - "29092:29092"
+    depends_on:
+      - zookeeper
+    healthcheck:
+      test: [ "CMD-SHELL", "nc -z localhost 8097" ]
+      interval: 30s
+      timeout: 10s
+      retries: 10
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:19092,EXTERNAL://tasks.kafka1:8097,DOCKER://host.docker.internal:29092
+      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-1 ]
+      restart_policy:
+        condition: on-failure
+
+
+  kafka2:
+    image: confluentinc/cp-kafka:7.3.2
+    networks:
+      - heidgaf
+    ports:
+      - "8098:8098"
+      - "29093:29093"
+    depends_on:
+      - zookeeper
+    healthcheck:
+      test: [ "CMD-SHELL", "nc -z localhost 8098" ]
+      interval: 30s
+      timeout: 10s
+      retries: 10
+    environment:
+      KAFKA_BROKER_ID: 2
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:19093,EXTERNAL://tasks.kafka2:8098,DOCKER://host.docker.internal:29093
+      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-1 ]
+      restart_policy:
+        condition: on-failure
+
+  kafka3:
+    image: confluentinc/cp-kafka:7.3.2
+    networks:
+      - heidgaf
+    ports:
+      - "8099:8099"
+      - "29094:29094"
+    depends_on:
+      - zookeeper
+    healthcheck:
+      test: [ "CMD-SHELL", "nc -z localhost 8099" ]
+      interval: 30s
+      timeout: 10s
+      retries: 10
+    environment:
+      KAFKA_BROKER_ID: 3
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT,DOCKER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:19094,EXTERNAL://tasks.kafka3:8099,DOCKER://host.docker.internal:29094
+      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-1 ]
+      restart_policy:
+        condition: on-failure
+
+networks:
+  heidgaf:
+    external: true

docker/docker-compose-swarm/docker-compose.swarm-monitoring.yml

@@ -0,0 +1,66 @@
+services:
+  clickhouse-server:
+    image: clickhouse/clickhouse-server:24.3.12.75-alpine
+    volumes:
+      - ../create_tables:/docker-entrypoint-initdb.d
+    networks:
+      - heidgaf
+    ports:
+      - "8123:8123"
+      - "9000:9000"
+#    healthcheck:
+#      test: [ "CMD-SHELL", "nc -z localhost 8123" ]
+#      interval: 10s
+#      timeout: 5s
+#      retries: 3
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-3 ]
+      restart_policy:
+        condition: on-failure
+
+  grafana:
+    image: grafana/grafana:11.2.2-security-01
+    networks:
+      - heidgaf
+    ports:
+      - "3000:3000"
+    volumes:
+      - ../grafana-provisioning/dashboards:/etc/grafana/provisioning/dashboards
+      - ../grafana-provisioning/dashboards/dashboards.yaml:/etc/grafana/provisioning/dashboards/dashboards.yaml
+      - ../grafana-provisioning/datasources.yaml:/etc/grafana/provisioning/datasources/datasources.yaml
+    environment:
+      - GF_SECURITY_ADMIN_USER=admin
+      - GF_SECURITY_ADMIN_PASSWORD=admin
+      - GF_INSTALL_PLUGINS=grafana-clickhouse-datasource
+#    healthcheck:
+#      test: [ "CMD-SHELL", "nc -z localhost 3000" ]
+#      interval: 10s
+#      timeout: 5s
+#      retries: 3
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-3 ]
+      restart_policy:
+        condition: on-failure
+
+  monitoring_agent:
+    image: stefan96/heidgaf-monitoring
+    networks:
+      - heidgaf
+    environment:
+      - GROUP_ID=monitoring_agent
+    depends_on:
+      - kafka1
+      - kafka2
+      - kafka3
+      - clickhouse-server
+    deploy:
+      placement:
+        constraints: [ node.hostname == heidgaf-3 ]
+      restart_policy:
+        condition: on-failure
+
+networks:
+  heidgaf:
+    external: true