SIENTIAPDE-1199

Update tests in Gates and OpcRepository to enhance logging and include metadata

- Removed unnecessary debug assertion in the Gates activity test for format prediction.
- Updated the test for last timestamp with no data to include a metadata dictionary.
- Modified OpcRepository tests to utilize custom_error logging with metadata for improved error context.

laborious/activities/activities.py

@@ -3,7 +3,7 @@ from temporalio import activity, workflow
 with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.postgres import Postgres
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
-    from sientia_do.temporal.utils.logger import Logger
+    from sientia_do.observability.logger import Logger
     from laborious.activities.mlflow import MLFlow
     from laborious.activities.gates import Gates
     from laborious.activities.opc import OPC

laborious/activities/gates.py

@@ -6,7 +6,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.base import BaseActivity
-    from sientia_do.temporal.utils.logger import Logger
+    from sientia_do.observability.logger import Logger
     from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter
     from typing import Any
     from laborious.utils.filters.conditional_filters import (
@@ -15,6 +15,7 @@ with workflow.unsafe.imports_passed_through():
     )
     from pandas import DataFrame
     from datetime import datetime
+    from laborious import metrics
 
 input_filter_functions = {
     'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
@@ -47,7 +48,8 @@ mlflow_content_filter_functions = {
 
 class Gates(BaseActivity):
     def __init__(self, logger: Logger, notification_handler: NotificationHandler):
-        BaseActivity.__init__(self, logger, notification_handler)
+        BaseActivity.__init__(
+            self, logger, notification_handler, set_error_counter=True)
 
     @activity.defn(name="input_gate")
     async def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
@@ -66,7 +68,7 @@ class Gates(BaseActivity):
         """
         metadata = input_data['metadata']
 
-        self.debug("Performing input gate...", metadata)
+        self.info("Performing input gate...", metadata)
 
         self.debug(f"Input data: {input_data}", metadata)
 
@@ -101,11 +103,11 @@ class Gates(BaseActivity):
 
         for path_flag in path_priority:
             if path_flag in filter_output:
-                self.debug(f"Input gate result: {path_flag}", metadata)
+                self.info(f"Input gate result: {path_flag}", metadata)
                 return path_flag, input_filter_functions['path_confidence'][path_flag], \
                     "Input data with bad quality"
 
-        self.debug("Nothing was filtered by the input gate", metadata)
+        self.info("Nothing was filtered by the input gate", metadata)
         return None, 0, ""
 
     @activity.defn(name="mlflow_response_gate")
@@ -126,7 +128,7 @@ class Gates(BaseActivity):
         """
 
         metadata = input_data['metadata']
-        self.debug("Performing mlflow response gate...", metadata)
+        self.info("Performing mlflow response gate...", metadata)
 
         filters = input_data['filters']
         data = input_data['data']
@@ -151,7 +153,7 @@ class Gates(BaseActivity):
                         notification_id=f"{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}",
                         message=data['content']['message'],
                         block="mlflow_gate",
-                        level=NotificationLevel.WARNING,
+                        level=NotificationLevel.ERROR,
                         attachment_content=data['content']['traceback']
                     )
             except Exception as e:
@@ -167,12 +169,12 @@ class Gates(BaseActivity):
 
         for path_flag in path_priority:
             if path_flag in filter_output:
-                self.debug(
+                self.info(
                     f"Mlflow response gate result: {path_flag}", metadata)
                 return path_flag, mlflow_response_filter_functions['path_confidence'][path_flag], \
                     ", ".join(comments)
 
-        self.debug("Nothing was filtered by the mlflow response gate", metadata)
+        self.info("Nothing was filtered by the mlflow response gate", metadata)
         return None, 0, ""
 
     @activity.defn(name="mlflow_content_gate")
@@ -193,7 +195,7 @@ class Gates(BaseActivity):
         """
 
         metadata = input_data['metadata']
-        self.debug("Performing mlflow content gate...", metadata)
+        self.info("Performing mlflow content gate...", metadata)
 
         filters = input_data['filters']
         data = DataFrame(input_data['data'])
@@ -232,12 +234,12 @@ class Gates(BaseActivity):
 
         for path_flag in path_priority:
             if path_flag in filter_output:
-                self.debug(
+                self.info(
                     f"Mlflow content gate result: {path_flag}", metadata)
                 return path_flag, mlflow_content_filter_functions['path_confidence'][path_flag], \
                     "Transformed data not passed the content filter"
 
-        self.debug("Nothing was filtered by the mlflow content gate", metadata)
+        self.info("Nothing was filtered by the mlflow content gate", metadata)
         return None, 0, ""
 
     @activity.defn(name="format_prediction")
@@ -254,7 +256,7 @@ class Gates(BaseActivity):
             dict: The formatted data.
         """
         metadata = input_data['metadata']
-        self.debug("Formatting prediction...", metadata)
+        self.info("Formatting prediction...", metadata)
 
         data = DataFrame(input_data['data'])
         data['timestamp'] = input_data['timestamp']
@@ -264,6 +266,8 @@ class Gates(BaseActivity):
         data['comments'] = ""
         data = data.sort_values(by='timestamp')
 
+        self.info(f"Prediction formatted: {data.size} rows", metadata)
+
         return data.to_dict()
 
     @activity.defn(name="format_default_prediction")
@@ -285,7 +289,7 @@ class Gates(BaseActivity):
         metadata = input_data['metadata']
         self.debug("Formatting default prediction...", metadata)
 
-        return DataFrame({
+        data = DataFrame({
             'prediction': [0],
             'response_time': [0],
             'timestamp': [input_data['timestamp']],
@@ -293,7 +297,10 @@ class Gates(BaseActivity):
             'prediction_confidence': [input_data['prediction_confidence']],
             'prediction_status': ['Bad'],
             'comments': [input_data['comment']]
-        }).to_dict()
+        })
+
+        self.info(f"Default prediction formatted: {data.size} rows", metadata)
+        return data.to_dict()
 
     @activity.defn(name="get_last_timestamp")
     async def get_last_timestamp(self, input_data: dict[str, Any]) -> str:
@@ -305,7 +312,53 @@ class Gates(BaseActivity):
         Returns:
             str: The last timestamp of the data.
         """
+        metadata = input_data['metadata']
+
+        self.info("Getting last timestamp...", metadata)
+
         data = DataFrame(input_data['data'])
+
         if data.empty:
             return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+
+        self.info(
+            f"Last timestamp: {max(data['timestamp'].values.tolist())}", metadata)
+
         return max(data['timestamp'].values.tolist())
+
+    @activity.defn(name="write_metrics")
+    async def write_metrics(self, input_data: dict[str, Any]):
+        """
+        Write metrics to the database.
+        input_data:
+            metadata: dict[str, Any]
+            prediction: dict[str, Any]
+        """
+        metadata = input_data['metadata']
+        prediction = DataFrame(input_data['prediction'])
+        prediction_confidence = prediction['prediction_confidence'].values[0]
+        response_time = prediction['response_time'].values[0]
+
+        self.info(
+            f"Writing metrics for model {metadata['model_name']}", metadata)
+
+        metrics.PREDICTIONS_WRITTEN_COUNT.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).inc()
+
+        metrics.PREDICTION_CONFIDENCE_MONITOR.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).set(prediction_confidence)
+
+        metrics.PREDICTION_RESPONSE_TIME_MONITOR.labels(
+            pod_id=self.pod_id,
+            model_name=metadata['model_name'],
+            pipeline_name=metadata['workflow_name']
+        ).observe(response_time)
+
+        self.info(
+            f"Metrics written for model {metadata['model_name']}", metadata)

laborious/activities/mlflow.py

@@ -1,3 +1,4 @@
+import json
 from temporalio import activity, workflow
 
 
@@ -5,7 +6,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.base import BaseActivity
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
-    from sientia_do.temporal.utils.logger import Logger
+    from sientia_do.observability.logger import Logger
     from laborious.utils.repository.model_repository import MLFlowRepository
     from typing import Any
     import numpy as np
@@ -16,14 +17,15 @@ with workflow.unsafe.imports_passed_through():
 class MLFlow(BaseActivity):
     def __init__(self, mlflow_host: str, mlflow_port: int, mlflow_username: str,
                  mlflow_password: str, logger: Logger, notification_handler: NotificationHandler):
-        BaseActivity.__init__(self, logger, notification_handler)
+        BaseActivity.__init__(
+            self, logger, notification_handler, set_error_counter=True)
         self.mlflow_host = mlflow_host
         self.mlflow_port = mlflow_port
         self.mlflow_username = mlflow_username
         self.mlflow_password = mlflow_password
 
         self.model_monitoring_repository = MLFlowRepository(
-            f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password
+            f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger.base_logger
         )
 
     @activity.defn(name="request_transform")
@@ -39,7 +41,7 @@ class MLFlow(BaseActivity):
             dict[str, Any]: The transformed data.
         """
         metadata = input_data['metadata']
-        self.debug('Transforming data...', metadata)
+        self.info('Transforming data...', metadata)
         data = DataFrame(input_data['data'])
         model_name = input_data['model_name']
         model_retention = input_data['model_retention']
@@ -65,8 +67,10 @@ class MLFlow(BaseActivity):
         response_data = self.model_monitoring_repository.transform(
             model_name, data, model_retention)
 
-        self.debug("Response data:", metadata)
+        self.debug("Transform response data:", metadata)
-        self.debug(response_data, metadata)
+        self.debug(json.dumps(response_data, indent=4), metadata)
+
+        self.info("Data transformed successfully", metadata)
 
         return response_data
 
@@ -83,7 +87,7 @@ class MLFlow(BaseActivity):
             dict[str, Any]: The predicted data.
         """
         metadata = input_data['metadata']
-        self.debug('Predicting data...', metadata)
+        self.info('Predicting data...', metadata)
         data = DataFrame(input_data['data'])
         model_name = input_data['model_name']
         model_retention = input_data['model_retention']
@@ -95,7 +99,10 @@ class MLFlow(BaseActivity):
         response_data = self.model_monitoring_repository.predict(
             model_name, data, model_retention)
 
-        self.debug(response_data, metadata)
+        self.debug("Prediction response data:", metadata)
+        self.debug(json.dumps(response_data, indent=4), metadata)
+
+        self.info("Prediction completed successfully", metadata)
 
         return response_data
 

laborious/activities/opc.py

@@ -5,7 +5,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.base import BaseActivity
-    from sientia_do.temporal.utils.logger import Logger
+    from sientia_do.observability.logger import Logger
     from laborious.utils.repository.opc_repository import OpcRepository
     from typing import Any
     import traceback
@@ -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,7 +55,6 @@ class OPC(BaseActivity):
                     attachment_content=error_data.get(
                         'attachment_content', None)
                 )
-        BaseActivity.__init__(self, logger, notification_handler)
 
     def write_data(self, server_id: str, tag: str, data: Any,
                    data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool:
@@ -96,6 +99,59 @@ class OPC(BaseActivity):
             )
             raise e
 
+    def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool:
+        if self.opc_repository.get(server_id) is None:
+            message = f"OPC server {server_id} not found to perform write operation."
+            self.send_notification(
+                metadata=metadata,
+                notification_id="OPC_SERVER_NOT_FOUND",
+                message=message,
+                block="write_opc_data",
+                level=NotificationLevel.ERROR,
+                attachment_content=f"OPC servers: {list(self.opc_repository.keys())}"
+            )
+            return False
+        return True
+
+    def manage_output_tags(
+            self, server_id: str, config: dict[str, Any], data: DataFrame,
+            metadata: dict[str, Any], success: bool) -> tuple[bool, int]:
+
+        count = 0
+        if 'prediction_tags' in config:
+            for tag, tag_config in config['prediction_tags'].items():
+                local_success = self.write_data(
+                    server_id=server_id,
+                    tag=tag,
+                    data=data.head(1)['prediction'].values[0],
+                    data_type=tag_config['data_type'],
+                    tag_type='prediction',
+                    metadata=metadata
+                )
+                if local_success:
+                    self.info(
+                        f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata)
+                    count += 1
+                success = success and local_success
+
+        if 'confidence_tags' in config:
+            for tag, tag_config in config['confidence_tags'].items():
+                local_success = self.write_data(
+                    server_id=server_id,
+                    tag=tag,
+                    data=data.head(1)['prediction_confidence'].values[0],
+                    data_type=tag_config['data_type'],
+                    tag_type='confidence',
+                    metadata=metadata
+                )
+                if local_success:
+                    self.info(
+                        f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata)
+                    count += 1
+                success = success and local_success
+
+        return success, count
+
     @activity.defn(name='write_opc_data')
     async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]:
         """
@@ -116,47 +172,28 @@ class OPC(BaseActivity):
 
         """
         metadata = input_data['metadata']
-        self.debug("Writing data to OPC servers...", metadata)
+        self.info("Writing data to OPC servers...", metadata)
         data = DataFrame(input_data['data'])
         opc_output_config = input_data['opc_output_config']
-        self.debug(data, metadata)
+        self.info(f"Data to write: {data.size} rows", metadata)
 
         success = True
 
+        success_count = 0
+
         for server_id, config in opc_output_config.items():
-            if self.opc_repository.get(server_id) is None:
+
-                message = f"OPC server {server_id} not found to perform write operation."
+            if not self.validate_server(server_id, metadata):
-                self.send_notification(
-                    metadata=metadata,
-                    notification_id="OPC_SERVER_NOT_FOUND",
-                    message=message,
-                    block="write_opc_data",
-                    level=NotificationLevel.ERROR,
-                    attachment_content=f"OPC servers: {list(self.opc_repository.keys())}"
-                )
                 success = False
+                continue
 
-            if 'prediction_tags' in config:
+            local_success, local_count = self.manage_output_tags(
-                for tag, tag_config in config['prediction_tags'].items():
+                server_id, config, data, metadata, success)
-                    success = success and self.write_data(
+            success = success and local_success
-                        server_id=server_id,
+            success_count += local_count
-                        tag=tag,
-                        data=data.head(1)['prediction'].values[0],
-                        data_type=tag_config['data_type'],
-                        tag_type='prediction',
-                        metadata=metadata
-                    )
 
-            if 'confidence_tags' in config:
+            self.info(
-                for tag, tag_config in config['confidence_tags'].items():
+                f"Data written to OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata)
-                    self.write_data(
-                        server_id=server_id,
-                        tag=tag,
-                        data=data.head(1)['prediction_confidence'].values[0],
-                        data_type=tag_config['data_type'],
-                        tag_type='confidence',
-                        metadata=metadata
-                    )
 
         return self.process_confidence(data, success, metadata)
 

laborious/metrics.py

@@ -0,0 +1,41 @@
+from prometheus_client import Gauge, Counter, Histogram
+
+APP_UP = Gauge(
+    "app_up",
+    "Indicates if the application is running (1) or shutting down (0)",
+    ["pod_id"],
+)
+
+CORE_LABELS = ["pod_id", "model_name", "pipeline_name"]
+
+PREDICTIONS_WRITTEN_COUNT = Counter(
+    "laborious_predictions_written_count",
+    "Number of predictions written to the database table predictions",
+    CORE_LABELS,
+)
+
+PREDICTION_CONFIDENCE_MONITOR = Gauge(
+    "laborious_prediction_confidence_monitor",
+    "Current confidence of each prediction",
+    CORE_LABELS,
+)
+
+PREDICTION_RESPONSE_TIME_MONITOR = Histogram(
+    "laborious_prediction_response_time_monitor",
+    "Current response time of each prediction",
+    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, "opc_server_id"],
+)
+
+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, "opc_server_id"],
+    buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0]
+)

laborious/utils/repository/model_repository.py

@@ -19,10 +19,11 @@ from sientia.ModelServing import ModelServing
 
 
 class MLFlowRepository():
-    def __init__(self, host, username, password):
+    def __init__(self, host, username, password, logger):
 
         self.model_serving = ModelServing(tracking_uri=host,
-                                          username=username, password=password)
+                                          username=username, password=password,
+                                          logger=logger)
 
     def transform(self, model_name: str, data: pd.DataFrame, model_retention: int):
         """
@@ -68,7 +69,7 @@ class MLFlowRepository():
         try:
             start_time = datetime.now()
             data = self.model_serving.get_cached_predict(
-                model_name, data, model_retention)[-1:]
+                model_name, data, model_retention)
 
             end_time = datetime.now()
             data = pd.DataFrame(data, columns=['prediction'])

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
@@ -8,7 +9,8 @@ from asyncua.ua import DataValue, Variant, VariantType, DateTime
 from regex import F
 from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
 from sientia_do.notifications.models import NotificationLevel
-from sientia_do.temporal.utils.logger import Logger
+from sientia_do.observability.logger import Logger
+from laborious import metrics
 
 data_type_map = {
     'float': {
@@ -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
@@ -51,6 +53,14 @@ class OpcRepository():
         self.last_reconnection_time = None
         self.notification_handler = notification_handler
         self.client = None
+        self.pod_id = pod_id
+
+        self.metadata = {
+            'model_name': '-',
+            'model_id': '-',
+            'workflow_name': 'opc_repository',
+            'schedule_name': '-'
+        }
 
     def set_security(self):
         """
@@ -81,7 +91,7 @@ class OpcRepository():
             self.server_cert_path) if self.server_cert_path else None
 
         self.client.application_uri = self.server_uri
-        self.logger.info('Setting security...')
+        self.logger.custom_info('Setting security...', self.metadata)
         self.client.set_security(
             SecurityPolicyBasic256,
             certificate=str(cert),
@@ -104,7 +114,8 @@ class OpcRepository():
         self.client = Client(self.url)
         if self.cert_path:
             self.set_security()
-        self.logger.info(f'Starting connection to OPC server {self.id}...')
+        self.logger.custom_info(
+            f'Starting connection to OPC server {self.id}...', self.metadata)
         return self.try_connect()
 
     def try_connect(self) -> tuple[bool, dict[str, Any]]:
@@ -121,7 +132,7 @@ class OpcRepository():
             return True, {}
         except Exception as e:
             trace = traceback.format_exc()
-            self.logger.error(trace)
+            self.logger.custom_error(trace, self.metadata)
 
             return False, {
                 "notification_id": f"OPC_CONNECTION_ERROR_{self.id}",
@@ -139,9 +150,11 @@ class OpcRepository():
             return
         try:
             self.client.disconnect()
-            self.logger.info('Disconnected from OPC server')
+            self.logger.custom_info(
+                'Disconnected from OPC server', self.metadata)
         except Exception as e:
-            self.logger.error(f"Failed to disconnect from OPC server: {e}")
+            self.logger.custom_error(
+                f"Failed to disconnect from OPC server: {e}", self.metadata)
         self.client = None
 
     def __del__(self):
@@ -151,7 +164,8 @@ class OpcRepository():
         try:
             self.disconnect()
         except Exception as e:
-            self.logger.error(f"Error in destructor: {e}")
+            self.logger.custom_error(
+                f"Error in destructor: {e}", self.metadata)
 
     def validate_connection(self) -> tuple[bool, dict[str, Any]]:
         """
@@ -168,32 +182,39 @@ class OpcRepository():
             return self.connect()
 
         if self.error_count > 5:
-            self.logger.warning(
+            self.logger.custom_warning(
-                f"OPC server {self.id} will be disconnected due to multiple errors")
+                f"OPC server {self.id} will be disconnected due to multiple errors", self.metadata)
             try:
                 self.disconnect()
             except Exception as e:
                 trace = traceback.format_exc()
-                self.logger.error(f"Failed to disconnect from OPC server: {e}")
+                self.logger.custom_error(
-                self.logger.error(trace)
+                    f"Failed to disconnect from OPC server: {e}", self.metadata)
-            self.logger.info(
+                self.logger.custom_error(trace, self.metadata)
-                f"Attempting to reconnect to OPC server {self.id}...")
+            self.logger.custom_info(
+                f"Attempting to reconnect to OPC server {self.id}...", self.metadata)
             return self.connect()
 
         if hasattr(self.client, 'aio_obj') and self.client.aio_obj.uaclient.protocol is None or \
                 (hasattr(self.client.aio_obj.uaclient, 'protocol') and
                  self.client.aio_obj.uaclient.protocol.state == "closed"):
 
-            self.logger.error(
+            self.logger.custom_error(
-                f"OPC server {self.id} is not connected")
+                f"OPC server {self.id} is not connected", self.metadata)
+
             if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds(
             ) > self.reconnection_interval:
                 self.disconnect()
-                self.logger.error(
+                self.logger.custom_info(
-                    f"Trying to reconnect to OPC server {self.id}...")
+                    f"Trying to reconnect to OPC server {self.id}...", self.metadata)
                 return self.connect()
 
-            return False, {}
+            return False, {
+                "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}",
+                "message": f"OPC server {self.id} is not connected, waiting for next reconnection window...",
+                "block": "opc_repository",
+                "level": NotificationLevel.WARNING
+            }
 
         return True, {}
 
@@ -215,6 +236,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:
@@ -239,7 +262,7 @@ class OpcRepository():
 
         data = data_type_map[data_type]['converter'](value)
         logger.custom_info(
-            f'Writing {data} - {type(data)} to {node}', metadata.get('schedule_name', 'N/A'))
+            f'Writing {data} - {type(data)} to {node}', metadata)
         now = datetime.now()
         ua_data = DataValue(
             Variant(data, data_type_map[data_type]['opc_type']),
@@ -256,9 +279,26 @@ 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'],
+                opc_server_id=self.id
+            ).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'],
+                opc_server_id=self.id
+            ).observe(response_time)
+
         except Exception as e:
             trace = traceback.format_exc()
-            logger.custom_error(trace, metadata.get('schedule_name', 'N/A'))
+            logger.custom_error(trace, metadata)
             self.error_count += 1
             return False, {
                 "notification_id": f"OPC_WRITE_DATA_ERROR_{self.id}",

laborious/worker/worker.py

@@ -1,6 +1,6 @@
 from temporalio import workflow, client
-from temporalio.worker import Worker
+from temporalio.worker import Worker, PollerBehaviorAutoscaling
-
+from temporalio.runtime import Runtime, TelemetryConfig, PrometheusConfig
 
 with workflow.unsafe.imports_passed_through():
     import os
@@ -19,16 +19,32 @@ with workflow.unsafe.imports_passed_through():
         build_mongodb_config
     )
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
-    from sientia_do.temporal.utils.logger import get_logger
+    from sientia_do.observability.logger import get_logger
+    from laborious import metrics
+    from prometheus_client import start_http_server
+
+POD_ID = os.getenv('POD_ID')
+SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', "9091"))
 
 
 async def main():
     host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
     logger = get_logger(__name__)
 
-    logger.info('Starting Worker...')
+    metadata = {
+        'pod_id': POD_ID,
+        'model_name': '-',
+        'model_id': '-',
+        'workflow_name': '-',
+        'schedule_name': '-',
+    }
 
-    logger.info('Starting Notification Handler...')
+    logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata)
+
+    logger.custom_info("Starting prometheus client...", metadata)
+    start_prometheus_server()
+
+    logger.custom_info('Starting Notification Handler...', metadata)
 
     mongo_config = build_mongodb_config()
     notification_handler = NotificationHandler(
@@ -38,7 +54,7 @@ async def main():
         project_name=os.getenv('PROJECT_NAME', 'laborious')
     )
 
