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, number_of_ingestors: int, poll_interval: int, logger: Logger): self.data_manager = DataManager(kafka_servers, logger) self.opc_managers = {} self.resource_manager = ResourceManager( redis_host, redis_port, lease_ttl, heartbeat_ttl, pod_id ) self.number_of_ingestors = number_of_ingestors 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): 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): self.resource_manager.ingestor_heartbeat() def get_active_ingestors(self) -> List[str]: ingestors = self.resource_manager.get_all_ingestors() return ingestors if ingestors else [] def get_slot_leases(self, max_slots: int = 1) -> Dict: acquired = {} for i in range(1, self.number_of_ingestors + 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): 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): 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: for id in ids: self.resource_manager.drop_tag_lease(id) def subscribe_to_tags(self, tags: Dict) -> None: # NOSONAR to_remove = [] self.logger.info(tags) for slot, slot_config in tags.items(): for server, server_config in slot_config.items(): 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." ) continue 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}" ) to_remove.append([slot, server]) continue 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) to_remove.append([slot, server]) for slot, server in to_remove: self.managed_tags[slot].pop(server, None)