Code import - branch 1.4.2

ingestor/app.py

@@ -0,0 +1,150 @@
+import asyncio
+import os
+import signal
+import traceback
+from threading import Event
+from time import time
+
+from prometheus_client import start_http_server
+
+import ingestor.metrics as metrics
+from ingestor.ingestor import Ingestor
+
+exit_signal = Event()
+POD_ID = os.getenv('HOSTNAME', 'localhost')
+
+
+async def main():  # NOSONAR
+    """
+    Main asynchronous function that orchestrates the OPC Ingestor application.
+
+    This function performs the following operations:
+    1. Starts the Prometheus metrics server for monitoring
+    2. Initializes the Ingestor instance
+    3. Prepares the ingestor (connects to services, acquires slot leases)
+    4. Runs the main processing loop until shutdown is requested
+    5. Handles graceful shutdown and cleanup
+
+    The main loop continuously:
+    - Processes OPC data from subscribed tags
+    - Manages slot leases and resource allocation
+    - Monitors OPC server connections
+    - Records metrics for monitoring and observability
+
+    Environment Variables:
+        HOSTNAME: Pod identifier for metrics labeling (default: "localhost")
+        HTTP_METRICS_PORT: Port for Prometheus metrics server (default: 9090)
+
+    Raises:
+        Exception: If ingestor preparation fails, the application will exit
+    """
+    start_prometheus_server()
+    ingestor = Ingestor()
+    try:
+        await ingestor.prepare_ingestor()
+    except Exception as e:
+        metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()  # Increment errors
+        ingestor.logger.error(f'Failed to prepare ingestor: {e}')
+        exit_signal.set()
+    ingestor.logger.info('Ingestor prepared. Starting main loop.')
+
+    while not exit_signal.is_set():  # NOSONAR
+        start_time = time()  # Start loop timer
+        try:
+            await ingestor.loop()
+            metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc()  # Increment loop counter
+
+            # Use asyncio.sleep instead of exit_signal.wait for better async compatibility
+            await asyncio.sleep(ingestor.poll_interval)  # NOSONAR
+
+        except KeyboardInterrupt:  # Handle Ctrl+C gracefully
+            print('KeyboardInterrupt received. Setting exit_signal flag.')
+            exit_signal.set()
+        except Exception:
+            print('Exception in main loop. Setting exit_signal flag.')
+            traceback.print_exc()
+            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()  # Increment errors
+            exit_signal.set()
+        finally:
+            # Record loop duration
+            duration = time() - start_time
+            metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration)
+
+    await ingestor.shutdown()
+    metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
+
+    ingestor.logger.info('Main loop exit_signaled.')
+
+    # Give Prometheus a chance to scrape one last time before exiting (optional)
+    await asyncio.sleep(5)
+
+    os._exit(0)
+
+
+def signal_handler(_signum, _frame):
+    """
+    Signal handler for graceful application shutdown.
+
+    This function handles system signals (SIGINT, SIGTERM, SIGHUP) by setting
+    the exit_signal flag, which triggers the main loop to complete its current
+    iteration and then shut down gracefully.
+
+    Args:
+        _signum: The signal number received
+        _frame: The current stack frame (unused)
+    """
+    print(f'Received signal {_signum}. Setting exit_signal flag.')
+    exit_signal.set()
+
+
+def start_prometheus_server():
+    """
+    Starts the Prometheus metrics HTTP server.
+
+    This function initializes a Prometheus metrics server on the configured port
+    to expose application metrics for monitoring and alerting. The server provides
+    metrics about application health, performance, and operational status.
+
+    Environment Variables:
+        HTTP_METRICS_PORT: Port number for the metrics server (default: 9090)
+
+    Raises:
+        Exception: If the server fails to start, the application will exit
+    """
+    try:
+        port = int(os.getenv('HTTP_METRICS_PORT', 9090))
+        start_http_server(port)
+        print(f'Prometheus server started on port {port}.')
+        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
+    except Exception as e:
+        print(f'Failed to start Prometheus server: {e}')
+        os._exit(1)
+
+
+def run_async_main():
+    """
+    Run the async main function with proper event loop setup.
+
+    This function sets up the asyncio event loop and runs the main async function.
+    It handles KeyboardInterrupt gracefully and ensures proper cleanup of the event loop.
+
+    The function is designed to work with both direct execution and containerized
+    environments, providing consistent behavior across different deployment scenarios.
+    """
+    try:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        print('KeyboardInterrupt received in main thread.')
+        exit_signal.set()
+    finally:
+        loop.close()
+
+
+if __name__ == '__main__':
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+    signal.signal(signal.SIGHUP, signal_handler)
+
+    run_async_main()

