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

from logging import Logger
import traceback
from typing import Dict, List
from copy import deepcopy
from ingestor.managers.data_manager import DataManager
from ingestor.managers.opc_manager import OpcManager
from ingestor.managers.resource_manager import ResourceManager


class IngestorManager():
    def __init__(self,
                 kafka_servers: str, redis_host: str, redis_port: int,
                 lease_ttl: int, heartbeat_ttl: int,  pod_id: str,
                 poll_interval: int, logger: Logger,
                 redis_username: str = None, redis_password: str = None):

        self.data_manager = DataManager(kafka_servers, logger)
        self.opc_managers = {}
        self.resource_manager = ResourceManager(
            redis_host, redis_port, lease_ttl, heartbeat_ttl, pod_id, redis_username, redis_password
        )
        self.number_of_slots = 0
        self.poll_interval = poll_interval
        self.logger = logger
        self.managed_tags = {}
        self.opc_servers = {}

    def initialize_opc_from_config(self, server_config: dict, data_manager: DataManager, logger: Logger) -> OpcManager | None:
        """
        Initializes an OPC Manager instance using the provided server configuration.
        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.
            data_manager (DataManager): An instance of the DataManager to handle data operations.
            logger (Logger): A logger instance for logging messages.
        Returns:
            OpcManager | None: An initialized OpcManager instance if successful,
            otherwise None if an error occurs during initialization.
        """

        try:
            manager = OpcManager(
                server_config['name'], server_config['url'], data_manager, logger, server_config['server_uri'],
                server_config.get('cert_path'), server_config.get(
                    'private_key_path'), server_config.get('server_cert_path')
            )

            manager.config = server_config
            manager.connect()
        except Exception as e:
            logger.error(f"Failed to initialize OpcManager: {e}")
            return None

        return manager

    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 = []
        for slot, slot_config in self.managed_tags.items():
            for server, server_config in slot_config.items():
                registered_servers.append(server)
                server_config = server_config.copy()
                server_config.pop('tags', None)
                server_instance = None
                if server not in self.opc_managers:

                    server_instance = self.initialize_opc_from_config(
                        server_config, self.data_manager, self.logger
                    )

                elif self.opc_managers[server].config != server_config:
                    del self.opc_managers[server]
                    server_instance = self.initialize_opc_from_config(
                        server_config, self.data_manager, self.logger
                    )

                if server_instance is not None:
                    self.opc_managers[server] = server_instance

        for server in list(self.opc_managers.keys()):
            if server not in registered_servers:
                self.logger.warning(
                    f"Server {server} not found in managed tags. "
                    f"Desconnecting from server."
                )
                self.opc_managers[server].disconnect()
                self.opc_managers.pop(server, None)

    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.
        """

        self.resource_manager.ingestor_heartbeat()

    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.
        """

        ingestors = self.resource_manager.get_all_ingestors()
        return ingestors if ingestors else []

    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.
        """

        leases = self.resource_manager.get_all_leases()
        self.number_of_slots = len(leases) if leases else 0
        return self.number_of_slots

    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.
        """

        slots = self.resource_manager.get_all_slots()
        self.number_of_slots = len(slots) if slots else 0
        return self.number_of_slots

    def get_slot_leases(self, max_slots: int = 1) -> Dict:
        """
        Acquires a specified number of resource slots by leasing them from the resource manager.
        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:
            - Iterates through available slots and attempts to lease them using the resource manager.
            - Logs the leasing of each slot.
            - Updates the `managed_tags` attribute with the acquired slots.
            - Stops leasing once the specified `max_slots` are acquired.
            - If unable to acquire the requested number of slots, logs a warning and returns the slots that were leased.
        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 self.resource_manager.lease_tag(str(i)):
                self.logger.info(f"Leased slot {i}")
                slots = self.resource_manager.get_tag_slot(str(i))
                if slots is None:
                    continue
                acquired[str(i)] = slots

                if len(acquired) >= max_slots:
                    self.managed_tags.update(acquired)
                    return acquired

        self.logger.warning(
            f"Unable to acquire {max_slots} slots. "
            f"Only {acquired} slots were leased."
        )
        self.managed_tags.update(acquired)
        return acquired

    def unsubscribe_slot(self, slot: str):
        """
        Unsubscribes a specific slot from all associated OPC servers.
        Args:
            slot (str): The name of the slot to unsubscribe.
        Raises:
            KeyError: If the specified slot does not exist in the managed tags.
        """

        for server in self.managed_tags[slot].keys():
            if server in self.opc_managers:
                self.opc_managers[server].unsubscribe(slot)

    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 = []
        update = {}
        for slot, slot_config in self.managed_tags.items():
            self.resource_manager.renew_tag_lease(slot)
            update = self.resource_manager.get_tag_slot(slot)
            if update is None:
                self.logger.warning(
                    f"Slot {slot} configuration not found. "
                    f"Removing slot from managed tags."
                )
                self.unsubscribe_slot(slot)
                removed_slots.append(slot)

                continue

            if update != slot_config:
                self.logger.info(
                    f"Slot {slot} configuration updated. "
                    f"Old: {slot_config}, New: {update}"
                )
                self.managed_tags[slot] = update

                self.unsubscribe_slot(slot)
                self.update_opc_servers()
                self.subscribe_to_tags({slot: update})

        for slot in removed_slots:
            del self.managed_tags[slot]
        self.update_opc_servers()

    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.
        Args:
            ids (List[str]): A list of slot IDs for which the leases 
                             should be released.
        Returns:
            None
        """

        for id in ids:
            self.resource_manager.drop_tag_lease(id)

    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.
        """

        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:
                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
            )
            self.opc_managers[server].subscribe(
                slot, deepcopy(tags_to_sub), self.poll_interval
            )
            self.logger.info(
                tags_to_sub
            )
        except Exception as e:
            self.logger.error(
                f"Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}"
            )
            self.logger.error(traceback.format_exc())
            self.logger.warning(
                "Removing subscription from server "
                f"{server} for slot {slot}"
            )
            self.opc_managers[server].unsubscribe(slot)
            return 2
        return 0

    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.
        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.
        """

        to_remove = []
        self.logger.info(tags)
        for slot, slot_config in tags.items():
            for server, server_config in slot_config.items():
                response = self.manage_server(
                    slot, server, server_config, tags
                )
                if response == 2:
                    to_remove.append([slot, server])

        for slot, server in to_remove:
            self.managed_tags[slot].pop(server, None)