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

SIENTIAPDE-1325

Browse Source 
Refactor monitoring and metrics integration across various components

- Removed coverage options from `pyproject.toml`.
- Updated prediction metrics in `README.md` to replace `pipeline_name` with `workflow_name`.
- Upgraded `sientia-dataops-library` dependency version in `requirements-light.txt` and `requirements.txt`.
- Enhanced metrics handling in `laborious` activities, including `Activities`, `Gates`, `MLFlow`, and `OPC`, to utilize a new `MetricsController`.
- Refactored metric emission methods to improve clarity and consistency across the codebase.
- Updated tests to reflect changes in metrics handling and ensure proper functionality.
This commit is contained in:
vitor-aignosi
2025-11-04 16:49:10 -03:00
parent 77550d49a6
commit a3da800cab
22 changed files with 1350 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
laborious/activities/activities.py
View File

@@ -1,3 +1,4 @@
from sientia_do.observability.metrics_controller import MetricsController
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
@@ -62,6 +63,8 @@ class Activities(Storage, MLFlow, Gates, OPC):
Raises:
Exception: If any parent class initialization fails
"""
metrics_controller = MetricsController(logger=logger)
# Initialize parent classes
Storage.__init__(
self,
@@ -75,6 +78,7 @@ class Activities(Storage, MLFlow, Gates, OPC):
minio_config=minio_config,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
MLFlow.__init__(
@@ -86,12 +90,22 @@ class Activities(Storage, MLFlow, Gates, OPC):
minio_config=minio_config,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Gates.__init__(self, logger=logger, notification_handler=notification_handler)
Gates.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
OPC.__init__(
self, opc_servers=opc_config, logger=logger, notification_handler=notification_handler
self,
opc_servers=opc_config,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
async def shutdown(self):
@@ -107,4 +121,6 @@ class Activities(Storage, MLFlow, Gates, OPC):
proper resource cleanup and prevent resource leaks.
"""
Storage.close(self)
await OPC.shutdown(self)
MLFlow.close(self)
Gates.close(self)
await OPC.close(self)

113
laborious/activities/gates.py
View File

@@ -10,7 +10,8 @@ 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.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now
from laborious import metrics
@@ -62,7 +63,7 @@ mlflow_content_path_confidence: Mapping[str, int] = {
}
class Gates(BaseActivity):
class Gates(SientiaMonitoring):
"""
Data quality gates and filtering activities for the Laborious system.
@@ -82,7 +83,12 @@ class Gates(BaseActivity):
mlflow_content_filter_functions (dict): Mapping of MLFlow content filter names to functions
"""
def __init__(self, logger: Logger, notification_handler: NotificationHandler):
def __init__(
self,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
"""
Initialize data quality gates with logging and notification capabilities.
@@ -93,7 +99,16 @@ class Gates(BaseActivity):
Raises:
Exception: If BaseActivity initialization fails
"""
BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True)
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
def close(self) -> None:
"""
Close the gates activity and clean up resources.
"""
SientiaMonitoring.shutdown(self)
def __del__(self):
self.close()
@activity.defn(name='input_gate')
async def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
@@ -153,7 +168,7 @@ class Gates(BaseActivity):
filter_output.append(config['policy'])
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=f'INTPUT_GATE_ERROR__{fil}',
message=f'Error in filter {fil}:{config}: \n {e}',
@@ -225,7 +240,7 @@ class Gates(BaseActivity):
if mlflow_response_filter_functions[fil](data, config):
filter_output.append(config['policy'])
comments.append(data['content']['message'])
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}',
message=data['content']['message'],
@@ -235,7 +250,7 @@ class Gates(BaseActivity):
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=f'MLFLOW_GATE_RESPONSE_FILTER__{fil}',
message=f'Error in filter {fil}:{config}: \n {e}',
@@ -305,7 +320,7 @@ class Gates(BaseActivity):
try:
if mlflow_content_filter_functions[fil](data, config):
filter_output.append(config['policy'])
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=f'{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}',
message=f'Data not passed the content filter {fil}:{config}',
@@ -315,7 +330,7 @@ class Gates(BaseActivity):
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=f'MLFLOW_GATE_CONTENT_FILTER__{fil}',
message=f'Error in filter {fil}:{config}: \n {e}',
@@ -608,39 +623,61 @@ class Gates(BaseActivity):
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'],
pipeline_name=metadata['workflow_name'],
).inc()
await self.emit_metric(
metric_object=metrics.PREDICTIONS_WRITTEN_COUNT,
tags={
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
},
)
metrics.PREDICTION_CONFIDENCE_MONITOR.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
).set(prediction_confidence)
await self.emit_metric(
metric_object=metrics.PREDICTION_CONFIDENCE_MONITOR,
method='set',
tags={
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
},
value=prediction_confidence,
)
metrics.PREDICTION_RESPONSE_TIME_MONITOR.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
).observe(response_time)
await self.emit_metric(
metric_object=metrics.PREDICTION_RESPONSE_TIME_MONITOR,
method='observe',
tags={
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
},
value=response_time,
)
for server_id, tags in opc_metrics.items():
for tag, response_time in tags.items():
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=server_id,
tag=tag,
).observe(response_time)
metrics.PREDICTION_OPC_WRITING_COUNT.labels(
pod_id=self.pod_id,
model_name=metadata['model_name'],
pipeline_name=metadata['workflow_name'],
opc_server_id=server_id,
tag=tag,
).inc()
await self.emit_metric(
metric_object=metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR,
method='observe',
tags={
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
'opc_server_id': server_id,
'tag': tag,
},
value=response_time,
)
await self.emit_metric(
metric_object=metrics.PREDICTION_OPC_WRITING_COUNT,
tags={
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
'opc_server_id': server_id,
'tag': tag,
},
)
self.info(f'Metrics written for model {metadata["model_name"]}', metadata)

