Try fixing kafka and clickhouse taking up too much space

README.md

@@ -82,6 +82,12 @@ environment variables such as `HAMSTRING_IMAGE_REGISTRY`,
 `DETECTOR_PLACEMENT_CONSTRAINT`. Remove the stack with
 `docker stack rm hamstring`.
 
+Before deploying on small nodes, plan Kafka/ClickHouse placement and configure
+bounded host storage. See the
+[storage-constrained Swarm deployment guide](./docs/storage_operations.rst) for
+Docker log rotation, Kafka retention semantics,
+ClickHouse TTL policy, filesystem isolation, and 50 GB-per-node budget.
+
 <p align="center">
   <img src="./assets/hamstring_terminal.gif" alt="Terminal example"/>
 </p>

config.yaml

@@ -290,11 +290,11 @@ pipeline:
   zeek:
     sensors:
       zeek:
-        static_analysis: true
+        static_analysis: false
         protocols:
           - dns
         interfaces:
-          - wlp3s0
+          - eth1
 
 environment:
   kafka_brokers:
@@ -314,40 +314,76 @@ environment:
     # Allow long-running detector batches without Kafka forcing a group rebalance.
     # Default librdkafka value is 300000 ms (5 minutes)
     max_poll_interval_ms: 300000
+  kafka_producer:
+    # Compress JSON pipeline and monitoring batches before replication.
+    compression_type: zstd
   kafka_transaction_batch:
     # Each stage processes this many Kafka records before atomically publishing
     # all outputs and committing the corresponding source offsets.
     size: 250
     timeout_ms: 50
   kafka_topics:
+    # Topics without a stage/exact override are low-volume monitoring topics.
+    # Kafka retention limits apply per partition; keep these topics at one
+    # partition so their aggregate disk footprint remains predictable.
+    partitions: 1
     replication_factor: 3
     auto_expand_partitions: true
+    config:
+      cleanup.policy: delete
+      retention.ms: 900000
+      retention.bytes: 67108864
+      segment.bytes: 16777216
     stages:
       logserver_in:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 7200000
+          retention.bytes: 536870912
       logserver_to_collector:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 1800000
+          retention.bytes: 134217728
       batch_sender_to_prefilter:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 1800000
+          retention.bytes: 134217728
       prefilter_to_inspector:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 1800000
+          retention.bytes: 134217728
       inspector_to_detector:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 1800000
+          retention.bytes: 134217728
       detector_to_alerter:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 3600000
+          retention.bytes: 268435456
       detector_to_detector:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 3600000
+          retention.bytes: 268435456
     topics:
       hamstring_alerts:
-        partitions: 12
+        partitions: 3
         replication_factor: 3
+        config:
+          retention.ms: 86400000
+          retention.bytes: 536870912
   kafka_topics_prefix:
     pipeline:
       logserver_in: "hamstring_input"

docker/clickhouse/config.d/retention.xml

@@ -2,12 +2,26 @@
     <!--
       MergeTree TTL deletion is performed during background merges.  The
       upstream default can defer a TTL merge for four hours; run eligible TTL
-      merges every 15 minutes so monitoring data is reclaimed promptly.
+      merges every five minutes so monitoring data is reclaimed promptly on
+      storage-constrained nodes.
     -->
     <merge_tree>
-        <merge_with_ttl_timeout>900</merge_with_ttl_timeout>
+        <merge_with_ttl_timeout>300</merge_with_ttl_timeout>
     </merge_tree>
 
+    <!--
+      Refuse new ClickHouse allocations before the shared host filesystem is
+      completely exhausted.  This is a safety reserve, not a ClickHouse size
+      quota: use a dedicated filesystem/LVM volume for hard isolation.
+    -->
+    <storage_configuration>
+        <disks>
+            <default>
+                <keep_free_space_bytes>5368709120</keep_free_space_bytes>
+            </default>
+        </disks>
+    </storage_configuration>
+
     <!-- Keep ClickHouse's own diagnostic logs from growing without bound. -->
     <logger>
         <size>100M</size>

