Dev

README.md

@@ -87,6 +87,9 @@ 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.
+For transactional consumer behavior, automatic recovery after max-poll or
+group-membership loss, and rebalance handling, see the
+[Kafka consumer recovery guide](./docs/kafka_recovery.rst).
 
 <p align="center">
   <img src="./assets/hamstring_terminal.gif" alt="Terminal example"/>

VERSION

@@ -1 +1 @@
-v2.3.1
+v2.3.2-dev

config.yaml

@@ -134,7 +134,10 @@ pipeline:
       relevance_method: check_dga_relevance
       collector_name: dga_collector
     - name: "domainator_filter"
-      relevance_method: no_relevance_check
+      # Forward only DNS names with at least one label above the registrable
+      # domain. For example: api.example.com passes; example.com and
+      # amazon.co.uk do not.
+      relevance_method: check_domainator_relevance
       collector_name: domainator_collector
 
   data_inspection:
@@ -290,7 +293,7 @@ pipeline:
   zeek:
     sensors:
       zeek:
-        static_analysis: false
+        static_analysis: true
         protocols:
           - dns
         interfaces:
@@ -318,10 +321,20 @@ environment:
     # 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.
+    # Global fallback. Exact topic settings override stage settings, which
+    # override these defaults. Environment variables remain final overrides.
     size: 250
     timeout_ms: 50
+    stages:
+      # CPU-heavy inference must return to Kafka much more frequently.
+      detector:
+        size: 10
+      alerter:
+        size: 25
+    # Keys are complete consumed topic names. A topic can override either value.
+    # topics:
+    #   pipeline-inspector_to_detector-domainator:
+    #     size: 5
   kafka_topics:
     # Topics without a stage/exact override are low-volume monitoring topics.
     # Kafka retention limits apply per partition; keep these topics at one

docs/configuration.rst

@@ -471,7 +471,10 @@ The following list shows the available configuration options.
    * - relevance_method
      - The name of the method used to to check if a given logline is relevant for further inspection.
        This check can be skipped by choosing ``"no_relevance_check"``.
-       Avalable configurations are: ``"no_relevance_check"``, ``"check_dga_relevance"``
+       Available configurations are: ``"no_relevance_check"``, ``"check_dga_relevance"``, and
+       ``"check_domainator_relevance"``. The Domainator method uses an offline Public Suffix List and forwards only
+       names with a subdomain above the registrable domain; for example, ``api.amazon.co.uk`` passes while
+       ``amazon.co.uk`` does not.
    * - collector_name
      - The name of the collector configuration the prefilter consumes data from. The same collector name can be referenced in multiple prefilter configurations.
 
@@ -644,8 +647,17 @@ The following parameters control the infrastructure of the software.
      - Kafka topic name prefixes given as strings. These prefix name are used to construct the actual topic names based on the instance name (e.g. a collector instance name) that produces for the given stage.
        (e.g. a prefilter instance name is added as suffix to the prefilter_to_inspector prefix for the inspector to know where to consume.)
    * - 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.
+     - ``1800000`` fallback; ``300000`` in the supplied ``config.yaml``
+     - Maximum time in milliseconds between Kafka consumer polls before Kafka removes the consumer from its group. Increase this for long-running detector batches and keep batch processing comfortably below the interval. See :doc:`kafka_recovery` for automatic membership recovery, rebalance handling, and delivery guarantees.
+   * - kafka_transaction_batch.size / timeout_ms
+     - ``250`` / ``50`` in the supplied ``config.yaml``
+     - Global Kafka consume/transaction batch defaults. ``size`` counts Kafka records and ``timeout_ms`` controls how long a consumer waits while collecting a batch.
+   * - kafka_transaction_batch.stages
+     - Detector ``size: 10``; alerter ``size: 25``
+     - Per-consuming-stage overrides. Short names such as ``detector`` and fully-qualified names such as ``data_analysis.detector`` are supported.
+   * - kafka_transaction_batch.topics
+     - Domainator input topic ``size: 5``
+     - Per-topic overrides keyed by the complete consumed topic name. Exact-topic settings override stage settings, which override the global defaults. See :doc:`kafka_recovery` for examples and environment-variable precedence.
    * - 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.

