diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 38ee483..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]: @@ -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: 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..f3730a4 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,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." - 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(): - 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 - ) + success = success and self.manage_output_tags( + server_id, config, data, metadata, success) return self.process_confidence(data, success, metadata) diff --git a/laborious/metrics.py b/laborious/metrics.py index a3adcf9..796e159 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, "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] +) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 031e171..730ec15 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 @@ -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')) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 9540258..47463f5 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(): @@ -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...') 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/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, {})) diff --git a/values.yaml b/values.yaml index dfd27ad..8fe6b27 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: @@ -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 \