Feature/general improvements

.gitignore

@@ -8,7 +8,6 @@ results/*
 *.so
 *.html
 *.csv
-*.json
 *.txt
 *.parquet
 .DS_Store
@@ -77,6 +76,7 @@ instance/
 
 # Sphinx documentation
 docs/_build/
+docs/output/
 
 # PyBuilder
 .pybuilder/
@@ -324,7 +324,9 @@ cython_debug/
 
 # Others
 docs/api/
+!/docs/api/
 !/docs/api/index.rst
+!/docs/api/*.rst
 
 # requirements.txt
 !*/requirements.*.txt

config.yaml

@@ -13,7 +13,7 @@ logging:
     data_inspection.inspector:
       debug: false
     data_analysis.detector:
-      debug: false
+      debug: true
 
 pipeline:
   scaling:
@@ -23,50 +23,58 @@ pipeline:
     modules:
       log_storage.logserver:
         executor: thread
-        max_workers: 1
+        max_workers: 4
       log_collection.collector:
         executor: thread
         max_workers: 1
         instances:
           dga_collector:
-            max_workers: 1
+            max_workers: 2
           domainator_collector:
-            max_workers: 1
+            max_workers: 2
       log_filtering.prefilter:
         executor: thread
-        max_workers: 1
+        max_workers: 2
         instances:
           dga_filter:
-            max_workers: 1
+            max_workers: 2
           no_filter:
-            max_workers: 1
+            max_workers: 2
       data_inspection.inspector:
         executor: thread
-        max_workers: 1
+        max_workers: 2
         instances:
           dga_inspector:
-            max_workers: 1
+            max_workers: 2
           no_inspector:
-            max_workers: 1
+            max_workers: 2
       data_analysis.detector:
         executor: thread
-        max_workers: 1
+        max_workers: 2
         instances:
           RF-dga_detector:
-            max_workers: 1
+            max_workers: 2
           domainator:
+            max_workers: 3
+          domainator_attributor:
+            max_workers: 1
+          domainator_attributor_behaviour:
+            max_workers: 1
+          domainator_attributor_identification_behaviour:
+            max_workers: 1
+          domainator_attributor_identification:
             max_workers: 1
       pipeline.alerter:
         executor: thread
-        max_workers: 1
+        max_workers: 2
         instances:
           generic:
-            max_workers: 1
+            max_workers: 2
           attributor:
-            max_workers: 1
+            max_workers: 2
       monitoring.agent:
         executor: thread
-        max_workers: 1
+        max_workers: 2
   log_storage:
     logserver:
       input_file: "/opt/file.txt"
@@ -144,33 +152,65 @@ pipeline:
       detector_module_name: "dga_detector"
       detector_class_name: "DGADetector"
       model: rf
+      use_scaler: false
       checksum: 5db8bfb617e80361362c33b1d1afc6d762c28e9fa9275fb11514a3bdef76bb88
       base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
       threshold: 0.5
       consume_from: inspector
       inspector_name: dga_inspector
-      next_detectors: domainator
+      next_detectors: ""
       send_to_alerter: true
       produce_topics: ""
     - name: "domainator"
       detector_module_name: "domainator_detector"
       detector_class_name: "DomainatorDetector"
       model: domainator
-      checksum: 9d86d66b4976c9b325bed0934a9a9eb3a20960b08be9afe491454624cc0aaa6c
+      use_scaler: false
+      checksum: a4aac4c585f1e614c3cf0d737e80b960c5de6e87b253f7cdd07125d9ce486476
       base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
-      threshold: 0.5
+      threshold: 0.05
       consume_from: inspector
       inspector_name: "domainator_inspector"
-      next_detectors: "domainator_attributor"
+      next_detectors: 
+        - "domainator_attributor_behaviour"
+        - "domainator_attributor_identification_behaviour"
+        - "domainator_attributor_identification"
       send_to_alerter: true
       produce_topics: ""
-    - name: "domainator_attributor"
+    - name: "domainator_attributor_behaviour"
       detector_module_name: "domainator_attributor"
       detector_class_name: "DomainatorAttributor"
