import traceback from typing import Dict, List from copy import deepcopy from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity from sientia_do.temporal.utils.logger import Logger from ingestor.managers.data_manager import DataManager from ingestor.managers.opc_manager import OpcManager from ingestor.managers.resource_manager import ResourceManager import ingestor.metrics as metrics class IngestorManager(BaseActivity): def __init__(self, kafka_servers: str, redis_data: dict, lease_ttl: int, heartbeat_ttl: int, poll_interval: int, mongo_connection_string: str, mongo_database: str, metadata: dict, logger: Logger, notification_handler: NotificationHandler, export_to_kafka: bool = False): redis_host = redis_data.get('host') redis_port = redis_data.get('port') redis_username = redis_data.get('username', None) redis_password = redis_data.get('password', None) self.data_manager = DataManager( kafka_servers=kafka_servers, mongo_connection_string=mongo_connection_string, mongo_database=mongo_database, export_to_kafka=export_to_kafka, metadata=metadata, logger=logger, notification_handler=notification_handler ) self.opc_managers = {} 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, ) self.number_of_slots = 0 self.poll_interval = poll_interval self.managed_tags = {} self.opc_servers = {} self.metadata = metadata BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler, set_error_counter=True) def initialize_opc_from_config(self, server_config: dict) -> 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: self.logger.info( f"Initializing OpcManager at {server_config['url']}") manager = OpcManager( name=server_config['name'], url=server_config['url'], 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') ) manager.config = server_config 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 def shutdown(self): for _server_name, server in self.opc_managers.items(): server.disconnect() self.data_manager.shutdown() def __del__(self): self.shutdown() 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.debug( f"Initializing OPC manager for server {server}" ) server_instance = self.initialize_opc_from_config( server_config ) elif server_instance.config != server_config: self.logger.warning( f"Reinitializing OPC manager for server {server}" ) server_instance.disconnect() del self.opc_managers[server] server_instance = 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." ) _a = [self.managed_tags[slot].pop(server, None) for slot, _value in current_managed_tags.items()] 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. " f"Desconnecting from server." ) self.opc_managers[server].disconnect() self.opc_managers.pop(server, None) metrics.OPC_MANAGERS_ACTIVE.labels( pod_id=self.pod_id).set(len(self.opc_managers)) def check_opc_servers_integrity(self): """ Checks the integrity of the OPC servers and updates the OPC servers if necessary. """ for server, opc_manager in self.opc_managers.items(): opc_manager.check_cycles() is_lost = opc_manager.check_opc_listenning() if is_lost: self.logger.warning( f"OPC server {server} is lost. " f"Server will be disconnected." ) for slot, _config in self.managed_tags.items(): self.managed_tags[slot].pop(server, None) metrics.OPC_MANAGERS_ACTIVE.labels( pod_id=self.pod_id).set(len(self.opc_managers)) 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 metrics.LEASES_TOTAL.set(self.number_of_slots) 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 metrics.SLOTS_TOTAL.set(self.number_of_slots) 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 metrics.SLOTS_ACQUIRED.labels(pod_id=self.pod_id).inc() if len(acquired) >= max_slots: self.managed_tags.update(acquired) metrics.SLOTS_MANAGED.labels( pod_id=self.pod_id).set(len(self.managed_tags)) return acquired self.logger.warning( f"Unable to acquire {max_slots} slots. " f"Only {acquired} slots were leased." ) self.managed_tags.update(acquired) metrics.SLOTS_MANAGED.labels( pod_id=self.pod_id).set(len(self.managed_tags)) 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: removed_slots.append(slot) continue self.managed_tags[slot] = update for slot in removed_slots: self.managed_tags.pop(slot, None) 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 lease_id in ids: self.resource_manager.drop_tag_lease(lease_id) metrics.SLOTS_RELEASED.labels(pod_id=self.pod_id).inc() 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: metrics.OPC_SUBSCRIPTION_ERRORS.labels( pod_id=self.pod_id, server=server, slot=slot).inc() trace = traceback.format_exc() self.send_notification( 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( "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)