diff --git a/README.md b/README.md index e38284d..6d02c5f 100644 --- a/README.md +++ b/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. +

Terminal example

diff --git a/VERSION b/VERSION index b1d18bc..2322eef 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -v2.3.0 +v2.3.1-dev diff --git a/config.yaml b/config.yaml index a54de89..0127379 100644 --- a/config.yaml +++ b/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" diff --git a/docker/clickhouse/config.d/retention.xml b/docker/clickhouse/config.d/retention.xml index 4107339..733091c 100644 --- a/docker/clickhouse/config.d/retention.xml +++ b/docker/clickhouse/config.d/retention.xml @@ -2,12 +2,26 @@ - 900 + 300 + + + + + 5368709120 + + + + 100M diff --git a/docker/create_tables/alerts.sql b/docker/create_tables/alerts.sql index 4e21886..a153ce2 100644 --- a/docker/create_tables/alerts.sql +++ b/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; diff --git a/docker/create_tables/batch_timestamps.sql b/docker/create_tables/batch_timestamps.sql index 57b23e5..920d9c0 100644 --- a/docker/create_tables/batch_timestamps.sql +++ b/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; diff --git a/docker/create_tables/batch_tree.sql b/docker/create_tables/batch_tree.sql index ec8540e..88d2dc6 100644 --- a/docker/create_tables/batch_tree.sql +++ b/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; diff --git a/docker/create_tables/failed_loglines.sql b/docker/create_tables/failed_loglines.sql index 9370205..71a82a1 100644 --- a/docker/create_tables/failed_loglines.sql +++ b/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; diff --git a/docker/create_tables/fill_levels.sql b/docker/create_tables/fill_levels.sql index fc87e1a..6f35e6b 100644 --- a/docker/create_tables/fill_levels.sql +++ b/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; diff --git a/docker/create_tables/logline_timestamps.sql b/docker/create_tables/logline_timestamps.sql index 3ecfd9e..0e7dc93 100644 --- a/docker/create_tables/logline_timestamps.sql +++ b/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; diff --git a/docker/create_tables/logline_to_batches.sql b/docker/create_tables/logline_to_batches.sql index 8d68b65..277a958 100644 --- a/docker/create_tables/logline_to_batches.sql +++ b/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; diff --git a/docker/create_tables/loglines.sql b/docker/create_tables/loglines.sql index 0b01c17..270fef3 100644 --- a/docker/create_tables/loglines.sql +++ b/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; diff --git a/docker/create_tables/server_log_terminal_events.sql b/docker/create_tables/server_log_terminal_events.sql index d4d9dae..01d206e 100644 --- a/docker/create_tables/server_log_terminal_events.sql +++ b/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; diff --git a/docker/create_tables/server_log_to_logline.sql b/docker/create_tables/server_log_to_logline.sql index e96ce7a..56df56f 100644 --- a/docker/create_tables/server_log_to_logline.sql +++ b/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; diff --git a/docker/create_tables/server_logs.sql b/docker/create_tables/server_logs.sql index 8f803cc..0a0e467 100644 --- a/docker/create_tables/server_logs.sql +++ b/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; diff --git a/docker/create_tables/server_logs_timestamps.sql b/docker/create_tables/server_logs_timestamps.sql index 81e5a71..0bce762 100644 --- a/docker/create_tables/server_logs_timestamps.sql +++ b/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; diff --git a/docker/create_tables/suspicious_batch_timestamps.sql b/docker/create_tables/suspicious_batch_timestamps.sql index c1dc908..9e2d15d 100644 --- a/docker/create_tables/suspicious_batch_timestamps.sql +++ b/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; diff --git a/docker/create_tables/suspicious_batches_to_batch.sql b/docker/create_tables/suspicious_batches_to_batch.sql index 4ddfeea..c81a4c1 100644 --- a/docker/create_tables/suspicious_batches_to_batch.sql +++ b/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; diff --git a/docker/create_tables/zz_monitoring_rollups.sql b/docker/create_tables/zz_monitoring_rollups.sql index 38e08c8..f5c127e 100644 --- a/docker/create_tables/zz_monitoring_rollups.sql +++ b/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; diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 840d0e8..5b74a92 100644 --- a/docker/docker-compose.yml +++ b/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 diff --git a/docker/docker-compose/base/docker-compose.kafka.yml b/docker/docker-compose/base/docker-compose.kafka.yml index 7434380..7782a90 100644 --- a/docker/docker-compose/base/docker-compose.kafka.yml +++ b/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: diff --git a/docker/docker-compose/prod/docker-compose.monitoring.yml b/docker/docker-compose/prod/docker-compose.monitoring.yml index 37178aa..4800c5a 100644 --- a/docker/docker-compose/prod/docker-compose.monitoring.yml +++ b/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 diff --git a/docker/docker-compose/prod/docker-compose.pipeline.yml b/docker/docker-compose/prod/docker-compose.pipeline.yml index 33e65d3..e78fc6a 100644 --- a/docker/docker-compose/prod/docker-compose.pipeline.yml +++ b/docker/docker-compose/prod/docker-compose.pipeline.yml @@ -1,6 +1,6 @@ services: logserver: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" networks: hamstring: @@ -10,7 +10,7 @@ services: - GROUP_ID=log_storage - HAMSTRING_WAIT_FOR=kafka logcollector: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" volumes: - ../../../config.yaml:/app/config.yaml @@ -21,7 +21,7 @@ services: - HAMSTRING_WAIT_FOR=kafka prefilter: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" volumes: - ../../../config.yaml:/app/config.yaml @@ -35,7 +35,7 @@ services: - HAMSTRING_WAIT_FOR=kafka inspector: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" volumes: - ../../../config.yaml:/app/config.yaml @@ -57,7 +57,7 @@ services: - HAMSTRING_WAIT_FOR=kafka detector: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" volumes: - ../../../config.yaml:/app/config.yaml @@ -71,7 +71,7 @@ services: - HAMSTRING_WAIT_FOR=kafka detector-gpu: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" volumes: - ../../../config.yaml:/app/config.yaml @@ -93,7 +93,7 @@ services: - NVIDIA_DRIVER_CAPABILITIES=compute,utility alerter: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.3.1}" restart: "unless-stopped" volumes: - ../../../config.yaml:/app/config.yaml diff --git a/docker/docker_swarm/docker-compose.swarm.yml b/docker/docker_swarm/docker-compose.swarm.yml index 5009920..9454b17 100644 --- a/docker/docker_swarm/docker-compose.swarm.yml +++ b/docker/docker_swarm/docker-compose.swarm.yml @@ -11,16 +11,24 @@ x-kafka-broker-env: &kafka-broker-env KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "${KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR:-3}" KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: "${KAFKA_TRANSACTION_STATE_LOG_MIN_ISR:-2}" KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "${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 applies only to closed segments and retention.bytes is per # partition replica. Frequent, small segments make the retention window # effective and keep the replicated pipeline topics bounded on disk. - 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}" +x-default-logging: &default-logging + driver: "${DOCKER_LOG_DRIVER:-local}" + options: + max-size: "${DOCKER_LOG_MAX_SIZE:-10m}" + max-file: "${DOCKER_LOG_MAX_FILE:-3}" + x-hamstring-kafka-wait-env: &hamstring-kafka-wait-env HAMSTRING_WAIT_FOR: kafka HAMSTRING_WAIT_INITIAL_DELAY_SECONDS: "${HAMSTRING_WAIT_INITIAL_DELAY_SECONDS:-30}" @@ -46,7 +54,8 @@ services: KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka1:19092,EXTERNAL://${KAFKA_EXTERNAL_HOST:-localhost}:${KAFKA1_EXTERNAL_PORT:-8097}" KAFKA_LISTENERS: INTERNAL://0.0.0.0:19092,EXTERNAL://0.0.0.0:8097,CONTROLLER://0.0.0.0:29093 volumes: - - kafka-broker-data1:/var/lib/kafka/data + - "${KAFKA1_DATA_SOURCE:-kafka-broker-data1}:/var/lib/kafka/data" + logging: *default-logging deploy: placement: constraints: @@ -69,7 +78,8 @@ services: KAFKA_LISTENERS: INTERNAL://0.0.0.0:19093,EXTERNAL://0.0.0.0:8098,CONTROLLER://0.0.0.0:29093 KAFKA_LOG4J_LOGGERS: "${KAFKA_LOG4J_LOGGERS:-kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO}" volumes: - - kafka-broker-data2:/var/lib/kafka/data + - "${KAFKA2_DATA_SOURCE:-kafka-broker-data2}:/var/lib/kafka/data" + logging: *default-logging deploy: placement: constraints: @@ -92,7 +102,8 @@ services: KAFKA_LISTENERS: INTERNAL://0.0.0.0:19094,EXTERNAL://0.0.0.0:8099,CONTROLLER://0.0.0.0:29093 KAFKA_LOG4J_LOGGERS: "${KAFKA_LOG4J_LOGGERS:-kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO}" volumes: - - kafka-broker-data3:/var/lib/kafka/data + - "${KAFKA3_DATA_SOURCE:-kafka-broker-data3}:/var/lib/kafka/data" + logging: *default-logging deploy: placement: constraints: @@ -119,13 +130,14 @@ services: volumes: - "${HAMSTRING_ROOT:-../..}/docker/create_tables:/docker-entrypoint-initdb.d" - "${HAMSTRING_ROOT:-../..}/docker/clickhouse/config.d/retention.xml:/etc/clickhouse-server/config.d/retention.xml:ro" - - ch_data:/var/lib/clickhouse/ - - ch_logs:/var/log/clickhouse-server/ + - "${CLICKHOUSE_DATA_SOURCE:-ch_data}:/var/lib/clickhouse/" + - "${CLICKHOUSE_LOG_SOURCE:-ch_logs}:/var/log/clickhouse-server/" healthcheck: test: ["CMD-SHELL", "nc -z 127.0.0.1 8123"] interval: 10s timeout: 5s retries: 3 + logging: *default-logging deploy: placement: constraints: @@ -158,6 +170,7 @@ services: timeout: 30s start_period: 120s retries: 3 + logging: *default-logging deploy: placement: constraints: @@ -178,6 +191,7 @@ services: - source: prometheus_config target: /etc/prometheus/prometheus.yml mode: 0444 + logging: *default-logging deploy: placement: constraints: @@ -198,6 +212,7 @@ services: published: "${KAFKA_EXPORTER_PORT:-9308}" protocol: tcp mode: ingress + logging: *default-logging deploy: placement: constraints: @@ -206,7 +221,7 @@ services: condition: any monitoring_agent: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-monitoring:${HAMSTRING_MONITORING_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-monitoring:${HAMSTRING_MONITORING_IMAGE_TAG:-v2.3.1}" command: ["src/monitoring/monitoring_agent.py"] networks: - hamstring @@ -218,6 +233,7 @@ services: HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" CLICKHOUSE_USER: "${CLICKHOUSE_USER:-default}" CLICKHOUSE_PASSWORD: "${CLICKHOUSE_PASSWORD:-hamstring}" + logging: *default-logging deploy: placement: constraints: @@ -226,7 +242,7 @@ services: condition: any logserver: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.3.1}" command: ["src/logserver/server.py"] networks: - hamstring @@ -237,6 +253,7 @@ services: <<: *hamstring-kafka-wait-env GROUP_ID: "${LOGSERVER_GROUP_ID:-log_storage}" HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" + logging: *default-logging deploy: mode: replicated replicas: ${LOGSERVER_REPLICAS:-1} @@ -247,7 +264,7 @@ services: condition: any logcollector: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.3.1}" command: ["src/logcollector/collector.py"] networks: - hamstring @@ -257,6 +274,7 @@ services: <<: *hamstring-kafka-wait-env GROUP_ID: "${LOGCOLLECTOR_GROUP_ID:-log_collection}" HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" + logging: *default-logging deploy: mode: replicated replicas: ${LOGCOLLECTOR_REPLICAS:-1} @@ -267,7 +285,7 @@ services: condition: any prefilter: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.3.1}" command: ["src/prefilter/prefilter.py"] networks: - hamstring @@ -277,6 +295,7 @@ services: <<: *hamstring-kafka-wait-env GROUP_ID: "${PREFILTER_GROUP_ID:-log_filtering}" HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" + logging: *default-logging deploy: mode: replicated replicas: ${PREFILTER_REPLICAS:-1} @@ -287,7 +306,7 @@ services: condition: any inspector: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.3.1}" command: ["src/inspector/inspector.py"] networks: - hamstring @@ -298,6 +317,7 @@ services: GROUP_ID: "${INSPECTOR_GROUP_ID:-data_inspection}" NUMBER_OF_INSTANCES: "${INSPECTOR_NUMBER_OF_INSTANCES:-1}" HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" + logging: *default-logging deploy: mode: replicated replicas: ${INSPECTOR_REPLICAS:-1} @@ -315,7 +335,7 @@ services: condition: any detector: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.3.1}" command: ["src/detector/detector.py"] networks: - hamstring @@ -327,6 +347,7 @@ services: NVIDIA_DRIVER_CAPABILITIES: "${NVIDIA_DRIVER_CAPABILITIES:-compute,utility}" NVIDIA_VISIBLE_DEVICES: "${NVIDIA_VISIBLE_DEVICES:-all}" HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" + logging: *default-logging deploy: mode: replicated replicas: ${DETECTOR_REPLICAS:-1} @@ -344,7 +365,7 @@ services: condition: any alerter: - image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.3.0}" + image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.3.1}" command: ["src/alerter/alerter.py"] networks: - hamstring @@ -355,6 +376,7 @@ services: <<: *hamstring-kafka-wait-env GROUP_ID: "${ALERTER_GROUP_ID:-data_alerting}" HAMSTRING_CONFIG_CHECKSUM: "${HAMSTRING_CONFIG_CHECKSUM:-d9bc812b76c2b58c633b412a6c07f981f3c4d787}" + logging: *default-logging deploy: mode: replicated replicas: ${ALERTER_REPLICAS:-1} @@ -364,7 +386,7 @@ services: restart_policy: condition: any 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 @@ -378,6 +400,7 @@ services: # environment.kafka_brokers[*].node_ip/external_port. CONTAINER_NAME: "${ZEEK_CONTAINER_NAME:-zeek}" HAMSTRING_ZEEK_KAFKA_WAIT_INTERVAL_SECONDS: "${HAMSTRING_ZEEK_KAFKA_WAIT_INTERVAL_SECONDS:-5}" + logging: *default-logging deploy: mode: replicated replicas: ${ZEEK_REPLICAS:-1} diff --git a/docs/configuration.rst b/docs/configuration.rst index c9bbfa3..b9dab39 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -646,18 +646,27 @@ The following parameters control the infrastructure of the software. * - kafka_consumer.max_poll_interval_ms - ``1800000`` - Maximum time in milliseconds between Kafka consumer polls before Kafka removes the consumer from its group. Increase this for long-running detector batches. + * - kafka_producer.compression_type + - ``zstd`` + - Compression codec used by HAMSTRING's Python Kafka producers. Zeek is an external producer and must be configured separately if its image supports compression settings. * - kafka_topics.replication_factor - ``3`` - Replication factor used when creating new Kafka topics. At runtime this is capped to the number of configured Kafka brokers. + * - kafka_topics.partitions + - ``1`` + - Default partition count for topics without a stage or exact-topic override. The supplied storage-constrained profile uses one partition for monitoring topics. + * - kafka_topics.config + - See ``config.yaml`` + - Kafka-native creation settings inherited by every new topic, such as ``cleanup.policy``, ``retention.ms``, ``retention.bytes``, and ``segment.bytes``. Values are passed to Kafka when the topic is first created; existing topic settings are not altered. * - kafka_topics.auto_expand_partitions - ``true`` - If enabled, existing HAMSTRING topics with fewer than the desired partition count are automatically expanded on consumer startup. Kafka does not support shrinking partition counts, so topics that are already larger are left unchanged. * - kafka_topics.stages - See ``config.yaml`` - - Per-pipeline-stage topic settings. Keys match ``environment.kafka_topics_prefix.pipeline`` keys. Each stage can set ``partitions`` and ``replication_factor`` for topics whose names use that stage prefix. + - Per-pipeline-stage topic settings. Keys match ``environment.kafka_topics_prefix.pipeline`` keys. Each stage can set ``partitions``, ``replication_factor``, and a Kafka-native ``config`` mapping for topics whose names use that stage prefix. * - kafka_topics.topics - See ``config.yaml`` - - Exact per-topic settings for topics that are not represented by a pipeline prefix, for example external alert topics. Topics without a stage or exact entry use 12 partitions and the default replication factor. + - Exact per-topic settings for topics that are not represented by a pipeline prefix, for example external alert topics. The entry can set ``partitions``, ``replication_factor``, and a Kafka-native ``config`` mapping. Topics without a stage or exact entry use ``kafka_topics.partitions`` and the default replication factor/configuration. * - monitoring.clickhouse_server.hostname - ``clickhouse-server`` - Hostname of the ClickHouse server. Used by Grafana. @@ -751,8 +760,8 @@ Application service variables - ``1`` - Swarm override for inspector ``NUMBER_OF_INSTANCES``. * - ``KAFKA_TOPIC_PARTITIONS`` - - ``12`` - - Default partition count requested for new HAMSTRING Kafka topics. + - from ``environment.kafka_topics.partitions`` (``1`` in the supplied configuration) + - Default partition count requested for new HAMSTRING Kafka topics without a stage or exact override. * - ``KAFKA_TOPIC_REPLICATION_FACTOR`` - from ``environment.kafka_topics.replication_factor`` - Replication factor requested for new HAMSTRING Kafka topics. At runtime this is capped to the @@ -760,6 +769,9 @@ Application service variables * - ``KAFKA_TOPIC_MIN_PARTITIONS`` - ``1`` - Runtime lower bound for topic partition creation and expansion. + * - ``KAFKA_PRODUCER_COMPRESSION_TYPE`` + - from ``environment.kafka_producer.compression_type`` (``zstd`` in the supplied configuration) + - Compression codec for HAMSTRING's Python Kafka producers. * - ``HAMSTRING_CONFIG_CHECKSUM`` - current compose value - Optional deployment marker used to force Swarm service updates when ``config.yaml`` changes. @@ -885,14 +897,20 @@ Infrastructure variables - ``3`` - Kafka offsets topic replication factor. * - ``KAFKA_LOG_RETENTION_HOURS`` - - ``4`` + - ``1`` - Kafka log retention in hours. * - ``KAFKA_LOG_RETENTION_BYTES`` - - ``10737418240`` - - Kafka log retention size. + - ``67108864`` + - Kafka fallback retention size per partition replica. * - ``KAFKA_LOG_SEGMENT_BYTES`` - - ``1073741824`` + - ``16777216`` - Kafka log segment size. + * - ``KAFKA_LOG_SEGMENT_MS`` + - ``300000`` + - Maximum time before a Kafka segment is rolled. + * - ``KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS`` + - ``60000`` + - Interval between Kafka retention eligibility scans. * - ``KAFKA_LOG_CLEANUP_POLICY`` - ``delete`` - Kafka log cleanup policy. @@ -951,6 +969,15 @@ Path and mount variables - Host file mounted to ``/opt/file.txt`` for the logserver. The exact relative default differs between Compose fragments and the Swarm file but resolves to ``docker/default_input`` from the repository. + * - ``KAFKA1_DATA_SOURCE``, ``KAFKA2_DATA_SOURCE``, ``KAFKA3_DATA_SOURCE`` + - corresponding Kafka named volume + - Swarm volume or host bind-mount source for each broker's ``/var/lib/kafka/data``. Use a dedicated filesystem path for hard disk isolation. + * - ``CLICKHOUSE_DATA_SOURCE`` + - ``ch_data`` + - Swarm volume or host bind-mount source for ``/var/lib/clickhouse``. + * - ``CLICKHOUSE_LOG_SOURCE`` + - ``ch_logs`` + - Swarm volume or host bind-mount source for ClickHouse file logs. * - ``ALERTER_LOGS_PATH`` - ``/opt/logs`` in prod, ``../../../logs`` in dev - Host directory mounted to ``/opt/logs`` by Docker Compose. The Swarm stack uses a named volume @@ -959,9 +986,9 @@ Path and mount variables Swarm scheduling variables ^^^^^^^^^^^^^^^^^^^^^^^^^^ -Docker Swarm placement constraints default to ``node.platform.os == linux``. Set the corresponding -variable to pin a service to a labeled node, for example -``KAFKA1_PLACEMENT_CONSTRAINT='node.labels.kafka1 == true'``. +All services default to ``node.platform.os == linux``. Stateful services can optionally be pinned by overriding their +constraints with ``node.hostname == ``. Verify broker distribution and local-volume ownership +after every deployment. The complete procedure is in :doc:`storage_operations`. .. list-table:: Swarm placement and replica variables :header-rows: 1 diff --git a/docs/index.rst b/docs/index.rst index 93cc0c5..7865179 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -24,6 +24,7 @@ Contents usage pipeline monitoring + storage_operations training developer_guide api/index diff --git a/docs/monitoring.rst b/docs/monitoring.rst index 0c9d8e0..15426df 100644 --- a/docs/monitoring.rst +++ b/docs/monitoring.rst @@ -36,21 +36,22 @@ them with the data in `ClickHouse`. The dashboard views can then be observed on Storage retention ----------------- -HAMSTRING treats Kafka as transient transport. By default each Kafka partition replica has a four-hour / 16 MiB -retention target, whichever limit is reached first. Kafka can delete only *closed* log segments, so the Compose files -also roll a segment every 15 minutes (or after 4 MiB); this makes time-based retention work for low-volume topics. -All Kafka values can be overridden at startup, for example: +HAMSTRING treats Kafka as transient transport. New monitoring topics default to one partition with a 15-minute / 64 MiB +retention target, whichever limit is reached first. Pipeline topics use three partitions and larger class-specific +limits from ``config.yaml``. Kafka can delete only *closed* log segments, so the Compose files also roll a segment +every five minutes (or after 16 MiB); this makes time-based retention work for low-volume topics. Broker fallback +values can be overridden at startup, for example: .. code-block:: console - KAFKA_LOG_RETENTION_HOURS=12 KAFKA_LOG_RETENTION_BYTES=67108864 \\ + KAFKA_LOG_RETENTION_HOURS=2 KAFKA_LOG_RETENTION_BYTES=134217728 \\ HOST_IP=127.0.0.1 docker compose -f docker/docker-compose.yml --profile prod up ``KAFKA_LOG_RETENTION_BYTES`` is a limit **per partition replica**, rather than a per-broker or per-volume limit. Increasing the partition count, replication factor, or this value increases the possible disk usage accordingly. -ClickHouse keeps detailed monitoring events and per-message latency state for one day (raw alerts are retained for -60 days). It then preserves compact monitoring history at progressively lower resolution: +ClickHouse keeps detailed monitoring events for six hours and per-message latency state for one day (raw alerts are +retained for 30 days). It then preserves compact monitoring history at progressively lower resolution: * one-minute aggregates for seven days; * fifteen-minute aggregates for 30 days; @@ -62,9 +63,11 @@ p95, p99, and maximum latency without retaining message, logline, or batch ident views, so selecting an older time range automatically uses the appropriate resolution. Recent latency data remains exact, while latency older than one day is represented by its bucketed p50 value in existing dashboard panels. -TTL deletion happens in background merges; the supplied ClickHouse configuration checks for TTL work every 15 -minutes. ClickHouse server logs are rotated at 100 MiB with three retained files. Increasing a raw-table TTL has a -much larger storage impact than increasing an aggregate-table TTL. +TTL deletion happens in background merges; the supplied ClickHouse configuration checks for TTL work every five +minutes and stops ClickHouse allocations while 5 GiB remains free on its data filesystem. ClickHouse server logs are +rotated at 100 MiB with three retained files. Increasing a raw-table TTL has a much larger storage impact than +increasing an aggregate-table TTL. See :doc:`storage_operations` for the complete disk budget, hard-isolation options, +and the limits of this free-space reserve. The retention schema is reconciled when the monitoring agent starts. New aggregate tables are backfilled once from whatever source data is still available; data already removed by an older TTL cannot be recovered. Retention changes diff --git a/docs/storage_operations.rst b/docs/storage_operations.rst new file mode 100644 index 0000000..d8678d0 --- /dev/null +++ b/docs/storage_operations.rst @@ -0,0 +1,258 @@ +Storage-constrained Swarm deployments +===================================== + +This guide describes the storage policy for a three-node HAMSTRING Swarm in +which each node has approximately 50 GB of local storage. It is intentionally +conservative: Kafka, ClickHouse, container images, container logs, and the +operating system must all fit on the same small hosts unless separate +filesystems are provided. + +Why a node can fill even with Kafka retention +---------------------------------------------- + +Kafka's ``retention.bytes`` is a rolling limit per partition replica, not a +broker-wide disk quota. Kafka also needs active segments, indexes, internal +topics, and temporary deletion headroom. With replication factor three and +three brokers, each broker normally stores one replica of every partition, so +each node carries approximately one complete logical copy of the retained +pipeline. + +Other host storage is independent of Kafka retention. In particular: + +* Docker's unbounded ``json-file`` log driver can create multi-gigabyte + ``*-json.log`` files under ``/var/lib/docker/containers``. +* Container image layers and snapshots may be stored under + ``/var/lib/containerd`` and are not included by a size check limited to + ``/var/lib/docker``. +* ClickHouse TTL deletion happens during background merges and can temporarily + use more space than the steady-state retained data. +* Swarm's default local volumes remain on the node where they were created. + Rescheduling a stateful service can leave old data behind and create a new + local volume elsewhere. + +For a 48 GB usable root filesystem, use the following operating budget: + +.. list-table:: Per-node storage budget + :header-rows: 1 + :widths: 55 20 + + * - Consumer + - Target maximum + * - Operating system, container runtime, and current images + - 10 GB + * - Kafka + - 12 GB + * - ClickHouse on its assigned node + - 10 GB + * - Container and system logs + - 1 GB + * - Other writable data and temporary files + - 5 GB + * - Free emergency headroom + - 10 GB + +The targets are operational thresholds, not guarantees imposed by a single +Kafka or ClickHouse setting. Separate filesystems, LVM logical volumes, or +filesystem quotas are required for hard isolation. + +Stateful service placement +-------------------------- + +Kafka and ClickHouse default to ``node.platform.os == linux`` and can therefore +be scheduled on any Linux node. On a three-node production cluster, verify +after every deployment that the three brokers run on separate physical nodes +and that ClickHouse has not moved away from its intended local data. + +If deterministic placement is desired, override the constraints with the +existing Swarm node hostnames during deployment: + +.. code-block:: console + + export KAFKA1_PLACEMENT_CONSTRAINT='node.hostname == hamstring-1' + export KAFKA2_PLACEMENT_CONSTRAINT='node.hostname == hamstring-2' + export KAFKA3_PLACEMENT_CONSTRAINT='node.hostname == hamstring-3' + export CLICKHOUSE_PLACEMENT_CONSTRAINT='node.hostname == hamstring-1' + +Use the actual Swarm node hostnames. These overrides are optional. Moving a +Kafka node ID onto another node with an unrelated local volume remains unsafe. + +Hard storage isolation +---------------------- + +By default, the stack still uses Docker named volumes under Docker's data +root. On a fresh host, prefer dedicated, quota-controlled mount points and +provide them as bind-mount sources: + +.. code-block:: console + + export KAFKA1_DATA_SOURCE=/srv/hamstring/kafka + export KAFKA2_DATA_SOURCE=/srv/hamstring/kafka + export KAFKA3_DATA_SOURCE=/srv/hamstring/kafka + export CLICKHOUSE_DATA_SOURCE=/srv/hamstring/clickhouse + export CLICKHOUSE_LOG_SOURCE=/srv/hamstring/clickhouse-logs + +Only the broker assigned to a node uses that node's +``/srv/hamstring/kafka`` directory. Create the directories with ownership and +permissions appropriate for the container images before deploying. On the +ClickHouse node, a workable 48 GB layout is an 18 GB root/runtime filesystem, +a 15 GB Kafka filesystem, and a 15 GB ClickHouse filesystem. Keep Kafka below +12 GB and ClickHouse below 10 GB in normal operation; their remaining space is +emergency headroom. Nodes without ClickHouse can leave more space for the +root/runtime filesystem. This layout requires the active container image +footprint to remain within the documented runtime budget. + +Container logging +----------------- + +Every service in ``docker/docker_swarm/docker-compose.swarm.yml`` uses Docker's +``local`` logging driver with these defaults: + +.. code-block:: yaml + + logging: + driver: local + options: + max-size: 10m + max-file: "3" + +This limits ordinary stdout/stderr history to approximately 30 MB per +container before compression. Override the values with +``DOCKER_LOG_DRIVER``, ``DOCKER_LOG_MAX_SIZE``, and ``DOCKER_LOG_MAX_FILE``. +Do not select ``json-file`` without also configuring rotation. + +The stack-level setting protects HAMSTRING services. Set the host-wide Docker +default as well so unrelated containers cannot fill the same filesystem. Add +or merge the following keys in ``/etc/docker/daemon.json`` on every node: + +.. code-block:: json + + { + "log-driver": "local", + "log-opts": { + "max-size": "10m", + "max-file": "3" + } + } + +Restarting Docker is disruptive. Plan the restart one node at a time, and +recreate existing containers afterward because changing the daemon default +does not retrofit their logging configuration. Docker documents the local +driver and its rotation behavior at +https://docs.docker.com/engine/logging/drivers/local/. + +Verify a deployed service's logging policy with: + +.. code-block:: console + + docker service inspect hamstring_zeek \ + --format '{{json .Spec.TaskTemplate.ContainerSpec.LogDriver}}' + +Kafka topic and disk policy +--------------------------- + +``config.yaml`` now supplies Kafka-native topic creation options under each +``config`` mapping. New topics use the following policy: + +.. list-table:: Kafka topic defaults + :header-rows: 1 + :widths: 35 15 20 20 + + * - Topic class + - Partitions + - Byte limit per partition + - Time limit + * - Default monitoring/telemetry topic + - 1 + - 64 MiB + - 15 minutes + * - Zeek input + - 3 + - 512 MiB + - 2 hours + * - Ordinary intermediate stage + - 3 + - 128 MiB + - 30 minutes + * - Detector fan-out/alerter stage + - 3 + - 256 MiB + - 1 hour + * - External alert topic + - 3 + - 512 MiB + - 24 hours + +Both limits apply; Kafka can delete a closed segment when either limit is +exceeded. The recoverable outage is therefore bounded by +``retention bytes / actual topic byte rate`` and may be shorter than the time +limit. If a consumer falls behind the retained window, the missing records +cannot be recovered from Kafka. + +The broker defaults use RF=3, minimum ISR=2, unclean leader election disabled, +16 MiB segments, five-minute segment rolling, and a one-minute retention scan. +This keeps the cluster writable with one broker unavailable while avoiding an +unsafe leader election. + +HAMSTRING's Python producers use Zstandard compression by default to reduce +broker disk and replication traffic for JSON pipeline and monitoring records. +This does not automatically configure the separately packaged Zeek producer; +enable a supported Kafka compression option in that image independently. + +Topic creation settings apply when HAMSTRING creates a topic. They do not +shrink or reconfigure an already existing topic, and Kafka cannot reduce an +existing partition count. Use a fresh cluster or an explicitly planned topic +migration when changing from the old 12-partition layout. + +ClickHouse retention and limits +------------------------------- + +High-cardinality raw monitoring tables retain six hours instead of one day and +use daily partitions so expired data can be reclaimed more efficiently. The +``alerts`` table retains 30 days. Lower-cardinality rollups preserve longer +dashboard history. Eligible TTL merges are checked every five minutes. + +ClickHouse is also configured with ``keep_free_space_bytes=5368709120``. This +causes ClickHouse to refuse new allocations before it consumes the final 5 GiB +visible on its data filesystem. It is not a 10 GB ClickHouse quota, and it +does not reserve that space against Kafka, containerd, or other root processes. +Use the dedicated ClickHouse mount described above for a hard boundary. + +TTL deletion is asynchronous and may briefly require extra merge space. Check +the live table footprint with: + +.. code-block:: sql + + SELECT + database, + table, + formatReadableSize(sum(bytes_on_disk)) AS disk_size + FROM system.parts + WHERE active + GROUP BY database, table + ORDER BY sum(bytes_on_disk) DESC; + +Image and host storage maintenance +---------------------------------- + +Measure the whole runtime footprint, not only Docker volumes: + +.. code-block:: console + + df -h / + du -xhd1 /var/lib | sort -h + du -xhd1 /var/lib/docker | sort -h + du -xhd1 /var/lib/containerd | sort -h + docker system df -v + journalctl --disk-usage + lsof +L1 + +``lsof +L1`` reveals deleted files that are still open and therefore counted +by ``df`` but not by ``du``. Review ``docker system df -v`` after deployments +and remove superseded, unused image versions through a controlled maintenance +procedure. Do not run broad pruning automatically on production nodes without +checking what is reclaimable. + +Alert at 70%, 80%, and 90% filesystem use. At 90%, stop or shed ingress before +the root filesystem reaches 100%; a small local disk cannot guarantee both +indefinite live capture and lossless recovery during an extended downstream +outage. diff --git a/docs/usage.rst b/docs/usage.rst index e852ac7..85033b7 100644 --- a/docs/usage.rst +++ b/docs/usage.rst @@ -42,6 +42,11 @@ policies, resource limits, and placement constraints: Set ``--advertise-addr`` to the manager IP address that worker nodes and published services should use. +The default deployment uses generic Linux-node constraints. Follow +:doc:`storage_operations` to verify stateful-service placement, bound container +logs, and choose dedicated data mounts before starting traffic on +storage-constrained hosts. + Configure the stack with environment variables before deployment. Common options include ``HAMSTRING_IMAGE_REGISTRY`` and per-service image tags, replica counts such as ``LOGCOLLECTOR_REPLICAS`` and ``DETECTOR_REPLICAS``, published ports such diff --git a/src/base/kafka/config.py b/src/base/kafka/config.py index 2988c56..0a91ffa 100644 --- a/src/base/kafka/config.py +++ b/src/base/kafka/config.py @@ -18,6 +18,11 @@ KAFKA_CONSUMER_MAX_POLL_INTERVAL_MS = int( KAFKA_CONSUMER_CONFIG.get("max_poll_interval_ms", 1_800_000) ) +KAFKA_PRODUCER_CONFIG = CONFIG["environment"].get("kafka_producer", {}) +KAFKA_PRODUCER_COMPRESSION_TYPE = os.getenv( + "KAFKA_PRODUCER_COMPRESSION_TYPE", + KAFKA_PRODUCER_CONFIG.get("compression_type", "zstd"), +) KAFKA_TRANSACTION_BATCH_CONFIG = CONFIG["environment"].get( "kafka_transaction_batch", {} @@ -36,7 +41,9 @@ ) KAFKA_TOPIC_CONFIG = CONFIG["environment"].get("kafka_topics", {}) -KAFKA_TOPIC_DEFAULT_PARTITIONS = int(os.getenv("KAFKA_TOPIC_PARTITIONS", 12)) +KAFKA_TOPIC_DEFAULT_PARTITIONS = int( + os.getenv("KAFKA_TOPIC_PARTITIONS", KAFKA_TOPIC_CONFIG.get("partitions", 12)) +) KAFKA_TOPIC_REPLICATION_FACTOR = int( os.getenv( "KAFKA_TOPIC_REPLICATION_FACTOR", @@ -46,6 +53,7 @@ KAFKA_TOPIC_AUTO_EXPAND_PARTITIONS = KAFKA_TOPIC_CONFIG.get( "auto_expand_partitions", True ) +KAFKA_TOPIC_DEFAULT_CONFIG = KAFKA_TOPIC_CONFIG.get("config", {}) KAFKA_TOPIC_STAGE_CONFIG = KAFKA_TOPIC_CONFIG.get("stages", {}) KAFKA_TOPIC_EXACT_CONFIG = KAFKA_TOPIC_CONFIG.get("topics", {}) KAFKA_PIPELINE_TOPIC_PREFIXES = ( diff --git a/src/base/kafka/producer.py b/src/base/kafka/producer.py index 4ea0ae9..ca57faf 100644 --- a/src/base/kafka/producer.py +++ b/src/base/kafka/producer.py @@ -95,6 +95,7 @@ def __init__(self): "bootstrap.servers": self.brokers, "enable.idempotence": False, "acks": "1", + "compression.type": kafka_config.KAFKA_PRODUCER_COMPRESSION_TYPE, "message.max.bytes": 1_000_000_000, } ) @@ -141,6 +142,7 @@ def __init__(self): "bootstrap.servers": self.brokers, "enable.idempotence": False, "acks": "1", + "compression.type": kafka_config.KAFKA_PRODUCER_COMPRESSION_TYPE, "message.max.bytes": 1_000_000_000, "linger.ms": 10, "batch.num.messages": 1000, @@ -190,6 +192,7 @@ def __init__(self): "bootstrap.servers": self.brokers, "transactional.id": f"{kafka_config.HOSTNAME}-{uuid.uuid4()}", "enable.idempotence": True, + "compression.type": kafka_config.KAFKA_PRODUCER_COMPRESSION_TYPE, "message.max.bytes": 1_000_000_000, } ) diff --git a/src/base/kafka/topics.py b/src/base/kafka/topics.py index 258b7e3..af4a990 100644 --- a/src/base/kafka/topics.py +++ b/src/base/kafka/topics.py @@ -87,6 +87,16 @@ def _topic_replication_factor( return min(configured_replication_factor, broker_count) +def _topic_creation_config(topic: str | None = None) -> dict[str, str]: + """Return Kafka-native creation settings with per-topic overrides applied.""" + creation_config = dict(kafka_config.KAFKA_TOPIC_DEFAULT_CONFIG) + creation_config.update(_topic_config(topic).get("config", {})) + return { + str(key): str(value).lower() if isinstance(value, bool) else str(value) + for key, value in creation_config.items() + } + + def topic_partition_count(cluster_metadata, topic: str) -> int | None: topics_metadata = getattr(cluster_metadata, "topics", {}) if isinstance(topics_metadata, dict): @@ -120,6 +130,9 @@ def ensure( topic: _topic_replication_factor(topic, replication_factor) for topic in normalized_topics } + config_by_topic = { + topic: _topic_creation_config(topic) for topic in normalized_topics + } should_expand = ( _as_bool(kafka_config.KAFKA_TOPIC_AUTO_EXPAND_PARTITIONS) if auto_expand_partitions is None @@ -151,6 +164,7 @@ def ensure( topic, target_by_topic[topic], replication_by_topic[topic], + config=config_by_topic[topic], ) for topic in missing_topics ] diff --git a/tests/kafka/test_exactly_once_kafka_produce_handler.py b/tests/kafka/test_exactly_once_kafka_produce_handler.py index 8631b09..10506bb 100644 --- a/tests/kafka/test_exactly_once_kafka_produce_handler.py +++ b/tests/kafka/test_exactly_once_kafka_produce_handler.py @@ -35,6 +35,7 @@ def test_init(self, mock_producer, mock_uuid): "bootstrap.servers": "127.0.0.1:9999,127.0.0.2:9998,127.0.0.3:9997", "transactional.id": f"test_transactional_id-{mock_uuid.uuid4.return_value}", "enable.idempotence": True, + "compression.type": "zstd", "message.max.bytes": 1000000000, } @@ -79,6 +80,7 @@ def test_init_retries_until_transactions_initialize( "bootstrap.servers": "127.0.0.1:9999,127.0.0.2:9998,127.0.0.3:9997", "transactional.id": f"default_tid-{mock_uuid.uuid4.return_value}", "enable.idempotence": True, + "compression.type": "zstd", "message.max.bytes": 1000000000, } diff --git a/tests/kafka/test_kafka_consume_handler.py b/tests/kafka/test_kafka_consume_handler.py index d904b06..1683749 100644 --- a/tests/kafka/test_kafka_consume_handler.py +++ b/tests/kafka/test_kafka_consume_handler.py @@ -12,6 +12,7 @@ from src.base.kafka.topics import ( _desired_topic_partitions, _topic_config, + _topic_creation_config, _topic_replication_factor, ) @@ -45,6 +46,7 @@ def test_build_consumer_group_id_for_multiple_topics(self): class TestTopicReconciliation(unittest.TestCase): + @patch("src.base.kafka.config.KAFKA_TOPIC_DEFAULT_CONFIG", {}) @patch("src.base.kafka.topics.NewTopic") def test_missing_topic_is_created_with_target_partitions(self, mock_new_topic): admin_client = MagicMock() @@ -53,10 +55,13 @@ def test_missing_topic_is_created_with_target_partitions(self, mock_new_topic): _metadata({"test_topic": 4}), ] admin_client.create_topics.return_value = {"test_topic": MagicMock()} - mock_new_topic.side_effect = lambda topic, partitions, replication_factor: ( - topic, - partitions, - replication_factor, + mock_new_topic.side_effect = ( + lambda topic, partitions, replication_factor, config: ( + topic, + partitions, + replication_factor, + config, + ) ) target_partitions_by_topic = ensure_topics( @@ -67,7 +72,7 @@ def test_missing_topic_is_created_with_target_partitions(self, mock_new_topic): ) self.assertEqual({"test_topic": 4}, target_partitions_by_topic) - admin_client.create_topics.assert_called_once_with([("test_topic", 4, 2)]) + admin_client.create_topics.assert_called_once_with([("test_topic", 4, 2, {})]) admin_client.create_partitions.assert_not_called() @patch("src.base.kafka.topics.NewPartitions") @@ -188,6 +193,32 @@ def test_exact_topic_config_wins(self): self.assertEqual(5, _desired_topic_partitions("hamstring_alerts")) self.assertEqual(2, _topic_replication_factor("hamstring_alerts")) + @patch( + "src.base.kafka.config.KAFKA_TOPIC_DEFAULT_CONFIG", + {"cleanup.policy": "delete", "retention.ms": 900000}, + ) + @patch( + "src.base.kafka.config.KAFKA_PIPELINE_TOPIC_PREFIXES", + {"logserver_in": "hamstring_input"}, + ) + @patch( + "src.base.kafka.config.KAFKA_TOPIC_STAGE_CONFIG", + { + "logserver_in": { + "config": {"retention.ms": 7200000, "retention.bytes": 536870912} + } + }, + ) + def test_topic_creation_config_merges_defaults_and_stringifies_values(self): + self.assertEqual( + { + "cleanup.policy": "delete", + "retention.ms": "7200000", + "retention.bytes": "536870912", + }, + _topic_creation_config("hamstring_input-dns"), + ) + class TestInit(unittest.TestCase): @patch("src.base.kafka.config.CONSUMER_GROUP_ID", "test_group_id") diff --git a/tests/kafka/test_simple_kafka_produce_handler.py b/tests/kafka/test_simple_kafka_produce_handler.py index afe086b..2f5871b 100644 --- a/tests/kafka/test_simple_kafka_produce_handler.py +++ b/tests/kafka/test_simple_kafka_produce_handler.py @@ -33,6 +33,7 @@ def test_successful(self): "bootstrap.servers": "127.0.0.1:9999,127.0.0.2:9998,127.0.0.3:9997", "enable.idempotence": False, "acks": "1", + "compression.type": "zstd", "message.max.bytes": 1000000000, }