-    logger.info('Starting Activities...')
+    logger.custom_info('Starting Activities...', metadata)
 
     activities = Activities(
         postgres_config=build_postgres_config(),
@@ -48,14 +64,25 @@ async def main():
         notification_handler=notification_handler
     )
 
-    logger.info('Starting Temporal Client...')
+    logger.custom_info(
+        f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
+
+    new_runtime = Runtime(
+        telemetry=TelemetryConfig(
+            metrics=PrometheusConfig(
+                bind_address=f"0.0.0.0:{SDK_METRICS_PORT}")
+        )
+    )
+
+    logger.custom_info('Starting Temporal Client...', metadata)
 
     temporal_client = await client.Client.connect(
         target_host=host,
-        namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious')
+        namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious'),
+        runtime=new_runtime
     )
 
-    logger.info('Starting Workers...')
+    logger.custom_info('Starting Workers...', metadata)
 
     workers = [
         Worker(
@@ -68,11 +95,12 @@ async def main():
                 activities.update_production_model,
                 activities.export_data_to_postgres
             ],
-            max_concurrent_workflow_tasks=100,
+            max_concurrent_workflow_tasks=50,
-            max_concurrent_activities=100,
+            max_concurrent_activities=50,
-            max_concurrent_local_activities=100,
+            max_concurrent_local_activities=50,
-            max_concurrent_workflow_task_polls=100,
+            max_cached_workflows=200,
-            max_cached_workflows=50,
+            workflow_task_poller_behavior=PollerBehaviorAutoscaling(),
+            activity_task_poller_behavior=PollerBehaviorAutoscaling()
         ),
         Worker(
             temporal_client,
@@ -95,13 +123,15 @@ async def main():
                 # Postgres
                 activities.load_custom_query,
                 activities.repeat_last_prediction,
-                activities.export_data_to_postgres
+                activities.export_data_to_postgres,
+                activities.write_metrics
             ],
-            max_concurrent_workflow_tasks=100,
+            max_concurrent_workflow_tasks=50,
-            max_concurrent_activities=100,
+            max_concurrent_activities=50,
-            max_concurrent_local_activities=100,
+            max_concurrent_local_activities=50,
-            max_concurrent_workflow_task_polls=100,
+            max_cached_workflows=200,
-            max_cached_workflows=50,
+            workflow_task_poller_behavior=PollerBehaviorAutoscaling(),
+            activity_task_poller_behavior=PollerBehaviorAutoscaling()
         )
     ]
 