docs/index.rst

@@ -24,6 +24,7 @@ Contents
    usage
    pipeline
    monitoring
+   kafka_recovery
    storage_operations
    training
    developer_guide

docs/kafka_recovery.rst

@@ -0,0 +1,144 @@
+Kafka consumer recovery
+=======================
+
+HAMSTRING's pipeline workers use Kafka transactions to commit their output
+records and their consumed input offsets atomically. This prevents a committed
+Kafka output from being separated from the input offset that produced it.
+
+Consumer membership and transactions
+------------------------------------
+
+A Kafka consumer owns an assignment of topic partitions only while it remains
+a valid member of its consumer group. Kafka can change that assignment when a
+worker starts or stops, a service is scaled, a broker or coordinator restarts,
+or a worker exceeds ``environment.kafka_consumer.max_poll_interval_ms``.
+
+If a worker finishes an old batch after losing membership, Kafka rejects
+``send_offsets_to_transaction`` with an error such as
+``UNKNOWN_MEMBER_ID``, ``ILLEGAL_GENERATION``, or
+``REBALANCE_IN_PROGRESS``. The transaction must then be aborted. Delivery
+callbacks can report ``_PURGE_QUEUE`` while records from that aborted
+transaction are discarded; this is a consequence of the abort, not Kafka
+topic retention deleting input records.
+
+Automatic recovery
+------------------
+
+Consumer membership errors do not use HAMSTRING's producer-only retry path.
+HAMSTRING instead performs the following recovery:
+
+#. abort the current Kafka transaction;
+#. discard output records belonging to that transaction;
+#. close the consumer with stale group metadata;
+#. create and subscribe a fresh consumer;
+#. consume again from the group's last committed offsets.
+
+The failed input offsets were not committed, so Kafka can redeliver the batch.
+The replacement transaction uses current group metadata. This recovery is
+shared by the logserver, collector, prefilter, inspector, detector, and alerter
+stages.
+
+HAMSTRING also handles librdkafka's local ``_MAX_POLL_EXCEEDED`` consumer error
+through the same consumer-reset path.
+
+Rebalance tracking
+------------------
+
+Every HAMSTRING consumer registers Kafka's assignment callbacks:
+
+``on_assign``
+   Records the newly owned partitions and starts a new local assignment epoch.
+
+``on_revoke``
+   Invalidates the current epoch when Kafka begins taking partitions away.
+
+``on_lost``
+   Invalidates the current epoch when ownership has already been lost and an
+   offset commit is no longer safe.
+
+Each consumed record carries the local assignment epoch under which it was
+fetched. Before opening a transaction, HAMSTRING verifies that every record in
+the batch still belongs to the current epoch and currently assigned
+partitions. A stale batch is discarded before any output transaction begins;
+the consumer is reset so that its uncommitted records are fetched again.
+
+These callbacks are delivered when the application serves Kafka events through
+``consume`` or ``poll``. They therefore improve correctness and diagnostics
+around observed rebalances, but they do not make unbounded processing safe. A
+CPU-bound detector that does not return to Kafka before
+``max_poll_interval_ms`` can still lose membership. The transaction error
+recovery remains the final safety net for that case.
+
+Preventing repeated membership loss
+-----------------------------------
+
+Choose ``max_poll_interval_ms`` with enough margin for the slowest expected
+batch, and keep the transaction batch small enough that normal processing stays
+well below that interval. The supplied ``config.yaml`` currently selects five
+minutes explicitly; the Python configuration fallback is 30 minutes when the
+setting is omitted. Increasing the interval reduces false membership loss for
+long-running work, but also lets a live yet stuck worker retain its assignment
+for longer.
+
+Remember that ``kafka_transaction_batch.size`` counts Kafka records, not the
+number of application events nested inside each record. Upstream application
+batches may need their own bound when detector work is expensive.
+
+Per-stage and per-topic batch settings
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Transaction batch size and collection timeout can be configured globally, by
+the consuming pipeline stage, or by the complete consumed topic name:
+
+.. code-block:: yaml
+
+   environment:
+     kafka_transaction_batch:
+       size: 250
+       timeout_ms: 50
+       stages:
+         detector:
+           size: 10
+         alerter:
+           size: 25
+       topics:
+         pipeline-inspector_to_detector-domainator:
+           size: 3
+           timeout_ms: 25
+
+The resolution order, from highest to lowest priority, is:
+
+#. ``KAFKA_TRANSACTION_BATCH_SIZE`` and
+   ``KAFKA_TRANSACTION_BATCH_TIMEOUT_MS`` environment variables;
+#. an exact entry under ``topics``;
+#. an entry under ``stages``;
+#. the global ``size`` and ``timeout_ms`` values.
+
+Stage keys can use the short component name, such as ``detector``, or the full
+internal name, such as ``data_analysis.detector``. A full name overrides the
+corresponding short name. Exact topic keys must contain the full topic name
+consumed by the worker. If one consumer subscribes to multiple topics, it uses
+the smallest effective size and timeout across those topics.
+
+The supplied configuration sets the detector stage to 10 records, the alerter
+stage to 25 records, and the initial Domainator topic to 5 records. Other stages
+inherit the global value. A caller that explicitly supplies ``max_messages`` or
+``timeout_ms`` to ``consume_batch`` still overrides the resolved defaults; the
+collector uses this for its application-level batching configuration.
+
+Operationally, alert on:
+
+* ``_MAX_POLL_EXCEEDED`` and ``UNKNOWN_MEMBER_ID``;
+* assignment revocation or loss followed by repeated recovery;
+* transaction aborts while consumer lag increases;
+* no successful output transaction while input offsets continue to arrive.
+
+Delivery guarantees outside Kafka
+---------------------------------
+
+The consume-process-produce transaction provides exactly-once behavior for
+Kafka records and Kafka offsets. File writes, ClickHouse inserts, webhooks, and
+other external actions are not part of that transaction. A recovered batch is
+processed again and can repeat those external effects. Integrations requiring
+deduplication should use a stable identifier such as ``server_message_id`` or
+``logline_id``.

