From e35eb300cb71f416035272d67591d17a4539062f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 13:58:43 -0300 Subject: [PATCH] SIENTIAPDE-1169 Implement OPC writing metrics and enhance OPC class initialization - Added metrics for counting predictions written to the OPC server and monitoring their response times. - Enhanced the OPC class initialization to include the pod ID for better tracking. - Updated the write method to increment the prediction count and observe response times. --- laborious/activities/opc.py | 6 ++++-- laborious/metrics.py | 13 ++++++++++++ laborious/utils/repository/opc_repository.py | 22 +++++++++++++++++++- 3 files changed, 38 insertions(+), 3 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 23e3ac4..4834288 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -22,6 +22,9 @@ class OPC(BaseActivity): self.notification_handler = notification_handler self.opc_servers = opc_servers + BaseActivity.__init__( + self, logger, notification_handler, set_error_counter=True) + self.opc_repository: dict[str, OpcRepository] = {} for id, server in opc_servers.items(): self.opc_repository[id] = OpcRepository( @@ -34,6 +37,7 @@ class OPC(BaseActivity): server_cert_path=server['server_cert_path'], notification_handler=self.notification_handler, reconnection_interval=server['reconnection_interval'], + pod_id=self.pod_id ) is_connected, error_data = self.opc_repository[id].connect() if not is_connected: @@ -51,8 +55,6 @@ class OPC(BaseActivity): attachment_content=error_data.get( 'attachment_content', None) ) - BaseActivity.__init__( - self, logger, notification_handler, set_error_counter=True) def write_data(self, server_id: str, tag: str, data: Any, data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: diff --git a/laborious/metrics.py b/laborious/metrics.py index a3adcf9..6580eef 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -26,3 +26,16 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram( CORE_LABELS, buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] ) + +PREDICTION_OPC_WRITING_COUNT = Counter( + "laborious_prediction_opc_writing_count", + "Number of predictions written to the OPC server", + CORE_LABELS, +) + +PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR = Histogram( + "laborious_prediction_opc_writing_response_time_monitor", + "Current response time of each prediction written to the OPC server", + CORE_LABELS, + buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] +) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 031e171..d96d811 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -1,4 +1,5 @@ import traceback +import time from datetime import datetime from pathlib import Path from typing import Any @@ -6,6 +7,7 @@ from asyncua.sync import Client from asyncua.crypto.security_policies import SecurityPolicyBasic256 from asyncua.ua import DataValue, Variant, VariantType, DateTime from regex import F +import metrics from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.utils.logger import Logger @@ -38,7 +40,7 @@ class OpcRepository(): def __init__(self, id: str, url: str, logger: Logger, notification_handler: NotificationHandler, reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None, - private_key_path: str = None, server_cert_path: str = None): + private_key_path: str = None, server_cert_path: str = None, pod_id: str = None): self.url = url self.id = id self.server_uri = server_uri @@ -90,6 +92,7 @@ class OpcRepository(): ) self.client.secure_channel_timeout = 10000000 self.client.session_timeout = 10000000 + self.pod_id = pod_id def connect(self) -> tuple[bool, dict[str, Any]]: """ @@ -215,6 +218,8 @@ class OpcRepository(): if not is_connected: return False, error + start_time = time.time() + try: node = self.client.get_node(node) except Exception as e: @@ -256,6 +261,21 @@ class OpcRepository(): try: node.write_value(ua_data) + + metrics.PREDICTION_OPC_WRITING_COUNT.labels( + pod_id=self.pod_id, + model_name=metadata['model_name'], + pipeline_name=metadata['workflow_name'] + ).inc() + + end_time = time.time() + response_time = end_time - start_time + metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels( + pod_id=self.pod_id, + model_name=metadata['model_name'], + pipeline_name=metadata['workflow_name'] + ).observe(response_time) + except Exception as e: trace = traceback.format_exc() logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))