Add monitoring solution using Grafana

.github/workflows/build_publish_docker.yml

@@ -25,7 +25,8 @@ jobs:
                      "inspector",
                      "logcollector",
                      "logserver",
-                     "prefilter"
+                     "prefilter",
+                     "monitoring"
         ]
     permissions:
       contents: read

config.yaml

@@ -32,8 +32,8 @@ pipeline:
         - [ "response_ip", IpAddress ]
         - [ "size", RegEx, '^\d+b$' ]
     batch_handler:
-      batch_size: 1000
-      batch_timeout: 20.0
+      batch_size: 10000
+      batch_timeout: 30.0
       subnet_id:
         ipv4_prefix_length: 24
         ipv6_prefix_length: 64
@@ -65,25 +65,25 @@ pipeline:
 
   monitoring:
     clickhouse_connector:
-      batch_size: 10000
+      batch_size: 50  # do not set higher
       batch_timeout: 2.0
 
 environment:
   timestamp_format: "%Y-%m-%dT%H:%M:%S.%fZ"
   kafka_brokers:
-    - hostname: 172.27.0.3
+    - hostname: kafka1
       port: 8097
-    - hostname: 172.27.0.4
+    - hostname: kafka2
       port: 8098
-    - hostname: 172.27.0.5
+    - hostname: kafka3
       port: 8099
   kafka_topics:
     pipeline:
-      logserver_in: "pipeline.logserver_in"
-      logserver_to_collector: "pipeline.logserver_to_collector"
-      batch_sender_to_prefilter: "pipeline.batch_sender_to_prefilter"
-      prefilter_to_inspector: "pipeline.prefilter_to_inspector"
-      inspector_to_detector: "pipeline.inspector_to_detector"
+      logserver_in: "pipeline-logserver_in"
+      logserver_to_collector: "pipeline-logserver_to_collector"
+      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
+      hostname: clickhouse-server

docker/.env

@@ -1 +1 @@
-MOUNT_PATH=./default.txt
+MOUNT_PATH=../../default.txt

docker/benchmark_tests/Dockerfile.run_test

@@ -0,0 +1,17 @@
+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 marshmallow_dataclass colorlog pyYAML confluent_kafka numpy polars scikit-learn torch
+
+COPY src/base ./src/base
+COPY src/train ./src/train
+COPY config.yaml .
+COPY docker/benchmark_tests .
+COPY data ./data
+
+RUN rm -rf /root/.cache
+
+CMD [ "python", "run_test.py"]

docker/benchmark_tests/docker-compose.run_test.yml

@@ -0,0 +1,20 @@
+services:
+  benchmark_test_run:
+    build:
+      context: ../..
+      dockerfile: docker/benchmark_tests/Dockerfile.run_test
+      network: host
+    networks:
+      docker_heidgaf:
+    deploy:
+      resources:
+        limits:
+          cpus: '2'
+          memory: 512m
+        reservations:
+          cpus: '1'
+          memory: 256m
+
+networks:
+  docker_heidgaf:
+    external: true

docker/benchmark_tests/run_test.py