docs/pipeline.rst

@@ -500,6 +500,11 @@ We currently support the following relevance methods:
     |                           | and responses are forwarded that include a  **NXDOMAIN**    |
     |                           | status code.                                                |
     +---------------------------+-------------------------------------------------------------+
+    | ``check_domainator_``     | Public-suffix-aware Domainator filter. Only DNS names with  |
+    | ``relevance``             | at least one label above the registrable domain are         |
+    |                           | forwarded. ``api.amazon.co.uk`` passes; ``amazon.co.uk``    |
+    |                           | and ``google.com`` are filtered out.                        |
+    +---------------------------+-------------------------------------------------------------+
 
 - **Example Configuration**:
 

requirements/requirements.logcollector.txt

@@ -3,3 +3,4 @@ colorlog
 confluent-kafka
 marshmallow_dataclass
 clickhouse_connect
+tldextract

requirements/requirements.prefilter.txt

@@ -3,3 +3,4 @@ colorlog
 confluent-kafka
 marshmallow_dataclass
 clickhouse_connect
+tldextract

src/alerter/alerter.py

@@ -70,7 +70,9 @@ def __init__(self, alerter_config, consume_topic) -> None:
         self.alert_data = None
         self.key = None
 
-        self.kafka_consume_handler = ExactlyOnceKafkaConsumeHandler(self.consume_topic)
+        self.kafka_consume_handler = ExactlyOnceKafkaConsumeHandler(
+            self.consume_topic, stage=module_name
+        )
         self.kafka_produce_handler = None
         self.server_log_terminal_events = ClickHouseKafkaSender(
             "server_log_terminal_events"

src/base/kafka/__init__.py

@@ -7,6 +7,7 @@
     SimpleKafkaConsumeHandler,
 )
 from src.base.kafka.errors import (
+    KafkaConsumerMembershipLost,
     KafkaMessageFetchException,
     TooManyFailedAttemptsError,
 )
