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

@@ -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:

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]
+)

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'))