41
laborious/activities/mlflow.py
View File

@@ -10,7 +10,8 @@ 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.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.temporal.constants import (
DATETIME_FORMAT,
DATETIME_FORMAT_MS_WITH_TZ,
@@ -22,7 +23,7 @@ with workflow.unsafe.imports_passed_through():
from laborious.utils.repository.model_repository import MLFlowRepository
class MLFlow(BaseActivity):
class MLFlow(SientiaMonitoring):
"""
MLFlow integration activities for model inference operations.
@@ -50,6 +51,7 @@ class MLFlow(BaseActivity):
mlflow_password: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
"""
Initialize MLFlow activities with server configuration.
@@ -65,14 +67,19 @@ class MLFlow(BaseActivity):
Raises:
Exception: If MLFlowRepository initialization fails
"""
BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True)
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
self.mlflow_host = mlflow_host
self.mlflow_port = mlflow_port
self.mlflow_username = mlflow_username
self.mlflow_password = mlflow_password
self.model_monitoring_repository = MLFlowRepository(
f'{mlflow_host}:{mlflow_port}', mlflow_username, mlflow_password, logger
f'{mlflow_host}:{mlflow_port}',
mlflow_username,
mlflow_password,
logger,
notification_handler,
metrics_controller,
)
if not hasattr(self, 'minio_repository'):
@@ -87,8 +94,18 @@ class MLFlow(BaseActivity):
minio_secret_key=minio_config['secret_key'],
minio_region_name=minio_config['region_name'],
minio_default_bucket=minio_config['default_bucket'],
metrics_controller=metrics_controller,
)
def close(self) -> None:
"""
Close the MLFlow activity and clean up resources.
"""
SientiaMonitoring.shutdown(self)
def __del__(self):
self.close()
@activity.defn(name='request_transform')
async def request_transform(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
@@ -143,7 +160,7 @@ class MLFlow(BaseActivity):
self.debug(data.head(5).to_string(), metadata)
# Request transformation from MLFlow model
response_data = self.model_monitoring_repository.transform(
response_data = await self.model_monitoring_repository.transform(
model_name, data, model_config, metadata
)
@@ -208,7 +225,7 @@ class MLFlow(BaseActivity):
).dt.strftime(DATETIME_FORMAT)
# Request prediction from MLFlow model
response_data = self.model_monitoring_repository.predict(
response_data = await self.model_monitoring_repository.predict(
model_name, data, model_config, metadata
)
@@ -263,12 +280,12 @@ class MLFlow(BaseActivity):
self.info(f'Loading retrain data from Key: {object_key}', metadata)
try:
data = self.minio_repository.get_parquet_as_dataframe(
data = await self.minio_repository.get_parquet_as_dataframe(
object_key=object_key, metadata=metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='ERROR_LOADING_RETRAIN_DATA',
message=f'Error loading retrain data: {e}',
@@ -316,13 +333,13 @@ class MLFlow(BaseActivity):
data.columns.name = None
retrain_output = self.model_monitoring_repository.retrain_model(
retrain_output = await self.model_monitoring_repository.retrain_model(
data=data, model_name=model_name, model_config=model_config, metadata=metadata
)
if not retrain_output['success']:
trace = retrain_output['traceback']
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='RETRAIN_MODEL_ERROR',
message=f'Error retraining model {model_name}: {retrain_output["message"]}',
@@ -379,7 +396,7 @@ class MLFlow(BaseActivity):
)
try:
response = self.model_monitoring_repository.update_production_model(
response = await self.model_monitoring_repository.update_production_model(
experiment=experiment, model_name=model_name, metadata=metadata
)
@@ -388,7 +405,7 @@ class MLFlow(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='UPDATE_PRODUCTION_MODEL_ERROR',
message=f'Error updating production model {model_name}: {e}',

23
laborious/activities/opc.py
View File

@@ -8,14 +8,15 @@ 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.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from laborious.utils.repository.opc_repository import OpcRepository
OPC_WRITTING_ERROR_CONFIDENCE = 12
class OPC(BaseActivity):
class OPC(SientiaMonitoring):
"""
OPC server integration activities for real-time data export.
@@ -39,12 +40,13 @@ class OPC(BaseActivity):
opc_servers: dict[str, dict[str, Any]],
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
self.logger = logger
self.notification_handler = notification_handler
self.opc_servers = opc_servers
BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True)
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
self.opc_repository: dict[str, OpcRepository] = {}
self.opc_servers = opc_servers
@@ -85,10 +87,11 @@ class OPC(BaseActivity):
server_cert_path=server['server_cert_path'],
notification_handler=self.notification_handler,
reconnection_interval=server['reconnection_interval'],
metrics_controller=self.metrics_controller,
)
is_connected, error_data = await self.opc_repository[opc_id].connect()
if not is_connected:
self.send_notification(
await self.send_notification_async(
metadata={
'model_id': '-',
'model_name': '-',
@@ -137,7 +140,7 @@ class OPC(BaseActivity):
tag, data, data_type, self.logger, metadata
)
if not is_success:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=info_data['notification_id'],
message=info_data['message'],
@@ -149,7 +152,7 @@ class OPC(BaseActivity):
return info_data['response_time']
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=f'WRITE_OPC_{tag_type.upper()}_ERROR',
message=f'Error writing data to OPC server: {e}',
@@ -159,7 +162,7 @@ class OPC(BaseActivity):
)
raise e
def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool:
async def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool:
"""
Validate that an OPC server is available and configured for write operations.
@@ -182,7 +185,7 @@ class OPC(BaseActivity):
"""
if self.opc_repository.get(server_id) is None:
message = f'OPC server {server_id} not found to perform write operation.'
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='OPC_SERVER_NOT_FOUND',
message=message,
@@ -299,7 +302,7 @@ class OPC(BaseActivity):
metrics: dict[str, dict[str, float | None]] = {}
for server_id, config in opc_output_config.items():
if not self.validate_server(server_id, metadata):
if not await self.validate_server(server_id, metadata):
success = False
continue
@@ -358,7 +361,7 @@ class OPC(BaseActivity):
return data.to_dict()
async def shutdown(self):
async def close(self):
"""
Gracefully shutdown all OPC server connections and cleanup resources.

8
laborious/activities/storage.py
View File

@@ -9,6 +9,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.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.temporal.constants import DATETIME_FORMAT_FILENAME, now
@@ -33,6 +34,7 @@ class Storage(Postgres):
minio_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
super().__init__(
host=host,
@@ -44,6 +46,7 @@ class Storage(Postgres):
max_connections=max_connections,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
if not hasattr(self, 'minio_repository'):
@@ -58,6 +61,7 @@ class Storage(Postgres):
minio_secret_key=minio_config['secret_key'],
minio_region_name=minio_config['region_name'],
minio_default_bucket=minio_config['default_bucket'],
metrics_controller=metrics_controller,
)
@activity.defn(name='query_to_minio')
@@ -95,14 +99,14 @@ class Storage(Postgres):
data = pd.DataFrame(data)
# Write parquet to memory and upload via persistent client
self.minio_repository.store_dataframe_as_parquet(
await self.minio_repository.store_dataframe_as_parquet(
dataframe=data, uri=uri, object_name=object_name, metadata=metadata
)
return {'success': True, 'object_key': object_name, 'uri': uri}
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='ERROR_STORING_QUERY_TO_MINIO',
message=f'Error storing query to MinIO: {e}',