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

@@ -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
+        Writes prediction values and confidence scores to PI Web API using configured
-        using configured web IDs. It handles errors gracefully by setting error
+        web IDs. Processes responses to validate writes and emit metrics. Handles errors
-        confidence values when prediction writes fail and sending notifications
+        gracefully by setting error confidence values when writes fail and sending
-        for both prediction and confidence write errors.
+        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(

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,
+)

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

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']
+    )