Code import - branch main

ingestor/managers/data_manager.py

@@ -0,0 +1,288 @@
+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
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.repository.mongodb_repository import MongoDBRepository
+from sientia_do.temporal.constants import now
+
+import ingestor.metrics as metrics
+
+
+class DataManager(SientiaMonitoring):
+    """
+    Manages data persistence and export 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
+        metadata (dict): Application metadata
+    """
+
+    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
+
+        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,
+            logger=logger,
+            metrics_controller=metrics_controller,
+            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}'
+        )
+
+        self.connection_string = mongo_connection_string
+        self.database = mongo_database
+
+        self.mongo_repository = MongoDBRepository(
+            connection_string=self.connection_string,
+            database_name=self.database,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        self.metadata = metadata
+
+        logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
+
+    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:
+            self.logger.error(f'Error closing MongoDB client: {e}')
+
+    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.
+
+        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.
+
+        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,
+                document={**data, 'inserted_at': now()},
+                metadata=self.metadata,
+            )
+            self.logger.debug(f'Message inserted into MongoDB collection {topic}: {data}')
+
+            await self.emit_metric(
+                metric_object=metrics.TAG_WRITTEN_COUNT,
+                tags={
+                    'pod_id': self.pod_id,
+                    'tag_name': data['name'],
+                    'collection_name': topic,
+                },
+            )
+
+        except Exception as e:
+            trace = traceback.format_exc()
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'MONGO_PRODUCER_ERROR_{topic}',
+                message=f'Error inserting message to MongoDB: {e}',
+                block='mongo_producer',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.logger.error(trace)

ingestor/managers/ingestor_manager.py

