From 26f776f8e1a497d6893438beaaa7a8ddf9def530 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 1 Sep 2025 12:47:39 -0300 Subject: [PATCH] SIENTIAPDE-1084 Enhance documentation and improve shutdown methods for DataManager and IngestorManager - Updated README.md to clarify persistent storage capabilities. - Enhanced docstrings in DataManager and IngestorManager for shutdown processes, detailing resource cleanup and error handling. - Improved comments in various methods to provide clearer explanations of functionality and side effects. --- README.md | 8 +- ingestor/managers/data_manager.py | 33 ++++++- ingestor/managers/ingestor_manager.py | 124 ++++++++++++++++++++++---- ingestor/managers/opc_manager.py | 62 +++++++++++-- 4 files changed, 196 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 8ab24d8..d7ef4ad 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ A high-performance, scalable OPC UA data ingestion system designed for industria - **OPC UA Integration**: Native support for OPC UA servers with secure and unsecured connections - **Automatic Load Balancing**: Slot-based architecture for horizontal scaling across multiple instances - **Real-time Data Streaming**: Kafka integration for historical data streaming -- **Persistent Storage**: MongoDB integration for historical data streaming +- **Persistent Storage**: MongoDB integration for historical data persistence - **Health Monitoring**: Comprehensive Prometheus metrics and health checks - **Fault Tolerance**: Automatic failover, reconnection, and error recovery @@ -19,12 +19,12 @@ A high-performance, scalable OPC UA data ingestion system designed for industria - **Notification System**: Integrated alerting and notification management - **Performance Optimization**: Configurable polling intervals and data collection frequencies -## ️ Architecture +## Architecture -The OPC Ingestor uses a modular, manager-based architecture: +The OPC Ingestor uses a modular, manager-based architecture designed for scalability and fault tolerance: ``` -┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ +┌──────────────────────────────────────────── │ Main App │ │ IngestorManager │ │ OPC Manager │ │ │◄──►│ │◄──►│ │ │ - Signal Hand. │ │ - Slot Mgmt │ │ - Connections │ diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index f9d7ca1..e186321 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -149,7 +149,17 @@ class DataManager(BaseActivity): set_error_counter=True) def shutdown(self): - """Closes the Kafka producer connection.""" + """ + Gracefully shuts down the DataManager and closes all connections. + + This method ensures proper cleanup of: + - Kafka producer connection with message flushing + - MongoDB client connection + - Metrics recording for connection status + + The method handles connection closure gracefully, logging any errors + that occur during shutdown while ensuring all resources are properly released. + """ if self.kafka_producer: try: self.kafka_producer.flush(timeout=10) @@ -176,13 +186,30 @@ class DataManager(BaseActivity): self.shutdown() def delivery_report(self, msg: str): - """Callback for delivery reports from Kafka.""" + """ + Callback for successful Kafka message delivery reports. + + This method is called by the Kafka producer when a message is successfully + delivered to a topic. It logs the delivery details including topic, partition, + and offset information for debugging and monitoring purposes. + + Args: + msg: Kafka message object containing delivery details + """ self.logger.debug( f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}" ) def delivery_error(self, err: str): - """Callback for delivery reports from Kafka.""" + """ + Callback for Kafka message delivery error reports. + + This method is called by the Kafka producer when a message delivery fails. + It logs the error details for debugging and monitoring purposes. + + Args: + err: Error information from the failed delivery attempt + """ self.logger.error(f"Delivery failed for record : {err}") def publish(self, topic: str, data: dict) -> None: diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index edaaf2e..7206b5f 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -164,6 +164,17 @@ class IngestorManager(BaseActivity): 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() @@ -252,6 +263,17 @@ class IngestorManager(BaseActivity): 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 """ for server, opc_manager in self.opc_managers.items(): opc_manager.check_cycles() @@ -272,8 +294,14 @@ class IngestorManager(BaseActivity): 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. + `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 """ self.resource_manager.ingestor_heartbeat() @@ -281,10 +309,15 @@ class IngestorManager(BaseActivity): 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 = self.resource_manager.get_all_ingestors() @@ -293,10 +326,16 @@ class IngestorManager(BaseActivity): 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 = self.resource_manager.get_all_leases() @@ -307,10 +346,16 @@ class IngestorManager(BaseActivity): 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 = self.resource_manager.get_all_slots() @@ -321,22 +366,29 @@ class IngestorManager(BaseActivity): 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: - - 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. + - 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. + (i.e., `get_tag_slot` returns None), that slot is skipped. """ acquired = {} @@ -367,10 +419,20 @@ class IngestorManager(BaseActivity): 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(): @@ -381,6 +443,7 @@ class IngestorManager(BaseActivity): """ 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. @@ -389,15 +452,18 @@ class IngestorManager(BaseActivity): 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. + - 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. + 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. + - Logs warnings for removed slots. + - Logs informational messages for updated slot configurations. """ removed_slots = [] @@ -417,14 +483,22 @@ class IngestorManager(BaseActivity): 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. + 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: self.resource_manager.drop_tag_lease(lease_id) @@ -433,28 +507,38 @@ class IngestorManager(BaseActivity): 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( @@ -511,19 +595,27 @@ class IngestorManager(BaseActivity): 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 = [] diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 2685155..2e8d4f4 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -194,16 +194,22 @@ class OpcManager(BaseActivity): async def create_subscription(self, name: str, period: int = 500): """ 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.subscription`. + - Creates a subscription and assigns it to `self.subscriptions[name]`. - Logs the creation of the subscription. + - Increments subscription creation metrics. """ if not self.client: @@ -222,18 +228,26 @@ class OpcManager(BaseActivity): 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: - nodes (dict): A dictionary where keys are node identifiers (e.g., node - IDs or paths) and values are configurations for each node. Each - configuration must include a 'frequency' key indicating the - frequency of data collection in Hz. + 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): @@ -261,11 +275,17 @@ class OpcManager(BaseActivity): 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. @@ -283,12 +303,20 @@ class OpcManager(BaseActivity): async def disconnect(self): """ Disconnects from the OPC UA server. - This method handles the disconnection process by deleting the subscription + + 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') @@ -319,14 +347,17 @@ class OpcManager(BaseActivity): 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. @@ -365,12 +396,15 @@ class OpcManager(BaseActivity): def check_cycles(self): """ Checks the cycle counts for all monitored nodes and sends - notifications if thresholds are exceeded. + 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[ @@ -389,8 +423,20 @@ class OpcManager(BaseActivity): 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, False otherwise. + 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