diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index b4604d8..1951973 100644 --- a/laborious/activities/opc.py +++ b/laborious/activities/opc.py @@ -85,7 +85,6 @@ 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 = await self.opc_repository[opc_id].connect() if not is_connected: diff --git a/laborious/activities/storage.py b/laborious/activities/storage.py index 8aec756..c04b6a4 100644 --- a/laborious/activities/storage.py +++ b/laborious/activities/storage.py @@ -10,14 +10,11 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger from sientia_do.temporal.activities.postgres import Postgres - from sientia_do.temporal.constants import now + from sientia_do.temporal.constants import DATETIME_FORMAT_FILENAME, now from laborious.utils.repository.minio_repository import MinioRepository -DATETIME_FILENAME_FORMAT = '%Y-%m-%d_%H-%M-%S' - - class Storage(Postgres): """ Extensions for Postgres activities with a helper to export query results @@ -84,7 +81,7 @@ class Storage(Postgres): metadata = input_data.get('metadata', {}) object_prefix = input_data.get('object_prefix', 'datasets/retrain') - timestamp = now().strftime(DATETIME_FILENAME_FORMAT) + timestamp = now().strftime(DATETIME_FORMAT_FILENAME) object_name = f'{object_prefix}_{timestamp}.parquet' uri = f's3://{self.minio_repository.minio_bucket}/{object_name}' diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 3361d9a..7400502 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -175,7 +175,12 @@ class MLFlowRepository: if experiment is None: if create_if_not_exists: - experiment = mlflow.create_experiment(experiment_name) + experiment_id = mlflow.create_experiment(experiment_name) + experiment = mlflow.get_experiment(experiment_id) + if experiment is None: + raise ValueError( + f'Experiment {experiment_name} not found after creation, unknown reason' + ) else: raise ValueError(f'Experiment {experiment_name} not found') @@ -765,6 +770,8 @@ class MLFlowRepository: data_path = f'{model_temp_path}/retrain_data.csv' + makedirs(model_temp_path, exist_ok=True) + data.to_csv(data_path, index=True) self.logger.custom_info( diff --git a/laborious/utils/repository/opc_repository.py b/laborious/utils/repository/opc_repository.py index be4edb5..8ab8563 100644 --- a/laborious/utils/repository/opc_repository.py +++ b/laborious/utils/repository/opc_repository.py @@ -1,3 +1,5 @@ +import asyncio +import json import time import traceback from datetime import datetime @@ -10,6 +12,7 @@ from asyncua.ua import DataValue, DateTime, Variant, VariantType from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger +from sientia_do.temporal.activities.base import BaseActivity from laborious import metrics @@ -37,7 +40,7 @@ data_type_map = { } -class OpcRepository: +class OpcRepository(BaseActivity): def __init__( self, opc_id: str, @@ -49,7 +52,6 @@ class OpcRepository: cert_path: str | None = None, private_key_path: str | None = None, server_cert_path: str | None = None, - pod_id: str | None = None, ): self.url = url self.id = opc_id @@ -63,7 +65,8 @@ class OpcRepository: self.last_reconnection_time: None | datetime = None self.notification_handler = notification_handler self.client: None | Client = None - self.pod_id = pod_id + + BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True) self.metadata = { 'model_name': '-', @@ -125,7 +128,14 @@ class OpcRepository: Exception: If the connection to the OPC server fails. """ - self.client = Client(self.url) + self.client = Client(self.url, timeout=10, watchdog_intervall=3600000) # type: ignore[attr-defined] + + self.client.name = self.pod_id + self.client.application_name = self.pod_id + pod_uri = self.pod_id.replace('-', ':') + self.client.application_uri = pod_uri + self.client.product_uri = pod_uri + if self.cert_path: await self.set_security() self.logger.custom_info(f'Starting connection to OPC server {self.id}...', self.metadata) @@ -158,6 +168,8 @@ class OpcRepository: await self.client.connect() return True, {} except Exception as e: + self.disconnect() + trace = traceback.format_exc() self.logger.custom_error(trace, self.metadata) @@ -169,6 +181,32 @@ class OpcRepository: 'attachment_content': trace, } + async def disconnection_fallback(self) -> list: + """ + Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try. + """ + + assert self.client is not None + error_stack = [] + for i in range(5): + try: + self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5') + await self.client.disconnect() + return [] + except Exception as e: + self.logger.error( + f'Failed to disconnect from OPC UA server in attempt {i + 1} of 5: {e}' + ) + error_stack.append( + { + 'attempt': i + 1, + 'error': str(e), + 'traceback': traceback.format_exc(), + } + ) + await asyncio.sleep(0.1 * i) + return error_stack + async def disconnect(self): """ Gracefully disconnect from the OPC server. @@ -179,11 +217,20 @@ class OpcRepository: """ if self.client is None: return - try: - await self.client.disconnect() - self.logger.custom_info('Disconnected from OPC server', self.metadata) - except Exception as e: - self.logger.custom_error(f'Failed to disconnect from OPC server: {e}', self.metadata) + + errors = await self.disconnection_fallback() + if errors: + self.send_notification( + metadata=self.metadata, + notification_id=f'OPC_DISCONNECTION_ERROR_{self.id}', + message='Failed to disconnect from OPC server in 5 attempts.', + block='opc_repository', + level=NotificationLevel.ERROR, + attachment_content=json.dumps(errors, indent=4), + ) + else: + self.logger.warning(f'Disconnected from OPC server {self.id} successfully') + self.client = None async def validate_connection(self) -> tuple[bool, dict[str, Any]]: diff --git a/requirements-light.txt b/requirements-light.txt index c145097..278fb58 100644 --- a/requirements-light.txt +++ b/requirements-light.txt @@ -3,7 +3,7 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.7 prometheus-client botocore boto3 diff --git a/requirements.txt b/requirements.txt index 3ba54e8..7122a3e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,7 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.7 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 prometheus-client botocore diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index 815df27..a9bbe67 100644 --- a/tests/laborious/activities/test_opc.py +++ b/tests/laborious/activities/test_opc.py @@ -105,7 +105,6 @@ async def test_init_opc(mock_send_notification, mock_opc_repository): server_cert_path='', notification_handler=mock_notification_handler, reconnection_interval=60, - pod_id='localhost', ), ] ) @@ -121,7 +120,6 @@ async def test_init_opc(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_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index a437653..c21608d 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -157,10 +157,13 @@ def test_get_experiment_none_create(mlflow, mlflow_repository): mlflow.get_experiment_by_name.return_value = None - mlflow.create_experiment.return_value = experiment + mlflow.get_experiment.return_value = experiment output = mlflow_repository.get_experiment('test', create_if_not_exists=True) + mlflow.create_experiment.assert_called_once_with('test') + mlflow.get_experiment.assert_called_once_with(mlflow.create_experiment.return_value) + assert output == experiment diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py index 6ec680d..2a47f13 100644 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ b/tests/laborious/utils/repository/test_opc_repository.py @@ -1,3 +1,4 @@ +import json from datetime import datetime from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch @@ -15,7 +16,7 @@ def mock_logger(): @pytest.fixture def opc_repository(mock_logger): - return OpcRepository( + repository = OpcRepository( opc_id='test_repo', url='opc.tcp://localhost:4840', logger=mock_logger, @@ -26,6 +27,8 @@ def opc_repository(mock_logger): private_key_path='/path/to/key.pem', server_cert_path='/path/to/server_cert.pem', ) + repository.send_notification = MagicMock() + return repository @pytest.fixture @@ -162,11 +165,37 @@ async def test_try_connect_no_client(opc_repository): @pytest.mark.asyncio -async def test_disconnect(opc_repository, mock_client): +async def test_disconnection_fallback_success(opc_repository, mock_client): opc_repository.client = mock_client - await opc_repository.disconnect() + mock_client.disconnect.return_value = True + result = await opc_repository.disconnection_fallback() mock_client.disconnect.assert_called_once() + assert result == [] + + +@pytest.mark.asyncio +async def test_disconnection_fallback_fail(opc_repository, mock_client): + opc_repository.client = mock_client + mock_client.disconnect.side_effect = Exception('Test error') + result = await opc_repository.disconnection_fallback() + assert result == [ + {'attempt': 1, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 2, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 3, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 4, 'error': 'Test error', 'traceback': ANY}, + {'attempt': 5, 'error': 'Test error', 'traceback': ANY}, + ] + assert mock_client.disconnect.call_count == 5 + + +@pytest.mark.asyncio +async def test_disconnect(opc_repository, mock_client): + opc_repository.client = mock_client + opc_repository.disconnection_fallback = AsyncMock(return_value=[]) + await opc_repository.disconnect() + + opc_repository.disconnection_fallback.assert_called_once() assert opc_repository.client is None @@ -179,11 +208,21 @@ async def test_disconnect_no_client(opc_repository): @pytest.mark.asyncio async def test_disconnect_error(opc_repository, mock_client): opc_repository.client = mock_client - mock_client.disconnect.side_effect = Exception('Test error') + opc_repository.disconnection_fallback = AsyncMock( + return_value=[{'attempt': 1, 'error': 'Test error', 'traceback': 'text'}] + ) await opc_repository.disconnect() - opc_repository.logger.custom_error.assert_called_once_with( - 'Failed to disconnect from OPC server: Test error', ANY + opc_repository.disconnection_fallback.assert_called_once() + opc_repository.send_notification.assert_called_once_with( + metadata=opc_repository.metadata, + notification_id=f'OPC_DISCONNECTION_ERROR_{opc_repository.id}', + message='Failed to disconnect from OPC server in 5 attempts.', + block='opc_repository', + level=NotificationLevel.ERROR, + attachment_content=json.dumps( + [{'attempt': 1, 'error': 'Test error', 'traceback': 'text'}], indent=4 + ), ) assert opc_repository.client is None diff --git a/values.yaml b/values.yaml index 1300546..e4b3165 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.0.3" + tag: "1.0.1" 0# 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-1231-ajustar-o-retreino-do-courier-no-laborious + value: "SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados" - name: PYTHON_APP value: "laborious.worker.worker"