Dev

README.md

@@ -65,6 +65,23 @@ HOST_IP=127.0.0.1 docker compose -f docker/docker-compose.yml --profile prod up
 ```sh
 HOST_IP=127.0.0.1 docker compose -f docker/docker-compose.yml --profile dev up
 ```
+
+#### Deploy **HAMSTRING** with Docker Swarm:
+Use the Swarm stack file when you want to use Hamstring in a production grade environment,
+as it provides scaling and error handling capabilities:
+
+```sh
+docker swarm init --advertise-addr <manager-ip>
+HAMSTRING_ROOT="$PWD" docker stack deploy -c docker/docker_swarm/docker-compose.swarm.yml hamstring
+```
+Set `--advertise-addr` to the manager IP address that worker nodes and published
+services should use.
+Set image tags, replica counts, ports, and placement constraints through
+environment variables such as `HAMSTRING_IMAGE_REGISTRY`,
+`HAMSTRING_DETECTOR_IMAGE_TAG`, `DETECTOR_REPLICAS`, `GRAFANA_PORT`, and
+`DETECTOR_PLACEMENT_CONSTRAINT`. Remove the stack with
+`docker stack rm hamstring`.
+
 <p align="center">
   <img src="./assets/hamstring_terminal.gif" alt="Terminal example"/>
 </p>

VERSION

@@ -1 +1 @@
-v2.2.0
+v2.2.1-dev

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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.2.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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.2.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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.2.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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.2.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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.2.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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.2.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.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.2.1}"
     restart: "unless-stopped"
     volumes:
       - ../../../config.yaml:/app/config.yaml

docker/docker_swarm/docker-compose.swarm.yml

@@ -200,7 +200,7 @@ services:
         condition: any
 
   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.2.1}"
     command: ["src/monitoring/monitoring_agent.py"]
     networks:
       - hamstring
@@ -220,7 +220,7 @@ services:
         condition: any
 
   logserver:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logserver:${HAMSTRING_LOGSERVER_IMAGE_TAG:-v2.2.1}"
     command: ["src/logserver/server.py"]
     networks:
       - hamstring
@@ -241,7 +241,7 @@ services:
         condition: any
 
   logcollector:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-logcollector:${HAMSTRING_LOGCOLLECTOR_IMAGE_TAG:-v2.2.1}"
     command: ["src/logcollector/collector.py"]
     networks:
       - hamstring
@@ -261,7 +261,7 @@ services:
         condition: any
 
   prefilter:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-prefilter:${HAMSTRING_PREFILTER_IMAGE_TAG:-v2.2.1}"
     command: ["src/prefilter/prefilter.py"]
     networks:
       - hamstring
@@ -281,7 +281,7 @@ services:
         condition: any
 
   inspector:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-inspector:${HAMSTRING_INSPECTOR_IMAGE_TAG:-v2.2.1}"
     command: ["src/inspector/inspector.py"]
     networks:
       - hamstring
@@ -309,7 +309,7 @@ services:
         condition: any
 
   detector:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-detector:${HAMSTRING_DETECTOR_IMAGE_TAG:-v2.2.1}"
     command: ["src/detector/detector.py"]
     networks:
       - hamstring
@@ -338,7 +338,7 @@ services:
         condition: any
 
   alerter:
-    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.2.0}"
+    image: "${HAMSTRING_IMAGE_REGISTRY:-ghcr.io/astraos-de}/hamstring-alerter:${HAMSTRING_ALERTER_IMAGE_TAG:-v2.2.1}"
     command: ["src/alerter/alerter.py"]
     networks:
       - hamstring

docs/configuration.rst

@@ -377,6 +377,24 @@ flag and keep ``NUMBER_OF_INSTANCES`` aligned with the replica count:
 With this example and the hybrid detector config shown above, the detector starts
 ``2 Docker replicas * 2 processes * 4 threads_per_process = 16`` Kafka consumers.
 
