gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f415207d5bbe884a7d56477cb2a4a064c7e0cbfe
sientia-dataops-opc-ingestor/ingestor/managers/opc_manager.py
vitor-aignosi 6d2a0e46be Refactor OPC server integrity checks to be asynchronous 
Updated the Ingestor and IngestorManager classes to use async methods for checking OPC server integrity and shutting down OPC managers. Adjusted related tests to ensure proper mocking of asynchronous behavior.
2025-10-23 11:46:02 -03:00

523 lines
21 KiB
Python
Raw Blame History

import asyncio
import json
import traceback
from pathlib import Path
from asyncua import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
import ingestor.metrics as metrics
from ingestor.managers.data_manager import DataManager
class OpcManager(BaseActivity):
"""
Manages OPC UA server connections and tag subscriptions.
The OpcManager is responsible for:
- Establishing and maintaining secure connections to OPC UA servers
- Managing tag subscriptions and data collection
- Handling server reconnection and error recovery
- Processing OPC data and forwarding it to the data manager
- Monitoring connection health and performance metrics
The manager supports both secure and unsecured connections, with optional
certificate-based authentication for enhanced security.
Args:
name (str): Unique identifier for the OPC server
url (str): OPC UA server endpoint URL
data_manager (DataManager): Manager for data persistence and export
logger (Logger): Logger instance for application logging
server_uri (str): OPC UA server application URI
notification_handler (NotificationHandler): Handler for sending notifications
metadata (dict): Application metadata for notifications and tracking
cert_path (str, optional): Path to client certificate file for secure connections
private_key_path (str, optional): Path to client private key file
server_cert_path (str, optional): Path to server certificate file for validation
Attributes:
url (str): OPC UA server endpoint URL
name (str): Unique identifier for the OPC server
server_uri (str): OPC UA server application URI
data_queue (dict): Queue for buffering OPC data before processing
non_receive_count (int): Counter for cycles without data reception
client (Client): OPC UA client instance
cert_path (str): Path to client certificate file
private_key_path (str): Path to client private key file
server_cert_path (str): Path to server certificate file
nodes (dict): Dictionary of OPC node references
subscriptions (dict): Active OPC subscriptions
data_manager (DataManager): Manager for data persistence and export
metadata (dict): Application metadata
"""
def __init__(
self,
name: str,
url: str,
subscription_period_ms: int,
data_manager: DataManager,
logger: Logger,
server_uri: str,
notification_handler: NotificationHandler,
metadata: dict,
cert_path: str | None = None,
private_key_path: str | None = None,
server_cert_path: str | None = None,
):
self.url = url
self.name = name
self.server_uri = server_uri
self.data_queue: dict = {}
self.non_receive_count = 0
self.client: Client | None = None
self.subscription_period_ms = subscription_period_ms
self.cert_path = cert_path
self.private_key_path = private_key_path
self.server_cert_path = server_cert_path
self.nodes: dict = {}
self.subscriptions: dict = {}
self.data_manager = data_manager
self.metadata = metadata
BaseActivity.__init__(
self, logger=logger, notification_handler=notification_handler, set_error_counter=True
)
metrics.OPC_CONNECTION_STATUS.labels(
pod_id=self.pod_id, server_name=self.name, server_url=self.url
).set(0)
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0)
def __str__(self):
"""
String representation of the OPC Manager.
Returns:
str: Human-readable representation showing server details and current state.
"""
return (
f'OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n'
f'nodes={self.nodes}, subscriptions={self.subscriptions}'
)
def __del__(self):
asyncio.run(self.shutdown())
async def shutdown(self):
"""
Comprehensive cleanup method for graceful shutdown.
This method ensures proper cleanup of all OPC UA resources:
- Closes active subscriptions
- Disconnects from the OPC server
- Releases allocated resources
Should be called before the application terminates to prevent resource leaks
and ensure clean disconnection from OPC servers.
"""
try:
await self.disconnect()
except Exception as e:
self.logger.error(f'Error during cleanup: {e}')
async def set_security(self):
"""
Configures the security settings for the OPC UA client.
This method sets up the security policy, certificates, and timeouts
required for establishing a secure connection with the OPC UA server.
It implements Basic256 security policy with certificate-based authentication.
Raises:
ValueError: If either the certificate path or private key path is not provided.
Security Settings:
- Security Policy: Basic256
- Secure Channel Timeout: 10,000,000 ms
- Session Timeout: 10,000,000 ms
The method configures:
- Client application URI
- Certificate-based authentication
- Server certificate validation (if provided)
- Connection timeouts for stability
"""
if not all([self.cert_path, self.private_key_path]):
raise ValueError(
'Certificate and private key paths must be provided for secure connection.'
)
cert = Path(self.cert_path) if self.cert_path else None
private_key = Path(self.private_key_path) if self.private_key_path else None
server_cert = Path(self.server_cert_path) if self.server_cert_path else None
if self.client:
await self.client.set_application_uri(self.server_uri)
self.logger.info('Setting security...')
await self.client.set_security(
SecurityPolicyBasic256,
certificate=str(cert),
private_key=str(private_key),
server_certificate=str(server_cert),
)
await self.client.set_secure_channel_timeout(10000000)
await self.client.set_session_timeout(10000000)
async def connect(self):
"""
Establishes a connection to the OPC server.
This method initializes the OPC client using the provided URL and
sets up security if a certificate path is specified. It then
attempts to connect to the server and logs the connection status.
The connection process includes:
1. Client initialization with server URL
2. Security configuration (if certificates are provided)
3. Connection establishment
4. Metrics recording for monitoring
Raises:
Exception: If the connection to the OPC server fails.
Metrics:
- OPC_CONNECTIONS_TOTAL: Incremented on connection attempt
- OPC_CONNECTION_STATUS: Set to 1 on successful connection
"""
metrics.OPC_CONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc()
try:
self.client = Client(self.url, timeout=10, watchdog_intervall=3600000)
assert self.client is not None # Informa ao mypy que client não é None
self.client.name = self.pod_id
self.client.application_name = self.pod_id
pod_uri = self.pod_id.replace('-', ':')
self.client.application_uri = pod_uri
self.client.product_uri = pod_uri
if self.cert_path:
await self.set_security()
self.logger.info(f'Starting connection to {self.name}...')
await self.client.connect()
metrics.OPC_CONNECTION_STATUS.labels(
pod_id=self.pod_id, server_name=self.name, server_url=self.url
).set(1)
self.logger.info(f'Connection to {self.name} successful.')
except Exception:
await self.disconnect()
raise
async def create_subscription(self, name: str):
"""
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.subscriptions[name]`.
- Logs the creation of the subscription.
- Increments subscription creation metrics.
"""
if not self.client:
raise ValueError('Client not connected. Call connect first.')
try:
self.subscriptions[name] = await self.client.create_subscription(
self.subscription_period_ms, self
)
self.logger.info(f'Subscription {name} created on {self.name}.')
metrics.OPC_SUBSCRIPTIONS_CREATED.labels(
pod_id=self.pod_id, server_name=self.name, slot_name=name
).inc()
except Exception as e:
self.logger.error(f'Failed to create subscription {name} on {self.name}: {e}')
raise
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:
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):
raise ValueError('Subscription not created. Call create_subscription first.')
self.logger.info(f'Subscribing to {subscription} on {self.name}...')
self.logger.info(f'Subscribing to nodes: {nodes}')
assert self.client is not None # Informa ao mypy que client não é None
addr_nodes = [self.client.get_node(n) for n in nodes]
self.logger.debug(f'Addr nodes: {addr_nodes}')
self.nodes.update(nodes)
self.logger.debug(f'Nodes: {self.nodes}')
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(
len(self.nodes)
)
self.collect_period = collect_period
for node, config in self.nodes.items():
self.nodes[node]['cycle_rule'] = {
'cycle_increment': collect_period * 1000 / float(config['frequency']),
'cycle_count': 0,
}
await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
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.
- If the subscription does not exist, no action is taken.
"""
if not self.subscriptions.get(subscription):
self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
return
await self.subscriptions[subscription].delete()
del self.subscriptions[subscription]
self.logger.info(f'Unsubscribed from {subscription}.')
async def disconnection_fallback(self) -> list:
"""
Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try.
"""
assert self.client is not None
error_stack = []
for i in range(5):
try:
self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5')
await self.client.disconnect()
return []
except Exception as e:
self.logger.error(
f'Failed to disconnect from OPC UA serve in attempt {i + 1} of 5: {e}'
)
error_stack.append(
{
'attempt': i + 1,
'error': str(e),
'traceback': traceback.format_exc(),
}
)
await asyncio.sleep(0.1 * i)
return error_stack
async def disconnect(self):
"""
Disconnects from the OPC UA server.
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')
if self.client is None:
self.logger.warning('Client already disconnected.')
return
try:
for sub in self.subscriptions:
await self.subscriptions[sub].delete()
self.logger.warning('Deleted all subscriptions.')
except Exception as sub_error:
self.logger.error(f'Failed to clean up subscription: {sub_error}')
errors = await self.disconnection_fallback()
if errors:
self.send_notification(
metadata=self.metadata,
notification_id=f'OPC_DISCONNECTION_ERROR_{self.name}',
message=f'Failed to disconnect from OPC UA server {self.name} after 5 attempts',
block='opc_manager',
level=NotificationLevel.ERROR,
attachment_content=json.dumps(errors, indent=4),
)
else:
self.logger.warning('Disconnected from OPC UA server.')
del self.client
self.client = None
metrics.OPC_CONNECTION_STATUS.labels(
pod_id=self.pod_id, server_name=self.name, server_url=self.url
).set(0)
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0)
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.
- Resets the non-receive count.
- Constructs a data dictionary containing the tag, tag name, timestamp, and value.
- Publishes the data to all topics associated with the node.
"""
# get data value
monitored_item = data.monitored_item
value = monitored_item.Value.Value.Value
# source_timestamp
source_timestamp = monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE)
tag = str(node)
self.logger.debug(
f'Data change notification received for tag:'
f'{tag} after {self.nodes[tag]["cycle_rule"]["cycle_count"]} cycles'
)
self.nodes[tag]['cycle_rule']['cycle_count'] = 0
self.non_receive_count = 0
metrics.OPC_CYCLES_WITHOUT_DATA.labels(pod_id=self.pod_id, server_name=self.name).set(0)
data = {
'tag': tag,
'name': self.nodes[str(node)]['tag_name'],
'timestamp': source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ),
'value': value,
}
for topic in self.nodes[tag]['topics']:
self.data_manager.publish(topic, data)
def check_cycles(self):
"""
Checks the cycle counts for all monitored nodes and sends
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['cycle_rule']['cycle_increment']
if self.nodes[node]['cycle_rule']['cycle_count'] >= 5:
name = config['tag_name']
cycles = self.nodes[node]['cycle_rule']['cycle_count']
self.send_notification(
metadata=self.metadata,
notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED',
message=f'{cycles} cycles without receive from {node}:{name}',
block='opc_manager',
level=NotificationLevel.WARNING,
)
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 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
metrics.OPC_CYCLES_WITHOUT_DATA.labels(pod_id=self.pod_id, server_name=self.name).set(
self.non_receive_count
)
if self.non_receive_count >= 5:
self.send_notification(
metadata=self.metadata,
notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
message=f'{self.non_receive_count} cycles without '
f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
block='opc_manager',
level=NotificationLevel.ERROR,
)
if self.non_receive_count >= 15:
metrics.OPC_RECONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc()
self.send_notification(
metadata=self.metadata,
notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
message=f'Retrying to connect to server {self.name}',
block='opc_manager',
level=NotificationLevel.ERROR,
)
return True
return False
Reference in New Issue View Git Blame Copy Permalink