@@ -0,0 +1,689 @@
+import asyncio
+import traceback
+from copy import deepcopy
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+import ingestor.metrics as metrics
+from ingestor.managers.data_manager import DataManager
+from ingestor.managers.opc_manager import OpcManager
+from ingestor.managers.resource_manager import ResourceManager
+
+
+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
+    - OPC Managers: Manage individual OPC UA server connections
+    - ResourceManager: Coordinates slot leasing and load balancing
+
+    This class implements a slot-based architecture where:
+    - Each slot represents a collection of OPC tags from one or more servers
+    - Slots are distributed across multiple ingestor instances for load balancing
+    - Dynamic slot allocation ensures optimal resource utilization
+
+    Key Responsibilities:
+    - Slot lease management and distribution
+    - OPC server connection lifecycle management
+    - Tag subscription coordination
+    - System health monitoring and integrity checks
+    - 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
+        poll_interval (int): Main loop polling interval in seconds
+        mongo_connection_string (str): MongoDB connection string
+        mongo_database (str): MongoDB database name
+        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
+        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
+        poll_interval (int): Main loop polling interval
+        managed_tags (dict): Currently managed tags organized by slot
+        opc_servers (dict): OPC server configurations
+        metadata (dict): Application metadata
+    """
+
+    def __init__(
+        self,
+        kafka_servers: str,
+        redis_data: dict,
+        lease_ttl: int,
+        heartbeat_ttl: int,
+        poll_interval: int,
+        mongo_connection_string: str,
+        mongo_database: str,
+        metadata: dict,
+        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'])
+        redis_username: str | None = redis_data.get('username', None)
+        redis_password: str | None = redis_data.get('password', None)
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        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,
+            metrics_controller=metrics_controller,
+        )
+        self.opc_managers: dict = {}
+        self.resource_manager = ResourceManager(
+            host=redis_host,
+            port=redis_port,
+            lease_ttl=lease_ttl,
+            heartbeat_ttl=heartbeat_ttl,
+            metadata=metadata,
+            logger=logger,
+            notification_handler=notification_handler,
+            username=redis_username,
+            password=redis_password,
+            metrics_controller=metrics_controller,
+        )
+        self.number_of_slots = 0
+        self.poll_interval = poll_interval
+        self.managed_tags: dict = {}
+        self.opc_servers: dict = {}
+
+        self.metadata = metadata
+
+    async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None:
+        """
+        Initializes an OPC Manager instance using the provided server configuration.
+
+        This method creates and configures an OPC Manager for a specific OPC UA server,
+        establishing the connection and preparing it for tag subscriptions.
+
+        Args:
+            server_config (dict): A dictionary containing the OPC server configuration.
+                Expected keys include:
+                    - 'name' (str): The name of the OPC server.
+                    - 'url' (str): The URL of the OPC server.
+                    - 'server_uri' (str): The URI of the OPC server.
+                    - 'cert_path' (str, optional): Path to the client certificate file.
+                    - 'private_key_path' (str, optional): Path to the private key file.
+                    - 'server_cert_path' (str, optional): Path to the server certificate file.
+
+        Returns:
+            OpcManager | None: An initialized OpcManager instance if successful,
+            otherwise None if an error occurs during initialization.
+
+        Raises:
+            Exception: If OPC manager initialization fails, the error is logged and
+                      a notification is sent, but the method returns None to allow
+                      the system to continue operating with other servers.
+        """
+
+        try:
+            self.logger.info(f'Initializing OpcManager at {server_config["url"]}')
+            manager = OpcManager(
+                name=server_config['name'],
+                url=server_config['url'],
+                subscription_period_ms=server_config['subscription_period_ms'],
+                data_manager=self.data_manager,
+                logger=self.logger,
+                server_uri=server_config['server_uri'],
+                notification_handler=self.notification_handler,
+                metadata=self.metadata,
+                cert_path=server_config.get('cert_path'),
+                private_key_path=server_config.get('private_key_path'),
+                server_cert_path=server_config.get('server_cert_path'),
+                metrics_controller=self.metrics_controller,
+            )
+
+            manager.config = server_config
+            await manager.connect()
+        except Exception as e:
+            trace = traceback.format_exc()
+            self.send_notification(
+                metadata=self.metadata,
+                notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}',
+                message=f'Error initializing OPC manager: {e}',
+                block='opc_manager',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.logger.error(trace)
+
+            return None
+
+        return manager
+
+    async def shutdown(self):
+        """
+        Gracefully shuts down the IngestorManager and all its components.
+
+        This method ensures proper cleanup of:
+        - All OPC manager instances and their connections
+        - Data manager connections and resources
+        - Active subscriptions and server connections
+
+        The shutdown process is performed asynchronously to allow proper cleanup
+        of all managed resources before termination.
+        """
+        for _server_name, server in self.opc_managers.items():
+            await server.shutdown()
+        self.data_manager.shutdown()
+
+    def __del__(self):
+        asyncio.run(self.shutdown())
+
+    async def remove_server(self, server: str):
+        """
+        Removes an OPC server from the ingestor.
+        """
+        if server in self.opc_managers:
+            await self.opc_managers[server].shutdown()
+            del self.opc_managers[server]
+        for slot, _config in self.managed_tags.items():
+            self.managed_tags[slot].pop(server, None)
+
+    async def update_opc_servers(self):
+        """
+        Updates the OPC (OLE for Process Control) server connections managed by the ingestor.
+        This method ensures that the OPC servers defined in `self.managed_tags` are properly
+        initialized and updated. It performs the following tasks:
+        - Registers new OPC servers based on the configuration in `self.managed_tags`.
+        - Updates existing OPC server instances if their configuration has changed.
+        - Disconnects and removes OPC servers that are no longer present in `self.managed_tags`.
+        Steps:
+        1. Iterates through the `self.managed_tags` dictionary to identify and register servers.
+        2. Initializes new OPC server instances if they are not already managed.
+        3. Reinitializes OPC server instances if their configuration has changed.
+        4. Disconnects and removes OPC servers that are no longer registered.
+        Attributes:
+            self.managed_tags (dict): A nested dictionary containing slot and server configurations.
+            self.opc_managers (dict): A dictionary mapping server names to their
+                OPC manager instances.
+            self.data_manager: An object responsible for managing data operations.
+            self.logger: A logging object for recording warnings and other messages.
+        Raises:
+            Any exceptions raised during OPC server initialization or disconnection.
+        Logs:
+            - Warnings for servers that are no longer found in `self.managed_tags`.
+        """
+
+        registered_servers = []
+        current_managed_tags = deepcopy(self.managed_tags)
+        for _slot, slot_config in current_managed_tags.items():
+            for server, server_config in slot_config.items():
+                registered_servers.append(server)
+                server_config = deepcopy(server_config)
+                server_config.pop('tags', None)
+                server_instance = self.opc_managers.get(server, None)
+                if server_instance is None:
+                    self.logger.info(f'Initializing OPC manager for server {server}')
+                    server_instance = await self.initialize_opc_from_config(server_config)
+
+                elif server_instance.config != server_config:
+                    self.logger.warning(f'Reinitializing OPC manager for server {server}')
+                    await server_instance.shutdown()
+                    del self.opc_managers[server]
+                    server_instance = await self.initialize_opc_from_config(server_config)
+                else:
+                    self.logger.debug(
+                        f'OPC manager for server {server} is already initialized and up to date'
+                    )
+
+                if server_instance is not None:
+                    self.opc_managers[server] = server_instance
+                else:
+                    self.logger.warning(
+                        f'Failed to initialize OPC manager for server {server}, '
+                        f'removing server from managed tags.'
+                    )
+
+                    await self.remove_server(server)
+
+        servers = list(self.opc_managers.keys())
+        for server in servers:
+            if server not in registered_servers:
+                self.logger.warning(
+                    f'Server {server} not found in managed tags. Desconnecting from server.'
+                )
+                await self.remove_server(server)
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_MANAGERS_ACTIVE,
+            method='set',
+            value=len(self.opc_managers),
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+
+    async def check_opc_servers_integrity(self):
+        """
+        Checks the integrity of the OPC servers and updates the OPC servers if necessary.
+
+        This method performs health checks on all managed OPC servers by:
+        - Checking cycle counts for data reception
+        - Monitoring connection health and data flow
+        - Triggering reconnection for lost servers
+        - Updating metrics for active OPC managers
+
+        Side Effects:
+            - Updates cycle monitoring for all nodes
+            - Removes lost servers from managed tags
+            - Updates OPC manager metrics
+        """
+        to_disconnect: list[str] = []
+        for server, opc_manager in self.opc_managers.items():
+            await opc_manager.check_cycles()
+
+            is_lost = await opc_manager.check_opc_listenning()
+            if is_lost:
+                self.logger.warning(f'OPC server {server} is lost. Server will be disconnected.')
+
+                to_disconnect.append(server)
+
+        for server in to_disconnect:
+            await self.remove_server(server)
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_MANAGERS_ACTIVE,
+            method='set',
+            value=len(self.opc_managers),
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+
+    async def declare_active(self):
+        """
+        Declares the ingestor as active by sending a heartbeat signal to the resource manager.
+
+        This method ensures that the ingestor is marked as active by invoking the
+        `ingestor_heartbeat` method of the associated resource manager. The heartbeat
+        mechanism enables load balancers and monitoring systems to track active instances.
+
+        Side Effects:
+            - Updates Redis with current instance heartbeat
+            - Enables load balancing and health monitoring
+        """
+
+        await self.resource_manager.ingestor_heartbeat()
+
+    async def get_active_ingestors(self) -> list[str]:
+        """
+        Retrieve a list of active ingestors.
+
+        This method fetches all ingestors from the resource manager and returns them.
+        If no ingestors are found, an empty list is returned.
+
+        Returns:
+            List[str]: A list of active ingestor names, or an empty list if none are found.
+
+        The method queries Redis for all active ingestor heartbeats and extracts
+        the pod identifiers for load balancing and coordination purposes.
+        """
+
+        ingestors = await self.resource_manager.get_all_ingestors()
+        return ingestors if ingestors else []
+
+    async def get_number_of_leases(self) -> int:
+        """
+        Retrieves the number of leases managed by the resource manager.
+
+        This method fetches all available leases from the resource manager,
+        calculates their count, and updates the `number_of_slots` attribute.
+
+        Returns:
+            int: The total number of leases. Returns 0 if no leases are available.
+
+        Side Effects:
+            - Updates internal slot count tracking
+            - Updates Prometheus metrics for total leases
+        """
+
+        leases = await self.resource_manager.get_all_leases()
+        self.number_of_slots = len(leases) if leases else 0
+        await self.emit_metric(
+            metric_object=metrics.LEASES_TOTAL,
+            method='set',
+            value=self.number_of_slots,
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+        return self.number_of_slots
+
+    async def get_number_of_slots(self) -> int:
+        """
+        Retrieves the number of slots managed by the resource manager.
+
+        This method fetches all available slots from the resource manager,
+        calculates their count, and updates the `number_of_slots` attribute.
+
+        Returns:
+            int: The total number of slots. Returns 0 if no slots are available.
+
+        Side Effects:
+            - Updates internal slot count tracking
+            - Updates Prometheus metrics for total slots
+        """
+
+        slots = await self.resource_manager.get_all_slots()
+        self.number_of_slots = len(slots) if slots else 0
+        await self.emit_metric(
+            metric_object=metrics.SLOTS_TOTAL,
+            method='set',
+            value=self.number_of_slots,
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+        return self.number_of_slots
+
+    async def get_slot_leases(self, max_slots: int = 1) -> dict:
+        """
+        Acquires a specified number of resource slots by leasing them from the resource manager.
+
+        This method implements the slot acquisition logic for load balancing:
+        - Iterates through available slots and attempts to lease them
+        - Logs the leasing of each slot
+        - Updates the `managed_tags` attribute with the acquired slots
+        - Stops leasing once the specified `max_slots` are acquired
+
+        Args:
+            max_slots (int): The maximum number of slots to lease. Defaults to 1.
+
+        Returns:
+            Dict: A dictionary where the keys are the slot identifiers (as strings)
+                  and the values are the leased slot details.
+
+        Behavior:
+            - Attempts to lease slots sequentially starting from slot 1
+            - Skips slots that cannot be retrieved after leasing
+            - Logs warnings if unable to acquire the requested number of slots
+            - Updates metrics for acquired slots and managed slots count
+
+        Notes:
+            - If a slot is leased but its details cannot be retrieved
+              (i.e., `get_tag_slot` returns None), that slot is skipped.
+        """
+
+        acquired = {}
+        for i in range(1, self.number_of_slots + 1):
+            if await self.resource_manager.lease_tag(str(i)):
+                self.logger.info(f'Leased slot {i}')
+                slots = await self.resource_manager.get_tag_slot(str(i))
+                if slots is None:
+                    continue
+                acquired[str(i)] = slots
+                await self.emit_metric(
+                    metric_object=metrics.SLOTS_ACQUIRED,
+                    method='inc',
+                    value=1,
+                    tags={
+                        'pod_id': self.pod_id,
+                    },
+                )
+
+                if len(acquired) >= max_slots:
+                    self.managed_tags.update(acquired)
+                    await self.emit_metric(
+                        metric_object=metrics.SLOTS_MANAGED,
+                        method='set',
+                        value=len(self.managed_tags),
+                        tags={
+                            'pod_id': self.pod_id,
+                        },
+                    )
+                    return acquired
+
+        self.logger.warning(
+            f'Unable to acquire {max_slots} slots. Only {acquired} slots were leased.'
+        )
+        self.managed_tags.update(acquired)
+        await self.emit_metric(
+            metric_object=metrics.SLOTS_MANAGED,
+            method='set',
+            value=len(self.managed_tags),
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+        return acquired
+
+    async def unsubscribe_slot(self, slot: str):
+        """
+        Unsubscribes a specific slot from all associated OPC servers.
+
+        This method removes all subscriptions for a given slot across all
+        OPC servers that were managing it. It ensures clean cleanup of
+        resources when slots are released or reconfigured.
+
+        Args:
+            slot (str): The name of the slot to unsubscribe.
+
+        Raises:
+            KeyError: If the specified slot does not exist in the managed tags.
+
+        Side Effects:
+            - Removes subscriptions from all OPC servers for the specified slot
+            - Cleans up subscription resources on the OPC servers
+        """
+
+        for server in self.managed_tags[slot].keys():
+            if server in self.opc_managers:
+                await self.opc_managers[server].unsubscribe(slot)
+
+    async def update_slot_config(self):
+        """
+        Updates the configuration of managed slots by renewing their leases,
+        fetching the latest configurations, and handling any changes or removals.
+
+        This method performs the following steps:
+        1. Renews the lease for each managed slot using the resource manager.
+        2. Fetches the latest configuration for each slot.
+        3. Logs and removes slots whose configurations are no longer available.
+        4. Updates the configuration of slots if changes are detected.
+        5. Unsubscribes and re-subscribes to slots with updated configurations.
+        6. Removes slots from the managed tags if they are no longer valid.
+        7. Updates the OPC servers after processing all slots.
+
+        Side Effects:
+            - Modifies the `managed_tags` dictionary to reflect the latest slot configurations.
+            - Updates OPC server subscriptions based on the current state of managed slots.
+
+        Raises:
+            None explicitly, but relies on the behavior of `resource_manager` and
+            other dependencies for error handling.
+
+        Logging:
+            - Logs warnings for removed slots.
+            - Logs informational messages for updated slot configurations.
+        """
+
+        removed_slots: list[str] = []
+        for slot, _slot_config in self.managed_tags.items():
+            await self.resource_manager.renew_tag_lease(slot)
+            update = await self.resource_manager.get_tag_slot(slot)
+            if update is None:
+                removed_slots.append(slot)
+                continue
+
+            self.managed_tags[slot] = update
+
+        for slot in removed_slots:
+            self.managed_tags.pop(slot, None)
+
+    async def drop_slot_leases(self, ids: list[str]) -> None:
+        """
+        Releases the leases associated with the specified slot IDs.
+
+        This method iterates through a list of slot IDs and calls the
+        `drop_tag_lease` method of the `resource_manager` to release
+        the lease for each ID. It's used during load balancing and
+        graceful shutdown scenarios.
+
+        Args:
+            ids (List[str]): A list of slot IDs for which the leases
+                             should be released.
+
+        Returns:
+            None
+
+        Side Effects:
+            - Releases Redis-based leases for specified slots
+            - Updates metrics for released slots count
+        """
+        for lease_id in ids:
+            await self.resource_manager.drop_tag_lease(lease_id)
+            await self.emit_metric(
+                metric_object=metrics.SLOTS_RELEASED,
+                method='inc',
+                value=1,
+                tags={
+                    'pod_id': self.pod_id,
+                },
+            )
+
+    async def manage_server(self, slot: str, server: str, server_config: dict, tags: dict) -> int:
+        """
+        Manages the subscription of tags to a specified OPC server and slot.
+
+        This method ensures that the specified server and slot have an active subscription
+        for the provided tags. If the server or slot is not properly configured, or if
+        subscription fails, appropriate error handling is performed.
+
+        Args:
+            slot (str): The slot identifier for the subscription.
+            server (str): The name of the OPC server.
+            server_config (dict): Configuration dictionary for the server, which includes
+                the tags to be subscribed under the key 'tags'.
+            tags (dict): A dictionary of tags to be subscribed.
+
+        Returns:
+            int: Status code indicating the result of the operation:
+                - 0: Subscription was successful.
+                - 1: Server not found in `opc_managers`.
+                - 2: Subscription creation or tag subscription failed.
+
+        Logs:
+            - Logs informational messages about the subscription process.
+            - Logs errors if the server is not found, subscription creation fails, or
+              tag subscription fails.
+            - Logs a warning if a subscription is removed due to failure.
+
+        Raises:
+            Exception: Any unexpected exceptions during subscription creation or tag
+            subscription are logged but not propagated.
+
+        Side Effects:
+            - Creates or updates OPC subscriptions
+            - Manages tag subscriptions on OPC servers
+            - Updates error metrics and notifications
+        """
+
+        self.logger.info(f'Subscribing to tags from {slot}:{server}')
+        tags_to_sub = server_config.get('tags')
+        if server not in self.opc_managers:
+            self.logger.error(f'Server {server} not found in opc_managers.')
+            return 1
+        if slot not in self.opc_managers[server].subscriptions:
+            try:
+                await self.opc_managers[server].create_subscription(slot)
+            except Exception as e:
+                self.logger.error(f'Failed to create subscription for slot {slot}: {e}')
+                return 2
+        try:
+            self.logger.info(tags_to_sub)
+            await self.opc_managers[server].subscribe(
+                slot, deepcopy(tags_to_sub), self.poll_interval
+            )
+            self.logger.info(tags_to_sub)
+        except Exception as e:
+            await self.emit_metric(
+                metric_object=metrics.OPC_SUBSCRIPTION_ERRORS,
+                tags={
+                    'pod_id': self.pod_id,
+                    'server': server,
+                    'slot': slot,
+                },
+            )
+            trace = traceback.format_exc()
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'OPC_SUBSCRIPTION_ERROR_{slot}:{server}',
+                message=f'Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}',
+                block='opc_manager',
+                level=NotificationLevel.ERROR,
+                attachment_content=trace,
+            )
+            self.logger.error(trace)
+
+            self.logger.warning(f'Removing subscription from server {server} for slot {slot}')
+            await self.opc_managers[server].unsubscribe(slot)
+            return 2
+        return 0
+
+    async def subscribe_to_tags(self, tags: dict) -> None:
+        """
+        Subscribes to a set of tags and manages their configurations.
+
+        This method processes a dictionary of tags, iterating through each slot and server
+        configuration. It attempts to manage the server configurations and removes any
+        servers that return a specific response code.
+
+        Args:
+            tags (Dict): A dictionary containing tag configurations. The structure is
+                         expected to be {slot: {server: server_config}}.
+
+        Side Effects:
+            - Logs the provided tags for debugging purposes.
+            - Updates the `managed_tags` attribute by removing servers that meet the
+              removal criteria.
+            - Establishes OPC subscriptions for all configured tags.
+
+        Removal Criteria:
+            - If the `manage_server` method returns a response code of 2 for a given
+              slot and server, that server is removed from the `managed_tags` attribute.
+
+        The method ensures that only successfully configured servers remain in the
+        managed tags, maintaining system stability and preventing subscription errors.
+        """
+
+        to_remove = []
+        self.logger.info(tags)
+        for slot, slot_config in tags.items():
+            for server, server_config in slot_config.items():
+                response = await self.manage_server(slot, server, server_config, tags)
+                if response == 2:
+                    to_remove.append([slot, server])
+
+        for slot, server in to_remove:
+            if server in self.opc_managers:
+                await self.opc_managers[server].shutdown()
+                del self.opc_managers[server]
+            self.managed_tags[slot].pop(server, None)

ingestor/managers/opc_manager.py

@@ -0,0 +1,590 @@
+import asyncio
+import json
+import traceback
+from pathlib import Path
+
+from asyncua import Client
+from asyncua.crypto.security_policies import SecurityPolicyBasic256
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.notifications.models import NotificationLevel
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
+
+import ingestor.metrics as metrics
+from ingestor.managers.data_manager import DataManager
+
+
+class OpcManager(SientiaMonitoring):
+    """
+    Manages OPC UA server connections and tag subscriptions.
+
+    The OpcManager is responsible for:
+    - Establishing and maintaining secure connections to OPC UA servers
+    - Managing tag subscriptions and data collection
+    - Handling server reconnection and error recovery
+    - Processing OPC data and forwarding it to the data manager
+    - Monitoring connection health and performance metrics
+
+    The manager supports both secure and unsecured connections, with optional
+    certificate-based authentication for enhanced security.
+
+    Args:
+        name (str): Unique identifier for the OPC server
+        url (str): OPC UA server endpoint URL
+        data_manager (DataManager): Manager for data persistence and export
+        logger (Logger): Logger instance for application logging
+        server_uri (str): OPC UA server application URI
+        notification_handler (NotificationHandler): Handler for sending notifications
+        metadata (dict): Application metadata for notifications and tracking
+        cert_path (str, optional): Path to client certificate file for secure connections
+        private_key_path (str, optional): Path to client private key file
+        server_cert_path (str, optional): Path to server certificate file for validation
+
+    Attributes:
+        url (str): OPC UA server endpoint URL
+        name (str): Unique identifier for the OPC server
+        server_uri (str): OPC UA server application URI
+        data_queue (dict): Queue for buffering OPC data before processing
+        non_receive_count (int): Counter for cycles without data reception
+        client (Client): OPC UA client instance
+        cert_path (str): Path to client certificate file
+        private_key_path (str): Path to client private key file
+        server_cert_path (str): Path to server certificate file
+        nodes (dict): Dictionary of OPC node references
+        subscriptions (dict): Active OPC subscriptions
+        data_manager (DataManager): Manager for data persistence and export
+        metadata (dict): Application metadata
+    """
+
+    def __init__(
+        self,
+        name: str,
+        url: str,
+        subscription_period_ms: int,
+        data_manager: DataManager,
+        logger: Logger,
+        server_uri: str,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        metadata: dict,
+        cert_path: str | None = None,
+        private_key_path: str | None = None,
+        server_cert_path: str | None = None,
+    ):
+        self.url = url
+        self.name = name
+        self.server_uri = server_uri
+        self.data_queue: dict = {}
+        self.non_receive_count = 0
+        self.client: Client | None = None
+        self.subscription_period_ms = subscription_period_ms
+        self.cert_path = cert_path
+        self.private_key_path = private_key_path
+        self.server_cert_path = server_cert_path
+        self.nodes: dict = {}
+        self.subscriptions: dict = {}
+        self.data_manager = data_manager
+        self.metadata = metadata
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        metrics.OPC_CONNECTION_STATUS.labels(
+            pod_id=self.pod_id, server_name=self.name, server_url=self.url
+        ).set(0)
+        metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0)
+
+    def __str__(self):
+        """
+        String representation of the OPC Manager.
+
+        Returns:
+            str: Human-readable representation showing server details and current state.
+        """
+        return (
+            f'OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n'
+            f'nodes={self.nodes}, subscriptions={self.subscriptions}'
+        )
+
+    def __del__(self):
+        asyncio.run(self.shutdown())
+
+    async def shutdown(self):
+        """
+        Comprehensive cleanup method for graceful shutdown.
+
+        This method ensures proper cleanup of all OPC UA resources:
+        - Closes active subscriptions
+        - Disconnects from the OPC server
+        - Releases allocated resources
+
+        Should be called before the application terminates to prevent resource leaks
+        and ensure clean disconnection from OPC servers.
+        """
+        try:
+            await self.disconnect()
+        except Exception as e:
+            self.logger.error(f'Error during cleanup: {e}')
+
+    async def set_security(self):
+        """
+        Configures the security settings for the OPC UA client.
+
+        This method sets up the security policy, certificates, and timeouts
+        required for establishing a secure connection with the OPC UA server.
+        It implements Basic256 security policy with certificate-based authentication.
+
+        Raises:
+            ValueError: If either the certificate path or private key path is not provided.
+
+        Security Settings:
+            - Security Policy: Basic256
+            - Secure Channel Timeout: 10,000,000 ms
+            - Session Timeout: 10,000,000 ms
+
+        The method configures:
+            - Client application URI
+            - Certificate-based authentication
+            - Server certificate validation (if provided)
+            - Connection timeouts for stability
+        """
+
+        if not all([self.cert_path, self.private_key_path]):
+            raise ValueError(
+                'Certificate and private key paths must be provided for secure connection.'
+            )
+        cert = str(Path(self.cert_path)) if self.cert_path else None
+        private_key = str(Path(self.private_key_path)) if self.private_key_path else None
+        server_cert = str(Path(self.server_cert_path)) if self.server_cert_path else None
+
+        if self.client:
+            self.client.application_uri = self.server_uri
+            self.logger.info('Setting security...')
+            await self.client.set_security(
+                SecurityPolicyBasic256,
+                certificate=cert,
+                private_key=private_key,
+                server_certificate=server_cert,
+            )
+            self.client.secure_channel_timeout = 10000000
+            self.client.session_timeout = 10000000
+
+    async def connect(self):
+        """
+        Establishes a connection to the OPC server.
+
+        This method initializes the OPC client using the provided URL and
+        sets up security if a certificate path is specified. It then
+        attempts to connect to the server and logs the connection status.
+
+        The connection process includes:
+        1. Client initialization with server URL
+        2. Security configuration (if certificates are provided)
+        3. Connection establishment
+        4. Metrics recording for monitoring
+
+        Raises:
+            Exception: If the connection to the OPC server fails.
+
+        Metrics:
+            - OPC_CONNECTIONS_TOTAL: Incremented on connection attempt
+            - OPC_CONNECTION_STATUS: Set to 1 on successful connection
+        """
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_CONNECTIONS_TOTAL,
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.name,
+            },
+        )
+        try:
+            self.client = Client(self.url, timeout=10, watchdog_intervall=3600000)
+            assert self.client is not None  # Informa ao mypy que client não é None
+
+            self.client.name = self.pod_id
+            self.client.application_name = self.pod_id
+            pod_uri = self.pod_id.replace('-', ':')
+            self.client.application_uri = pod_uri
+            self.client.product_uri = pod_uri
+
+            if self.cert_path:
+                await self.set_security()
+            self.logger.info(f'Starting connection to {self.name}...')
+            await self.client.connect()
+            await self.emit_metric(
+                metric_object=metrics.OPC_CONNECTION_STATUS,
+                method='set',
+                value=1,
+                tags={
+                    'pod_id': self.pod_id,
+                    'server_name': self.name,
+                    'server_url': self.url,
+                },
+            )
+            self.logger.info(f'Connection to {self.name} successful.')
+        except Exception:
+            await self.disconnect()
+
+            raise
+
+    async def create_subscription(self, name: str):
+        """
+        Creates a subscription with the specified monitoring period.
+
+        This method establishes a subscription to monitor data changes or events
+        from the OPC UA server. If the client is not connected, an exception is raised.
+
+        Args:
+            name (str): The name identifier for the subscription
+            period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms.
+
+        Raises:
+            ValueError: If the client is not connected.
+
+        Side Effects:
+            - Sets the `self.period` attribute to the specified or default period.
+            - Creates a subscription and assigns it to `self.subscriptions[name]`.
+            - Logs the creation of the subscription.
+            - Increments subscription creation metrics.
+        """
+
+        if not self.client:
+            raise ValueError('Client not connected. Call connect first.')
+        try:
+            self.subscriptions[name] = await self.client.create_subscription(
+                self.subscription_period_ms, self
+            )
+            self.logger.info(f'Subscription {name} created on {self.name}.')
+            await self.emit_metric(
+                metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
+                method='inc',
+                value=1,
+                tags={
+                    'pod_id': self.pod_id,
+                    'server_name': self.name,
+                    'slot_name': name,
+                },
+            )
+        except Exception as e:
+            self.logger.error(f'Failed to create subscription {name} on {self.name}: {e}')
+            raise
+
+    async def subscribe(self, subscription: str, nodes: dict, collect_period: int):
+        """
+        Subscribes to a set of OPC UA nodes for data change notifications.
+
+        This method adds the specified nodes to the subscription and configures
+        their data collection rules based on the provided collection period and
+        node-specific frequency.
+
+        Args:
+            subscription (str): The name of the subscription to use
+            nodes (dict): A dictionary where keys are node identifiers and values are
+                configurations for each node. Each configuration must include a 'frequency'
+                key indicating the frequency of data collection in Hz.
+            collect_period (int): The data collection period in seconds.
+
+        Raises:
+            ValueError: If the subscription has not been created by calling
+                `create_subscription` prior to this method.
+
+        Side Effects:
+            - Updates internal node tracking and cycle rules
+            - Establishes data change monitoring for specified nodes
+            - Updates metrics for subscribed tags count
+        """
+
+        if not self.subscriptions.get(subscription):
+            raise ValueError('Subscription not created. Call create_subscription first.')
+
+        self.logger.info(f'Subscribing to {subscription} on {self.name}...')
+        self.logger.info(f'Subscribing to nodes: {nodes}')
+        assert self.client is not None  # Informa ao mypy que client não é None
+        addr_nodes = [self.client.get_node(n) for n in nodes]
+        self.logger.debug(f'Addr nodes: {addr_nodes}')
+        self.nodes.update(nodes)
+        self.logger.debug(f'Nodes: {self.nodes}')
+
+        self.collect_period = collect_period
+
+        for node, config in self.nodes.items():
+            self.nodes[node]['cycle_rule'] = {
+                'cycle_increment': collect_period * 1000 / float(config['frequency']),
+                'cycle_count': 0,
+            }
+
+        await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
+
+        await self.emit_metric(
+            metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+            method='set',
+            value=len(self.nodes),
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.name,
+            },
+        )
+
+    async def unsubscribe(self, subscription: str):
+        """
+        Unsubscribes from a given subscription.
+
+        This method removes the specified subscription and cleans up associated
+        resources. It handles cases where the subscription doesn't exist gracefully.
+
+        Args:
+            subscription (str): The name of the subscription to unsubscribe from.
+
+        Logs:
+            - A warning if the specified subscription does not exist.
+            - An info message upon successful unsubscription.
+
+        Behavior:
+            - If the subscription exists, it is deleted and removed from the
+              subscriptions dictionary.
+            - If the subscription does not exist, no action is taken.
+        """
+
+        if not self.subscriptions.get(subscription):
+            self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
+            return
+        await self.subscriptions[subscription].delete()
+        del self.subscriptions[subscription]
+        self.logger.info(f'Unsubscribed from {subscription}.')
+
+    async def disconnection_fallback(self) -> list:
+        """
+        Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try.
+        """
+
+        assert self.client is not None
+        error_stack = []
+        for i in range(5):
+            try:
+                self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5')
+                await self.client.disconnect()
+                return []
+            except Exception as e:
+                self.logger.error(
+                    f'Failed to disconnect from OPC UA serve  in attempt {i + 1} of 5: {e}'
+                )
+                error_stack.append(
+                    {
+                        'attempt': i + 1,
+                        'error': str(e),
+                        'traceback': traceback.format_exc(),
+                    }
+                )
+                await asyncio.sleep(0.1 * i)
+        return error_stack
+
+    async def disconnect(self):
+        """
+        Disconnects from the OPC UA server.
+
+        This method handles the disconnection process by deleting all subscriptions
+        and disconnecting the client from the OPC UA server. It logs the disconnection
+        process and handles any exceptions that may occur during cleanup.
+
+        Raises:
+            Exception: If an error occurs while deleting the subscription or disconnecting
+                       from the OPC UA server, it logs the error details.
+
+        Side Effects:
+            - Deletes all active subscriptions
+            - Disconnects the OPC client
+            - Updates connection status metrics
+            - Clears internal client reference
+        """
+
+        self.logger.warning('Disconnecting from OPC server')
+        if self.client is None:
+            self.logger.warning('Client already disconnected.')
+            return
+        try:
+            for sub in self.subscriptions:
+                await self.subscriptions[sub].delete()
+            self.logger.warning('Deleted all subscriptions.')
+        except Exception as sub_error:
+            self.logger.error(f'Failed to clean up subscription: {sub_error}')
+
+        errors = await self.disconnection_fallback()
+
+        if errors:
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'OPC_DISCONNECTION_ERROR_{self.name}',
+                message=f'Failed to disconnect from OPC UA server {self.name} after 5 attempts',
+                block='opc_manager',
+                level=NotificationLevel.ERROR,
+                attachment_content=json.dumps(errors, indent=4),
+            )
+        else:
+            self.logger.warning('Disconnected from OPC UA server.')
+
+        del self.client
+        self.client = None
+        await self.emit_metric(
+            metric_object=metrics.OPC_CONNECTION_STATUS,
+            method='set',
+            value=0,
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.name,
+                'server_url': self.url,
+            },
+        )
+        await self.emit_metric(
+            metric_object=metrics.OPC_TAGS_SUBSCRIBED,
+            method='set',
+            value=0,
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.name,
+            },
+        )
+
+    async def datachange_notification(self, node, _val, data):
+        """
+        Handles data change notifications for monitored OPC UA nodes.
+
+        This method is triggered when a monitored node's value changes. It processes
+        the notification, updates internal state, and publishes the data to the
+        appropriate topics.
+
+        Args:
+            node (NodeId): The OPC UA node that triggered the data change notification.
+            _val (Any): The new value of the node (unused in this implementation).
+            data (DataChangeNotification): The data change notification object containing
+                details about the change.
+
+        Behavior:
+            - Extracts the value and source timestamp from the monitored item.
+            - Resets the cycle count for the node's cycle rule.
+            - Resets the non-receive count.
+            - Constructs a data dictionary containing the tag, tag name, timestamp, and value.
+            - Publishes the data to all topics associated with the node.
+        """
+
+        # get data value
+        monitored_item = data.monitored_item
+        value = monitored_item.Value.Value.Value
+        # source_timestamp
+        source_timestamp = monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE)
+        tag = str(node)
+
+        self.logger.debug(
+            f'Data change notification received for tag:'
+            f'{tag} after {self.nodes[tag]["cycle_rule"]["cycle_count"]} cycles'
+        )
+
+        data = {
+            'tag': tag,
+            'name': self.nodes[str(node)]['tag_name'],
+            'timestamp': source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ),
+            'value': value,
+        }
+
+        for topic in self.nodes[tag]['topics']:
+            await self.data_manager.publish(topic, data)
+
+        self.nodes[tag]['cycle_rule']['cycle_count'] = 0
+        self.non_receive_count = 0
+        await self.emit_metric(
+            metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
+            method='set',
+            value=0,
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.name,
+            },
+        )
+
+    async def check_cycles(self):
+        """
+        Checks the cycle counts for all monitored nodes and sends
+        notifications if thresholds are exceeded.
+
+        This method iterates through all monitored nodes and updates their cycle counts based on
+        configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers
+        a warning notification.
+
+        Side Effects:
+            - Updates cycle counts for all monitored nodes
+            - Sends warning notifications for nodes exceeding cycle thresholds
+        """
+        for node, config in self.nodes.items():
+            self.nodes[node]['cycle_rule']['cycle_count'] += config['cycle_rule']['cycle_increment']
+            if self.nodes[node]['cycle_rule']['cycle_count'] >= 5:
+                name = config['tag_name']
+                cycles = self.nodes[node]['cycle_rule']['cycle_count']
+                await self.send_notification_async(
+                    metadata=self.metadata,
+                    notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED',
+                    message=f'{cycles} cycles without receive from {node}:{name}',
+                    block='opc_manager',
+                    level=NotificationLevel.WARNING,
+                )
+
+    async def check_opc_listenning(self) -> bool:
+        """
+        Checks the OPC connection and triggers notifications if the connection is lost.
+
+        This method monitors the data reception health by tracking cycles without
+        data. It sends notifications at different thresholds and can trigger
+        reconnection attempts.
+
+        Returns:
+            bool: True if the connection is lost and reconnection should be attempted,
+                  False otherwise.
+
+        Side Effects:
+            - Increments non-receive count
+            - Updates metrics for cycles without data
+            - Sends warning notifications at 5 cycles
+            - Sends error notifications and triggers reconnection at 15 cycles
+        """
+
+        self.non_receive_count += 1
+        await self.emit_metric(
+            metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
+            method='set',
+            value=self.non_receive_count,
+            tags={
+                'pod_id': self.pod_id,
+                'server_name': self.name,
+            },
+        )
+        if self.non_receive_count >= 5:
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
+                message=f'{self.non_receive_count} cycles without '
+                f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
+                block='opc_manager',
+                level=NotificationLevel.ERROR,
+            )
+        if self.non_receive_count >= 15:
+            await self.emit_metric(
+                metric_object=metrics.OPC_RECONNECTIONS_TOTAL,
+                tags={
+                    'pod_id': self.pod_id,
+                    'server_name': self.name,
+                },
+            )
+            await self.send_notification_async(
+                metadata=self.metadata,
+                notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
+                message=f'Retrying to connect to server {self.name}',
+                block='opc_manager',
+                level=NotificationLevel.ERROR,
+            )
+            return True
+        return False

ingestor/managers/resource_manager.py

@@ -0,0 +1,315 @@
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import Logger
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+from sientia_do.repository.redis_repository import RedisRepository
+
+
+class ResourceManager(SientiaMonitoring):
+    """
+    Manages Redis-based resource coordination and slot leasing for the OPC Ingestor.
+
+    The ResourceManager is responsible for:
+    - Coordinating slot allocation across multiple ingestor instances
+    - Managing lease lifecycles and heartbeats for load balancing
+    - Providing distributed locking and resource management
+    - Monitoring Redis operations and connection health
+
+    The manager implements a sophisticated slot leasing system that enables:
+    - Dynamic load distribution across multiple ingestor instances
+    - Automatic failover and recovery from instance failures
+    - Fair resource allocation based on system capacity
+    - Real-time monitoring of system health and performance
+
+    Args:
+        host (str): Redis server hostname
+        port (int): Redis server port
+        lease_ttl (int): Time-to-live for slot leases in seconds
+        heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds
+        metadata (dict): Application metadata for notifications and tracking
+        logger (Logger): Logger instance for application logging
+        notification_handler (NotificationHandler): Handler for sending notifications
+        username (str, optional): Redis username for authentication
+        password (str, optional): Redis password for authentication
+
+    Attributes:
+        redis (Redis): Redis client instance
+        lease_ttl (int): Time-to-live for slot leases
+        heartbeat_ttl (int): Time-to-live for heartbeat signals
+        metadata (dict): Application metadata
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        lease_ttl: int,
+        heartbeat_ttl: int,
+        metadata: dict,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+        username: str | None = None,
+        password: str | None = None,
+    ) -> None:
+        """
+        Initializes the ResourceManager with Redis connection and configuration.
+
+        This constructor establishes a connection to Redis and verifies connectivity
+        by performing a ping operation. It sets up the connection with optional
+        authentication and records the connection status in metrics.
+
+        Args:
+            host (str): Redis server hostname
+            port (int): Redis server port
+            lease_ttl (int): Time-to-live for slot leases in seconds
+            heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds
+            metadata (dict): Application metadata
+            logger (Logger): Logger instance
+            notification_handler (NotificationHandler): Notification handler
+            username (str, optional): Redis username for authentication
+            password (str, optional): Redis password for authentication
+
+        Raises:
+            Exception: If Redis connection fails, the error is logged and metrics
+                      are updated before re-raising the exception.
+
+        Metrics:
+            - REDIS_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure
+        """
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+        try:
+            self.redis_repository = RedisRepository(
+                host=host,
+                port=port,
+                username=username,
+                password=password,
+                logger=logger,
+                notification_handler=notification_handler,
+                metrics_controller=metrics_controller,
+            )
+            self.redis_repository.redis_client.ping()
+        except Exception as e:
+            self.logger.error(f'Failed to connect to Redis: {e}')
+            raise
+
+        self.lease_ttl = lease_ttl
+        self.heartbeat_ttl = heartbeat_ttl
+        self.metadata = metadata
+
+    async def get_tag_slot(self, tag_id: str) -> dict | None:
+        """
+        Retrieve the tag slot information for a given ID.
+
+        This method constructs the Redis key for a tag slot and retrieves
+        the associated configuration information.
+
+        Args:
+            id (str): The unique identifier of the tag slot to retrieve.
+
+        Returns:
+            dict: A dictionary containing the tag slot information associated
+                  with the given ID, or None if not found.
+
+        The method constructs the key using the pattern "slot:opc_tags:{id}"
+        and delegates to the get() method for the actual Redis operation.
+        """
+
+        self.info(f'Getting tag slot for tag_id: {tag_id}', metadata=self.metadata)
+
+        slot = await self.redis_repository.get(f'slot:opc_tags:{tag_id}', metadata=self.metadata)
+        self.info(f'Tag slot for tag_id: {tag_id} is: {slot}', metadata=self.metadata)
+        return slot
+
+    async def ingestor_heartbeat(self) -> None:
+        """
+        Sends a heartbeat signal to Redis to indicate that the ingestor is active.
+
+        This method sets a key in Redis with a specific format that includes the
+        ingestor's pod ID. The key is set with a value of 1 and an expiration
+        time defined by `self.heartbeat_ttl`. This allows monitoring systems to
+        track the activity and health of the ingestor.
+
+        The heartbeat mechanism enables:
+        - Load balancers to identify active ingestor instances
+        - Health monitoring systems to detect failed instances
+        - Automatic failover and recovery mechanisms
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="set"
+            - REDIS_OPERATIONS_DURATION: Records timing for heartbeat operations
+        """
+
+        await self.redis_repository.set(
+            f'heartbeat:ingestor:{self.pod_id}', 1, ttl=self.heartbeat_ttl, metadata=self.metadata
+        )
+
+    async def lease_tag(self, tag_id: str) -> bool:
+        """
+        Attempts to lease a tag by setting a key in Redis with a specified TTL.
+
+        This method uses the Redis `SET` command with the `NX` option to ensure that
+        the key is only set if it does not already exist. The key is set with an
+        expiration time defined by `lease_ttl`. This implements a distributed
+        locking mechanism for tag allocation.
+
+        Args:
+            tag_id (str): The unique identifier of the tag to be leased.
+
+        Returns:
+            bool: True if the lease was successfully acquired, False if the tag
+                  is already leased by another ingestor.
+
+        The leasing mechanism ensures:
+        - Only one ingestor can process a specific tag at a time
+        - Automatic lease expiration prevents deadlocks
+        - Fair distribution of tags across available ingestor instances
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="set_nx"
+            - REDIS_OPERATIONS_DURATION: Records timing for lease operations
+        """
+
+        return await self.redis_repository.set(
+            f'lease:opc_tags:{tag_id}',
+            self.pod_id,
+            ttl=self.lease_ttl,
+            nx=True,
+            metadata=self.metadata,
+        )
+
+    async def renew_tag_lease(self, tag_id: str) -> bool:
+        """
+        Renews the lease for a specific OPC tag if the current pod holds the lease.
+
+        This method checks if the current pod (identified by `self.pod_id`) holds
+        the lease for the given OPC tag. If so, it extends the lease by resetting
+        its expiration time in Redis to the configured lease TTL.
+
+        Args:
+            tag_id (str): The identifier of the OPC tag whose lease is to be renewed.
+
+        Returns:
+            bool: True if the lease was successfully renewed, False if the current
+                  pod doesn't hold the lease or renewal failed.
+
+        Lease renewal is essential for:
+        - Maintaining continuous tag processing without interruptions
+        - Preventing lease expiration during long-running operations
+        - Ensuring system stability and reliability
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="get" and "expire"
+            - REDIS_OPERATIONS_DURATION: Records timing for renewal operations
+        """
+
+        current = await self.redis_repository.get(
+            f'lease:opc_tags:{tag_id}', metadata=self.metadata
+        )
+        if current == self.pod_id:
+            await self.redis_repository.expire(
+                f'lease:opc_tags:{tag_id}', self.lease_ttl, metadata=self.metadata
+            )
+            return True
+        return False
+
+    async def drop_tag_lease(self, tag_id: str) -> None:
+        """
+        Drops the lease for a specific OPC tag.
+
+        This method removes the lease for the given OPC tag by deleting the
+        corresponding key in Redis. This is typically called when an ingestor
+        is shutting down or when it needs to release a tag for reallocation.
+
+        Args:
+            tag_id (str): The identifier of the OPC tag whose lease is to be dropped.
+
+        Lease dropping enables:
+        - Graceful shutdown of ingestor instances
+        - Dynamic reallocation of tags for load balancing
+        - Recovery from failed or unresponsive ingestor instances
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="delete"
+            - REDIS_OPERATIONS_DURATION: Records timing for lease dropping operations
+        """
+
+        await self.redis_repository.delete(f'lease:opc_tags:{tag_id}', metadata=self.metadata)
+
+    async def get_all_ingestors(self) -> list[str]:
+        """
+        Retrieves all active ingestors from Redis.
+
+        This method fetches all keys in Redis that match the pattern for ingestor
+        heartbeats and returns a list of active ingestor identifiers. The method
+        uses the pattern "heartbeat:ingestor:*" to find all active instances.
+
+        Returns:
+            List[str]: A list of active ingestor identifiers, extracted from
+                      the Redis keys by removing the "heartbeat:ingestor:" prefix.
+
+        This information is used for:
+        - Load balancing calculations
+        - System health monitoring
+        - Resource allocation decisions
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="keys"
+            - REDIS_OPERATIONS_DURATION: Records timing for ingestor discovery
+        """
+
+        return await self.redis_repository.keys('heartbeat:ingestor:*', metadata=self.metadata)
+
+    async def get_all_slots(self) -> list[str]:
+        """
+        Retrieves all available slots from Redis.
+
+        This method fetches all keys in Redis that match the pattern for OPC tag
+        slots and returns a list of slot identifiers. The method uses the pattern
+        "slot:opc_tags:*" to find all configured slots.
+
+        Returns:
+            List[str]: A list of slot identifiers, extracted from the Redis keys
+                      by removing the "slot:opc_tags:" prefix.
+
+        Slot information is used for:
+        - Resource allocation planning
+        - Load balancing across ingestor instances
+        - System capacity monitoring
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="keys"
+            - REDIS_OPERATIONS_DURATION: Records timing for slot discovery
+        """
+
+        return await self.redis_repository.keys('slot:opc_tags:*', metadata=self.metadata)
+
+    async def get_all_leases(self) -> list[str]:
+        """
+        Retrieves all active leases from Redis.
+
+        This method fetches all keys in Redis that match the pattern for OPC tag
+        leases and returns a list of lease identifiers. The method uses the pattern
+        "lease:opc_tags:*" to find all active leases.
+
+        Returns:
+            List[str]: A list of lease identifiers, extracted from the Redis keys
+                      by removing the "lease:opc_tags:" prefix.
+
+        Lease information is used for:
+        - Current resource utilization monitoring
+        - Load balancing calculations
+        - System health and performance analysis
+
+        Metrics:
+            - REDIS_OPERATIONS_TOTAL: Incremented with operation="keys"
+            - REDIS_OPERATIONS_DURATION: Records timing for lease discovery
+        """
+
+        return await self.redis_repository.keys('lease:opc_tags:*', metadata=self.metadata)