From da10f900a835f60e8095439cfc449b6ccd0abb83 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 15:25:04 -0300 Subject: [PATCH 01/13] SIENTIAPDE-1169 Update dependencies and configuration for improved functionality - Updated sientia-dataops-library version from 1.3.7 to 1.3.8 in requirements.txt. - Changed GITHUB_BRANCH in values.yaml to reflect new testing focus: SIENTIAPDE-1169. - Modified notification level in gates.py from WARNING to ERROR for better error handling. --- laborious/activities/gates.py | 2 +- requirements.txt | 2 +- values.yaml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 38ee483..b958566 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -152,7 +152,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: diff --git a/requirements.txt b/requirements.txt index 38a0818..8bccc94 100644 --- a/requirements.txt +++ b/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 diff --git a/values.yaml b/values.yaml index dfd27ad..3502f24 100644 --- a/values.yaml +++ b/values.yaml @@ -144,7 +144,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" From 7f3457e318669acdad9fd94600e1da79eb7dfc50 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 15:41:03 -0300 Subject: [PATCH 02/13] SIENTIAPDE-1169 Enhance BaseActivity initialization across multiple activities to include error counter - Updated the initialization of the BaseActivity in Gates, MLFlow, and OPC classes to set the error counter to True, improving error tracking and handling capabilities. --- laborious/activities/gates.py | 3 ++- laborious/activities/mlflow.py | 3 ++- laborious/activities/opc.py | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index b958566..dbc1396 100644 --- a/laborious/activities/gates.py +++ b/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]: diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index af53a22..0d94d13 100644 --- a/laborious/activities/mlflow.py +++ b/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 diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 4ee3092..23e3ac4 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -51,7 +51,8 @@ class OPC(BaseActivity): attachment_content=error_data.get( 'attachment_content', None) ) - BaseActivity.__init__(self, logger, notification_handler) + 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: From a2a9f95aa575ebf88f6d71c09b216096a09704ee Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 09:41:17 -0300 Subject: [PATCH 03/13] SIENTIAPDE-1169 SIENTIAPDE-1174 Add SDK metrics configuration and update worker for telemetry - Introduced sdk-metrics service in values.yaml with ClusterIP configuration. - Updated worker.py to integrate SDK metrics telemetry using the new HTTP_SDK_METRICS_PORT environment variable. - Enhanced Prometheus configuration to bind SDK metrics to the specified port. --- laborious/worker/worker.py | 13 +++++++++++-- values.yaml | 29 +++++++++++++++++++---------- 2 files changed, 30 insertions(+), 12 deletions(-) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 9540258..fa8c57d 100644 --- a/laborious/worker/worker.py +++ b/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(): @@ -57,9 +58,17 @@ async def main(): logger.info('Starting Temporal Client...') + new_runtime = Runtime( + telemetry=TelemetryConfig( + metrics=PrometheusConfig( + bind_address=f"0.0.0.0:{SDK_METRICS_PORT}") + ) + ) + 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...') diff --git a/values.yaml b/values.yaml index 3502f24..46e72e2 100644 --- a/values.yaml +++ b/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 - # O path do endpoint de métricas na sua aplicação. - path: /metrics - # Labels adicionais para o recurso ServiceMonitor. - # Essencial para que o Prometheus Operator o descubra. Se você usa o helm chart kube-prometheus-stack, - # ele procura por ServiceMonitors com o label "release: kube-prometheus-stack". + endpoints: + - port: metrics + path: /metrics + interval: 30s + relabelings: [] + - 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: @@ -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" From e82b25582c7c2160e2b257a3bb84e7efc7a566b3 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 09:45:54 -0300 Subject: [PATCH 04/13] SIENTIAPDE-1169 Update helm chart version in values.yaml and modify logging in worker.py - Updated helm upgrade command in values.yaml to version 0.5.0-uat. - Changed log message in worker.py to indicate the start of the SDK Metrics Server, while retaining the original log for the Temporal Client. --- laborious/worker/worker.py | 4 +++- values.yaml | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index fa8c57d..47463f5 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -56,7 +56,7 @@ async def main(): notification_handler=notification_handler ) - logger.info('Starting Temporal Client...') + logger.info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...') new_runtime = Runtime( telemetry=TelemetryConfig( @@ -65,6 +65,8 @@ async def main(): ) ) + logger.info('Starting Temporal Client...') + temporal_client = await client.Client.connect( target_host=host, namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious'), diff --git a/values.yaml b/values.yaml index 46e72e2..588c690 100644 --- a/values.yaml +++ b/values.yaml @@ -221,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-uat # kubectl create secret generic git-ssh-key-sientia-laborious-worker \ # --namespace sientia \ From 6eb71739dbbfb8b29023386a6d2ce87595d7f10c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 13:47:50 -0300 Subject: [PATCH 05/13] SIENTIAPDE-1169 Update helm chart version in values.yaml to 0.5.0 --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 588c690..8fe6b27 100644 --- a/values.yaml +++ b/values.yaml @@ -221,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.5.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 \ From e35eb300cb71f416035272d67591d17a4539062f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 13:58:43 -0300 Subject: [PATCH 06/13] 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')) From f56755b69e66e6f1d8e5942fc52ebdd0cd5a2f33 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 14:04:02 -0300 Subject: [PATCH 07/13] SIENTIAPDE-1169 Refactor opc_repository.py to import metrics module - Added import statement for the metrics module to enhance functionality in the OPC repository. - This change supports the integration of metrics tracking for OPC operations. --- laborious/utils/repository/opc_repository.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index d96d811..1564189 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -7,10 +7,10 @@ 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 +from laborious import metrics data_type_map = { 'float': { From 366d40668928153b72a6d3f2a034578328cf490d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 14:08:00 -0300 Subject: [PATCH 08/13] SIENTIAPDE-1169 Enhance OpcRepository initialization to include pod_id for improved tracking --- laborious/utils/repository/opc_repository.py | 1 + 1 file changed, 1 insertion(+) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 1564189..e22288e 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -53,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): """ From e567b8d0a7e690686fd68194df5382d06909509f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 14:15:22 -0300 Subject: [PATCH 09/13] SIENTIAPDE-1169 Update OPC metrics to include opc_server_id for enhanced tracking --- laborious/metrics.py | 4 ++-- laborious/utils/repository/opc_repository.py | 6 ++++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/laborious/metrics.py b/laborious/metrics.py index 6580eef..9d422fb 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -30,12 +30,12 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram( PREDICTION_OPC_WRITING_COUNT = Counter( "laborious_prediction_opc_writing_count", "Number of predictions written to the OPC server", - CORE_LABELS, + [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, + [CORE_LABELS, "opc_server_id"], 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 e22288e..212ea5d 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -266,7 +266,8 @@ class OpcRepository(): metrics.PREDICTION_OPC_WRITING_COUNT.labels( pod_id=self.pod_id, model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'] + pipeline_name=metadata['workflow_name'], + opc_server_id=self.id ).inc() end_time = time.time() @@ -274,7 +275,8 @@ class OpcRepository(): metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels( pod_id=self.pod_id, model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'] + pipeline_name=metadata['workflow_name'], + opc_server_id=self.id ).observe(response_time) except Exception as e: From 37a31a51a47f40560bfdecdbc06ea71f7c17bb07 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 14:17:43 -0300 Subject: [PATCH 10/13] SIENTIAPDE-1169 Refactor OPC metrics to use unpacking for CORE_LABELS in metrics.py - Updated the definition of prediction OPC writing metrics to utilize unpacking for CORE_LABELS, enhancing code clarity and maintainability. --- laborious/metrics.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/laborious/metrics.py b/laborious/metrics.py index 9d422fb..796e159 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -30,12 +30,12 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram( PREDICTION_OPC_WRITING_COUNT = Counter( "laborious_prediction_opc_writing_count", "Number of predictions written to the OPC server", - [CORE_LABELS, "opc_server_id"], + [*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"], + [*CORE_LABELS, "opc_server_id"], buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] ) From cc7efba8870d35acd13a3da29ff20832e43a3116 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 18 Aug 2025 16:30:33 -0300 Subject: [PATCH 11/13] SIENTIAPDE-1169 Refactor OPC data writing to improve success tracking - Updated the OPC class to store the success status of data writing operations for both prediction and confidence tags. - Added logging for successful and failed writes to the OPC server, enhancing traceability of data operations. --- laborious/activities/opc.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 4834288..e39a35d 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -141,7 +141,7 @@ class OPC(BaseActivity): if 'prediction_tags' in config: for tag, tag_config in config['prediction_tags'].items(): - success = success and self.write_data( + local_success = self.write_data( server_id=server_id, tag=tag, data=data.head(1)['prediction'].values[0], @@ -149,10 +149,15 @@ class OPC(BaseActivity): tag_type='prediction', metadata=metadata ) + if local_success: + self.info( + f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata) + else: + success = success and local_success if 'confidence_tags' in config: for tag, tag_config in config['confidence_tags'].items(): - self.write_data( + local_success = self.write_data( server_id=server_id, tag=tag, data=data.head(1)['prediction_confidence'].values[0], @@ -160,6 +165,11 @@ class OPC(BaseActivity): tag_type='confidence', metadata=metadata ) + if local_success: + self.info( + f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata) + else: + success = success and local_success return self.process_confidence(data, success, metadata) From fc6c95c2644f79e66b392ed7556131ab59cdd184 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 08:31:58 -0300 Subject: [PATCH 12/13] SIENTIAPDE-1169 Refactor OPC data writing and repository initialization - Simplified success tracking logic in the OPC class for writing prediction and confidence data. - Removed unused pod_id attribute from OpcRepository initialization. - Updated test cases to include pod_id for improved metrics tracking during data writing operations. --- laborious/activities/opc.py | 6 ++--- laborious/utils/repository/opc_repository.py | 1 - tests/laborious/activities/test_opc.py | 2 ++ .../utils/repository/test_opc_repository.py | 22 +++++++++++++++++-- 4 files changed, 24 insertions(+), 7 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index e39a35d..7a0c28f 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -152,8 +152,7 @@ class OPC(BaseActivity): if local_success: self.info( f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata) - else: - success = success and local_success + success = success and local_success if 'confidence_tags' in config: for tag, tag_config in config['confidence_tags'].items(): @@ -168,8 +167,7 @@ class OPC(BaseActivity): if local_success: self.info( f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata) - else: - success = success and local_success + success = success and local_success return self.process_confidence(data, success, metadata) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 212ea5d..730ec15 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -93,7 +93,6 @@ 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]]: """ diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index 6540a16..37ae391 100644 --- a/tests/laborious/activities/test_opc.py +++ b/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' ) ]) diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index c399139..f80cf40 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/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, {})) From 3ee076214c7309e8693055a3da99605a1862ed36 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 09:07:05 -0300 Subject: [PATCH 13/13] SIENTIAPDE-1169 Implement server validation and output tag management in OPC class - Added a new method to validate the existence of OPC servers before writing data, improving error handling. - Introduced a method to manage writing of prediction and confidence tags, streamlining the data writing process. - Refactored the write_opc_data method to utilize the new validation and management methods for better code organization and clarity. --- laborious/activities/opc.py | 93 +++++++++++++++++++++---------------- 1 file changed, 54 insertions(+), 39 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 7a0c28f..f3730a4 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -99,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]: """ @@ -127,47 +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." - 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())}" - ) + + if not self.validate_server(server_id, metadata): success = False + continue - 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 + success = success and self.manage_output_tags( + server_id, config, data, metadata, success) return self.process_confidence(data, success, metadata)