diff --git a/README.md b/README.md index b8650b60..e38284d0 100644 --- a/README.md +++ b/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 +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`. +

Terminal example

diff --git a/docker/docker-compose/prod/docker-compose.pipeline.yml b/docker/docker-compose/prod/docker-compose.pipeline.yml index 33c82d46..0793f239 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.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 diff --git a/docker/docker_swarm/docker-compose.swarm.yml b/docker/docker_swarm/docker-compose.swarm.yml index f1dfdf79..784c76d1 100644 --- a/docker/docker_swarm/docker-compose.swarm.yml +++ b/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 diff --git a/docs/configuration.rst b/docs/configuration.rst index 8805f126..c9bbfa33 100644 --- a/docs/configuration.rst +++ b/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 + $ 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`` ^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/docs/usage.rst b/docs/usage.rst index 5d67d800..e852ac73 100644 --- a/docs/usage.rst +++ b/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 + $ 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 --------------------------- diff --git a/src/base/clickhouse_kafka_sender.py b/src/base/clickhouse_kafka_sender.py index 514b1f60..56145edb 100644 --- a/src/base/clickhouse_kafka_sender.py +++ b/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) )() diff --git a/src/detector/detector.py b/src/detector/detector.py index dd121f36..287ca6e3 100644 --- a/src/detector/detector.py +++ b/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( diff --git a/src/inspector/inspector.py b/src/inspector/inspector.py index a3e5d8a6..4cdabdd9 100644 --- a/src/inspector/inspector.py +++ b/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( diff --git a/src/logcollector/batch_handler.py b/src/logcollector/batch_handler.py index 2ffdf63c..b5ed112a 100644 --- a/src/logcollector/batch_handler.py +++ b/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) diff --git a/src/logcollector/collector.py b/src/logcollector/collector.py index 7dd1d738..de287ace 100644 --- a/src/logcollector/collector.py +++ b/src/logcollector/collector.py @@ -8,7 +8,9 @@ sys.path.append(os.getcwd()) from src.base.clickhouse_kafka_sender import ClickHouseKafkaSender -from src.base.kafka_handler import ExactlyOnceKafkaConsumeHandler +from src.base.kafka_handler import ( + ExactlyOnceKafkaConsumeHandler, +) from src.base.logline_handler import LoglineHandler from src.base import utils from src.base.execution import ( @@ -69,20 +71,33 @@ def __init__( self.kafka_consume_handler = ExactlyOnceKafkaConsumeHandler(consume_topic) self.batch_configuration = utils.get_batch_configuration(collector_name) self.loglines = asyncio.Queue() + self.monitoring_kafka_producer = ClickHouseKafkaSender.create_shared_producer() self.batch_handler = BufferedBatchSender( - produce_topics=produce_topics, collector_name=collector_name + produce_topics=produce_topics, + collector_name=collector_name, + monitoring_kafka_producer=self.monitoring_kafka_producer, ) self.logline_handler = LoglineHandler(validation_config) # databases - self.failed_protocol_loglines = ClickHouseKafkaSender("failed_loglines") - self.protocol_loglines = ClickHouseKafkaSender("loglines") - self.logline_timestamps = ClickHouseKafkaSender("logline_timestamps") - self.server_log_to_logline = ClickHouseKafkaSender("server_log_to_logline") + self.failed_protocol_loglines = ClickHouseKafkaSender( + "failed_loglines", self.monitoring_kafka_producer + ) + self.protocol_loglines = ClickHouseKafkaSender( + "loglines", self.monitoring_kafka_producer + ) + self.logline_timestamps = ClickHouseKafkaSender( + "logline_timestamps", self.monitoring_kafka_producer + ) + self.server_log_to_logline = ClickHouseKafkaSender( + "server_log_to_logline", self.monitoring_kafka_producer + ) self.server_log_terminal_events = ClickHouseKafkaSender( - "server_log_terminal_events" + "server_log_terminal_events", self.monitoring_kafka_producer + ) + self.fill_levels = ClickHouseKafkaSender( + "fill_levels", self.monitoring_kafka_producer ) - self.fill_levels = ClickHouseKafkaSender("fill_levels") self.fill_levels.insert( dict( diff --git a/src/logserver/server.py b/src/logserver/server.py index 8a48a2f6..66c7755a 100644 --- a/src/logserver/server.py +++ b/src/logserver/server.py @@ -54,10 +54,15 @@ def __init__(self, consume_topic, produce_topics) -> None: self.kafka_consume_handler = ExactlyOnceKafkaConsumeHandler(consume_topic) self.kafka_produce_handler = ExactlyOnceKafkaProduceHandler() + self.monitoring_kafka_producer = ClickHouseKafkaSender.create_shared_producer() # databases - self.server_logs = ClickHouseKafkaSender("server_logs") - self.server_logs_timestamps = ClickHouseKafkaSender("server_logs_timestamps") + self.server_logs = ClickHouseKafkaSender( + "server_logs", self.monitoring_kafka_producer + ) + self.server_logs_timestamps = ClickHouseKafkaSender( + "server_logs_timestamps", self.monitoring_kafka_producer + ) async def start(self) -> None: """Starts the tasks to both fetch messages from Kafka and read them from the input file.""" diff --git a/src/prefilter/prefilter.py b/src/prefilter/prefilter.py index 4e3349de..f8e7b3b7 100644 --- a/src/prefilter/prefilter.py +++ b/src/prefilter/prefilter.py @@ -83,12 +83,21 @@ def __init__( self.logline_handler = LoglineHandler(validation_config) 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.logline_timestamps = ClickHouseKafkaSender("logline_timestamps") - self.fill_levels = ClickHouseKafkaSender("fill_levels") + self.batch_tree = ClickHouseKafkaSender( + "batch_tree", self.monitoring_kafka_producer + ) + self.batch_timestamps = ClickHouseKafkaSender( + "batch_timestamps", 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.fill_levels.insert( dict( diff --git a/tests/clickhouse/test_clickhouse_kafka_sender.py b/tests/clickhouse/test_clickhouse_kafka_sender.py index 9bb5dbc0..d19a44dd 100644 --- a/tests/clickhouse/test_clickhouse_kafka_sender.py +++ b/tests/clickhouse/test_clickhouse_kafka_sender.py @@ -21,6 +21,21 @@ def test_init(self, mock_produce_handler, mock_marshmallow): self.assertEqual(mock_produce_handler_instance, sut.kafka_producer) mock_produce_handler.assert_called_once() + @patch("src.base.clickhouse_kafka_sender.marshmallow_dataclass") + @patch("src.base.clickhouse_kafka_sender.SimpleKafkaProduceHandler") + def test_init_uses_provided_producer(self, mock_produce_handler, mock_marshmallow): + # Arrange + table_name = "test_table" + kafka_producer = object() + + # Act + sut = ClickHouseKafkaSender(table_name, kafka_producer) + + # Assert + self.assertEqual(table_name, sut.table_name) + self.assertIs(kafka_producer, sut.kafka_producer) + mock_produce_handler.assert_not_called() + class TestInsert(unittest.TestCase): @patch("src.base.clickhouse_kafka_sender.marshmallow_dataclass")