Add ClickHouse connection

config.yaml

@@ -19,7 +19,6 @@ pipeline:
   log_storage:
     logserver:
       input_file: "/opt/file.txt"
-      max_number_of_connections: 1000
 
   log_collection:
     collector:
@@ -64,6 +63,11 @@ pipeline:
       base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
       threshold: 0.5
 
+  monitoring:
+    clickhouse_connector:
+      batch_size: 10000
+      batch_timeout: 2.0
+
 environment:
   timestamp_format: "%Y-%m-%dT%H:%M:%S.%fZ"
   kafka_brokers:
@@ -80,3 +84,6 @@ environment:
       batch_sender_to_prefilter: "pipeline.batch_sender_to_prefilter"
       prefilter_to_inspector: "pipeline.prefilter_to_inspector"
       inspector_to_detector: "pipeline.inspector_to_detector"
+  monitoring:
+    clickhouse_server:
+      hostname: 172.27.0.11

docker/dev-query.py

@@ -0,0 +1,51 @@
+import datetime
+import os
+import sys
+import uuid
+
+import clickhouse_connect
+
+sys.path.append(os.getcwd())
+from src.base.data_classes.clickhouse_connectors import TABLE_NAME_TO_TYPE
+
+
+def get_tables():
+    tables = {}
+
+    for table_name in TABLE_NAME_TO_TYPE:
+        tables[table_name] = []
+
+    return tables
+
+
+def query_once(client, tables):
+    for table_name in tables.keys():
+        tables[table_name] = client.query(f"SELECT * FROM {table_name};")
+
+    return tables
+
+
+def main():
+    client = clickhouse_connect.get_client(host="172.27.0.11", port=8123)
+    tables = get_tables()
+
+    client.insert(
+        "server_logs",
+        [[uuid.uuid4(), datetime.datetime.now(), "This is a logline"]],
+        ["message_id", "timestamp_in", "message_text"],
+    )
+
+    results = query_once(client, tables)
+
+    for key in results:
+        print(f"'{key}':")
+
+        if results[key].result_rows:
+            for row in results[key].result_rows:
+                print("\t", row)
+        else:
+            print("\t -")
+
+
+if __name__ == "__main__":
+    main()

docker/docker-compose.dev-query.yml

@@ -0,0 +1,56 @@
+include:
+  - "docker-compose.kafka.yml"
+
+services:
+  sandbox:
+    build:
+      context: ..
+      dockerfile: docker/dockerfiles/Dockerfile.dev-query
+      network: host
+    depends_on:
+      kafka1:
+        condition: service_healthy
+      kafka2:
+        condition: service_healthy
+      kafka3:
+        condition: service_healthy
+    networks:
+      heidgaf:
+        ipv4_address: 172.27.0.100
+    memswap_limit: 768m
+    deploy:
+      resources:
+        limits:
+          cpus: '2'
+          memory: 512m
+        reservations:
+          cpus: '1'
+          memory: 256m
+    volumes:
+      - "${MOUNT_PATH:?MOUNT_PATH not set}:/opt/file.txt"
+
+  clickhouse-server:
+    image: clickhouse/clickhouse-server:24.3.12.75-alpine
+    container_name: clickhouse-server
+    networks:
+      heidgaf:
+        ipv4_address: 172.27.0.11
+    restart: "unless-stopped"
+    ports:
+      - "8123:8123"
+      - "9000:9000"
+    healthcheck:
+      test: [ "CMD-SHELL", "nc -z 127.0.0.1 8123" ]
+      interval: 10s
+      timeout: 5s
+      retries: 3
+
+
+networks:
+  heidgaf:
+    driver: bridge
+    ipam:
+      driver: default
+      config:
+        - subnet: 172.27.0.0/16
+          gateway: 172.27.0.1

docker/docker-compose.yml

@@ -184,6 +184,25 @@ services:
       timeout: 5s
       retries: 3
 
+  monitoring_agent:
+    build:
+      context: ..
+      dockerfile: docker/dockerfiles/Dockerfile.monitoring
+      network: host
+    restart: "unless-stopped"
+    depends_on:
+      kafka1:
+        condition: service_healthy
+      kafka2:
+        condition: service_healthy
+      kafka3:
+        condition: service_healthy
+      clickhouse-server:
+        condition: service_healthy
+    networks:
+      heidgaf:
+        ipv4_address: 172.27.0.12
+
 networks:
   heidgaf:
     driver: bridge

docker/dockerfiles/Dockerfile.dev-query

