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

SIENTIAPDE-1712

Browse Source 
Refactor metrics and API handling for improved consistency and clarity

- Removed the SIENTIA_CORE_LABELS constant and replaced it with CORE_LABELS for uniformity across metrics.
- Updated the API class to ensure operation_type is always included in core labels for PI Web API metrics.
- Simplified metric tag handling in the Gates class by consolidating common tags into a single core_tags dictionary.
- Enhanced the PredictionProcess class to improve error handling and variable naming for clarity.
This commit is contained in:
vitor-aignosi
2026-03-23 10:17:57 -03:00
parent 44558b4415
commit e17824eb85
5 changed files with 51 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
laborious/activities/api.py
View File

@@ -74,34 +74,26 @@ class API(SientiaMonitoring):
def get_pi_web_api_core_labels( def get_pi_web_api_core_labels(
self, self,
metadata: dict[str, Any], metadata: dict[str, Any],
operation_type: str | None = None, operation_type: str = 'write_pi_web_api_data',
) -> dict[str, Any]: ) -> dict[str, Any]:
""" """
Generate core labels for metrics, optionally including operation_type. Generate core labels for PI Web API metrics.
This override keeps compatibility with the base implementation while adding PI Web API metrics in laborious use the shared ``CORE_LABELS`` from
a convenience overload behavior: ``sientia_do``, which includes ``operation_type``. For this reason,
- When operation_type is provided, it behaves exactly like the base class, operation_type must always be present in emitted labels.
returning labels that include the operation_type key.
- When operation_type is omitted (None), it removes the operation_type key
from the resulting labels. This is useful for metrics, such as the PI Web
API metrics, that are defined without the operation_type label.
Args: Args:
- metadata (dict[str, Any]): Workflow execution metadata used to derive labels - metadata (dict[str, Any]): Workflow execution metadata used to derive labels.
- operation_type (str | None): Optional operation type label. If None, the - operation_type (str): Operation type label for metric cardinality.
operation_type key will be removed from the returned labels.
Return: Return:
dict[str, Any]: Core labels dictionary, with operation_type only when provided dict[str, Any]: Core labels dictionary including operation_type.
""" """
base_labels = super().get_core_labels( return super().get_core_labels(
metadata=metadata, metadata=metadata,
operation_type=operation_type or '-', operation_type=operation_type,
) )
if operation_type is None:
base_labels.pop('operation_type', None)
return base_labels
def close(self) -> None: def close(self) -> None:
""" """

32
laborious/activities/gates.py
View File

