gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #26 from Aignosi/SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados

Browse Source 
SIENTIAPDE-1312: Refactor Error Handling, Enhance OPC Functionality, and Update Dependencies
This commit is contained in:
Matheus Demoner
2025-10-21 17:24:35 -03:00
committed by GitHub
parent 8b5fbfe0ab c5fb921654
commit 2cecd8eefb
10 changed files with 119 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
laborious/activities/opc.py
View File

@@ -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:

7
laborious/activities/storage.py
View File

@@ -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}'

9
laborious/utils/repository/model_repository.py
View File

@@ -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(

65
laborious/utils/repository/opc_repository.py
View File

@@ -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]]:

2
requirements-light.txt
View File

@@ -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

2
requirements.txt
View File

@@ -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

2
tests/laborious/activities/test_opc.py
View File

@@ -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',
)
]
)

5
tests/laborious/utils/repository/test_model_repository.py
View File

@@ -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

51
tests/laborious/utils/repository/test_opc_repository.py
View File

@@ -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

4
values.yaml
View File

@@ -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"