@@ -0,0 +1,248 @@
+import datetime
+import ipaddress
+import os
+import random
+import sys
+import time
+
+import polars as pl
+from confluent_kafka import KafkaError
+
+sys.path.append(os.getcwd())
+from src.base.kafka_handler import SimpleKafkaProduceHandler
+from src.train.dataset import Dataset, DatasetLoader
+from src.base.log_config import get_logger
+from src.base.utils import setup_config
+
+logger = get_logger()
+config = setup_config()
+
+PRODUCE_TO_TOPIC = config["environment"]["kafka_topics"]["pipeline"]["logserver_in"]
+
+
+class DatasetGenerator:
+    """Generates log lines and datasets."""
+
+    def __init__(self, data_base_path: str = "./data"):
+        datasets = DatasetLoader(base_path=data_base_path, max_rows=10000)
+
+        dataset = Dataset(
+            data_path="",
+            data=pl.concat(
+                [
+                    datasets.dgta_dataset.data,
+                    # datasets.cic_dataset.data,
+                    # datasets.bambenek_dataset.data,
+                    # datasets.dga_dataset.data,
+                    # datasets.dgarchive_dataset.data,
+                ]
+            ),
+            max_rows=1000,
+        )
+
+        self.domains = dataset.data
+
+    def generate_random_logline(
+        self, statuses: list[str] = None, record_types: list[str] = None
+    ):
+        """Generates a (mostly) random logline."""
+        if record_types is None:
+            record_types = 6 * ["AAAA"] + 10 * ["A"] + ["PR", "CNAME"]
+
+        if statuses is None:
+            statuses = ["NOERROR", "NXDOMAIN"]
+
+        # choose timestamp
+        timestamp = (
+            datetime.datetime.now() + datetime.timedelta(0, 0, random.randint(0, 900))
+        ).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+        # choose status code
+        status = random.choice(statuses)
+
+        # choose client IP address
+        number_of_subnets = 50
+        client_ip = (
+            f"192.168.{random.randint(0, number_of_subnets)}.{random.randint(1, 255)}"
+        )
+
+        # choose server IP address
+        server_ip = f"10.10.0.{random.randint(1, 100)}"
+
+        # choose random domain (can be malicious or benign)
+        domain = self.get_random_domain()
+
+        # choose random record type
+        record_type = random.choice(record_types)
+
+        # choose random response IP address
+        def _get_random_ipv4():
+            max_ipv4 = ipaddress.IPv4Address._ALL_ONES  # 2 ** 32 - 1
+            return ipaddress.IPv4Address._string_from_ip_int(
+                random.randint(0, max_ipv4)
+            )
+
+        def _get_random_ipv6():
+            max_ipv6 = ipaddress.IPv6Address._ALL_ONES  # 2 ** 128 - 1
+            return ipaddress.IPv6Address._string_from_ip_int(
+                random.randint(0, max_ipv6)
+            )
+
+        ip_address_choices = [_get_random_ipv4(), _get_random_ipv6()]
+        response_ip_address = random.choice(ip_address_choices)
+
+        # choose random size
+        size = f"{random.randint(50, 255)}b"
+
+        return f"{timestamp} {status} {client_ip} {server_ip} {domain} {record_type} {response_ip_address} {size}"
+
+    def get_random_domain(self) -> str:
+        random_domain = self.domains.sample(n=1)
+        return random_domain["query"].item()
+
+    def generate_dataset(self, number_of_elements: int) -> list[str]:
+        dataset = []
+
+        for _ in range(number_of_elements):
+            logline = self.generate_random_logline()
+            dataset.append(logline)
+
+        return dataset
+
+
+class ScalabilityTest:
+    """Base class for tests that focus on the scalability of the software."""
+
+    def __init__(self):
+        self.dataset_generator = DatasetGenerator()
+        self.kafka_producer = SimpleKafkaProduceHandler()
+
+        self.interval_lengths = None
+        self.msg_per_sec_in_intervals = None
+
+    def execute(self):
+        """Executes the test with the configured parameters."""
+        logger.warning(f"Start at: {datetime.datetime.now()}")
+
+        cur_index = 0
+        for i in range(len(self.msg_per_sec_in_intervals)):
+            cur_index = self._execute_one_interval(
+                cur_index=cur_index,
+                msg_per_sec=self.msg_per_sec_in_intervals[i],
+                length_in_sec=self.interval_lengths[i],
+            )
+
+        logger.warning(f"Stop at: {datetime.datetime.now()}")
+
+    def _execute_one_interval(
+        self, cur_index: int, msg_per_sec: float | int, length_in_sec: float | int
+    ) -> int:
+        start_of_interval_timestamp = datetime.datetime.now()
+        logger.warning(
+            f"Start interval with {msg_per_sec} msg/s at {start_of_interval_timestamp}"
+        )
+
+        while (
+            datetime.datetime.now() - start_of_interval_timestamp
+            < datetime.timedelta(seconds=length_in_sec)
+        ):
+            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 / msg_per_sec)
+
+        logger.warning(f"Finish interval with {msg_per_sec} msg/s")
+        return cur_index
+
+
+class RampUpTest(ScalabilityTest):
+    """Starts with a low rate and increases the rate in fixed intervals."""
+
+    def __init__(
+        self,
+        msg_per_sec_in_intervals: list[float | int],
+        interval_length_in_sec: int | float | list[int | float],
+    ):
+        super().__init__()
+        self.msg_per_sec_in_intervals = msg_per_sec_in_intervals
+
+        if type(interval_length_in_sec) is list:
+            self.interval_lengths = interval_length_in_sec
+        else:
+            self.interval_lengths = [
+                interval_length_in_sec for _ in range(len(msg_per_sec_in_intervals))
+            ]
+
+        if len(interval_length_in_sec) != len(msg_per_sec_in_intervals):
+            raise Exception("Different lengths of interval lists. Must be equal.")
+
+
+class BurstTest(ScalabilityTest):
+    """Starts with a normal rate, sends a high rate for a short period, then returns to normal rate. Repeats the
+    process for a defined number of times."""
+
+    def __init__(
+        self,
+        normal_rate_msg_per_sec: float | int,
+        burst_rate_msg_per_sec: float | int,
+        normal_rate_interval_length: float | int,
+        burst_rate_interval_length: float | int,
+        number_of_intervals: int = 1,
+    ):
+        super().__init__()
+
+        self.msg_per_sec_in_intervals = [normal_rate_msg_per_sec]
+        self.interval_lengths = [normal_rate_interval_length]
+
+        for _ in range(number_of_intervals):
+            self.msg_per_sec_in_intervals.append(burst_rate_msg_per_sec)
+            self.msg_per_sec_in_intervals.append(normal_rate_msg_per_sec)
+
+            self.interval_lengths.append(burst_rate_interval_length)
+            self.interval_lengths.append(normal_rate_interval_length)
+
+
+class LongTermTest(ScalabilityTest):
+    """Starts with a low rate and increases the rate in fixed intervals."""
+
+    def __init__(self, full_length: float | int, msg_per_sec: float | int):
+        super().__init__()
+
+        self.msg_per_sec_in_intervals = [msg_per_sec]
+        self.interval_lengths = [full_length]
+
+
+def main():
+    """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()
+
+
+if __name__ == "__main__":
+    main()

docker/create_datatest_tables/dgta_dataset.sql

@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS dgta_dataset (
+    query String,
+    class Int32,
+    labels Array(String),
+    tld String,
+    fqdn String,
+    secondleveldomain String,
+    thirdleveldomain String
+)
+ENGINE = MergeTree
+PRIMARY KEY(query);

src/monitoring/create_tables/alerts.sql → docker/create_tables/alerts.sql

@@ -3,6 +3,7 @@ CREATE TABLE IF NOT EXISTS alerts (
     alert_timestamp DateTime64(6) NOT NULL,
     suspicious_batch_id UUID NOT NULL,
     overall_score Float32 NOT NULL,
+    domain_names String NOT NULL,
     result String,
 )
 ENGINE = MergeTree

src/monitoring/create_tables/dns_loglines.sql → docker/create_tables/dns_loglines.sql

@@ -5,7 +5,7 @@ CREATE TABLE IF NOT EXISTS dns_loglines (
     status_code String NOT NULL,
     client_ip String NOT NULL,
     record_type String NOT NULL,
-    additional_fields Nullable(String)
+    additional_fields String
 )
 ENGINE = MergeTree
 PRIMARY KEY (logline_id);

docker/create_tables/fill_levels.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS fill_levels (
+    timestamp DateTime64(6) NOT NULL,
+    stage String NOT NULL,
+    entry_type String NOT NULL,
+    entry_count UInt32 DEFAULT 0
+)
+ENGINE = MergeTree
+PRIMARY KEY (timestamp, stage, entry_type);