From b10c226e4e791704e1a840a01762163fd35f8116 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 12:25:29 -0300 Subject: [PATCH] SAM0123-231: Remove Kafka integration from Ingestor service. --- ingestor/ingestor.py | 14 +- ingestor/managers/data_manager.py | 160 +-------------- ingestor/managers/ingestor_manager.py | 10 +- ingestor/metrics.py | 17 -- tests/unit/managers/test_data_manager.py | 195 +------------------ tests/unit/managers/test_ingestor_manager.py | 6 - tests/unit/test_ingestor.py | 6 - tests/unit/test_metrics.py | 24 --- 8 files changed, 18 insertions(+), 414 deletions(-) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 4ddb063..2244d61 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -19,7 +19,7 @@ class Ingestor(SientiaMonitoring): - Managing slot leases for load balancing across multiple instances - Connecting to and monitoring OPC UA servers - Subscribing to OPC tags and collecting real-time data - - Distributing data to Kafka and MongoDB + - Distributing data to MongoDB - Providing health monitoring and metrics collection The ingestor uses a slot-based architecture where each slot represents @@ -27,8 +27,6 @@ class Ingestor(SientiaMonitoring): This allows for horizontal scaling and load distribution. Environment Variables: - KAFKA_SERVERS: Comma-separated list of Kafka server addresses (default: "localhost:9092") - EXPORT_TO_KAFKA: Enable/disable Kafka export (default: "false") REDIS_HOST: Redis server hostname (default: "localhost") REDIS_PORT: Redis server port (default: 6379) REDIS_USERNAME: Redis username (optional) @@ -43,7 +41,6 @@ class Ingestor(SientiaMonitoring): MONGODB_DATABASE: MongoDB database name (default: "sientia") Attributes: - export_to_kafka (bool): Whether to export data to Kafka redis_host (str): Redis server hostname redis_port (int): Redis server port redis_username (str): Redis username (optional) @@ -54,7 +51,6 @@ class Ingestor(SientiaMonitoring): poll_interval (int): Interval in seconds for polling operations mongo_database (str): MongoDB database name mongo_connection_string (str): Complete MongoDB connection string - kafka_servers (list): List of Kafka server addresses logger: Logger instance for application logging notification_handler: Handler for sending notifications metadata (dict): Application metadata for notifications and tracking @@ -66,17 +62,12 @@ class Ingestor(SientiaMonitoring): Initializes the ingestor with configuration values retrieved from environment variables. Sets up all necessary connections and configurations for: - - Kafka connectivity (if enabled) - Redis for slot management and coordination - MongoDB for data persistence and notifications - OPC UA server management - Metrics collection and monitoring """ - kafka_servers = getenv('KAFKA_SERVERS', 'localhost:9092') - export_to_kafka: bool = getenv('EXPORT_TO_KAFKA', 'false') == 'true' - - self.export_to_kafka = export_to_kafka self.redis_host = getenv('REDIS_HOST', 'localhost') self.redis_port = int(getenv('REDIS_PORT', '6379')) self.redis_username = getenv('REDIS_USERNAME', None) @@ -91,7 +82,6 @@ class Ingestor(SientiaMonitoring): self.mongo_database = getenv('MONGODB_DATABASE', 'sientia') self.mongo_connection_string = f'mongodb://{mongo_username}:{mongo_password}@{mongo_url}' - self.kafka_servers = kafka_servers.split(',') self.logger = get_logger(__name__) self.notification_handler = NotificationHandler( connection_string=self.mongo_connection_string, @@ -179,7 +169,6 @@ class Ingestor(SientiaMonitoring): """ self.ingestor_manager = IngestorManager( - kafka_servers=','.join(self.kafka_servers), redis_data={ 'host': self.redis_host, 'port': self.redis_port, @@ -195,7 +184,6 @@ class Ingestor(SientiaMonitoring): logger=self.logger, notification_handler=self.notification_handler, metrics_controller=self.metrics_controller, - export_to_kafka=self.export_to_kafka, ) assert self.ingestor_manager is not None diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 44d248e..ef3b5ac 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -1,10 +1,6 @@ -import json import os import traceback -from time import sleep -from kafka import KafkaProducer -from kafka.errors import NoBrokersAvailable from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger @@ -18,31 +14,22 @@ import ingestor.metrics as metrics class DataManager(SientiaMonitoring): """ - Manages data persistence and export operations for the OPC Ingestor. + Manages data persistence operations for the OPC Ingestor. The DataManager is responsible for: - Storing OPC data in MongoDB for historical analysis and persistence - - Exporting data to Kafka for real-time streaming and downstream processing - Managing database connections and ensuring data integrity - Providing data access interfaces for other components - The manager supports both MongoDB and Kafka operations, with Kafka export - being optional and configurable. It implements retry logic for connection - failures and provides comprehensive error handling and notification. - Args: - kafka_servers (str): Comma-separated string of Kafka server addresses mongo_connection_string (str): MongoDB connection string mongo_database (str): MongoDB database name - export_to_kafka (bool): Whether to enable Kafka export functionality metadata (dict): Application metadata for notifications and tracking logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for sending notifications Attributes: pod_id (str): Pod identifier for metrics labeling - kafka_producer (KafkaProducer): Kafka producer instance for data export - export_to_kafka (bool): Whether Kafka export is enabled connection_string (str): MongoDB connection string database (str): MongoDB database name mongo_client (MongoClient): MongoDB client instance @@ -51,47 +38,25 @@ class DataManager(SientiaMonitoring): def __init__( self, - kafka_servers: str, mongo_connection_string: str, mongo_database: str, - export_to_kafka: bool, metadata: dict, logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, ) -> None: """ - Initializes the DataManager instance with Kafka and MongoDB connections. - - This constructor attempts to establish connections to the specified services: - 1. Kafka: Initializes producer with retry logic (up to 3 attempts) - 2. MongoDB: Establishes connection and verifies server availability - - The initialization process includes: - - Kafka producer setup with JSON serialization - - MongoDB client initialization and connection testing - - Metrics recording for connection status - - Error handling with notifications + Initializes the DataManager instance with a MongoDB connection. Args: - kafka_servers (str): Comma-separated string of Kafka server addresses mongo_connection_string (str): MongoDB connection string mongo_database (str): MongoDB database name - export_to_kafka (bool): Whether to enable Kafka export metadata (dict): Application metadata logger (Logger): Logger instance notification_handler (NotificationHandler): Notification handler - - Raises: - NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts. - - Metrics: - - KAFKA_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure """ self.pod_id = os.getenv('HOSTNAME', 'localhost') - self.kafka_producer = None - self.export_to_kafka = export_to_kafka SientiaMonitoring.__init__( self, @@ -100,37 +65,6 @@ class DataManager(SientiaMonitoring): notification_handler=notification_handler, ) - if self.export_to_kafka: - for i in range(0, 3): - logger.info( - f'Trying ({i}) to initializing DataManager with Kafka servers: {kafka_servers}' - ) - try: - self.kafka_producer = KafkaProducer( - bootstrap_servers=kafka_servers, - value_serializer=lambda v: json.dumps(v).encode( - 'utf-8' - ), # Serialize JSON messages - key_serializer=lambda k: str(k).encode('utf-8') if k else None, - ) - # Kafka connected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1) - break - except NoBrokersAvailable: - logger.error(f'Kafka servers {kafka_servers} are not available. Retrying...') - sleep(5) - else: - # Kafka not connected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) - logger.error( - f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.' - ) - raise NoBrokersAvailable( - f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.' - ) - - logger.info(f'DataManager initialized with Kafka servers: {kafka_servers}') - logger.info( f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}' ) @@ -153,26 +87,7 @@ class DataManager(SientiaMonitoring): def shutdown(self): """ Gracefully shuts down the DataManager and closes all connections. - - This method ensures proper cleanup of: - - Kafka producer connection with message flushing - - MongoDB client connection - - Metrics recording for connection status - - The method handles connection closure gracefully, logging any errors - that occur during shutdown while ensuring all resources are properly released. """ - if self.kafka_producer: - try: - self.kafka_producer.flush(timeout=10) - self.kafka_producer.close() - # Mark as disconnected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) - except Exception as e: - self.logger.error(f'Error closing Kafka producer: {e}') - else: - self.logger.warning('Kafka producer is already closed or not initialized.') - try: self.mongo_repository.close() except Exception as e: @@ -181,83 +96,18 @@ class DataManager(SientiaMonitoring): def __del__(self): self.shutdown() - def delivery_report(self, msg): - """ - Callback for successful Kafka message delivery reports. - - This method is called by the Kafka producer when a message is successfully - delivered to a topic. It logs the delivery details including topic, partition, - and offset information for debugging and monitoring purposes. - - Args: - msg: Kafka message object containing delivery details - """ - self.logger.debug( - f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}' - ) - - def delivery_error(self, err): - """ - Callback for Kafka message delivery error reports. - - This method is called by the Kafka producer when a message delivery fails. - It logs the error details for debugging and monitoring purposes. - - Args: - err: Error information from the failed delivery attempt - """ - self.logger.error(f'Delivery failed for record : {err}') - async def publish(self, topic: str, data: dict) -> None: """ - Publishes a message to a specified Kafka topic. + Persists a message to MongoDB. Args: - topic (str): The name of the Kafka topic to which the message will be published. - data (dict): The message data to be sent to the Kafka topic. + topic (str): The name of the MongoDB collection to which the message will be written. + data (dict): The message data to be stored. Returns: None - - Raises: - Exception: If there is an error during message delivery, it will be handled by the `delivery_error` callback. """ - if self.export_to_kafka and self.kafka_producer: - try: - self.logger.debug(f'Publishing message to topic {topic}: {data}') - self.kafka_producer.send(topic=topic, value=data).add_callback( - self.delivery_report - ).add_errback(self.delivery_error) - - self.kafka_producer.flush(timeout=10) - await self.emit_metric( - metric_object=metrics.KAFKA_MESSAGES_SENT, - tags={ - 'pod_id': self.pod_id, - 'topic': topic, - }, - ) - - except Exception as e: - await self.emit_metric( - metric_object=metrics.KAFKA_MESSAGES_ERRORS, - tags={ - 'pod_id': self.pod_id, - 'topic': topic, - }, - ) - trace = traceback.format_exc() - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'KAFKA_PRODUCER_ERROR_{topic}', - message=f'Error publishing message to topic {topic}: {e}', - block='kafka_producer', - level=NotificationLevel.ERROR, - attachment_content=trace, - ) - self.logger.error(trace) - try: await self.mongo_repository.insert( collection_name=topic, diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 030e305..34a09ca 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -19,7 +19,7 @@ class IngestorManager(SientiaMonitoring): Central coordinator for managing OPC data ingestion operations. The IngestorManager orchestrates the interaction between different components: - - DataManager: Handles data persistence and Kafka export + - DataManager: Handles data persistence - OPC Managers: Manage individual OPC UA server connections - ResourceManager: Coordinates slot leasing and load balancing @@ -36,7 +36,6 @@ class IngestorManager(SientiaMonitoring): - Load balancing across multiple ingestor instances Args: - kafka_servers (str): Comma-separated list of Kafka server addresses redis_data (dict): Redis connection parameters (host, port, username, password) lease_ttl (int): Time-to-live for slot leases in seconds heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds @@ -46,10 +45,9 @@ class IngestorManager(SientiaMonitoring): metadata (dict): Application metadata for notifications and tracking logger (Logger): Logger instance for application logging notification_handler (NotificationHandler): Handler for sending notifications - export_to_kafka (bool): Whether to export data to Kafka Attributes: - data_manager (DataManager): Manages data persistence and Kafka export + data_manager (DataManager): Manages data persistence opc_managers (dict): Dictionary of OPC managers keyed by server name resource_manager (ResourceManager): Manages Redis-based resource coordination number_of_slots (int): Total number of slots configured in the system @@ -61,7 +59,6 @@ class IngestorManager(SientiaMonitoring): def __init__( self, - kafka_servers: str, redis_data: dict, lease_ttl: int, heartbeat_ttl: int, @@ -72,7 +69,6 @@ class IngestorManager(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, - export_to_kafka: bool = False, ): redis_host: str = redis_data['host'] redis_port: int = int(redis_data['port']) @@ -87,10 +83,8 @@ class IngestorManager(SientiaMonitoring): ) self.data_manager = DataManager( - kafka_servers=kafka_servers, mongo_connection_string=mongo_connection_string, mongo_database=mongo_database, - export_to_kafka=export_to_kafka, metadata=metadata, logger=logger, notification_handler=notification_handler, diff --git a/ingestor/metrics.py b/ingestor/metrics.py index 503d299..cb18652 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -19,7 +19,6 @@ from prometheus_client import Counter, Gauge, Histogram # Metric label definitions for consistent labeling across all metrics POD_ID_LABEL = ['pod_id'] SERVER_LABELS = ['pod_id', 'server_name', 'server_url'] -KAFKA_LABELS = ['pod_id', 'topic'] REDIS_LABELS = ['pod_id', 'operation'] NOTIFICATION_LABELS = ['pod_id', 'level', 'block'] @@ -134,22 +133,6 @@ OPC_RECONNECTIONS_TOTAL = Counter( ['pod_id', 'server_name'], ) -# --- Data Manager (Kafka) Metrics --- -KAFKA_MESSAGES_SENT = Counter( - 'kafka_messages_sent_total', 'Total messages sent to Kafka', KAFKA_LABELS -) -KAFKA_MESSAGES_ERRORS = Counter( - 'kafka_messages_errors_total', - 'Total errors sending messages to Kafka', - KAFKA_LABELS, -) -KAFKA_CONNECTION_STATUS = Gauge( - 'kafka_connection_status', - 'Connection status with Kafka (1=connected, 0=disconnected)', - POD_ID_LABEL, -) - - # --- Notification Metrics --- NOTIFICATIONS_SENT = Counter( 'notifications_sent_total', diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 2cb4ae9..4c6b2b6 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -1,6 +1,5 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch -from kafka.errors import NoBrokersAvailable from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel @@ -18,14 +17,11 @@ metadata = { @fixture -@patch('ingestor.managers.data_manager.KafkaProducer') @patch('ingestor.managers.data_manager.MongoDBRepository') -def data_manager(mongodb_repository, kafka): +def data_manager(mongodb_repository): data_manager = DataManager( - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=True, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -38,141 +34,34 @@ def data_manager(mongodb_repository, kafka): return data_manager -@patch('ingestor.managers.data_manager.KafkaProducer') @patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___success(mongodb_repository, kafka): +def test___init___success(mongodb_repository): logger_mock = MagicMock() data_manager = DataManager( metadata=metadata['metadata'], - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=True, logger=logger_mock, notification_handler=MagicMock(), metrics_controller=MagicMock(), ) - kafka.assert_called_once_with( - bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY - ) - assert data_manager.kafka_producer is not None logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' + 'DataManager initialized with MongoDB servers: mongodb://localhost:27017' ) - logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092') logger_mock.error.assert_not_called() - assert logger_mock.info.call_count == 4 + assert data_manager.connection_string == 'mongodb://localhost:27017' -@patch('ingestor.managers.data_manager.KafkaProducer') -@patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___second_attempt(mongodb_repository, kafka): - kafka.side_effect = [NoBrokersAvailable, MagicMock()] - logger_mock = MagicMock() - - data_manager = DataManager( - kafka_servers='localhost:9092', - mongo_connection_string='mongodb://localhost:27017', - mongo_database='sientia', - export_to_kafka=True, - logger=logger_mock, - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=MagicMock(), - ) - - kafka.assert_any_call( - bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY - ) - assert kafka.call_count == 2 - assert data_manager.kafka_producer is not None - logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (1) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092') - logger_mock.error.assert_called_once_with( - 'Kafka servers localhost:9092 are not available. Retrying...' - ) - assert logger_mock.info.call_count == 5 - - -@patch('ingestor.managers.data_manager.KafkaProducer') -@patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___failure_max_attempts(mongodb_repository, kafka): - kafka.side_effect = NoBrokersAvailable - logger_mock = MagicMock() - - try: - DataManager( - kafka_servers='localhost:9092', - mongo_connection_string='mongodb://localhost:27017', - mongo_database='sientia', - export_to_kafka=True, - logger=logger_mock, - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=MagicMock(), - ) - except NoBrokersAvailable as e: - assert ( - str(e) - == 'NoBrokersAvailable: Failed to connect to Kafka servers localhost:9092 after 3 attempts.' - ) - - assert kafka.call_count == 3 - logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (1) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (2) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.error.assert_called_with( - 'Failed to connect to Kafka servers localhost:9092 after 3 attempts.' - ) - assert logger_mock.info.call_count == 3 - - else: - raise AssertionError('Expected NoBrokersAvailable exception was not raised.') - - -def test_shutdown_has_producer(data_manager): - flush_mock = MagicMock() +def test_shutdown(data_manager): close_mock = MagicMock() - - data_manager.kafka_producer.flush = flush_mock - data_manager.kafka_producer.close = close_mock + data_manager.mongo_repository.close = close_mock data_manager.shutdown() - flush_mock.assert_called_once() close_mock.assert_called_once() -def test_shutdown_no_producer(data_manager): - data_manager.kafka_producer = None - - data_manager.shutdown() - - data_manager.logger.warning.assert_any_call( - 'Kafka producer is already closed or not initialized.' - ) - - -def test_shutdown_exception(data_manager): - data_manager.kafka_producer.flush = MagicMock(side_effect=Exception('Test error')) - data_manager.kafka_producer.close = MagicMock() - - data_manager.shutdown() - data_manager.logger.error.assert_called_once_with('Error closing Kafka producer: Test error') - - def test_shutdown_exception_mongo(data_manager): data_manager.mongo_repository.close = MagicMock(side_effect=Exception('Test error')) @@ -187,88 +76,24 @@ def test___del__(data_manager): data_manager.shutdown.assert_called_once() -def test_delivery_report(data_manager): - msg = MagicMock() - msg.topic = 'test_topic' - msg.partition = 0 - msg.offset = 1 - - data_manager.delivery_report(msg) - - data_manager.logger.debug.assert_called_once_with( - f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}' - ) - - -def test_delivery_error(data_manager): - err = 'Test error' - data_manager.delivery_error(err) - - data_manager.logger.error.assert_called_once_with(f'Delivery failed for record : {err}') - - @mark.asyncio async def test_publish(data_manager): - topic = 'test_topic' - data = {'key': 'value'} - - # Mock the send method of the Kafka producer - send_mock = MagicMock() - data_manager.kafka_producer.send = send_mock - - # Call the publish method - await data_manager.publish(topic, data) - - # Check if the send method was called with the correct arguments - send_mock.assert_called_once_with(topic=topic, value=data) - - send_mock.return_value.add_callback.assert_called_once() - - data_manager.kafka_producer.flush.assert_called_once() - - -def test_publish_no_kafka(data_manager): - data_manager.export_to_kafka = False - topic = 'test_topic' - data = {'key': 'value'} - - data_manager.publish(topic, data) - - data_manager.kafka_producer.send.assert_not_called() - - -@patch('ingestor.managers.data_manager.traceback') -@mark.asyncio -async def test_publish_error(traceback, data_manager): topic = 'test_topic' data = {'key': 'value', 'name': 'test_tag'} - # Mock the send method of the Kafka producer to raise an exception - send_mock = MagicMock(side_effect=Exception('Test error')) - data_manager.kafka_producer.send = send_mock + data_manager.mongo_repository.insert = AsyncMock() - data_manager.mongo_repository = AsyncMock() - - # Call the publish method await data_manager.publish(topic, data) - # Check if the send method was called with the correct arguments - send_mock.assert_called_once_with(topic=topic, value=data) - - # Check if the error was logged - data_manager.send_notification_async.assert_called_once_with( - notification_id=f'KAFKA_PRODUCER_ERROR_{topic}', - message=f'Error publishing message to topic {topic}: Test error', - block='kafka_producer', - level=NotificationLevel.ERROR, - attachment_content=traceback.format_exc.return_value, + data_manager.mongo_repository.insert.assert_called_once_with( + collection_name=topic, + document={**data, 'inserted_at': ANY}, metadata=metadata['metadata'], ) @mark.asyncio async def test_publish_error_mongo(data_manager): - data_manager.export_to_kafka = False data_manager.mongo_repository.insert = AsyncMock(side_effect=Exception('Test error')) await data_manager.publish('test_topic', {'key': 'value'}) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 32ac827..9dcdd21 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -20,14 +20,12 @@ metadata = { @patch('ingestor.managers.ingestor_manager.ResourceManager') def ingestor_manager(data_manager_mock, resource_manager_mock): ingestor = IngestorManager( - kafka_servers='localhost:9092', redis_data={'host': 'localhost', 'port': 6379}, lease_ttl=60, heartbeat_ttl=60, poll_interval=5, mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -49,14 +47,12 @@ def test___init__( notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock ): ingestor = IngestorManager( - kafka_servers='localhost:9092', redis_data={'host': 'localhost', 'port': 6379}, lease_ttl=60, heartbeat_ttl=60, poll_interval=5, mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -65,10 +61,8 @@ def test___init__( opc_manager_mock.assert_not_called() data_manager_mock.assert_called_once_with( - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, metadata=metadata['metadata'], logger=ingestor.logger, notification_handler=ingestor.notification_handler, diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 63ea638..8bbccc4 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -9,8 +9,6 @@ from ingestor.ingestor import Ingestor @patch('ingestor.ingestor.NotificationHandler') def test___init__(notification_handler, getenv): getenv.side_effect = [ - 'localhost:9092,localhost:35', # KAFKA_SERVERS - 'true', # EXPORT_TO_KAFKA 'localhost', # REDIS_HOST '63790', # REDIS_PORT 'user', # REDIS_USERNAME @@ -27,7 +25,6 @@ def test___init__(notification_handler, getenv): ingestor = Ingestor() - getenv.assert_any_call('KAFKA_SERVERS', 'localhost:9092') getenv.assert_any_call('REDIS_HOST', 'localhost') getenv.assert_any_call('REDIS_PORT', '6379') getenv.assert_any_call('REDIS_USERNAME', None) @@ -37,7 +34,6 @@ def test___init__(notification_handler, getenv): getenv.assert_any_call('HOSTNAME', 'localhost') getenv.assert_any_call('POLL_INTERVAL', '5') - assert ingestor.kafka_servers == ['localhost:9092', 'localhost:35'] assert ingestor.redis_host == 'localhost' assert ingestor.redis_port == 63790 assert ingestor.redis_username == 'user' @@ -124,7 +120,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor): await ingestor.prepare_ingestor() ingestor_manager_mock.assert_called_once_with( - kafka_servers=','.join(ingestor.kafka_servers), redis_data={ 'host': ingestor.redis_host, 'port': ingestor.redis_port, @@ -139,7 +134,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor): metadata=ingestor.metadata, logger=ingestor.logger, notification_handler=ingestor.notification_handler, - export_to_kafka=ingestor.export_to_kafka, metrics_controller=ingestor.metrics_controller, ) ingestor_manager.declare_active.assert_called_once() diff --git a/tests/unit/test_metrics.py b/tests/unit/test_metrics.py index 809abdf..3c50dea 100644 --- a/tests/unit/test_metrics.py +++ b/tests/unit/test_metrics.py @@ -183,30 +183,6 @@ def test_opc_reconnections_total(): assert set(metrics.OPC_RECONNECTIONS_TOTAL._labelnames) == {'pod_id', 'server_name'} -def test_kafka_messages_sent(): - """Verify the definition of KAFKA_MESSAGES_SENT.""" - assert metrics.KAFKA_MESSAGES_SENT is not None - assert isinstance(metrics.KAFKA_MESSAGES_SENT, Counter) - assert metrics.KAFKA_MESSAGES_SENT._name == 'kafka_messages_sent' # REMOVED _total - assert set(metrics.KAFKA_MESSAGES_SENT._labelnames) == {'pod_id', 'topic'} - - -def test_kafka_messages_errors(): - """Verify the definition of KAFKA_MESSAGES_ERRORS.""" - assert metrics.KAFKA_MESSAGES_ERRORS is not None - assert isinstance(metrics.KAFKA_MESSAGES_ERRORS, Counter) - assert metrics.KAFKA_MESSAGES_ERRORS._name == 'kafka_messages_errors' # REMOVED _total - assert set(metrics.KAFKA_MESSAGES_ERRORS._labelnames) == {'pod_id', 'topic'} - - -def test_kafka_connection_status(): - """Verify the definition of KAFKA_CONNECTION_STATUS.""" - assert metrics.KAFKA_CONNECTION_STATUS is not None - assert isinstance(metrics.KAFKA_CONNECTION_STATUS, Gauge) - assert metrics.KAFKA_CONNECTION_STATUS._name == 'kafka_connection_status' - assert set(metrics.KAFKA_CONNECTION_STATUS._labelnames) == {'pod_id'} - - def test_notifications_sent(): """Verify the definition of NOTIFICATIONS_SENT.""" assert metrics.NOTIFICATIONS_SENT is not None