From 7891a7ca6ef1bd48cf9f92572d920475fbd7ff08 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 11:33:00 -0300 Subject: [PATCH] SIENTIAPDE-1478 SIENTIAPDE-1478 Implement PI Web API response processing and metrics tracking - Added a new method in the API class to process responses from the PI Web API, validating tag writes and emitting metrics for success and errors. - Enhanced error handling for missing WebIds and tag names in responses, with appropriate logging and notifications. - Updated tests to cover various scenarios for processing PI Web API responses, ensuring robust functionality and metrics emission. - Refactored existing methods to integrate the new response processing logic, improving overall code clarity and maintainability. --- laborious/activities/api.py | 124 +++++++++++++++++- laborious/metrics.py | 21 ++- requirements.txt | 3 +- tests/laborious/activities/test_api.py | 173 ++++++++++++++++++++++++- 4 files changed, 312 insertions(+), 9 deletions(-) diff --git a/laborious/activities/api.py b/laborious/activities/api.py index be8ffb3..95a22dc 100644 --- a/laborious/activities/api.py +++ b/laborious/activities/api.py @@ -1,6 +1,7 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): + import json import traceback from typing import Any @@ -12,6 +13,8 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.repository.pi_web_api_client import PIWebAPIClient + from laborious import metrics + PI_WEB_API_PREDICTION_ERROR_CONFIDENCE = 13 @@ -65,19 +68,106 @@ class API(SientiaMonitoring): def close(self) -> None: """ Close the PI Web API client and shutdown monitoring services. + + This method properly closes all connections and resources associated + with the PI Web API client and monitoring services. """ self.pi_web_api_client.close() SientiaMonitoring.shutdown(self) + async def process_pi_web_api_response( + self, + response_data: dict[str, Any], + tags: dict[str, str], + core_labels: dict[str, str], + metadata: dict[str, Any], + ) -> int: + """ + Process the response data from PI Web API write operation. + + Validates that all tags were successfully written, emits metrics for each tag + (success or error), and returns the appropriate prediction confidence value. + Sets error confidence if any tag write fails or if the number of written tags + doesn't match the expected count. + + Args: + - response_data (dict[str, Any]): The response data from the PI Web API write operation. + - tags (dict[str, str]): The tags that were written to the PI Web API. + - core_labels (dict[str, str]): The core labels of the workflow execution. + - metadata (dict[str, Any]): The metadata of the workflow execution. + Returns: + int: Prediction confidence value (0 for success, 13 for errors) + """ + + # Convert tags from name:webid to webid:name + tags = {w: t for t, w in tags.items()} + + tag_names = list[str](tags.values()) + + confidence = 0 + + # Evaluate response for each tag + written_tags = [] + response_items = response_data.get('Items', []) + for item in response_items: + web_id = item.get('WebId') + if not web_id: + self.error('The response did not contain some WebIds', metadata) + continue + errors = item.get('Errors', []) + tag_name = tags.get(web_id) + if not tag_name: + self.error( + f'The response did not contain the tag name for WebId {web_id}', metadata + ) + continue + if errors: + self.error( + f'Error writing tag {tag_name}:{web_id} to PI Web API: {errors}', metadata + ) + await self.emit_metric( + metric_object=metrics.PI_WEB_API_PREDICTION_WRITTEN_ERROR_COUNT, + tags={ + **core_labels, + 'tag_name': tag_name, + }, + ) + confidence = PI_WEB_API_PREDICTION_ERROR_CONFIDENCE + else: + await self.emit_metric( + metric_object=metrics.PI_WEB_API_PREDICTION_WRITTEN_COUNT, + tags={ + **core_labels, + 'tag_name': tag_name, + }, + ) + written_tags.append(tag_name) + + if len(written_tags) != len(tag_names): + self.error( + f'The number of written tags does not match the number of tag names: Expected {tag_names} tags, but {written_tags} tags were written', + metadata, + ) + await self.send_notification_async( + metadata=metadata, + notification_id='WRITE_PI_WEB_API_PREDICTION_ERROR', + message=f'The number of written tags does not match the number of tag names: Expected {tag_names} tags, but {written_tags} tags were written.\nResponse:\n {json.dumps(response_data, indent=4)}\nTags:\n {json.dumps(tags, indent=4)}', + block='write_pi_web_api_data', + level=NotificationLevel.ERROR, + ) + confidence = PI_WEB_API_PREDICTION_ERROR_CONFIDENCE + + return confidence + @activity.defn(name='write_pi_web_api_data') async def write_pi_web_api_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: """ Write prediction and confidence data to PI Web API. - This method writes prediction values and confidence scores to PI Web API - using configured web IDs. It handles errors gracefully by setting error - confidence values when prediction writes fail and sending notifications - for both prediction and confidence write errors. + Writes prediction values and confidence scores to PI Web API using configured + web IDs. Processes responses to validate writes and emit metrics. Handles errors + gracefully by setting error confidence values when writes fail and sending + notifications for both prediction and confidence write errors. Args: input_data (dict[str, Any]): The input data containing: @@ -104,11 +194,16 @@ class API(SientiaMonitoring): prediction_tags = list[str](raw_prediction_tags.values()) confidence_tags = list[str](raw_confidence_tags.values()) + core_labels = { + **self.get_core_labels(metadata), + 'url_path': f'{self.pi_web_api_client.base_url}{endpoint}', + } + prediction_value = data.head(1)['prediction'].values[0] confidence_value = data.head(1)['prediction_confidence'].values[0] try: - await self.pi_web_api_client.write_value( + prediction_response = await self.pi_web_api_client.write_value( web_ids=prediction_tags, value={ 'Timestamp': data.head(1)['timestamp'].values[0], @@ -118,6 +213,15 @@ class API(SientiaMonitoring): metadata=metadata, ) + confidence = await self.process_pi_web_api_response( + response_data=prediction_response, + tags=raw_prediction_tags, + core_labels=core_labels, + metadata=metadata, + ) + + data['prediction_confidence'] = confidence + except Exception as e: trace = traceback.format_exc() await self.send_notification_async( @@ -134,7 +238,7 @@ class API(SientiaMonitoring): return data.to_dict() try: - await self.pi_web_api_client.write_value( + confidence_response = await self.pi_web_api_client.write_value( web_ids=confidence_tags, value={ 'Timestamp': data.head(1)['timestamp'].values[0], @@ -143,6 +247,14 @@ class API(SientiaMonitoring): endpoint=endpoint, metadata=metadata, ) + + await self.process_pi_web_api_response( + response_data=confidence_response, + tags=raw_confidence_tags, + core_labels=core_labels, + metadata=metadata, + ) + except Exception as e: trace = traceback.format_exc() await self.send_notification_async( diff --git a/laborious/metrics.py b/laborious/metrics.py index be5604d..cdd44db 100644 --- a/laborious/metrics.py +++ b/laborious/metrics.py @@ -24,7 +24,9 @@ Metric Labels: """ from prometheus_client import Counter, Gauge, Histogram -from sientia_do.observability.metrics import CORE_LABELS as SIENTIA_CORE_LABELS +from sientia_do.observability.metrics import ( + CORE_LABELS as SIENTIA_CORE_LABELS, +) # Application health metric APP_UP = Gauge( @@ -188,3 +190,20 @@ MODEL_ANALYZE_ERROR_COUNT = Counter( 'Number of errors during analyze operations', SIENTIA_CORE_LABELS, ) + + +# ================== PI Web API metrics ================== + +PI_WEB_API_LABELS = [*CORE_LABELS, 'tag_name'] + +PI_WEB_API_PREDICTION_WRITTEN_COUNT = Counter( + 'laborious_pi_web_api_prediction_written_count', + 'Number of predictions written to the PI Web API', + PI_WEB_API_LABELS, +) + +PI_WEB_API_PREDICTION_WRITTEN_ERROR_COUNT = Counter( + 'laborious_pi_web_api_prediction_written_error_count', + 'Number of errors writing predictions to the PI Web API', + PI_WEB_API_LABELS, +) diff --git a/requirements.txt b/requirements.txt index c0ba2d7..ddb2406 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,7 +3,8 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.8.0 +#git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.8.0 +/home/grezewave/Documents/projects/sientia/sientia-dataops-library/ git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.40.6 prometheus-client botocore diff --git a/tests/laborious/activities/test_api.py b/tests/laborious/activities/test_api.py index eaf23f2..a1ff3bb 100644 --- a/tests/laborious/activities/test_api.py +++ b/tests/laborious/activities/test_api.py @@ -80,6 +80,7 @@ def api(mock_pi_web_api_client): mock_client = MagicMock() mock_client.write_value = AsyncMock() mock_client.close = MagicMock() + mock_client.base_url = 'https://test-pi-server.com' mock_pi_web_api_client.return_value = mock_client api_instance = API( @@ -92,6 +93,15 @@ def api(mock_pi_web_api_client): ) api_instance.send_notification_async = AsyncMock() api_instance.info = MagicMock() + api_instance.error = MagicMock() + api_instance.emit_metric = AsyncMock() + api_instance.get_core_labels = MagicMock( + return_value={ + 'pod_id': 'test_pod', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + } + ) return api_instance @@ -109,6 +119,12 @@ async def test_write_pi_web_api_data_success(mock_dataframe, api, base_input_dat mock_dataframe.return_value = _create_mock_dataframe() + # Mock successful responses + api.pi_web_api_client.write_value.side_effect = [ + {'Items': [{'WebId': 'web_id_1', 'Errors': []}, {'WebId': 'web_id_2', 'Errors': []}]}, + {'Items': [{'WebId': 'web_id_3', 'Errors': []}, {'WebId': 'web_id_4', 'Errors': []}]}, + ] + result = await api.write_pi_web_api_data(input_data) api.info.assert_called_once_with('Writing data to PI Web API...', metadata['metadata']) @@ -176,8 +192,9 @@ async def test_write_pi_web_api_data_prediction_error(mock_dataframe, api, base_ async def test_write_pi_web_api_data_confidence_error(mock_dataframe, api, base_input_data): mock_dataframe.return_value = _create_mock_dataframe() + # First call succeeds, second fails api.pi_web_api_client.write_value.side_effect = [ - None, + {'Items': [{'WebId': 'web_id_1', 'Errors': []}]}, Exception('Confidence write failed'), ] @@ -214,6 +231,12 @@ async def test_write_pi_web_api_data_empty_tags(mock_dataframe, api, base_input_ mock_dataframe.return_value = _create_mock_dataframe() + # Mock empty responses + api.pi_web_api_client.write_value.side_effect = [ + {'Items': []}, + {'Items': []}, + ] + result = await api.write_pi_web_api_data(input_data) api.pi_web_api_client.write_value.assert_has_calls( @@ -251,3 +274,151 @@ async def test_close(api): api.close() api.pi_web_api_client.close.assert_called_once() + + +@mark.asyncio +async def test_process_pi_web_api_response_success(api): + """Test successful processing of PI Web API response with all tags written.""" + response_data = { + 'Items': [ + {'WebId': 'web_id_1', 'Errors': []}, + {'WebId': 'web_id_2', 'Errors': []}, + ] + } + tags = {'tag1': 'web_id_1', 'tag2': 'web_id_2'} + core_labels = { + 'pod_id': 'test_pod', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + } + + confidence = await api.process_pi_web_api_response( + response_data=response_data, + tags=tags, + core_labels=core_labels, + metadata=metadata['metadata'], + ) + + assert confidence == 0 + assert api.emit_metric.call_count == 2 + # Verify that emit_metric was called with correct tags structure + call_args_list = api.emit_metric.call_args_list + assert len(call_args_list) == 2 + # Check that all calls include core_labels and tag_name + for call_args in call_args_list: + assert 'tag_name' in call_args.kwargs['tags'] + assert call_args.kwargs['tags']['tag_name'] in ['tag1', 'tag2'] + + +@mark.asyncio +async def test_process_pi_web_api_response_with_errors(api): + """Test processing response with errors in some tags.""" + response_data = { + 'Items': [ + {'WebId': 'web_id_1', 'Errors': ['Error writing tag']}, + {'WebId': 'web_id_2', 'Errors': []}, + ] + } + tags = {'tag1': 'web_id_1', 'tag2': 'web_id_2'} + core_labels = { + 'pod_id': 'test_pod', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + } + + confidence = await api.process_pi_web_api_response( + response_data=response_data, + tags=tags, + core_labels=core_labels, + metadata=metadata['metadata'], + ) + + assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE + assert api.emit_metric.call_count == 2 + api.error.assert_any_call( + "Error writing tag tag1:web_id_1 to PI Web API: ['Error writing tag']", metadata['metadata'] + ) + + +@mark.asyncio +async def test_process_pi_web_api_response_missing_tags(api): + """Test processing response when number of written tags doesn't match expected.""" + response_data = { + 'Items': [ + {'WebId': 'web_id_1', 'Errors': []}, + ] + } + tags = {'tag1': 'web_id_1', 'tag2': 'web_id_2'} + core_labels = { + 'pod_id': 'test_pod', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + } + + confidence = await api.process_pi_web_api_response( + response_data=response_data, + tags=tags, + core_labels=core_labels, + metadata=metadata['metadata'], + ) + + assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE + api.send_notification_async.assert_called_once() + call_args = api.send_notification_async.call_args + assert call_args.kwargs['notification_id'] == 'WRITE_PI_WEB_API_PREDICTION_ERROR' + assert call_args.kwargs['level'] == NotificationLevel.ERROR + + +@mark.asyncio +async def test_process_pi_web_api_response_missing_webid(api): + """Test processing response when WebId is missing in response item.""" + response_data = { + 'Items': [ + {'Errors': []}, + {'WebId': 'web_id_2', 'Errors': []}, + ] + } + tags = {'tag1': 'web_id_1', 'tag2': 'web_id_2'} + core_labels = { + 'pod_id': 'test_pod', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + } + + confidence = await api.process_pi_web_api_response( + response_data=response_data, + tags=tags, + core_labels=core_labels, + metadata=metadata['metadata'], + ) + + assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE + api.error.assert_any_call('The response did not contain some WebIds', metadata['metadata']) + + +@mark.asyncio +async def test_process_pi_web_api_response_missing_tag_name(api): + """Test processing response when tag name is not found for WebId.""" + response_data = { + 'Items': [ + {'WebId': 'unknown_web_id', 'Errors': []}, + ] + } + tags = {'tag1': 'web_id_1'} + core_labels = { + 'pod_id': 'test_pod', + 'model_name': 'test_model', + 'workflow_name': 'test_workflow', + } + + confidence = await api.process_pi_web_api_response( + response_data=response_data, + tags=tags, + core_labels=core_labels, + metadata=metadata['metadata'], + ) + + assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE + api.error.assert_any_call( + 'The response did not contain the tag name for WebId unknown_web_id', metadata['metadata'] + )