@@ -29,6 +30,7 @@
     "ExactlyOnceKafkaConsumeHandler",
     "ExactlyOnceKafkaProduceHandler",
     "KafkaConsumeHandler",
+    "KafkaConsumerMembershipLost",
     "KafkaHandler",
     "KafkaMessageFetchException",
     "KafkaProduceHandler",

src/base/kafka/config.py

@@ -27,6 +27,8 @@
 KAFKA_TRANSACTION_BATCH_CONFIG = CONFIG["environment"].get(
     "kafka_transaction_batch", {}
 )
+KAFKA_TRANSACTION_BATCH_STAGE_CONFIG = KAFKA_TRANSACTION_BATCH_CONFIG.get("stages", {})
+KAFKA_TRANSACTION_BATCH_TOPIC_CONFIG = KAFKA_TRANSACTION_BATCH_CONFIG.get("topics", {})
 KAFKA_TRANSACTION_BATCH_SIZE = int(
     os.getenv(
         "KAFKA_TRANSACTION_BATCH_SIZE",
@@ -40,6 +42,47 @@
     )
 )
 
+
+def _transaction_batch_stage_config(stage: str | None) -> dict:
+    """Resolve short and fully-qualified stage names, preferring the latter."""
+    if not stage:
+        return {}
+    short_stage = str(stage).rsplit(".", 1)[-1]
+    resolved = dict(KAFKA_TRANSACTION_BATCH_STAGE_CONFIG.get(short_stage, {}))
+    resolved.update(KAFKA_TRANSACTION_BATCH_STAGE_CONFIG.get(str(stage), {}))
+    return resolved
+
+
+def transaction_batch_settings(
+    topics: str | list[str], stage: str | None = None
+) -> tuple[int, int]:
+    """Resolve transaction size/timeout by topic, stage, then global defaults."""
+    normalized_topics = [topics] if isinstance(topics, str) else list(topics)
+    stage_config = _transaction_batch_stage_config(stage)
+    global_config = {
+        "size": KAFKA_TRANSACTION_BATCH_CONFIG.get("size", 100),
+        "timeout_ms": KAFKA_TRANSACTION_BATCH_CONFIG.get("timeout_ms", 50),
+    }
+
+    effective_configs = []
+    for topic in normalized_topics or [None]:
+        effective = {**global_config, **stage_config}
+        if topic is not None:
+            effective.update(KAFKA_TRANSACTION_BATCH_TOPIC_CONFIG.get(topic, {}))
+        effective_configs.append(effective)
+
+    # A multi-topic consumer uses the most conservative effective value.
+    size = min(max(1, int(item["size"])) for item in effective_configs)
+    timeout_ms = min(max(0, int(item["timeout_ms"])) for item in effective_configs)
+
+    # Deployment-level environment variables remain explicit final overrides.
+    if "KAFKA_TRANSACTION_BATCH_SIZE" in os.environ:
+        size = max(1, int(os.environ["KAFKA_TRANSACTION_BATCH_SIZE"]))
+    if "KAFKA_TRANSACTION_BATCH_TIMEOUT_MS" in os.environ:
+        timeout_ms = max(0, int(os.environ["KAFKA_TRANSACTION_BATCH_TIMEOUT_MS"]))
+    return size, timeout_ms
+
+
 KAFKA_TOPIC_CONFIG = CONFIG["environment"].get("kafka_topics", {})
 KAFKA_TOPIC_DEFAULT_PARTITIONS = int(
     os.getenv("KAFKA_TOPIC_PARTITIONS", KAFKA_TOPIC_CONFIG.get("partitions", 12))