+For Docker Swarm, deploy the dedicated stack file instead of the local Compose
+profiles:
+
+.. code-block:: console
+
+   $ docker swarm init --advertise-addr <manager-ip>
+   $ HAMSTRING_ROOT="$PWD" docker stack deploy -c docker/docker_swarm/docker-compose.swarm.yml hamstring
+
+Set ``--advertise-addr`` to the manager IP address that worker nodes and
+published services should use.
+
+The Swarm file reads replica counts such as ``LOGSERVER_REPLICAS``,
+``LOGCOLLECTOR_REPLICAS``, ``PREFILTER_REPLICAS``, ``INSPECTOR_REPLICAS``,
+``DETECTOR_REPLICAS``, ``ALERTER_REPLICAS``, and ``ZEEK_REPLICAS``. It also
+accepts image tags, published ports, and placement constraints through
+environment variables, for example ``HAMSTRING_DETECTOR_IMAGE_TAG``,
+``GRAFANA_PORT``, and ``DETECTOR_PLACEMENT_CONSTRAINT``.
+
 ``pipeline.log_storage``
 ^^^^^^^^^^^^^^^^^^^^^^^^
 

docs/usage.rst

@@ -27,6 +27,31 @@ For local development builds, use the development profile:
 Set ``HOST_IP`` to the host address that external Kafka clients should use. The default
 ``localhost`` works for single-host local runs.
 
+Docker Swarm Deployment
+-----------------------
+
+Use ``docker/docker_swarm/docker-compose.swarm.yml`` when deploying HAMSTRING as
+a Swarm stack. Swarm honors the ``deploy`` sections for replicas, restart
+policies, resource limits, and placement constraints:
+
+.. code-block:: console
+
+   $ docker swarm init --advertise-addr <manager-ip>
+   $ HAMSTRING_ROOT="$PWD" docker stack deploy -c docker/docker_swarm/docker-compose.swarm.yml hamstring
+
+Set ``--advertise-addr`` to the manager IP address that worker nodes and
+published services should use.
+
+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
+as ``GRAFANA_PORT`` and ``PROMETHEUS_PORT``, and placement constraints such as
+``DETECTOR_PLACEMENT_CONSTRAINT``. Remove the stack with:
+
+.. code-block:: console
+
+   $ docker stack rm hamstring
+
 Scaling With Docker Compose
 ---------------------------
 

src/base/clickhouse_kafka_sender.py

@@ -24,7 +24,15 @@ class ClickHouseKafkaSender:
     serialization for the specified ClickHouse table.
     """
 
-    def __init__(self, table_name: str):
+    @staticmethod
+    def create_shared_producer() -> SimpleKafkaProduceHandler:
+        return SimpleKafkaProduceHandler()
+
+    def __init__(
+        self,
+        table_name: str,
+        kafka_producer: SimpleKafkaProduceHandler | None = None,
+    ):
         """
         Args:
             table_name (str): Name of the ClickHouse table to send insert operations for.
@@ -33,7 +41,7 @@ def __init__(self, table_name: str):
             KeyError: If the specified table name is not found in TABLE_NAME_TO_TYPE mapping.
         """
         self.table_name = table_name
-        self.kafka_producer = SimpleKafkaProduceHandler()
+        self.kafka_producer = kafka_producer or SimpleKafkaProduceHandler()
         self.data_schema = marshmallow_dataclass.class_schema(
             TABLE_NAME_TO_TYPE.get(table_name)
         )()

src/detector/detector.py

@@ -241,15 +241,22 @@ def __init__(
         self.kafka_produce_handler = None
 
         self.model, self.scaler = self._get_model()
+        self.monitoring_kafka_producer = ClickHouseKafkaSender.create_shared_producer()
 
         # databases
-        self.batch_tree = ClickHouseKafkaSender("batch_tree")
+        self.batch_tree = ClickHouseKafkaSender(
+            "batch_tree", self.monitoring_kafka_producer
+        )
         self.suspicious_batch_timestamps = ClickHouseKafkaSender(
-            "suspicious_batch_timestamps"
+            "suspicious_batch_timestamps", self.monitoring_kafka_producer
+        )
+        self.alerts = ClickHouseKafkaSender("alerts", self.monitoring_kafka_producer)
+        self.logline_timestamps = ClickHouseKafkaSender(
+            "logline_timestamps", self.monitoring_kafka_producer
+        )
+        self.fill_levels = ClickHouseKafkaSender(
+            "fill_levels", self.monitoring_kafka_producer
         )
-        self.alerts = ClickHouseKafkaSender("alerts")
-        self.logline_timestamps = ClickHouseKafkaSender("logline_timestamps")
-        self.fill_levels = ClickHouseKafkaSender("fill_levels")
 
         self.fill_levels.insert(
             dict(

src/inspector/inspector.py

@@ -117,18 +117,27 @@ def __init__(self, consume_topic, produce_topics, config) -> None:
 
         self.kafka_consume_handler = ExactlyOnceKafkaConsumeHandler(self.consume_topic)
         self.kafka_produce_handler = ExactlyOnceKafkaProduceHandler()
+        self.monitoring_kafka_producer = ClickHouseKafkaSender.create_shared_producer()
 
         # databases
-        self.batch_tree = ClickHouseKafkaSender("batch_tree")
-        self.batch_timestamps = ClickHouseKafkaSender("batch_timestamps")
+        self.batch_tree = ClickHouseKafkaSender(
+            "batch_tree", self.monitoring_kafka_producer
+        )
+        self.batch_timestamps = ClickHouseKafkaSender(
+            "batch_timestamps", self.monitoring_kafka_producer
+        )
         self.suspicious_batch_timestamps = ClickHouseKafkaSender(
-            "suspicious_batch_timestamps"
+            "suspicious_batch_timestamps", self.monitoring_kafka_producer
         )
         self.suspicious_batches_to_batch = ClickHouseKafkaSender(
-            "suspicious_batches_to_batch"
+            "suspicious_batches_to_batch", self.monitoring_kafka_producer
+        )
+        self.logline_timestamps = ClickHouseKafkaSender(
+            "logline_timestamps", self.monitoring_kafka_producer
+        )
+        self.fill_levels = ClickHouseKafkaSender(
+            "fill_levels", self.monitoring_kafka_producer
         )
-        self.logline_timestamps = ClickHouseKafkaSender("logline_timestamps")
-        self.fill_levels = ClickHouseKafkaSender("fill_levels")
 
         self.fill_levels.insert(
             dict(

src/logcollector/batch_handler.py

@@ -33,17 +33,32 @@ class BufferedBatch:
     chronological processing. Tracks batch metadata including IDs, timestamps, and fill levels for monitoring.
     """
 