@@ -109,21 +139,34 @@ async def main():
     for w in workers:
         handlers.append(w.run())
 
-    logger.info('Workers started successfully')
+    logger.custom_info('Workers started successfully', metadata)
 
     try:
         # This will run the workers and wait for them to complete.
         # If an exception occurs in any of the worker handlers, it will be propagated here.
         await asyncio.gather(*handlers)
     except BaseException as e:  # NOSONAR
-        logger.error(f"An unhandled exception occurred: {e}")
+        logger.custom_error(f"An unhandled exception occurred: {e}", metadata)
     finally:
         if notification_handler:
             notification_handler.shutdown()
         if activities:
             activities.shutdown()
         # Exit with a non-zero status code to indicate failure to Kubernetes
+        metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
         sys.exit(1)
 
+
+def start_prometheus_server():
+    try:
+        port = int(os.getenv("HTTP_METRICS_PORT", 9090))
+        start_http_server(port)
+        print(f"Prometheus server started on port {port}.")
+        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
+    except Exception as e:
+        print(f"Failed to start Prometheus server: {e}")
+        os._exit(1)
+
+
 if __name__ == '__main__':
     asyncio.run(main())

laborious/workflows/minimal_retrain.py

@@ -3,7 +3,7 @@ from temporalio import workflow
 with workflow.unsafe.imports_passed_through():
     from laborious.activities.activities import Activities
     from typing import Any