@@ -703,34 +703,30 @@ class Gates(MinioManager):
self.info(f'Writing metrics for model {metadata["model_name"]}', metadata) self.info(f'Writing metrics for model {metadata["model_name"]}', metadata)
await self.emit_metric(
metric_object=metrics.PREDICTIONS_WRITTEN_COUNT, core_tags = {
tags={
'pod_id': self.pod_id, 'pod_id': self.pod_id,
'runtime': self.runtime,
'operation_type': 'predict',
'model_name': metadata['model_name'], 'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'], 'workflow_name': metadata['workflow_name'],
}, }
await self.emit_metric(
metric_object=metrics.PREDICTIONS_WRITTEN_COUNT,
tags=core_tags,
) )
await self.emit_metric( await self.emit_metric(
metric_object=metrics.PREDICTION_CONFIDENCE_MONITOR, metric_object=metrics.PREDICTION_CONFIDENCE_MONITOR,
method='set', method='set',
tags={ tags=core_tags,
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
},
value=prediction_confidence, value=prediction_confidence,
) )
await self.emit_metric( await self.emit_metric(
metric_object=metrics.PREDICTION_RESPONSE_TIME_MONITOR, metric_object=metrics.PREDICTION_RESPONSE_TIME_MONITOR,
method='observe', method='observe',
tags={ tags=core_tags,
'pod_id': self.pod_id,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
},
value=response_time, value=response_time,
) )
@@ -741,9 +737,7 @@ class Gates(MinioManager):
metric_object=metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR, metric_object=metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR,
method='observe', method='observe',
tags={ tags={
'pod_id': self.pod_id, **core_tags,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
'opc_server_id': server_id, 'opc_server_id': server_id,
'tag': tag, 'tag': tag,
}, },
@@ -753,9 +747,7 @@ class Gates(MinioManager):
await self.emit_metric( await self.emit_metric(
metric_object=metrics.PREDICTION_OPC_WRITING_COUNT, metric_object=metrics.PREDICTION_OPC_WRITING_COUNT,
tags={ tags={
'pod_id': self.pod_id, **core_tags,
'model_name': metadata['model_name'],
'workflow_name': metadata['workflow_name'],
'opc_server_id': server_id, 'opc_server_id': server_id,
'tag': tag, 'tag': tag,
}, },

62
laborious/metrics.py
View File

@@ -26,7 +26,7 @@ Metric Labels:
from prometheus_client import Counter, Gauge, Histogram from prometheus_client import Counter, Gauge, Histogram
from sientia_do.observability.metrics import ( from sientia_do.observability.metrics import (
CORE_LABELS as SIENTIA_CORE_LABELS, CORE_LABELS
) )
# Application health metric # Application health metric
@@ -36,9 +36,6 @@ APP_UP = Gauge(
['pod_id'], ['pod_id'],
) )
# Core labels used across multiple laborious metrics (aligned with ``SientiaMonitoring.labels`` subset)
CORE_LABELS = ['pod_id', 'runtime', 'model_name', 'workflow_name']
# Prediction operation metrics # Prediction operation metrics
PREDICTIONS_WRITTEN_COUNT = Counter( PREDICTIONS_WRITTEN_COUNT = Counter(
'laborious_predictions_written_count', 'laborious_predictions_written_count',
@@ -61,45 +58,6 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram(
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
) )
# ================== MinIO metrics ==================
MINIO_READ_LAG = Histogram(
'laborious_minio_read_lag',
'Lag between the last write to MinIO and the last read from MinIO',
[*SIENTIA_CORE_LABELS, 'bucket_name', 'object_name'],
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
)
MINIO_WRITE_LAG = Histogram(
'laborious_minio_write_lag',
'Lag between the last write to MinIO and the last read from MinIO',
[*SIENTIA_CORE_LABELS, 'bucket_name', 'object_name'],
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
)
MINIO_READ_COUNT = Counter(
'laborious_minio_read_count',
'Number of reads from MinIO',
[*SIENTIA_CORE_LABELS, 'bucket_name', 'object_name'],
)
MINIO_WRITE_COUNT = Counter(
'laborious_minio_write_count',
'Number of writes to MinIO',
[*SIENTIA_CORE_LABELS, 'bucket_name', 'object_name'],
)
MINIO_READ_ERROR_COUNT = Counter(
'laborious_minio_read_error_count',
'Number of errors reading from MinIO',
[*SIENTIA_CORE_LABELS, 'bucket_name', 'object_name'],
)
MINIO_WRITE_ERROR_COUNT = Counter(
'laborious_minio_write_error_count',
'Number of errors writing to MinIO',
[*SIENTIA_CORE_LABELS, 'bucket_name', 'object_name'],
)
# ================== OPC metrics ================== # ================== OPC metrics ==================
@@ -138,58 +96,58 @@ OPC_CONNECTION_STATUS = Gauge(
MODEL_READ_LAG = Histogram( MODEL_READ_LAG = Histogram(
'laborious_model_read_lag', 'laborious_model_read_lag',
'Lag between the start and read of read operations', 'Lag between the start and read of read operations',
SIENTIA_CORE_LABELS, CORE_LABELS,
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
) )
MODEL_WRITE_LAG = Histogram( MODEL_WRITE_LAG = Histogram(
'laborious_model_write_lag', 'laborious_model_write_lag',
'Lag between the start and end of write operations', 'Lag between the start and end of write operations',
SIENTIA_CORE_LABELS, CORE_LABELS,
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
) )
MODEL_READ_COUNT = Counter( MODEL_READ_COUNT = Counter(
'laborious_model_read_count', 'laborious_model_read_count',
'Number of reads from the model', 'Number of reads from the model',
SIENTIA_CORE_LABELS, CORE_LABELS,
) )
MODEL_WRITE_COUNT = Counter( MODEL_WRITE_COUNT = Counter(
'laborious_model_write_count', 'laborious_model_write_count',
'Number of writes to the model', 'Number of writes to the model',
SIENTIA_CORE_LABELS, CORE_LABELS,
) )
MODEL_READ_ERROR_COUNT = Counter( MODEL_READ_ERROR_COUNT = Counter(
'laborious_model_read_error_count', 'laborious_model_read_error_count',
'Number of errors reading from the model', 'Number of errors reading from the model',
SIENTIA_CORE_LABELS, CORE_LABELS,
) )
MODEL_WRITE_ERROR_COUNT = Counter( MODEL_WRITE_ERROR_COUNT = Counter(
'laborious_model_write_error_count', 'laborious_model_write_error_count',
'Number of errors writing to the model', 'Number of errors writing to the model',
SIENTIA_CORE_LABELS, CORE_LABELS,
) )
MODEL_ANALYZE_LAG = Histogram( MODEL_ANALYZE_LAG = Histogram(
'laborious_model_analyze_lag', 'laborious_model_analyze_lag',
'Lag between the start and end of analyze operations', 'Lag between the start and end of analyze operations',
SIENTIA_CORE_LABELS, CORE_LABELS,
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0], buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0],
) )
MODEL_ANALYZE_COUNT = Counter( MODEL_ANALYZE_COUNT = Counter(
'laborious_model_analyze_count', 'laborious_model_analyze_count',
'Number of analyze operations', 'Number of analyze operations',
SIENTIA_CORE_LABELS, CORE_LABELS,
) )
MODEL_ANALYZE_ERROR_COUNT = Counter( MODEL_ANALYZE_ERROR_COUNT = Counter(
'laborious_model_analyze_error_count', 'laborious_model_analyze_error_count',
'Number of errors during analyze operations', 'Number of errors during analyze operations',
SIENTIA_CORE_LABELS, CORE_LABELS,
) )

22
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -98,13 +98,21 @@ class PredictionProcess:
model_config, model_config,
save_transform, save_transform,
) )
finally:
await workflow.execute_activity_method( await workflow.execute_activity_method(
Activities.cleanup_minio_objects_expired, Activities.cleanup_minio_objects_expired,
{**metadata, 'data': data}, {**metadata, 'data': data},
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=5), start_to_close_timeout=timedelta(minutes=5),
) )
except Exception as e:
await workflow.execute_activity_method(
Activities.cleanup_minio_objects_expired,
{**metadata, 'data': data},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=5),
)
raise e
async def _run_prediction_pipeline( async def _run_prediction_pipeline(
self, self,
@@ -140,7 +148,7 @@ class PredictionProcess:
return return
# Request MLFlow model transformation # Request MLFlow model transformation
response_data = await workflow.execute_local_activity_method( transformed_data = await workflow.execute_local_activity_method(
Activities.request_transform, Activities.request_transform,
{**metadata, 'data': data, 'model_name': model_name, 'model_config': model_config}, {**metadata, 'data': data, 'model_name': model_name, 'model_config': model_config},
retry_policy=retry_policy, retry_policy=retry_policy,
@@ -153,7 +161,7 @@ class PredictionProcess:
{ {
**metadata, **metadata,
'filters': input_data['mlflow_transform_filters'], 'filters': input_data['mlflow_transform_filters'],
'data': response_data, 'data': transformed_data,
'type': 'transform', 'type': 'transform',
'path_priority': input_data['path_priority'], 'path_priority': input_data['path_priority'],
}, },
@@ -167,8 +175,6 @@ class PredictionProcess:
): ):
return return
transformed_data = response_data['content']
path_flag, confidence, comment = await workflow.execute_local_activity_method( path_flag, confidence, comment = await workflow.execute_local_activity_method(
Activities.mlflow_content_gate, Activities.mlflow_content_gate,
{ {
@@ -187,7 +193,7 @@ class PredictionProcess:
): ):
return return
response_data = await workflow.execute_local_activity_method( predicted_data = await workflow.execute_local_activity_method(
Activities.request_predict, Activities.request_predict,
{ {
**metadata, **metadata,
@@ -205,7 +211,7 @@ class PredictionProcess:
{ {
**metadata, **metadata,
'filters': input_data['mlflow_predict_filters'], 'filters': input_data['mlflow_predict_filters'],
'data': response_data, 'data': predicted_data,
'type': 'predict', 'type': 'predict',
'path_priority': input_data['path_priority'], 'path_priority': input_data['path_priority'],
}, },
@@ -225,7 +231,7 @@ class PredictionProcess:
{ {
'metadata': metadata, 'metadata': metadata,
'path_flag': path_flag, 'path_flag': path_flag,
'data': response_data['content'], 'data': predicted_data,
'transformed_data': transformed_data if save_transform else None, 'transformed_data': transformed_data if save_transform else None,
'prediction_confidence': confidence, 'prediction_confidence': confidence,
'timestamp': last_timestamp, 'timestamp': last_timestamp,

9
tests/laborious/activities/test_api.py
View File

@@ -81,18 +81,17 @@ def test_get_pi_web_api_core_labels_without_operation_type(mock_pi_web_api_clien
'runtime': 'local', 'runtime': 'local',
'model_name': 'test_model', 'model_name': 'test_model',
'workflow_name': 'test_workflow', 'workflow_name': 'test_workflow',
'operation_type': '-', 'operation_type': 'write_pi_web_api_data',
}, },
): ):
labels = api_instance.get_pi_web_api_core_labels( labels = api_instance.get_pi_web_api_core_labels(metadata=metadata['metadata'])
metadata=metadata['metadata'], operation_type=None assert labels['operation_type'] == 'write_pi_web_api_data'
)
assert 'operation_type' not in labels
assert labels == { assert labels == {
'pod_id': 'test_pod', 'pod_id': 'test_pod',
'runtime': 'local', 'runtime': 'local',
'model_name': 'test_model', 'model_name': 'test_model',
'workflow_name': 'test_workflow', 'workflow_name': 'test_workflow',
'operation_type': 'write_pi_web_api_data',
} }