-    def __init__(self, collector_name):
+    def __init__(
+        self,
+        collector_name,
+        monitoring_kafka_producer=None,
+    ):
         self.name = f"buffered-batch-for-{collector_name}"
         self.batch = {}  # Batch for the latest messages coming in
         self.buffer = {}  # Former batch with previous messages
         self.batch_id = {}  # Batch ID per key
+        self.monitoring_kafka_producer = (
+            monitoring_kafka_producer or ClickHouseKafkaSender.create_shared_producer()
+        )
 
         # databases
-        self.logline_to_batches = ClickHouseKafkaSender("logline_to_batches")
-        self.batch_timestamps = ClickHouseKafkaSender("batch_timestamps")
-        self.fill_levels = ClickHouseKafkaSender("fill_levels")
-        self.batch_tree = ClickHouseKafkaSender("batch_tree")
+        self.logline_to_batches = ClickHouseKafkaSender(
+            "logline_to_batches", self.monitoring_kafka_producer
+        )
+        self.batch_timestamps = ClickHouseKafkaSender(
+            "batch_timestamps", self.monitoring_kafka_producer
+        )
+        self.fill_levels = ClickHouseKafkaSender(
+            "fill_levels", self.monitoring_kafka_producer
+        )
+        self.batch_tree = ClickHouseKafkaSender(
+            "batch_tree", self.monitoring_kafka_producer
+        )
         self.fill_levels.insert(
             dict(
                 timestamp=datetime.datetime.now(),
@@ -431,16 +446,26 @@ class BufferedBatchSender:
     via Kafka and tracks message timestamps for monitoring and debugging purposes.
     """
 
-    def __init__(self, produce_topics, collector_name):
+    def __init__(
+        self,
+        produce_topics,
+        collector_name,
+        monitoring_kafka_producer=None,
+    ):
         self.topics = produce_topics
         self.batch_configuration = get_batch_configuration(collector_name)
         self.timer = None
-        self.batch = BufferedBatch(collector_name)
+        self.monitoring_kafka_producer = (
+            monitoring_kafka_producer or ClickHouseKafkaSender.create_shared_producer()
+        )
+        self.batch = BufferedBatch(collector_name, self.monitoring_kafka_producer)
 
         self.kafka_produce_handler = ExactlyOnceKafkaProduceHandler()
 
         # databases
-        self.logline_timestamps = ClickHouseKafkaSender("logline_timestamps")
+        self.logline_timestamps = ClickHouseKafkaSender(
+            "logline_timestamps", self.monitoring_kafka_producer
+        )
 
     def __del__(self):
         timer = getattr(self, "timer", None)