-    from sientia_do.temporal.utils.policies import retry_policy
+    from sientia_do.temporal.policies import retry_policy
     from datetime import timedelta
 
 

laborious/workflows/predictions_batch.py

@@ -3,7 +3,7 @@ from temporalio import workflow
 with workflow.unsafe.imports_passed_through():
     from laborious.activities.activities import Activities
     from typing import Any
-    from sientia_do.temporal.utils.policies import retry_policy
+    from sientia_do.temporal.policies import retry_policy
     from datetime import timedelta
 
 

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -4,7 +4,7 @@ with workflow.unsafe.imports_passed_through():
     from laborious.activities.activities import Activities
     from typing import Any
     from datetime import timedelta
-    from sientia_do.temporal.utils.policies import retry_policy
+    from sientia_do.temporal.policies import retry_policy
 
 
 @workflow.defn(name="format_and_export_prediction")
@@ -95,3 +95,13 @@ class FormatAndExportPrediction():
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60)
         )
+
+        await workflow.execute_activity_method(
+            Activities.write_metrics,
+            {
+                **metadata,
+                'prediction': prediction
+            },
+            retry_policy=retry_policy,
+            start_to_close_timeout=timedelta(seconds=60)
+        )

laborious/workflows/sub_workflows/prediction_process.py