-      model: domainator
-      checksum: 9d86d66b4976c9b325bed0934a9a9eb3a20960b08be9afe491454624cc0aaa6c
+      model: domainator_attributor_behaviour
+      use_scaler: false
+      checksum: d8f302edc166ecc80985838a30b5dff16ccc83480ea3c2480652f49c8f6b5e9b
       base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
-      threshold: 0.5
+      threshold: 0.05
+      consume_from: detector
+      detector_name: "domainator"
+      next_detectors: ""
+      send_to_alerter: true
+      produce_topics: ""
+    - name: "domainator_attributor_identification_behaviour"
+      detector_module_name: "domainator_attributor"
+      detector_class_name: "DomainatorAttributor"
+      model: domainator_attributor_identification_behaviour
+      use_scaler: false
+      checksum: 9a0970b4160b22f4c3c5ac99760f0ace5500dd25c5a195ff13254ad3c11d5dcd
+      base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
+      threshold: 0.05
+      consume_from: detector
+      detector_name: "domainator"
+      next_detectors: ""
+      send_to_alerter: true
+      produce_topics: ""
+    - name: "domainator_attributor_identification"
+      detector_module_name: "domainator_attributor"
+      detector_class_name: "DomainatorAttributor"
+      model: domainator_attributor_identification
+      use_scaler: false
+      checksum: 360bd26881beabce7e7581963240915de807c48b5e4a3501a657139f2ecb8a8b
+      base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
+      threshold: 0.05
       consume_from: detector
       detector_name: "domainator"
       next_detectors: ""
@@ -180,6 +220,9 @@ pipeline:
   alerting:
     log_to_file: true
     log_file_path: "/opt/logs/alerts.txt"
+    log_rotation:
+      enabled: true
+      retention_days: 7
     log_to_kafka: true
     external_kafka_topic: "hamstring_alerts"
     plugins: []
@@ -212,6 +255,40 @@ environment:
       internal_port: 19094
       external_port: 8099
       node_ip: 127.0.0.1
+  kafka_consumer:
+    # Allow long-running detector batches without Kafka forcing a group rebalance.
+    # Default librdkafka value is 300000 ms (5 minutes), which can be too short
+    # for model inference plus downstream alert/monitoring writes.
+    max_poll_interval_ms: 1800000
+  kafka_topics:
+    replication_factor: 3
+    auto_expand_partitions: true
+    stages:
+      logserver_in:
+        partitions: 12
+        replication_factor: 3
+      logserver_to_collector:
+        partitions: 12
+        replication_factor: 3
+      batch_sender_to_prefilter:
+        partitions: 12
+        replication_factor: 3
+      prefilter_to_inspector:
+        partitions: 12
+        replication_factor: 3
+      inspector_to_detector:
+        partitions: 12
+        replication_factor: 3
+      detector_to_alerter:
+        partitions: 12
+        replication_factor: 3
+      detector_to_detector:
+        partitions: 12
+        replication_factor: 3
+    topics:
+      hamstring_alerts:
+        partitions: 12
+        replication_factor: 3
   kafka_topics_prefix:
     pipeline:
       logserver_in: "hamstring_input"

docker/create_tables/alerts.sql

