From 3a06f0ca39e6d6405a58051148b4a4d1e6441327 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 10:42:27 -0300 Subject: [PATCH 01/12] SIENTIAPDE-1199 Update dependencies and refactor logging imports for observability - Updated the sientia-dataops-library dependency version to 1.4.0 in requirements.txt. - Changed image tag in values.yaml from 0.3.2 to 0.4.1. - Refactored logging imports across multiple files to use the new observability module instead of the temporal.utils.logger. - Updated retry policy imports in workflow files to reflect the new module structure. --- laborious/activities/activities.py | 2 +- laborious/activities/gates.py | 2 +- laborious/activities/mlflow.py | 2 +- laborious/activities/opc.py | 2 +- laborious/utils/repository/opc_repository.py | 10 ++++++++-- laborious/worker/worker.py | 2 +- laborious/workflows/minimal_retrain.py | 2 +- laborious/workflows/predictions_batch.py | 2 +- .../sub_workflows/format_and_export_prediction.py | 2 +- .../workflows/sub_workflows/prediction_process.py | 2 +- requirements.txt | 2 +- values.yaml | 4 ++-- 12 files changed, 20 insertions(+), 14 deletions(-) diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index 3e47efe..3970a28 100644 --- a/laborious/activities/activities.py +++ b/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 diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index dbc1396..b4a85eb 100644 --- a/laborious/activities/gates.py +++ b/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 ( diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 0d94d13..0f89939 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -5,7 +5,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 diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index f3730a4..c01af45 100644 --- a/laborious/activities/opc.py +++ b/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 diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 730ec15..f2e8b75 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -9,7 +9,7 @@ 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 = { @@ -189,6 +189,7 @@ class OpcRepository(): self.logger.error( f"OPC server {self.id} is not connected") + if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( ) > self.reconnection_interval: self.disconnect() @@ -196,7 +197,12 @@ class OpcRepository(): f"Trying to reconnect to OPC server {self.id}...") 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, {} diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 47463f5..0621aa4 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -19,7 +19,7 @@ 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 diff --git a/laborious/workflows/minimal_retrain.py b/laborious/workflows/minimal_retrain.py index 3d6f8d6..b3da70a 100644 --- a/laborious/workflows/minimal_retrain.py +++ b/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 diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index 9d3bcd6..a4b14f0 100644 --- a/laborious/workflows/predictions_batch.py +++ b/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 diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index 8ed79fb..a5c3afd 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/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") diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index e20f7ab..da684fc 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/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 diff --git a/requirements.txt b/requirements.txt index 8bccc94..4a38e93 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.8 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.0 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.5 prometheus-client diff --git a/values.yaml b/values.yaml index 8fe6b27..937dfa3 100644 --- a/values.yaml +++ b/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.3.2" + tag: "0.4.1" # 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: @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1169-pensar-e-projetar-testes-de-breakdown-e-performance" + value: "SIENTIAPDE-1199-revisar-e-testar-observabilidade" - name: PYTHON_APP value: "laborious.worker.worker" From 8012ff778c6edae14383c0898d0456e039368a61 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 11:52:42 -0300 Subject: [PATCH 02/12] SIENTIAPDE-1199 Refactor logging in OpcRepository to include full metadata in error and info messages - Updated logging statements in the OpcRepository class to pass the complete metadata object instead of just the schedule name, enhancing the context available for debugging and monitoring. --- laborious/utils/repository/opc_repository.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index f2e8b75..91bdd8f 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -250,7 +250,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']), @@ -286,7 +286,7 @@ class OpcRepository(): 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}", From 46e21c75380e4514c0aa311b54442579d68441ff Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 12:12:15 -0300 Subject: [PATCH 03/12] SIENTIAPDE-1199 Enhance logging in MLFlow activity to include formatted JSON response data - Added JSON import for better formatting of response data in debug logs. - Updated logging statements to output transformed and prediction response data as pretty-printed JSON, improving readability and debugging context. --- laborious/activities/mlflow.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 0f89939..d3e918d 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -1,3 +1,4 @@ +import json from temporalio import activity, workflow @@ -66,8 +67,8 @@ class MLFlow(BaseActivity): response_data = self.model_monitoring_repository.transform( model_name, data, model_retention) - self.debug("Response data:", metadata) - self.debug(response_data, metadata) + self.debug("Transform response data:", metadata) + self.debug(json.dumps(response_data, indent=4), metadata) return response_data @@ -97,7 +98,7 @@ class MLFlow(BaseActivity): model_name, data, model_retention) self.debug("Prediction response data:", metadata) - self.debug(response_data, metadata) + self.debug(json.dumps(response_data, indent=4), metadata) return response_data From ef32167f575fdb910c59ea587f7ebccce047814a Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 16:05:13 -0300 Subject: [PATCH 04/12] SIENTIAPDE-1199 Update dependencies in requirements.txt and increment image tag in values.yaml - Updated sientia-dataops-library dependency version from 1.4.0 to 1.4.1 in requirements.txt. - Incremented image tag in values.yaml from 0.4.1 to 0.4.2. --- requirements.txt | 2 +- values.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 4a38e93..fdc9371 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.4.0 +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 diff --git a/values.yaml b/values.yaml index 937dfa3..70943c0 100644 --- a/values.yaml +++ b/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.4.1" + 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: From 89384d778344ecd70cf2a7c25396dfe9ba61ea3f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 19 Aug 2025 16:42:48 -0300 Subject: [PATCH 05/12] SIENTIAPDE-1199 Update test for validate_connection method in OpcRepository to include detailed response structure - Modified the test case to assert the response format when the connection is lost, ensuring it includes notification ID, message, block, and level attributes for better clarity in error handling. --- tests/laborious/utils/repository/test_opc_repository.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index f80cf40..768458e 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -182,7 +182,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) From 8b9bb8d720dc1f6438ba0a54d282a4cc43f5d7de Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 20 Aug 2025 10:38:08 -0300 Subject: [PATCH 06/12] SIENTIAPDE-1199 Refactor logging in Gates and MLFlow activities to use info level for key operations - Updated logging statements in the Gates class to replace debug logs with info logs for input and output gate operations, enhancing visibility. - Modified MLFlow class to use info logs for data transformation and prediction processes, improving clarity in the logging output. - Adjusted OPC class to return the count of successfully written tags, providing better insight into data writing operations. --- laborious/activities/gates.py | 44 ++++++++++++++++++++++++---------- laborious/activities/mlflow.py | 8 +++++-- laborious/activities/opc.py | 21 ++++++++++++---- laborious/worker/worker.py | 24 ++++++++++--------- 4 files changed, 67 insertions(+), 30 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index b4a85eb..32d1006 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -68,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) @@ -103,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") @@ -128,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'] @@ -169,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") @@ -195,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']) @@ -234,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") @@ -256,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'] @@ -266,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") @@ -287,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']], @@ -295,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: @@ -307,9 +312,18 @@ 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") @@ -325,6 +339,9 @@ class Gates(BaseActivity): 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'], @@ -342,3 +359,6 @@ class Gates(BaseActivity): model_name=metadata['model_name'], pipeline_name=metadata['workflow_name'] ).observe(response_time) + + self.info( + f"Metrics written for model {metadata['model_name']}", metadata) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index d3e918d..3ad9aae 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -41,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'] @@ -70,6 +70,8 @@ class MLFlow(BaseActivity): self.debug("Transform response data:", metadata) self.debug(json.dumps(response_data, indent=4), metadata) + self.info("Data transformed successfully", metadata) + return response_data @activity.defn(name="request_predict") @@ -85,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'] @@ -100,6 +102,8 @@ class MLFlow(BaseActivity): self.debug("Prediction response data:", metadata) self.debug(json.dumps(response_data, indent=4), metadata) + self.info("Prediction completed successfully", metadata) + return response_data @activity.defn(name="retrain_model") diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index c01af45..debb011 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -115,7 +115,9 @@ class OPC(BaseActivity): def manage_output_tags( self, server_id: str, config: dict[str, Any], data: DataFrame, - metadata: dict[str, Any], success: bool) -> bool: + 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( @@ -129,6 +131,7 @@ class OPC(BaseActivity): 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: @@ -144,9 +147,10 @@ class OPC(BaseActivity): 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 + return success, count @activity.defn(name='write_opc_data') async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: @@ -168,21 +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 not self.validate_server(server_id, metadata): success = False continue - success = success and self.manage_output_tags( + local_success, local_count = self.manage_output_tags( server_id, config, data, metadata, success) + success = success and local_success + success_count += local_count + + self.info( + 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) return self.process_confidence(data, success, metadata) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 0621aa4..85326df 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -1,5 +1,5 @@ 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(): @@ -86,11 +86,12 @@ async def main(): activities.update_production_model, activities.export_data_to_postgres ], - max_concurrent_workflow_tasks=100, - max_concurrent_activities=100, - max_concurrent_local_activities=100, - max_concurrent_workflow_task_polls=100, - max_cached_workflows=50, + max_concurrent_workflow_tasks=50, + max_concurrent_activities=50, + max_concurrent_local_activities=50, + max_cached_workflows=200, + workflow_task_poller_behavior=PollerBehaviorAutoscaling(), + activity_task_poller_behavior=PollerBehaviorAutoscaling() ), Worker( temporal_client, @@ -116,11 +117,12 @@ async def main(): activities.export_data_to_postgres, activities.write_metrics ], - max_concurrent_workflow_tasks=100, - max_concurrent_activities=100, - max_concurrent_local_activities=100, - max_concurrent_workflow_task_polls=100, - max_cached_workflows=50, + max_concurrent_workflow_tasks=50, + max_concurrent_activities=50, + max_concurrent_local_activities=50, + max_cached_workflows=200, + workflow_task_poller_behavior=PollerBehaviorAutoscaling(), + activity_task_poller_behavior=PollerBehaviorAutoscaling() ) ] From 40870b021a7c078f19e586f69799851c93ffcd72 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 20 Aug 2025 10:51:38 -0300 Subject: [PATCH 07/12] SIENTIAPDE-1199 Refactor logging in worker module to include metadata in custom info and error messages - Updated logging statements in the worker module to utilize custom_info and custom_error methods, incorporating a metadata dictionary for enhanced context. - Improved visibility of key operations such as starting the worker, notification handler, and activities, as well as error handling for unhandled exceptions. --- laborious/worker/worker.py | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 85326df..7f668ed 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -31,12 +31,20 @@ async def main(): host = os.getenv('TEMPORAL_HOST', 'localhost:7233') logger = get_logger(__name__) - logger.info(f'Starting Worker with POD_ID: {POD_ID}') + metadata = { + 'pod_id': POD_ID, + 'model_name': '-', + 'model_id': '-', + 'workflow_name': '-', + 'schedule_name': '-', + } - logger.info("Starting prometheus client...") + logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata) + + logger.custom_info("Starting prometheus client...", metadata) start_prometheus_server() - logger.info('Starting Notification Handler...') + logger.custom_info('Starting Notification Handler...', metadata) mongo_config = build_mongodb_config() notification_handler = NotificationHandler( @@ -46,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(), @@ -56,7 +64,8 @@ async def main(): notification_handler=notification_handler ) - logger.info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...') + logger.custom_info( + f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) new_runtime = Runtime( telemetry=TelemetryConfig( @@ -65,7 +74,7 @@ async def main(): ) ) - logger.info('Starting Temporal Client...') + logger.custom_info('Starting Temporal Client...', metadata) temporal_client = await client.Client.connect( target_host=host, @@ -73,7 +82,7 @@ async def main(): runtime=new_runtime ) - logger.info('Starting Workers...') + logger.custom_info('Starting Workers...', metadata) workers = [ Worker( @@ -130,14 +139,14 @@ 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() From 935a9e92791f10c32ac354cfd01d9d0289a968f1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 20 Aug 2025 11:35:42 -0300 Subject: [PATCH 08/12] SIENTIAPDE-1199 Enhance logging in OpcRepository to include metadata for all log messages - Introduced a metadata dictionary to the OpcRepository class for improved context in logging. - Updated logging statements to utilize custom_info, custom_warning, and custom_error methods, ensuring all key operations and error messages include relevant metadata. - Improved visibility of connection status, disconnection attempts, and error handling, facilitating better monitoring and debugging. --- laborious/utils/repository/opc_repository.py | 44 +++++++++++++------- 1 file changed, 28 insertions(+), 16 deletions(-) diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index 91bdd8f..4f97ef9 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -55,6 +55,13 @@ class OpcRepository(): self.client = None self.pod_id = pod_id + self.metadata = { + 'model_name': '-', + 'model_id': '-', + 'workflow_name': 'opc_repository', + 'schedule_name': '-' + } + def set_security(self): """ Configures the security settings for the OPC UA client. @@ -84,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), @@ -107,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]]: @@ -124,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}", @@ -142,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): @@ -154,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]]: """ @@ -171,30 +182,31 @@ class OpcRepository(): return self.connect() if self.error_count > 5: - self.logger.warning( - f"OPC server {self.id} will be disconnected due to multiple errors") + self.logger.custom_warning( + 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.error(trace) - self.logger.info( - f"Attempting to reconnect to OPC server {self.id}...") + self.logger.custom_error( + f"Failed to disconnect from OPC server: {e}", self.metadata) + self.logger.custom_error(trace, self.metadata) + 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( - f"OPC server {self.id} is not connected") + self.logger.custom_error( + 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( - f"Trying to reconnect to OPC server {self.id}...") + self.logger.custom_info( + f"Trying to reconnect to OPC server {self.id}...", self.metadata) return self.connect() return False, { From 7b5c519cde3ee01c05cf6a17c6499d1910f6ad1e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 21 Aug 2025 08:09:20 -0300 Subject: [PATCH 09/12] SIENTIAPDE-1199 Update replica count in values.yaml from 3 to 5 for increased scalability --- values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/values.yaml b/values.yaml index 70943c0..25180fb 100644 --- a/values.yaml +++ b/values.yaml @@ -3,7 +3,7 @@ # Declare variables to be passed into your templates. # This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/ -replicaCount: 3 +replicaCount: 5 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/ image: From a3b642dbef1e3521b3da3a0952399eaf935ccc83 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 21 Aug 2025 08:56:18 -0300 Subject: [PATCH 10/12] 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. --- tests/laborious/activities/test_gates.py | 4 ++-- tests/laborious/utils/repository/test_opc_repository.py | 9 +++++---- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index e1ba49d..2032cc6 100644 --- a/tests/laborious/activities/test_gates.py +++ b/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 diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 768458e..cf9b493 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/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( - "Failed to disconnect from OPC server: Test error" + opc_repository.logger.custom_error.assert_called_once_with( + "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), ] ) From baf52c4a16c330e246b53a680eba1b52ab8a9c05 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 21 Aug 2025 09:04:35 -0300 Subject: [PATCH 11/12] SIENTIAPDE-1199 Refactor OPC class to initialize success_count within the loop for better scope management --- laborious/activities/opc.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index debb011..7019cb2 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -179,10 +179,10 @@ class OPC(BaseActivity): success = True - success_count = 0 - for server_id, config in opc_output_config.items(): + success_count = 0 + if not self.validate_server(server_id, metadata): success = False continue From 3761cbd22fda3dff67250c80deb54b447e622669 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 21 Aug 2025 09:11:32 -0300 Subject: [PATCH 12/12] SIENTIAPDE-1199 Refactor OPC class to remove success_count variable and streamline output tag management --- laborious/activities/opc.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 7019cb2..5eaecf2 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -181,8 +181,6 @@ class OPC(BaseActivity): for server_id, config in opc_output_config.items(): - success_count = 0 - if not self.validate_server(server_id, metadata): success = False continue @@ -190,7 +188,6 @@ class OPC(BaseActivity): local_success, local_count = self.manage_output_tags( server_id, config, data, metadata, success) success = success and local_success - success_count += local_count self.info( 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)