@@ -3,7 +3,7 @@ from temporalio import workflow
 with workflow.unsafe.imports_passed_through():
     from laborious.activities.activities import Activities
     from typing import Any
-    from sientia_do.temporal.utils.policies import retry_policy
+    from sientia_do.temporal.policies import retry_policy
     from datetime import timedelta
 
 

requirements.txt

@@ -3,5 +3,6 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.5
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.1
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.5
+prometheus-client

tests.ipynb

@@ -0,0 +1,218 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "b10e5c25",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "[ 0.5479121  -0.12224312  0.71719584  0.39473606 -0.8116453   0.9512447\n",
+      "  0.5222794   0.57212861 -0.74377273 -0.09922812 -0.25840395  0.85352998\n",
+      "  0.28773024  0.64552323 -0.1131716  -0.54552256  0.10916957 -0.87236549\n",
+      "  0.65526234  0.2633288   0.51617548 -0.29094806  0.94139605  0.78624224\n",
+      "  0.55676699 -0.61072258 -0.06655799 -0.91239247 -0.69142102  0.36609791\n",
+      "  0.48952431  0.93501946 -0.34834928 -0.25908059 -0.06088838 -0.62105728\n",
+      " -0.74015699 -0.04859015 -0.5461813   0.33962799 -0.12569616  0.66535639\n",
+      "  0.4005302  -0.37526672  0.6645196   0.60952871 -0.22504324 -0.42334379\n",
+      "  0.36499101 -0.72049503 -0.6001836  -0.98527546  0.57384876  0.32970171\n",
+      "  0.41033076  0.56145806 -0.08216845  0.13748239 -0.720406   -0.77093985\n",
+      "  0.33680592 -0.05780759  0.13047221  0.52999771  0.26943664  0.1071588\n",
+      "  0.11841432 -0.3920998  -0.93836433 -0.12656522 -0.57083065 -0.18294271\n",
+      "  0.70680615 -0.53212103 -0.88339452 -0.43723222 -0.41281248  0.32383303\n",
+      "  0.1140643   0.56779642  0.32862708 -0.18722628  0.62804077 -0.66605416\n",
+      " -0.95457585 -0.81990428  0.4447187  -0.07624554 -0.67745644  0.00208955\n",
+      " -0.69537579  0.39264075 -0.10768745 -0.23795755 -0.39697582  0.26056519\n",
+      " -0.27637478 -0.82470016 -0.7639882   0.92379533  0.81716138  0.39941427\n",
+      " -0.46826008  0.93835275  0.55750181  0.43378038 -0.101277   -0.45551688\n",
+      " -0.80721808  0.80520479 -0.08844742 -0.59527327 -0.38808675  0.15843914\n",
+      " -0.64645443  0.71322857  0.51703906  0.43892591 -0.13581392  0.25461768\n",
+      "  0.16819594  0.2996932  -0.83111136 -0.1683852  -0.91677165 -0.01201836\n",
+      " -0.34027758 -0.71095162 -0.79319406  0.17528914 -0.65881406  0.85024024\n",
+      "  0.16212228 -0.30626039  0.18183098 -0.95439226  0.91711843 -0.03539313\n",
+      "  0.56547045 -0.83454    -0.02668334 -0.01858601  0.87565291  0.1434561\n",
+      " -0.0530212  -0.46604867 -0.33686201  0.0413448  -0.12217708 -0.95677584\n",
+      "  0.65258385  0.79232154 -0.71950182  0.10807229 -0.78284852  0.34448019\n",
+      " -0.43753243  0.31884527  0.45398923  0.53729498 -0.78451811  0.83202369\n",
+      " -0.53957202 -0.92517489  0.10970494 -0.25815543  0.65957949  0.61650294\n",
+      " -0.36572221  0.90579879 -0.41816432  0.03011426 -0.48806982  0.87208714\n",
+      " -0.67078436 -0.91017876 -0.12980588  0.98475113  0.78335453  0.49721604\n",
+      "  0.78158498  0.78689328  0.03771672 -0.3681419   0.54402486  0.32332253\n",
+      " -0.25268454 -0.81106666  0.49357922 -0.47507897  0.8736263  -0.51805885\n",
+      " -0.75448414  0.66222534 -0.69343137 -0.64146338  0.19876558  0.74912408\n",
+      " -0.60713067 -0.37935265  0.55480968  0.94365285  0.00148237 -0.71220499\n",
+      " -0.97212742 -0.54068794 -0.73635556  0.35531735 -0.75633499  0.01265986\n",
+      "  0.38852487  0.16223322 -0.6004487   0.60824905  0.43081426  0.47796801\n",
+      " -0.7378845  -0.75249239  0.8551251  -0.20484361 -0.39810262 -0.02283191\n",
+      "  0.32572843  0.91124651 -0.42710755  0.84961686 -0.95028102  0.11039608\n",
+      "  0.26795022 -0.78820519 -0.71932081 -0.16177136  0.93246382  0.19208511\n",
+      "  0.86604644  0.60872183 -0.0652368   0.5695269  -0.96432643 -0.78171201\n",
+      "  0.65885723  0.59363418 -0.53471852  0.06153918  0.21203164  0.73547791\n",
+      "  0.20621431 -0.17485686 -0.25163191 -0.14823583  0.30386205  0.73498126\n",
+      " -0.09220624 -0.50432087 -0.52667527  0.49202856  0.63313753 -0.78944384\n",
+      " -0.86688229  0.18886733 -0.70765351  0.64932838 -0.37933065 -0.71225613\n",
+      "  0.84194094 -0.66893655 -0.43055984 -0.69277321 -0.76901987 -0.95770397\n",
+      " -0.88920918 -0.65071706 -0.89323613  0.18228763  0.36142905 -0.21273909\n",
+      " -0.36401781  0.00905247  0.75000988  0.70226325 -0.91304988 -0.63700318\n",
+      " -0.52651026 -0.50122485  0.1424653  -0.16747515 -0.90149176 -0.25277172\n",
+      "  0.0475059  -0.79665619  0.66691711 -0.89607627  0.84968374 -0.80177372\n",
+      "  0.6871499   0.80530629  0.95914136  0.60405176  0.55895508]\n",
+      "[ 0.28496655  0.55799271 -0.73089558  0.07213607  0.02844574  0.71514429\n",
+      " -0.07440127 -0.22982101  0.27912654 -0.46707336 -0.72046318 -0.04424545\n",
+      " -0.16622126 -0.53486012 -0.26497638 -0.2672151  -0.34500887 -0.24107184\n",
+      "  0.37148669 -0.40624705  0.89771585  0.83269604 -0.03817914 -0.34327759\n",
+      "  0.07086958  0.69712098  0.30517468  0.60878366  0.06544455  0.26583526\n",
+      " -0.42368877  0.46978632 -0.59519081  0.38959626  0.72143814 -0.73579433\n",
+      "  0.22875948 -0.8098085   0.45143126 -0.83101356  0.87187965 -0.72518414\n",
+      "  0.91776049  0.60176835  0.18736401  0.56524821  0.59022968  0.89205413\n",
+      " -0.49323329  0.18015179 -0.8099016   0.2323314  -0.65741739  0.12990122\n",
+      "  0.14486103 -0.06802969  0.04526355  0.52784678  0.59848943 -0.01569357\n",
+      "  0.19918688  0.86247247 -0.76053282 -0.76579287 -0.82458198  0.31572657\n",
+      " -0.1627834   0.54864283  0.34246283 -0.33272448  0.79673309  0.52506429\n",
+      " -0.45893012 -0.27161596 -0.37112004 -0.6847767  -0.70443325  0.87225493\n",
+      " -0.12419193 -0.23336035  0.45937142  0.10598613  0.87227997  0.56060299\n",
+      " -0.04126087 -0.24728105  0.97326309  0.43552047  0.90238932 -0.76304285\n",
+      "  0.70106736  0.27414777 -0.75615664  0.176516    0.37219273 -0.97539463\n",
+      " -0.09136408  0.65079902 -0.40928195 -0.08290384 -0.11537175 -0.39614522\n",
+      "  0.83688379  0.56258807 -0.77882318  0.99406932  0.75840005 -0.43218312\n",
+      "  0.67379316 -0.78716094  0.99820946  0.33136947  0.30025003 -0.81911855\n",
+      "  0.7940668  -0.94200099 -0.51834388 -0.71395625  0.55353588 -0.60359155\n",
+      "  0.82127645  0.31253808 -0.92767458 -0.98914033 -0.89668417  0.21185036\n",
+      "  0.60296362 -0.52289436  0.69881769 -0.88553612  0.60192771  0.85559086\n",
+      "  0.5442168   0.39624157  0.67596044 -0.9196974  -0.59643578 -0.75015264\n",
+      "  0.00906198  0.49037626  0.26002369  0.7022622  -0.68957402  0.46924218\n",
+      " -0.61391702 -0.4584825   0.41980939  0.96040957  0.22308721 -0.89099937\n",
+      "  0.23261794 -0.9152989   0.76829142  0.41915657 -0.65374431 -0.81655799\n",
+      " -0.63293354  0.96005436 -0.08287872  0.5681619   0.27281668  0.1448263\n",
+      " -0.70973949  0.89204891 -0.39731473  0.15603443  0.39955189  0.29846631\n",
+      "  0.88118882 -0.70312202  0.01670548 -0.19193122 -0.05166254 -0.76156495\n",
+      " -0.73181078 -0.44384891 -0.39059079 -0.14419357  0.22197509  0.26925823\n",
+      " -0.17637821 -0.18243378 -0.56474295  0.1766125  -0.36591818 -0.92788033\n",
+      " -0.16319991 -0.05173465 -0.54881426  0.14491587  0.1315438   0.40400436\n",
+      "  0.29589696  0.30486611 -0.3675717   0.57486444  0.09828877 -0.13716361\n",
+      "  0.25202496 -0.27868533  0.02547849  0.47341138  0.77280577  0.84211439\n",
+      "  0.00726585  0.04055023  0.59974082 -0.37109862  0.67476472 -0.01171671\n",
+      " -0.76828655 -0.85588171  0.68398642 -0.88886417 -0.43877713 -0.33173992\n",
+      " -0.65401111 -0.37221326  0.48538513 -0.97063431  0.65434685  0.71309605\n",
+      " -0.25547685 -0.6927742   0.20168082 -0.76065489 -0.27016128  0.91685836\n",
+      "  0.99092895  0.54420978 -0.37807698  0.3753301   0.41081273 -0.22431661\n",
+      "  0.28177727 -0.97854471 -0.58188468  0.05017661 -0.67249739 -0.66818626\n",
+      "  0.67260858  0.97826601  0.11193886  0.67813946  0.98064333 -0.71680822\n",
+      " -0.10350877 -0.21485457 -0.83990143  0.51066035 -0.13244195 -0.06134613\n",
+      " -0.69865405 -0.6381467   0.81420724 -0.91070182 -0.53429543 -0.41588134\n",
+      " -0.01960492  0.17289035 -0.01342005 -0.83176933 -0.51266509  0.68717677\n",
+      "  0.2751774   0.2982981   0.34040651  0.52580604 -0.88378304 -0.26678323\n",
+      "  0.07905487 -0.32308703  0.68895775 -0.03485498  0.53725518  0.70403103\n",
+      "  0.00958297  0.81910449  0.17424788  0.7005486  -0.31881841 -0.00236608\n",
+      "  0.06282208 -0.79004057 -0.20289499  0.83467535  0.26166448 -0.64498684\n",
+      " -0.32228873 -0.61679398 -0.95035374  0.85492092 -0.10358534 -0.38492986\n",
+      "  0.19695438 -0.98537109 -0.44395579  0.40606693  0.26753955]\n",
+      "    Counter   Rollout  CounterPlusRollout           Timestamp\n",
+      "0  0.000000  0.000000            0.000000 2025-01-01 00:00:00\n",
+      "1  0.547912  0.284967            0.832879 2025-01-01 00:00:01\n",
+      "2  0.425669  0.842959            1.268628 2025-01-01 00:00:02\n",
+      "3  1.142865  0.112064            1.254928 2025-01-01 00:00:03\n",
+      "4  1.537601  0.184200            1.721801 2025-01-01 00:00:04\n"
+     ]
+    }
+   ],
+   "source": [
+    "import pandas as pd\n",
+    "import numpy as np\n",
+    "\n",
+    "# Set random seed for reproducibility\n",
+    "rng = np.random.default_rng(42)\n",
+    "\n",
+    "# Generate random walks starting at 0\n",
+    "counter = np.zeros(300)\n",
+    "rollout = np.zeros(300)\n",
+    "\n",
+    "# Generate random steps between -1 and 1\n",
+    "counter_steps = rng.uniform(-1, 1, 299)\n",
+    "rollout_steps = rng.uniform(-1, 1, 299)\n",
+    "\n",
+    "print(counter_steps)\n",
+    "print(rollout_steps)\n",
+    "\n",
+    "# Calculate cumulative sum and scale to -100 to 100 range\n",
+    "for i in range(1, 300):\n",
+    "    counter[i] = counter[i-1] + counter_steps[i-1]\n",
+    "    rollout[i] = rollout[i-1] + rollout_steps[i-1]\n",
+    "\n",
+    "# normalize values between -100 and 100, lowest value is -100, highest value is 100\n",
+    "counter = (counter - min(counter)) / (max(counter) - min(counter)) * 200 - 100\n",
+    "rollout = (rollout - min(rollout)) / (max(rollout) - min(rollout)) * 200 - 100\n",
+    "\n",
+    "# Create DataFrame\n",
+    "df = pd.DataFrame({\n",
+    "    'Counter': counter,\n",
+    "    'Rollout': rollout,\n",
+    "    'CounterPlusRollout': counter + rollout\n",
+    "})\n",
+    "\n",
+    "# add a timestamp column\n",
+    "df['Timestamp'] = pd.date_range(start='2025-01-01', periods=300, freq='1s')\n",
+    "\n",
+    "# Save to CSV\n",
+    "df.to_csv('random_walks.csv', index=False)\n",
+    "\n",
+    "# Display first few rows\n",
+    "print(df.head())\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "id": "e7c8eeb1",
+   "metadata": {},
+   "outputs": [
+    {
+     "ename": "TypeError",
+     "evalue": "MLFlowRepository.__init__() missing 1 required positional argument: 'logger'",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
+      "\u001b[31mTypeError\u001b[39m                                 Traceback (most recent call last)",
+      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[2]\u001b[39m\u001b[32m, line 3\u001b[39m\n\u001b[32m      1\u001b[39m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mlaborious\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mutils\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mrepository\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mmodel_repository\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m MLFlowRepository\n\u001b[32m----> \u001b[39m\u001b[32m3\u001b[39m mlflow_repository = \u001b[43mMLFlowRepository\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m      4\u001b[39m \u001b[43m    \u001b[49m\u001b[43mhost\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mhttp://localhost:5080/\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[32m      5\u001b[39m \u001b[43m    \u001b[49m\u001b[43musername\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43maignosi\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[32m      6\u001b[39m \u001b[43m    \u001b[49m\u001b[43mpassword\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43maignosi\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\n\u001b[32m      7\u001b[39m \u001b[43m)\u001b[49m\n\u001b[32m      9\u001b[39m mlflow_repository.get_experiment_by_run_id(\u001b[33m\"\u001b[39m\u001b[33m1\u001b[39m\u001b[33m\"\u001b[39m)\n",
+      "\u001b[31mTypeError\u001b[39m: MLFlowRepository.__init__() missing 1 required positional argument: 'logger'"
+     ]
+    }
+   ],
+   "source": [
+    "from laborious.utils.repository.model_repository import MLFlowRepository\n",
+    "\n",
+    "mlflow_repository = MLFlowRepository(\n",
+    "    host=\"http://localhost:5080/\",\n",
+    "    username=\"aignosi\",\n",
+    "    password=\"aignosi\"\n",
+    ")\n",
+    "\n",
+    "mlflow_repository.get_experiment_by_run_id(\"1\")"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "venv",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.13"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

tests/laborious/activities/test_gates.py

@@ -344,7 +344,6 @@ async def test_format_prediction(gates_activity):
     assert result['prediction_confidence'] == {0: 0.9}
     assert result['prediction_status'] == {0: 'Good'}
     assert result['comments'] == {0: ""}
-    gates_activity.debug.assert_called()
 
 
 @mark.asyncio
@@ -393,7 +392,8 @@ async def test_get_last_timestamp_with_data(gates_activity):
 async def test_get_last_timestamp_no_data(gates_activity):
     # Arrange
     input_data = {
-        'data': {}
+        'data': {},
+        **metadata
     }
 
     # Act
@@ -402,3 +402,42 @@ async def test_get_last_timestamp_no_data(gates_activity):
     # Assert
     assert isinstance(result, str)  # Should be a timestamp string
     assert len(result) > 0
+
+
+@mark.asyncio
+@patch('laborious.activities.gates.metrics')
+async def test_write_metrics(mock_metrics, gates_activity):
+    """Test write_metrics method."""
+    input_data = {
+        **metadata,
+        'prediction': {
+            'prediction': [1, 2, 3],
+            'prediction_confidence': [0.9, 0.8, 0.7],
+            'response_time': [0.1, 0.2, 0.3]
+        }
+    }
+    await gates_activity.write_metrics(input_data)
+    mock_metrics.PREDICTIONS_WRITTEN_COUNT.labels.assert_called_once_with(
+        pod_id=gates_activity.pod_id,
+        model_name=metadata['metadata']['model_name'],
+        pipeline_name=metadata['metadata']['workflow_name']
+    )
+    mock_metrics.PREDICTIONS_WRITTEN_COUNT.labels.return_value.inc.assert_called_once_with()
+
+    mock_metrics.PREDICTION_CONFIDENCE_MONITOR.labels.assert_called_once_with(
+        pod_id=gates_activity.pod_id,
+        model_name=metadata['metadata']['model_name'],
+        pipeline_name=metadata['metadata']['workflow_name']
+    )
+    mock_metrics.PREDICTION_CONFIDENCE_MONITOR.labels.return_value.set.assert_called_once_with(
+        0.9
+    )
+
+    mock_metrics.PREDICTION_RESPONSE_TIME_MONITOR.labels.assert_called_once_with(
+        pod_id=gates_activity.pod_id,
+        model_name=metadata['metadata']['model_name'],
+        pipeline_name=metadata['metadata']['workflow_name']
+    )
+    mock_metrics.PREDICTION_RESPONSE_TIME_MONITOR.labels.return_value.observe.assert_called_once_with(
+        0.1
+    )

tests/laborious/activities/test_mlflow.py

@@ -24,7 +24,7 @@ def test___init__(mock_mlflow_repository):
     assert mlflow.mlflow_password == "admin"
 
     mock_mlflow_repository.assert_called_once_with(
-        "http://localhost:5000", "admin", "admin"
+        "http://localhost:5000", "admin", "admin", ANY
     )
 
 

tests/laborious/activities/test_opc.py

@@ -91,6 +91,7 @@ def test___init__(mock_send_notification, mock_opc_repository):
             server_cert_path="",
             notification_handler=mock_notification_handler,
             reconnection_interval=60,
+            pod_id='localhost'
         ),
     ])
     mock_opc_repository.assert_has_calls([
@@ -104,6 +105,7 @@ def test___init__(mock_send_notification, mock_opc_repository):
             server_cert_path="",
             notification_handler=mock_notification_handler,
             reconnection_interval=60,
+            pod_id='localhost'
         )
     ])
 

tests/laborious/utils/repository/test_model_repository.py

@@ -16,7 +16,8 @@ def mlflow_repository():
         repo = MLFlowRepository(
             host='http://localhost:5000',
             username='admin',
-            password='admin'
+            password='admin',
+            logger=MagicMock()
         )
         return repo
 
@@ -71,7 +72,9 @@ def test_predict_success(mlflow_repository):
 
     assert output['success'] is True
     assert output['content'] == {'prediction': {
-        0: 3}, 'response_time': ANY}
+        0: 2,
+        1: 3
+    }, 'response_time': ANY}
 
 
 def test_predict_error(mlflow_repository):

tests/laborious/utils/repository/test_opc_repository.py

@@ -139,8 +139,9 @@ def test_disconnect_error(opc_repository, mock_client):
     mock_client.disconnect.side_effect = Exception("Test error")
     opc_repository.disconnect()
 
-    opc_repository.logger.error.assert_called_once_with(
+    opc_repository.logger.custom_error.assert_called_once_with(
-        "Failed to disconnect from OPC server: Test error"
+        "Failed to disconnect from OPC server: Test error",
+        ANY
     )
     assert opc_repository.client is None
 
@@ -163,9 +164,9 @@ def test_validate_connection_error_count_disconnect_error(opc_repository):
     assert response == opc_repository.connect.return_value
     opc_repository.disconnect.assert_called_once()
     opc_repository.connect.assert_called_once()
-    opc_repository.logger.error.assert_has_calls(
+    opc_repository.logger.custom_error.assert_has_calls(
         [
-            call("Failed to disconnect from OPC server: Test error"),
+            call("Failed to disconnect from OPC server: Test error", ANY),
         ]
     )
 
@@ -182,7 +183,12 @@ def test_validate_connection_lost_not_time_to_reconect(_mock_datetime, opc_repos
 
     response = opc_repository.validate_connection()
     opc_repository.try_connect.assert_not_called()
-    assert response == (False, {})
+    assert response == (False, {
+        "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}",
+        "message": f"OPC server {opc_repository.id} is not connected, waiting for next reconnection window...",
+        "block": "opc_repository",
+        "level": NotificationLevel.WARNING
+    })
 
 
 @patch('laborious.utils.repository.opc_repository.hasattr', return_value=True)
@@ -263,18 +269,36 @@ def test_write_data_invalid_data_type(opc_repository, mock_client):
     assert error_data.get('attachment_content') is None
 
 
-def test_write_data(opc_repository, mock_client):
+@patch('laborious.utils.repository.opc_repository.metrics')
+def test_write_data(mock_metrics, opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
     mock_node = MagicMock()
     mock_client.get_node.return_value = mock_node
 
     opc_repository.write_data("ns=2;s=TestNode", 42.0,
-                              "float", opc_repository.logger, metadata)
+                              "float", opc_repository.logger, metadata['metadata'])
 
     mock_client.get_node.assert_called_once_with("ns=2;s=TestNode")
     mock_node.write_value.assert_called_once()
 
+    mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.assert_called_once_with(
+        pod_id=opc_repository.pod_id,
+        model_name=metadata['metadata']['model_name'],
+        pipeline_name=metadata['metadata']['workflow_name'],
+        opc_server_id=opc_repository.id
+    )
+    mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.return_value.inc.assert_called_once_with()
+
+    mock_metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels.assert_called_once_with(
+        pod_id=opc_repository.pod_id,
+        model_name=metadata['metadata']['model_name'],
+        pipeline_name=metadata['metadata']['workflow_name'],
+        opc_server_id=opc_repository.id
+    )
+    mock_metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels.return_value.observe.assert_called_once_with(
+        ANY)
+
 
 def test_write_data_write_value_failed(opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))

tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py

@@ -79,7 +79,7 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
             start_to_close_timeout=ANY
         )])
 
-    assert workflow_mock.execute_activity_method.call_count == 2
+    assert workflow_mock.execute_activity_method.call_count == 3
     assert workflow_mock.execute_local_activity_method.call_count == 1
 
 
@@ -145,5 +145,5 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
         )
     ])
 
-    assert workflow_mock.execute_activity_method.call_count == 2
+    assert workflow_mock.execute_activity_method.call_count == 3
     assert workflow_mock.execute_local_activity_method.call_count == 1

values.yaml

@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.2.7"
+  tag: "0.4.2"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -111,11 +111,39 @@ tolerations: []
 
 affinity: {}
 
-service:
+services:
-  enabled: false
+  sdk-metrics:
-  type: ClusterIP
+    enabled: true
-  port: 4840
+    type: ClusterIP
-  targetPort: 4840
+    port: 9091
+    targetPort: 9091
+    name: sdk-metrics
+
+  metrics:
+    enabled: true
+    type: ClusterIP
+    port: 9090
+    targetPort: 9090
+    name: metrics
+
+# Configuração do ServiceMonitor para o Prometheus Operator
+# ref: https://github.com/prometheus-operator/prometheus-operator
+serviceMonitor:
+  # Se true, um recurso ServiceMonitor será criado.
+  enabled: true
+  # O intervalo no qual as métricas devem ser coletadas (ex: 30s, 1m).
+  endpoints:
+  - port: metrics
+    path: /metrics
+    interval: 30s
+    relabelings: []
+  - port: sdk-metrics
+    path: /metrics
+    interval: 30s
+    relabelings: []
+
+  additionalLabels:
+    release: kube-prometheus-stack
   
 
 env:
@@ -123,7 +151,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1172-criar-pipeline-de-alertas-orquestrador"
+    value: "SIENTIAPDE-1199-revisar-e-testar-observabilidade"
   - name: PYTHON_APP
     value: "laborious.worker.worker"
 
@@ -162,6 +190,10 @@ env:
 
   - name: LOG_LEVEL
     value: "DEBUG"
+  - name: HTTP_METRICS_PORT
+    value: "9090"
+  - name: HTTP_SDK_METRICS_PORT
+    value: "9091"
   - name: PROJECT_NAME
     value: "sientia-laborious"
 
@@ -189,7 +221,7 @@ ssh:
 
 # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp
 
-# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat
+# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.5.0
 
 # kubectl create secret generic git-ssh-key-sientia-laborious-worker \
 #   --namespace sientia \