docker/create_tables/alerts.sql

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

docker/create_tables/batch_timestamps.sql

@@ -8,6 +8,6 @@ CREATE TABLE IF NOT EXISTS batch_timestamps (
     is_active Bool NOT NULL
 )
 ENGINE = MergeTree
-PARTITION BY toYYYYMM(timestamp)
+PARTITION BY toDate(timestamp)
 ORDER BY (stage, status, timestamp, instance_name, batch_id)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp) + INTERVAL 6 HOUR;

docker/create_tables/batch_tree.sql

@@ -10,6 +10,6 @@ CREATE TABLE IF NOT EXISTS batch_tree (
     timestamp DateTime64(6) NOT NULL
 )
 ENGINE = MergeTree
-PARTITION BY toYYYYMM(timestamp)
+PARTITION BY toDate(timestamp)
 ORDER BY (stage, status, timestamp, instance_name, batch_row_id, parent_batch_row_id)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp) + INTERVAL 6 HOUR;

docker/create_tables/failed_loglines.sql

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

docker/create_tables/fill_levels.sql

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

docker/create_tables/logline_timestamps.sql

@@ -6,6 +6,6 @@ CREATE TABLE IF NOT EXISTS logline_timestamps (
     is_active Bool NOT NULL
 )
 ENGINE = MergeTree
-PARTITION BY toYYYYMM(timestamp)
+PARTITION BY toDate(timestamp)
 ORDER BY (stage, status, timestamp, logline_id)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp) + INTERVAL 6 HOUR;

docker/create_tables/logline_to_batches.sql

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

docker/create_tables/loglines.sql

@@ -6,6 +6,6 @@ CREATE TABLE IF NOT EXISTS loglines (
     additional_fields String
 )
 ENGINE = MergeTree
-PARTITION BY toYYYYMM(timestamp)
+PARTITION BY toDate(timestamp)
 ORDER BY (timestamp, src_ip, subnet_id, logline_id)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp) + INTERVAL 6 HOUR;

docker/create_tables/server_log_terminal_events.sql

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

docker/create_tables/server_log_to_logline.sql

@@ -5,5 +5,5 @@ CREATE TABLE IF NOT EXISTS server_log_to_logline (
 )
 ENGINE = MergeTree
 ORDER BY (timestamp, message_id, logline_id)
-PARTITION BY toYYYYMM(timestamp)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+PARTITION BY toDate(timestamp)
+TTL toDateTime(timestamp) + INTERVAL 6 HOUR;

docker/create_tables/server_logs.sql

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

docker/create_tables/server_logs_timestamps.sql

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

docker/create_tables/suspicious_batch_timestamps.sql

@@ -9,6 +9,6 @@ CREATE TABLE IF NOT EXISTS suspicious_batch_timestamps (
     is_active Bool NOT NULL
 )
 ENGINE = MergeTree
-PARTITION BY toYYYYMM(timestamp)
+PARTITION BY toDate(timestamp)
 ORDER BY (stage, status, timestamp, instance_name, suspicious_batch_id, src_ip)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp) + INTERVAL 6 HOUR;

docker/create_tables/suspicious_batches_to_batch.sql

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

docker/create_tables/zz_monitoring_rollups.sql

@@ -60,7 +60,7 @@ CREATE TABLE IF NOT EXISTS server_log_latencies (
     end_timestamp AggregateFunction(max, DateTime64(6))
 )
 ENGINE = AggregatingMergeTree
-PARTITION BY toYYYYMM(event_date)
+PARTITION BY event_date
 ORDER BY (event_date, message_id)
 TTL toDateTime(event_date) + INTERVAL 1 DAY;
 
@@ -94,7 +94,7 @@ CREATE TABLE IF NOT EXISTS logline_stage_latencies (
     end_timestamp AggregateFunction(max, DateTime64(6))
 )
 ENGINE = AggregatingMergeTree
-PARTITION BY toYYYYMM(event_date)
+PARTITION BY event_date
 ORDER BY (stage, event_date, logline_id)
 TTL toDateTime(event_date) + INTERVAL 1 DAY;
 
