gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1312

Browse Source 
SIENTIAPDE-1318 Enhance OpcManager disconnection handling

- Introduced a disconnection_fallback method to attempt multiple disconnects from the OPC UA server, improving reliability.
- Updated error handling during connection attempts to ensure proper logging and notification of disconnection failures.
- Refactored disconnect method to utilize the new fallback mechanism and maintain metrics accurately.
This commit is contained in:
vitor-aignosi
2025-10-20 15:27:05 -03:00
parent 44a9367663
commit 7001b5e9bd
1 changed files with 49 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
ingestor/managers/opc_manager.py
View File

@@ -1,5 +1,7 @@
import asyncio
import json
from pathlib import Path
import traceback
from asyncua import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256
@@ -198,7 +200,7 @@ class OpcManager(BaseActivity):
pod_uri = f'urn:sientia-do:pod:{self.pod_id}'
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}...')
@@ -208,11 +210,10 @@ class OpcManager(BaseActivity):
).set(1)
self.logger.info(f'Connection to {self.name} successful.')
except Exception as e:
if self.client:
try:
await self.client.disconnect()
except Exception as internal_e:
self.logger.error(f'And error occurred while creating connection from {self.name}, and an exception occurred while disconnecting: {internal_e}')
try:
await self.disconnect()
except Exception as internal_e:
self.logger.error(f'And error occurred while creating connection from {self.name}, and an exception occurred while disconnecting: {internal_e}')
del self.client
self.client = None
@@ -333,6 +334,28 @@ class OpcManager(BaseActivity):
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 server: {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.
@@ -363,21 +386,28 @@ class OpcManager(BaseActivity):
except Exception as sub_error:
self.logger.error(f'Failed to clean up subscription: {sub_error}')
# Inserir backoff para disconnect, emitindo métrica e alerta
try:
await self.client.disconnect()
except Exception as conn_error:
self.logger.error(f'Failed to disconnect from OPC UA server: {conn_error}')
finally:
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)
errors = 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.