ingestor/ingestor.py

@@ -0,0 +1,444 @@
+from copy import deepcopy
+from os import getenv
+from typing import Any
+
+from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+from sientia_do.observability.logger import get_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.ingestor_manager import IngestorManager
+
+
+class Ingestor(SientiaMonitoring):
+    """
+    Main OPC Ingestor class that orchestrates data collection from OPC UA servers.
+
+    The Ingestor is responsible for:
+    - 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 MongoDB
+    - Providing health monitoring and metrics collection
+
+    The ingestor uses a slot-based architecture where each slot represents
+    a collection of OPC tags that can be managed by a single ingestor instance.
+    This allows for horizontal scaling and load distribution.
+
+    Environment Variables:
+        REDIS_HOST: Redis server hostname (default: "localhost")
+        REDIS_PORT: Redis server port (default: 6379)
+        REDIS_USERNAME: Redis username (optional)
+        REDIS_PASSWORD: Redis password (optional)
+        LEASE_TTL: Time-to-live for slot leases in seconds (default: 10)
+        HEARTBEAT_TTL: Time-to-live for heartbeats in seconds (default: 20)
+        HOSTNAME: Pod identifier (default: "localhost")
+        POLL_INTERVAL: Main loop polling interval in seconds (default: 5)
+        MONGODB_URL: MongoDB server address (default: "localhost:27017")
+        MONGODB_USERNAME: MongoDB username (default: "sientia")
+        MONGODB_PASSWORD: MongoDB password (default: "sientia")
+        MONGODB_DATABASE: MongoDB database name (default: "sientia")
+
+    Attributes:
+        redis_host (str): Redis server hostname
+        redis_port (int): Redis server port
+        redis_username (str): Redis username (optional)
+        redis_password (str): Redis password (optional)
+        lease_ttl (int): Time-to-live for slot leases in seconds
+        heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds
+        pod_id (str): Identifier for the current pod or host
+        poll_interval (int): Interval in seconds for polling operations
+        mongo_database (str): MongoDB database name
+        mongo_connection_string (str): Complete MongoDB connection string
+        logger: Logger instance for application logging
+        notification_handler: Handler for sending notifications
+        metadata (dict): Application metadata for notifications and tracking
+        ingestor_manager: Manager instance for coordinating operations
+    """
+
+    def __init__(self):
+        """
+        Initializes the ingestor with configuration values retrieved from environment variables.
+
+        Sets up all necessary connections and configurations for:
+        - Redis for slot management and coordination
+        - MongoDB for data persistence and notifications
+        - OPC UA server management
+        - Metrics collection and monitoring
+        """
+
+        self.redis_host = getenv('REDIS_HOST', 'localhost')
+        self.redis_port = int(getenv('REDIS_PORT', '6379'))
+        self.redis_username = getenv('REDIS_USERNAME', None)
+        self.redis_password = getenv('REDIS_PASSWORD', None)
+        self.lease_ttl = int(getenv('LEASE_TTL', '10'))
+        self.heartbeat_ttl = int(getenv('HEARTBEAT_TTL', '20'))
+        self.pod_id = getenv('HOSTNAME', 'localhost')
+        self.poll_interval = int(getenv('POLL_INTERVAL', '5'))
+        mongo_url = getenv('MONGODB_URL', 'localhost:27017')
+        mongo_username = getenv('MONGODB_USERNAME', 'sientia')
+        mongo_password = getenv('MONGODB_PASSWORD', 'sientia')
+        self.mongo_database = getenv('MONGODB_DATABASE', 'sientia')
+        self.mongo_connection_string = f'mongodb://{mongo_username}:{mongo_password}@{mongo_url}'
+
+        self.logger = get_logger(__name__)
+        self.notification_handler = NotificationHandler(
+            connection_string=self.mongo_connection_string,
+            database=self.mongo_database,
+            logger=self.logger,
+            project_name='opc_ingestor',
+        )
+        self.metrics_controller = MetricsController(logger=self.logger)
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=self.logger,
+            notification_handler=self.notification_handler,
+            metrics_controller=self.metrics_controller,
+        )
+
+        self.metadata = {
+            'model_id': '-',
+            'model_name': '-',
+            'workflow_name': 'opc_ingestor',
+            'schema_name': 'opc_ingestor',
+            'pod_id': self.pod_id,
+        }
+        self.ingestor_manager: IngestorManager | None = None
+
+    async def shutdown(self):
+        """
+        Gracefully shuts down the ingestor and all its components.
+
+        This method ensures proper cleanup of:
+        - OPC UA connections and subscriptions
+        - Resource managers and data connections
+        - Active slot leases and heartbeats
+
+        Should be called before application termination to prevent resource leaks.
+        """
+        if self.ingestor_manager:
+            await self.ingestor_manager.shutdown()
+
+    async def handle_acquired_tags(self, acquired):
+        """
+        Handles the acquired tags by subscribing to them if available.
+
+        This method processes the tags that have been allocated to this ingestor
+        instance through the slot leasing system. It updates OPC server configurations
+        and establishes subscriptions to the allocated tags.
+
+        Args:
+            acquired (list): A list of acquired tags to be processed. If the list
+                             is empty or None, no action is taken other than logging
+                             a warning.
+
+        Behavior:
+            - If no tags are acquired, logs a warning about no slots being available
+            - If tags are acquired, updates OPC server configurations and subscribes
+              to the allocated tags for data collection
+        """
+
+        if not acquired:
+            self.logger.warning('No slots available')
+
+        else:
+            # Subscribe to acquired slots
+            if self.ingestor_manager:
+                await self.ingestor_manager.update_opc_servers()
+                await self.ingestor_manager.subscribe_to_tags(acquired)
+
+    async def prepare_ingestor(self):
+        """
+        Prepares the ingestor by initializing all components and acquiring initial slot leases.
+
+        This method performs the following steps:
+        1. Initializes the IngestorManager with all necessary configuration parameters
+        2. Declares the ingestor as active in the coordination system
+        3. Acquires slot leases for tag management
+        4. Processes the acquired tags and establishes OPC subscriptions
+
+        The preparation phase is critical for establishing the ingestor's role in
+        the distributed system and ensuring it can begin processing OPC data.
+
+        Raises:
+            Exception: If any error occurs during the initialization or lease acquisition process.
+                      This will cause the application to exit as the ingestor cannot function
+                      without proper initialization.
+        """
+
+        self.ingestor_manager = IngestorManager(
+            redis_data={
+                'host': self.redis_host,
+                'port': self.redis_port,
+                'username': self.redis_username,
+                'password': self.redis_password,
+            },
+            lease_ttl=self.lease_ttl,
+            heartbeat_ttl=self.heartbeat_ttl,
+            poll_interval=self.poll_interval,
+            mongo_connection_string=self.mongo_connection_string,
+            mongo_database=self.mongo_database,
+            metadata=self.metadata,
+            logger=self.logger,
+            notification_handler=self.notification_handler,
+            metrics_controller=self.metrics_controller,
+        )
+        assert self.ingestor_manager is not None
+
+        # Declare ingestor active
+        await self.ingestor_manager.declare_active()
+
+        # Get slot lease
+        acquired = await self.ingestor_manager.get_slot_leases()
+        self.logger.info(f'Acquired slots: {acquired}')
+
+        await self.handle_acquired_tags(acquired)
+
+        await self.emit_metric(
+            metric_object=metrics.SLOTS_MANAGED,
+            method='set',
+            value=len(self.ingestor_manager.managed_tags),
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+
+    async def manage_no_slots(self, number_of_slots: int):
+        """
+        Manages the scenario where there are no slots assigned to the ingestor.
+
+        This method handles the case where an ingestor is active but has no
+        allocated slots. It attempts to acquire a slot lease if slots are
+        available in the system.
+
+        Args:
+            number_of_slots (int): The number of available slots in the system.
+
+        Behavior:
+            - Only attempts to acquire slots if the ingestor is currently active
+              (has no managed tags) and there are slots available
+            - Requests a single slot lease to begin processing
+        """
+
+        if self.ingestor_manager and not self.ingestor_manager.managed_tags and number_of_slots > 0:
+            # This ingestor is active and has no slots, so we need to try to
+
+            # Get slot lease
+            await self.ingestor_manager.get_slot_leases(1)
+
+    async def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int):
+        """
+        Manages the allocation and deallocation of slot leases for ingestors.
+
+        This method implements the load balancing logic for distributing OPC tag
+        processing across multiple ingestor instances. It ensures optimal resource
+        utilization and fair distribution of work.
+
+        Args:
+            available_slots (int): The number of slots currently available for allocation.
+            lacking_ingestors (int): The number of ingestors that are active and without slots.
+            slot_diff (int): The difference between the total slots and the required slots.
+
+        Behavior:
+            - If there are available slots and lacking ingestors, attempts to acquire
+              slot leases for the available slots and processes the acquired tags.
+            - If there are no lacking ingestors but there are extra slots (slot_diff > 0),
+              releases the extra slot leases to ensure proper allocation.
+
+        Logs:
+            - Logs the number of available slots when attempting to acquire leases.
+            - Logs the number of extra slots when releasing leases.
+        """
+        if not self.ingestor_manager:
+            return
+
+        if available_slots > 0 and lacking_ingestors > 0:
+            # Some ingestors are inactive, so there are "available_slots" slots available
+            self.logger.info(f'Slots available: {available_slots}')
+
+            # Get slot lease
+            await self.ingestor_manager.get_slot_leases(available_slots)
+
+        elif lacking_ingestors <= 0 and slot_diff > 0:
+            self.logger.info(f'Extra slots available: {slot_diff}')
+            # There's enough slots for all ingestors, but this ingestor has more than one slot
+            # So we need to drop the extra leases
+
+            overleases = list(self.ingestor_manager.managed_tags.keys())[1:]
+
+            await self.ingestor_manager.drop_slot_leases(overleases)
+
+            for lease in overleases:
+                await self.ingestor_manager.unsubscribe_slot(lease)
+                self.ingestor_manager.managed_tags.pop(lease)
+
+            await self.emit_metric(
+                metric_object=metrics.SLOTS_MANAGED,
+                method='set',
+                value=len(self.ingestor_manager.managed_tags),
+                tags={
+                    'pod_id': self.pod_id,
+                },
+            )
+
+    async def update_ingestor_manager(self, old_managed_tags: dict[str, Any]):
+        """
+        Updates the ingestor manager with new managed tags and handles configuration changes.
+
+        This method compares the current managed tags with the previous state and
+        performs necessary operations to maintain synchronization:
+        - Subscribes to newly allocated tags
+        - Resubscribes to tags with changed configurations
+        - Unsubscribes from deallocated tags
+
+        Args:
+            old_managed_tags (Dict[str, Any]): The previous state of managed tags.
+
+        Behavior:
+            - Compares current and previous tag configurations
+            - Establishes subscriptions for new tags
+            - Updates subscriptions for modified tags
+            - Removes subscriptions for deallocated tags
+            - Updates metrics to reflect current state
+        """
+
+        if not self.ingestor_manager:
+            return
+
+        self.logger.debug(f'Current managed tags: {self.ingestor_manager.managed_tags}')
+
+        await self.ingestor_manager.update_opc_servers()
+        new_managed_tags = deepcopy(self.ingestor_manager.managed_tags)
+
+        self.logger.debug(
+            f'Comparing new managed tags {new_managed_tags} with old managed tags {old_managed_tags}'
+        )
+
+        keys = set(new_managed_tags) | set(old_managed_tags)
+
+        changes = {
+            k: (new_managed_tags.get(k), old_managed_tags.get(k))
+            for k in keys
+            if new_managed_tags.get(k) != old_managed_tags.get(k)
+        }
+
+        self.logger.debug(f'Changes: {changes}')
+
+        for slot, config in new_managed_tags.items():
+            if slot not in old_managed_tags:
+                self.logger.info(f'Subscribing to new slot {slot}')
+                await self.ingestor_manager.subscribe_to_tags({slot: config})
+                continue
+
+            if config != old_managed_tags[slot]:
+                self.logger.info(f'Resubscribing to slot {slot}')
+                await self.ingestor_manager.unsubscribe_slot(slot)
+                await self.ingestor_manager.subscribe_to_tags({slot: config})
+
+        for slot in old_managed_tags.keys():
+            if slot not in new_managed_tags:
+                self.logger.info(f'Unsubscribing from slot {slot}')
+                await self.ingestor_manager.unsubscribe_slot(slot)
+
+        await self.emit_metric(
+            metric_object=metrics.SLOTS_MANAGED,
+            method='set',
+            value=len(self.ingestor_manager.managed_tags),
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+
+    async def loop(self):
+        """
+        Executes the main processing loop for managing ingestors and slots.
+
+        This method is the core of the ingestor's operation, performing the following
+        tasks in each iteration:
+        1. Declares the ingestor as active to maintain its presence in the system
+        2. Polls for slot updates and manages resource allocation
+        3. Handles scenarios where no slots are available
+        4. Manages slot leases based on system load and available resources
+        5. Updates OPC server configurations and checks server integrity
+        6. Synchronizes managed tags with the current system state
+
+        The loop implements a sophisticated load balancing algorithm that:
+        - Distributes OPC tag processing across multiple ingestor instances
+        - Ensures optimal resource utilization
+        - Maintains system stability during scaling operations
+        - Provides real-time monitoring and metrics collection
+
+        This method is intended to be called repeatedly to ensure the ingestor
+        manager operates correctly and maintains synchronization with the slots
+        and OPC servers.
+        """
+
+        if not self.ingestor_manager:
+            return
+
+        await self.ingestor_manager.declare_active()
+
+        self.logger.info('Polling for slot updates...')
+        # Get active ingestors
+
+        current_managed_tags = deepcopy(self.ingestor_manager.managed_tags)
+
+        ingestors = await self.ingestor_manager.get_active_ingestors()
+        number_of_ingestors = len(ingestors)
+        number_of_leases = await self.ingestor_manager.get_number_of_leases()
+        number_of_slots = await self.ingestor_manager.get_number_of_slots()
+
+        # Update active ingestors gauge
+        await self.emit_metric(
+            metric_object=metrics.ACTIVE_INGESTORS,
+            method='set',
+            value=number_of_ingestors,
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+
+        # Handle no slots
+        self.logger.info('Managing no slots...')
+        await self.manage_no_slots(number_of_slots)
+
+        available_slots = number_of_slots - number_of_leases
+        lacking_ingestors = number_of_slots - number_of_ingestors
+        slot_diff = len(self.ingestor_manager.managed_tags) - 1
+
+        self.logger.info('Managing leases...')
+        await self.manage_leases(available_slots, lacking_ingestors, slot_diff)
+
+        # Update managed slots gauge
+        await self.emit_metric(
+            metric_object=metrics.SLOTS_MANAGED,
+            method='set',
+            value=len(self.ingestor_manager.managed_tags),
+            tags={
+                'pod_id': self.pod_id,
+            },
+        )
+
+        self.logger.debug(
+            f'Active ingestors: {ingestors}, '
+            f'Number of slots: {number_of_slots}, '
+            f'Number of leases: {number_of_leases}, '
+            f'Managed tags: {self.ingestor_manager.managed_tags}, '
+            f'Managed servers: {self.ingestor_manager.opc_managers}'
+        )
+        if not self.ingestor_manager.managed_tags:
+            # No slots acquired
+            self.logger.info('No slots acquired in this loop')
+
+        # Update opc servers
+        self.logger.info('Updating slot config...')
+        await self.ingestor_manager.update_slot_config()
+
+        # Check OPC cycles
+        self.logger.info('Checking OPC servers integrity...')
+        await self.ingestor_manager.check_opc_servers_integrity()
+
+        self.logger.info('Updating managed tags...')
+        await self.update_ingestor_manager(current_managed_tags)

ingestor/managers/data_manager.py

@@ -0,0 +1,138 @@
+import os
+import traceback
+
+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 operations for the OPC Ingestor.
+
+    The DataManager is responsible for:
+    - Storing OPC data in MongoDB for historical analysis and persistence
+    - Managing database connections and ensuring data integrity
+    - Providing data access interfaces for other components
+
+    Args:
+        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
+
+    Attributes:
+        pod_id (str): Pod identifier for metrics labeling
+        connection_string (str): MongoDB connection string
+        database (str): MongoDB database name
+        mongo_client (MongoClient): MongoDB client instance
+        metadata (dict): Application metadata
+    """
+
+    def __init__(
+        self,
+        mongo_connection_string: str,
+        mongo_database: str,
+        metadata: dict,
+        logger: Logger,
+        notification_handler: NotificationHandler,
+        metrics_controller: MetricsController,
+    ) -> None:
+        """
+        Initializes the DataManager instance with a MongoDB connection.
+
+        Args:
+            mongo_connection_string (str): MongoDB connection string
+            mongo_database (str): MongoDB database name
+            metadata (dict): Application metadata
+            logger (Logger): Logger instance
+            notification_handler (NotificationHandler): Notification handler
+        """
+
+        self.pod_id = os.getenv('HOSTNAME', 'localhost')
+
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            metrics_controller=metrics_controller,
+            notification_handler=notification_handler,
+        )
+
+        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.
+        """
+        try:
+            self.mongo_repository.close()
+        except Exception as e:
+            self.logger.error(f'Error closing MongoDB client: {e}')
+
+    def __del__(self):
+        self.shutdown()
+
+    async def publish(self, topic: str, data: dict) -> None:
+        """
+        Persists a message to MongoDB.
+
+        Args:
+            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
+        """
+
+        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,683 @@
+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
+    - 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:
+        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
+
+    Attributes:
+        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
+        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,
+        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,
+    ):
+        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(
+            mongo_connection_string=mongo_connection_string,
+            mongo_database=mongo_database,
+            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,591 @@
+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
+        assert cert is not None and private_key is not 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.description = 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)

ingestor/metrics.py

@@ -0,0 +1,141 @@
+"""
+Prometheus metrics configuration for the OPC Ingestor application.
+
+This module defines all the metrics used for monitoring and observability
+of the OPC Ingestor system. It includes metrics for:
+
+- Application health and performance
+- OPC server connections and subscriptions
+- Data processing and storage operations
+- Resource management and load balancing
+- Error tracking and notification systems
+
+All metrics follow Prometheus naming conventions and include appropriate
+labels for multi-dimensional analysis and alerting.
+"""
+
+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']
+REDIS_LABELS = ['pod_id', 'operation']
+NOTIFICATION_LABELS = ['pod_id', 'level', 'block']
+
+MAIN_LABELS = ['pod_id']
+
+# --- Reliability Metrics ---
+TAG_WRITTEN_COUNT = Counter(
+    'ingestor_tag_written_count',
+    'Number of writing process to the collection',
+    [*MAIN_LABELS, 'tag_name', 'collection_name'],
+)
+
+# --- General Application Metrics ---
+APP_LOOP_COUNT = Counter(
+    'app_main_loop_total',
+    'Total number of times the application main loop has run',
+    POD_ID_LABEL,
+)
+APP_LOOP_DURATION = Histogram(
+    'app_main_loop_duration_seconds',
+    'Duration of the application main loop in seconds',
+    POD_ID_LABEL,
+)
+APP_ERRORS_TOTAL = Counter(
+    'app_errors_total',
+    'Total number of unhandled errors in the main loop',
+    POD_ID_LABEL,
+)
+APP_UP = Gauge(
+    'app_up',
+    'Indicates if the application is running (1) or shutting down (0)',
+    POD_ID_LABEL,
+)
+
+# --- Ingestor Manager Metrics ---
+ACTIVE_INGESTORS = Gauge(
+    'ingestor_active_total',
+    'Number of active ingestors reported by Redis',
+    POD_ID_LABEL,
+)
+SLOTS_TOTAL = Gauge(
+    'ingestor_slots_total',
+    'Total number of slots configured in Redis',
+    POD_ID_LABEL,
+)
+LEASES_TOTAL = Gauge(
+    'ingestor_leases_total',
+    'Total number of leases (allocated slots) in Redis',
+    POD_ID_LABEL,
+)
+SLOTS_MANAGED = Gauge(
+    'ingestor_slots_managed_current',
+    'Number of slots currently managed by this ingestor instance',
+    POD_ID_LABEL,
+)
+SLOTS_ACQUIRED = Counter(
+    'ingestor_slots_acquired_total',
+    'Total number of slots acquired by this instance',
+    POD_ID_LABEL,
+)
+SLOTS_RELEASED = Counter(
+    'ingestor_slots_released_total',
+    'Total number of slots released by this instance',
+    POD_ID_LABEL,
+)
+
+# --- OPC Manager Metrics ---
+OPC_MANAGERS_ACTIVE = Gauge(
+    'ingestor_opc_managers_active',
+    'Number of active OPC Managers in this instance',
+    POD_ID_LABEL,
+)
+OPC_SUBSCRIPTION_ERRORS = Counter(
+    'ingestor_opc_subscription_errors_total',
+    'Errors when trying to subscribe to OPC tags',
+    ['pod_id', 'server', 'slot'],
+)
+
+OPC_CONNECTIONS_TOTAL = Counter(
+    'opc_connections_initiated_total',
+    'Total connection attempts to OPC servers',
+    ['pod_id', 'server_name'],
+)
+OPC_CONNECTIONS_FAILED = Counter(
+    'opc_connections_failed_total',
+    'Total failed connection attempts to OPC servers',
+    ['pod_id', 'server_name'],
+)
+OPC_CONNECTION_STATUS = Gauge(
+    'opc_connection_status',
+    'Connection status with the OPC server (1=connected, 0=disconnected)',
+    SERVER_LABELS,
+)
+OPC_SUBSCRIPTIONS_CREATED = Counter(
+    'opc_subscriptions_created_total',
+    'Total OPC subscriptions created',
+    ['pod_id', 'server_name', 'slot_name'],
+)
+OPC_TAGS_SUBSCRIBED = Gauge(
+    'opc_tags_subscribed_current',
+    'Current number of OPC tags subscribed on a server',
+    ['pod_id', 'server_name'],
+)
+OPC_CYCLES_WITHOUT_DATA = Gauge(
+    'opc_cycles_without_data',
+    'Current number of cycles without receiving data from a server',
+    ['pod_id', 'server_name'],
+)
+OPC_RECONNECTIONS_TOTAL = Counter(
+    'opc_reconnections_tried_total',
+    'Reconnection attempts to an OPC server after a loss',
+    ['pod_id', 'server_name'],
+)
+
+# --- Notification Metrics ---
+NOTIFICATIONS_SENT = Counter(
+    'notifications_sent_total',
+    'Total number of notifications sent',
+    NOTIFICATION_LABELS,
+)