diff --git a/laborious/activities/api.py b/laborious/activities/api.py index 366ecd2..c568e98 100644 --- a/laborious/activities/api.py +++ b/laborious/activities/api.py @@ -74,34 +74,26 @@ class API(SientiaMonitoring): def get_pi_web_api_core_labels( self, metadata: dict[str, Any], - operation_type: str | None = None, + operation_type: str = 'write_pi_web_api_data', ) -> 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 - a convenience overload behavior: - - When operation_type is provided, it behaves exactly like the base class, - 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. + PI Web API metrics in laborious use the shared ``CORE_LABELS`` from + ``sientia_do``, which includes ``operation_type``. For this reason, + operation_type must always be present in emitted labels. Args: - - metadata (dict[str, Any]): Workflow execution metadata used to derive labels - - operation_type (str | None): Optional operation type label. If None, the - operation_type key will be removed from the returned labels. + - metadata (dict[str, Any]): Workflow execution metadata used to derive labels. + - operation_type (str): Operation type label for metric cardinality. 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, - 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: """ diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index f9eeb16..75d4b93 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -703,34 +703,30 @@ class Gates(MinioManager): self.info(f'Writing metrics for model {metadata["model_name"]}', metadata) + + core_tags = { + 'pod_id': self.pod_id, + 'runtime': self.runtime, + 'operation_type': 'predict', + 'model_name': metadata['model_name'], + 'workflow_name': metadata['workflow_name'], + } 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'], - }, + tags=core_tags, ) 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'], - }, + tags=core_tags, value=prediction_confidence, ) 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'], - }, + tags=core_tags, value=response_time, ) @@ -741,9 +737,7 @@ class Gates(MinioManager): 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'], + **core_tags, 'opc_server_id': server_id, 'tag': tag, }, @@ -753,9 +747,7 @@ class Gates(MinioManager): 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'], + **core_tags, 'opc_server_id': server_id, 'tag': tag, }, diff --git a/laborious/metrics.py b/laborious/metrics.py index 6b3c30a..a67e574 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -26,7 +26,7 @@ Metric Labels: from prometheus_client import Counter, Gauge, Histogram from sientia_do.observability.metrics import ( - CORE_LABELS as SIENTIA_CORE_LABELS, + CORE_LABELS ) # Application health metric @@ -36,9 +36,6 @@ APP_UP = Gauge( ['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 PREDICTIONS_WRITTEN_COUNT = Counter( '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], ) -# ================== 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 ================== @@ -138,58 +96,58 @@ OPC_CONNECTION_STATUS = Gauge( MODEL_READ_LAG = Histogram( 'laborious_model_read_lag', '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], ) MODEL_WRITE_LAG = Histogram( 'laborious_model_write_lag', '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], ) MODEL_READ_COUNT = Counter( 'laborious_model_read_count', 'Number of reads from the model', - SIENTIA_CORE_LABELS, + CORE_LABELS, ) MODEL_WRITE_COUNT = Counter( 'laborious_model_write_count', 'Number of writes to the model', - SIENTIA_CORE_LABELS, + CORE_LABELS, ) MODEL_READ_ERROR_COUNT = Counter( 'laborious_model_read_error_count', 'Number of errors reading from the model', - SIENTIA_CORE_LABELS, + CORE_LABELS, ) MODEL_WRITE_ERROR_COUNT = Counter( 'laborious_model_write_error_count', 'Number of errors writing to the model', - SIENTIA_CORE_LABELS, + CORE_LABELS, ) MODEL_ANALYZE_LAG = Histogram( 'laborious_model_analyze_lag', '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], ) MODEL_ANALYZE_COUNT = Counter( 'laborious_model_analyze_count', 'Number of analyze operations', - SIENTIA_CORE_LABELS, + CORE_LABELS, ) MODEL_ANALYZE_ERROR_COUNT = Counter( 'laborious_model_analyze_error_count', 'Number of errors during analyze operations', - SIENTIA_CORE_LABELS, + CORE_LABELS, ) diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 00ed167..fbac978 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -98,13 +98,21 @@ class PredictionProcess: model_config, save_transform, ) - finally: await workflow.execute_activity_method( Activities.cleanup_minio_objects_expired, {**metadata, 'data': data}, retry_policy=retry_policy, 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( self, @@ -140,7 +148,7 @@ class PredictionProcess: return # Request MLFlow model transformation - response_data = await workflow.execute_local_activity_method( + transformed_data = await workflow.execute_local_activity_method( Activities.request_transform, {**metadata, 'data': data, 'model_name': model_name, 'model_config': model_config}, retry_policy=retry_policy, @@ -153,7 +161,7 @@ class PredictionProcess: { **metadata, 'filters': input_data['mlflow_transform_filters'], - 'data': response_data, + 'data': transformed_data, 'type': 'transform', 'path_priority': input_data['path_priority'], }, @@ -167,8 +175,6 @@ class PredictionProcess: ): return - transformed_data = response_data['content'] - path_flag, confidence, comment = await workflow.execute_local_activity_method( Activities.mlflow_content_gate, { @@ -187,7 +193,7 @@ class PredictionProcess: ): return - response_data = await workflow.execute_local_activity_method( + predicted_data = await workflow.execute_local_activity_method( Activities.request_predict, { **metadata, @@ -205,7 +211,7 @@ class PredictionProcess: { **metadata, 'filters': input_data['mlflow_predict_filters'], - 'data': response_data, + 'data': predicted_data, 'type': 'predict', 'path_priority': input_data['path_priority'], }, @@ -225,7 +231,7 @@ class PredictionProcess: { 'metadata': metadata, 'path_flag': path_flag, - 'data': response_data['content'], + 'data': predicted_data, 'transformed_data': transformed_data if save_transform else None, 'prediction_confidence': confidence, 'timestamp': last_timestamp, diff --git a/tests/laborious/activities/test_api.py b/tests/laborious/activities/test_api.py index 554459e..bf5b9f1 100644 --- a/tests/laborious/activities/test_api.py +++ b/tests/laborious/activities/test_api.py @@ -81,18 +81,17 @@ def test_get_pi_web_api_core_labels_without_operation_type(mock_pi_web_api_clien 'runtime': 'local', 'model_name': 'test_model', 'workflow_name': 'test_workflow', - 'operation_type': '-', + 'operation_type': 'write_pi_web_api_data', }, ): - labels = api_instance.get_pi_web_api_core_labels( - metadata=metadata['metadata'], operation_type=None - ) - assert 'operation_type' not in labels + labels = api_instance.get_pi_web_api_core_labels(metadata=metadata['metadata']) + assert labels['operation_type'] == 'write_pi_web_api_data' assert labels == { 'pod_id': 'test_pod', 'runtime': 'local', 'model_name': 'test_model', 'workflow_name': 'test_workflow', + 'operation_type': 'write_pi_web_api_data', }