@@ -0,0 +1,15 @@
+FROM python:3.11-slim-bookworm
+
+ENV PYTHONDONTWRITEBYTECODE=1
+
+WORKDIR /usr/src/app
+
+RUN pip --disable-pip-version-check install --no-cache-dir --no-compile clickhouse_connect marshmallow_dataclass colorlog pyYAML confluent_kafka
+
+COPY src/base ./src/base
+COPY config.yaml .
+COPY docker/dev-query.py .
+
+RUN rm -rf /root/.cache
+
+CMD [ "python", "dev-query.py"]

requirements/requirements.detector.txt

@@ -5,3 +5,4 @@ colorlog~=6.8.2
 PyYAML~=6.0.1
 confluent-kafka~=2.4.0
 marshmallow_dataclass~=8.7.1
+clickhouse_connect~=0.8.3

requirements/requirements.inspector.txt

@@ -4,3 +4,4 @@ colorlog~=6.8.2
 streamad~=0.3.1
 numpy~=1.26.4
 marshmallow_dataclass~=8.7.1
+clickhouse_connect~=0.8.3

requirements/requirements.logcollector.txt

@@ -2,3 +2,4 @@ PyYAML~=6.0.1
 colorlog~=6.8.2
 confluent-kafka~=2.4.0
 marshmallow_dataclass~=8.7.1
+clickhouse_connect~=0.8.3

requirements/requirements.logserver.txt

@@ -3,3 +3,4 @@ colorlog~=6.8.2
 confluent-kafka~=2.4.0
 marshmallow_dataclass~=8.7.1
 aiofiles~=24.1.0
+clickhouse_connect~=0.8.3

requirements/requirements.monitoring.txt

@@ -0,0 +1,5 @@
+clickhouse_connect~=0.8.3
+confluent-kafka~=2.4.0
+marshmallow_dataclass~=8.7.1
+colorlog~=6.8.2
+PyYAML~=6.0.1

requirements/requirements.prefilter.txt

@@ -2,3 +2,4 @@ PyYAML~=6.0.1
 colorlog~=6.8.2
 confluent-kafka~=2.4.0
 marshmallow_dataclass~=8.7.1
+clickhouse_connect~=0.8.3

src/base/__init__.py

@@ -1,19 +0,0 @@
-from typing import List
-from dataclasses import dataclass, field
-import marshmallow.validate
-import datetime
-
-
-@dataclass
-class Batch:
-    begin_timestamp: datetime.datetime = field(
-        metadata={
-            "marshmallow_field": marshmallow.fields.DateTime("%Y-%m-%dT%H:%M:%S.%fZ")
-        }
-    )
-    end_timestamp: datetime.datetime = field(
-        metadata={
-            "marshmallow_field": marshmallow.fields.DateTime("%Y-%m-%dT%H:%M:%S.%fZ")
-        }
-    )
-    data: List[dict] = field(default_factory=list)

src/base/clickhouse_kafka_sender.py

@@ -0,0 +1,26 @@
+import os
+import sys
+
+import marshmallow_dataclass
+
+sys.path.append(os.getcwd())
+from src.base.data_classes.clickhouse_connectors import TABLE_NAME_TO_TYPE
+from src.base.kafka_handler import SimpleKafkaProduceHandler
+from src.base.log_config import get_logger
+
+logger = get_logger()
+
+
+class ClickHouseKafkaSender:
+    def __init__(self, table_name: str):
+        self.table_name = table_name
+        self.kafka_producer = SimpleKafkaProduceHandler()
+        self.data_schema = marshmallow_dataclass.class_schema(
+            TABLE_NAME_TO_TYPE.get(table_name)
+        )()
+
+    def insert(self, data: dict):
+        self.kafka_producer.produce(
+            topic=f"clickhouse_{self.table_name}",
+            data=self.data_schema.dumps(data),
+        )

src/base/data_classes/batch.py

@@ -0,0 +1,24 @@
+import datetime
+import uuid
+from dataclasses import dataclass, field
+from typing import List
+
+import marshmallow.validate
+
+
+@dataclass
+class Batch:
+    batch_id: uuid.UUID = field(
+        metadata={"marshmallow_field": marshmallow.fields.UUID()}
+    )
+    begin_timestamp: datetime.datetime = field(
+        metadata={
+            "marshmallow_field": marshmallow.fields.DateTime("%Y-%m-%dT%H:%M:%S.%fZ")
+        }
+    )
+    end_timestamp: datetime.datetime = field(
+        metadata={
+            "marshmallow_field": marshmallow.fields.DateTime("%Y-%m-%dT%H:%M:%S.%fZ")
+        }
+    )
+    data: List = field(default_factory=list)