@@ -8,4 +8,5 @@ CREATE TABLE IF NOT EXISTS alerts (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(alert_timestamp)
-ORDER BY (alert_timestamp, src_ip, suspicious_batch_id);
+ORDER BY (alert_timestamp, src_ip, suspicious_batch_id)
+TTL toDateTime(alert_timestamp) + INTERVAL 60 DAY;

docker/create_tables/batch_timestamps.sql

@@ -9,4 +9,5 @@ CREATE TABLE IF NOT EXISTS batch_timestamps (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
-ORDER BY (stage, status, timestamp, instance_name, batch_id);
+ORDER BY (stage, status, timestamp, instance_name, batch_id)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/batch_tree.sql

@@ -11,4 +11,5 @@ CREATE TABLE IF NOT EXISTS batch_tree (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
-ORDER BY (stage, status, timestamp, instance_name, batch_row_id, parent_batch_row_id);
+ORDER BY (stage, status, timestamp, instance_name, batch_row_id, parent_batch_row_id)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/failed_loglines.sql

@@ -6,4 +6,5 @@ CREATE TABLE IF NOT EXISTS failed_loglines (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp_failed)
-ORDER BY (timestamp_failed, timestamp_in);
+ORDER BY (timestamp_failed, timestamp_in)
+TTL toDateTime(timestamp_failed) + INTERVAL 1 DAY;

docker/create_tables/fill_levels.sql

@@ -6,4 +6,5 @@ CREATE TABLE IF NOT EXISTS fill_levels (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
-ORDER BY (stage, entry_type, timestamp);
+ORDER BY (stage, entry_type, timestamp)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/logline_timestamps.sql

@@ -7,4 +7,5 @@ CREATE TABLE IF NOT EXISTS logline_timestamps (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
-ORDER BY (stage, status, timestamp, logline_id);
+ORDER BY (stage, status, timestamp, logline_id)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/logline_to_batches.sql

@@ -1,6 +1,11 @@
 CREATE TABLE IF NOT EXISTS logline_to_batches (
+    timestamp DateTime64(6) NOT NULL,
     logline_id UUID NOT NULL,
     batch_id UUID NOT NULL
 )
 ENGINE = MergeTree
-ORDER BY (batch_id, logline_id);
+ORDER BY (timestamp, batch_id, logline_id)
+PARTITION BY toYYYYMM(timestamp)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+
+

docker/create_tables/loglines.sql

@@ -7,4 +7,5 @@ CREATE TABLE IF NOT EXISTS loglines (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
-ORDER BY (timestamp, src_ip, subnet_id, logline_id);
+ORDER BY (timestamp, src_ip, subnet_id, logline_id)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/server_log_terminal_events.sql

@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS server_log_terminal_events (
+    message_id UUID NOT NULL,
+    stage LowCardinality(String) NOT NULL,
+    status LowCardinality(String) NOT NULL,
+    timestamp DateTime64(6) NOT NULL
+)
+ENGINE = MergeTree
+PARTITION BY toYYYYMM(timestamp)
+ORDER BY (stage, status, timestamp, message_id)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/server_log_to_logline.sql

@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS server_log_to_logline (
+    message_id UUID NOT NULL,
+    logline_id UUID NOT NULL,
+    timestamp DateTime64(6) NOT NULL,
+)
+ENGINE = MergeTree
+ORDER BY (timestamp, message_id, logline_id)
+PARTITION BY toYYYYMM(timestamp)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+

docker/create_tables/server_logs.sql

@@ -5,4 +5,5 @@ CREATE TABLE IF NOT EXISTS server_logs (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp_in)
-ORDER BY (timestamp_in, message_id);
+ORDER BY (timestamp_in, message_id)
+TTL toDateTime(timestamp_in) + INTERVAL 1 DAY;

docker/create_tables/server_logs_timestamps.sql

@@ -5,4 +5,5 @@ CREATE TABLE IF NOT EXISTS server_logs_timestamps (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(event_timestamp)
-ORDER BY (event, event_timestamp, message_id);
+ORDER BY (event, event_timestamp, message_id)
+TTL toDateTime(event_timestamp) + INTERVAL 1 DAY;

docker/create_tables/suspicious_batch_timestamps.sql

@@ -10,4 +10,5 @@ CREATE TABLE IF NOT EXISTS suspicious_batch_timestamps (
 )
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
-ORDER BY (stage, status, timestamp, instance_name, suspicious_batch_id, src_ip);
+ORDER BY (stage, status, timestamp, instance_name, suspicious_batch_id, src_ip)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docker/create_tables/suspicious_batches_to_batch.sql

@@ -1,6 +1,9 @@
 CREATE TABLE IF NOT EXISTS suspicious_batches_to_batch (
+    timestamp DateTime64(6) NOT NULL,
     suspicious_batch_id UUID NOT NULL,
     batch_id UUID NOT NULL
 )
 ENGINE = MergeTree
-ORDER BY (batch_id, suspicious_batch_id);
+ORDER BY (timestamp, batch_id, suspicious_batch_id)
+PARTITION BY toYYYYMM(timestamp)
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;