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

SIENTIAPDE-1110

Browse Source 
Refactor OPC Manager metrics logging for improved readability

- Reformatted metrics logging statements in the OpcManager class for better readability and consistency.
- Enhanced debug logging to provide clearer insights into subscription management and connection status.
This commit is contained in:
vitor-aignosi
2025-07-08 12:25:11 -03:00
parent 9d510c5936
commit 7391ed3ba5
1 changed files with 34 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
ingestor/managers/opc_manager.py
View File

@@ -29,8 +29,10 @@ class OpcManager():
self.data_manager = data_manager self.data_manager = data_manager
self.notification_handler = notification_handler self.notification_handler = notification_handler
self.pod_id = pod_id self.pod_id = pod_id
metrics.OPC_CONNECTION_STATUS.labels(pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0) metrics.OPC_CONNECTION_STATUS.labels(
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0) 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): def __str__(self):
return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \
@@ -85,18 +87,22 @@ class OpcManager():
Exception: If the connection to the OPC server fails. Exception: If the connection to the OPC server fails.
""" """
metrics.OPC_CONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc() metrics.OPC_CONNECTIONS_TOTAL.labels(
pod_id=self.pod_id, server_name=self.name).inc()
try: try:
self.client = Client(self.url) self.client = Client(self.url)
if self.cert_path: if self.cert_path:
self.set_security() self.set_security()
self.logger.info(f'Starting connection to {self.name}...') self.logger.info(f'Starting connection to {self.name}...')
self.client.connect() self.client.connect()
metrics.OPC_CONNECTION_STATUS.labels(pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(1) 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.') self.logger.info(f'Connection to {self.name} successful.')
except Exception as e: except Exception as e:
metrics.OPC_CONNECTION_STATUS.labels(pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0) metrics.OPC_CONNECTION_STATUS.labels(
metrics.OPC_CONNECTIONS_FAILED.labels(pod_id=self.pod_id, server_name=self.name).inc() pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0)
metrics.OPC_CONNECTIONS_FAILED.labels(
pod_id=self.pod_id, server_name=self.name).inc()
self.logger.error(f"Failed to connect to {self.name}: {e}") self.logger.error(f"Failed to connect to {self.name}: {e}")
raise raise
@@ -121,9 +127,11 @@ class OpcManager():
p = period if period is not None else 500 p = period if period is not None else 500
self.subscriptions[name] = self.client.create_subscription(p, self) self.subscriptions[name] = self.client.create_subscription(p, self)
self.logger.info(f'Subscription {name} created on {self.name}.') 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() metrics.OPC_SUBSCRIPTIONS_CREATED.labels(
pod_id=self.pod_id, server_name=self.name, slot_name=name).inc()
except Exception as e: except Exception as e:
self.logger.error(f"Failed to create subscription {name} on {self.name}: {e}") self.logger.error(
f"Failed to create subscription {name} on {self.name}: {e}")
raise raise
def subscribe(self, subscription: str, nodes: dict, collect_period: int): def subscribe(self, subscription: str, nodes: dict, collect_period: int):
@@ -144,13 +152,18 @@ class OpcManager():
""" """
if not self.subscriptions.get(subscription): if not self.subscriptions.get(subscription):
raise ValueError("Subscription not created. Call create_subscription first.") 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 {subscription} on {self.name}...")
self.logger.info(f"Subscribing to nodes: {nodes}") self.logger.info(f"Subscribing to nodes: {nodes}")
self.addr_nodes = [self.client.get_node(n) for n in nodes if n not in self.nodes] self.addr_nodes = [self.client.get_node(
n) for n in nodes if n not in self.nodes]
self.logger.debug(f"Addr nodes: {self.addr_nodes}")
self.nodes.update(nodes) self.nodes.update(nodes)
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(len(self.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 self.collect_period = collect_period
for node, config in self.nodes.items(): for node, config in self.nodes.items():
@@ -216,8 +229,10 @@ class OpcManager():
finally: finally:
del self.client del self.client
self.client = None 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_CONNECTION_STATUS.labels(
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0) 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)
self.logger.warning("Disconnected from OPC UA server.") self.logger.warning("Disconnected from OPC UA server.")
def datachange_notification(self, node, _val, data): def datachange_notification(self, node, _val, data):
@@ -252,7 +267,8 @@ class OpcManager():
self.nodes[tag]['cycle_rule']['cycle_count'] = 0 self.nodes[tag]['cycle_rule']['cycle_count'] = 0
self.non_receive_count = 0 self.non_receive_count = 0
metrics.OPC_CYCLES_WITHOUT_DATA.labels(pod_id=self.pod_id, server_name=self.name).set(0) metrics.OPC_CYCLES_WITHOUT_DATA.labels(
pod_id=self.pod_id, server_name=self.name).set(0)
data = { data = {
'tag': tag, 'tag': tag,
@@ -295,7 +311,8 @@ class OpcManager():
""" """
self.non_receive_count += 1 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) 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: if self.non_receive_count >= 5:
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
@@ -305,7 +322,8 @@ class OpcManager():
level=NotificationLevel.ERROR level=NotificationLevel.ERROR
) )
if self.non_receive_count >= 15: if self.non_receive_count >= 15:
metrics.OPC_RECONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc() metrics.OPC_RECONNECTIONS_TOTAL.labels(
pod_id=self.pod_id, server_name=self.name).inc()
self.notification_handler.build_and_send_notification( self.notification_handler.build_and_send_notification(
notification_id=f'OPC_CONNECTION_RETRY__{self.name}', notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
message=f'Retrying to connect to server {self.name}', message=f'Retrying to connect to server {self.name}',