@@ -119,7 +119,7 @@ CREATE TABLE IF NOT EXISTS batch_stage_latencies (
     end_timestamp AggregateFunction(max, DateTime64(6))
 )
 ENGINE = AggregatingMergeTree
-PARTITION BY toYYYYMM(event_date)
+PARTITION BY event_date
 ORDER BY (stage, event_date, instance_name, batch_id)
 TTL toDateTime(event_date) + INTERVAL 1 DAY;
 
@@ -145,7 +145,7 @@ CREATE TABLE IF NOT EXISTS suspicious_batch_stage_latencies (
     end_timestamp AggregateFunction(max, DateTime64(6))
 )
 ENGINE = AggregatingMergeTree
-PARTITION BY toYYYYMM(event_date)
+PARTITION BY event_date
 ORDER BY (stage, event_date, instance_name, suspicious_batch_id)
 TTL toDateTime(event_date) + INTERVAL 1 DAY;
 

docker/docker-compose.yml

@@ -142,7 +142,7 @@ services:
     profiles: ["prod", "prod-gpu"]
 
   zeek:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-zeek:${HAMSTRING_ZEEK_IMAGE_TAG:-2.0.4}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-zeek:${HAMSTRING_ZEEK_IMAGE_TAG:-2.0.5}"
     cap_add:
       - NET_ADMIN
       - NET_RAW

docker/docker-compose/base/docker-compose.kafka.yml

@@ -9,18 +9,20 @@ x-kafka-broker-env: &kafka-broker-env
   KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
   KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
   KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+  KAFKA_MIN_INSYNC_REPLICAS: "${KAFKA_MIN_INSYNC_REPLICAS:-2}"
+  KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE: "${KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE:-false}"
   # Retention is evaluated only for closed log segments.  Keep the segment
-  # smaller and roll it frequently so that the four-hour retention period is
-  # effective even for low-volume partitions.
+  # small and roll it frequently so that short topic-level retention periods
+  # are effective even for low-volume partitions.
   #
   # log.retention.bytes applies to *each partition replica*, not to the broker.
-  # With 12 partitions and replication factor 3, a 10 GiB value can consume
+  # With many partitions and replication factor 3, a large value can consume
   # tens of GiB across the three broker volumes.
-  KAFKA_LOG_RETENTION_HOURS: "${KAFKA_LOG_RETENTION_HOURS:-4}"
-  KAFKA_LOG_RETENTION_BYTES: "${KAFKA_LOG_RETENTION_BYTES:-16777216}"
-  KAFKA_LOG_SEGMENT_BYTES: "${KAFKA_LOG_SEGMENT_BYTES:-4194304}"
-  KAFKA_LOG_SEGMENT_MS: "${KAFKA_LOG_SEGMENT_MS:-900000}"
-  KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: "${KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS:-300000}"
+  KAFKA_LOG_RETENTION_HOURS: "${KAFKA_LOG_RETENTION_HOURS:-1}"
+  KAFKA_LOG_RETENTION_BYTES: "${KAFKA_LOG_RETENTION_BYTES:-67108864}"
+  KAFKA_LOG_SEGMENT_BYTES: "${KAFKA_LOG_SEGMENT_BYTES:-16777216}"
+  KAFKA_LOG_SEGMENT_MS: "${KAFKA_LOG_SEGMENT_MS:-300000}"
+  KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: "${KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS:-60000}"
   KAFKA_LOG_CLEANUP_POLICY: "${KAFKA_LOG_CLEANUP_POLICY:-delete}"
 
 services:

docker/docker-compose/prod/docker-compose.monitoring.yml

@@ -1,6 +1,6 @@
 services:
   monitoring_agent:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-monitoring:${HAMSTRING_MONITORING_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-monitoring:${HAMSTRING_MONITORING_IMAGE_TAG:-v2.3.1}"
     restart: "unless-stopped"
     volumes:
       - ../../../config.yaml:/app/config.yaml