Merge pull request #15 from Aignosi/SIENTIAPDE-1169-pensar-e-projetar-testes-de-breakdown-e-performance

Sientiapde 1169 pensar e projetar testes de breakdown e performance

laborious/activities/gates.py

@@ -48,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]:
@@ -152,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:

laborious/activities/mlflow.py

@@ -16,7 +16,8 @@ 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

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,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,55 @@ 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) -> bool:
+        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)
+                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)
+                success = success and local_success
+
+        return success
+
     @activity.defn(name='write_opc_data')
     async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]:
         """
@@ -124,39 +176,13 @@ class OPC(BaseActivity):
         success = True
 
         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:
+            success = success and 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(
-                        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 'confidence_tags' in config:
-                for tag, tag_config in config['confidence_tags'].items():
-                    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

@@ -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, "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/opc_repository.py

@@ -1,4 +1,5 @@
 import traceback
+import time
 from datetime import datetime
 from pathlib import Path
 from typing import Any
@@ -9,6 +10,7 @@ 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 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,7 @@ class OpcRepository():
         self.last_reconnection_time = None
         self.notification_handler = notification_handler
         self.client = None
+        self.pod_id = pod_id
 
     def set_security(self):
         """
@@ -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,23 @@ 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'))

laborious/worker/worker.py

@@ -1,6 +1,6 @@
 from temporalio import workflow, client
 from temporalio.worker import Worker
-
+from temporalio.runtime import Runtime, TelemetryConfig, PrometheusConfig
 
 with workflow.unsafe.imports_passed_through():
     import os
@@ -24,6 +24,7 @@ with workflow.unsafe.imports_passed_through():
     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():
@@ -55,11 +56,21 @@ async def main():
         notification_handler=notification_handler
     )
 
+    logger.info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...')
+
+    new_runtime = Runtime(
+        telemetry=TelemetryConfig(
+            metrics=PrometheusConfig(
+                bind_address=f"0.0.0.0:{SDK_METRICS_PORT}")
+        )
+    )
+
     logger.info('Starting Temporal Client...')
 
     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...')

requirements.txt

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

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_opc_repository.py

@@ -263,18 +263,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, {}))

values.yaml

@@ -112,6 +112,13 @@ tolerations: []
 affinity: {}
 
 services:
+  sdk-metrics:
+    enabled: true
+    type: ClusterIP
+    port: 9091
+    targetPort: 9091
+    name: sdk-metrics
+
   metrics:
     enabled: true
     type: ClusterIP
@@ -125,18 +132,18 @@ serviceMonitor:
   # Se true, um recurso ServiceMonitor será criado.
   enabled: true
   # O intervalo no qual as métricas devem ser coletadas (ex: 30s, 1m).
-  interval: 30s
+  endpoints:
-  # O path do endpoint de métricas na sua aplicação.
+  - port: metrics
     path: /metrics
-  # Labels adicionais para o recurso ServiceMonitor.
+    interval: 30s
-  # Essencial para que o Prometheus Operator o descubra. Se você usa o helm chart kube-prometheus-stack,
+    relabelings: []
-  # ele procura por ServiceMonitors com o label "release: kube-prometheus-stack".
+  - port: sdk-metrics
+    path: /metrics
+    interval: 30s
+    relabelings: []
+
   additionalLabels:
     release: kube-prometheus-stack
-  # Configurações de relabeling adicionais, se necessário.
-  # ref: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config
-  relabelings: []
-  port: metrics
   
 
 env:
@@ -144,7 +151,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1174-mapear-e-implementar-metricas-a-serem-criadas"
+    value: "SIENTIAPDE-1169-pensar-e-projetar-testes-de-breakdown-e-performance"
   - name: PYTHON_APP
     value: "laborious.worker.worker"
 
@@ -185,6 +192,8 @@ env:
     value: "DEBUG"
   - name: HTTP_METRICS_PORT
     value: "9090"
+  - name: HTTP_SDK_METRICS_PORT
+    value: "9091"
   - name: PROJECT_NAME
     value: "sientia-laborious"
 
@@ -212,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
+# 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 \