sientia-dataops-opc-ingestor/ingestor/managers/resource_manager.py

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)