SIENTIAPDE-1314

Implement tag value registration in Gates activity and update related tests. Enhanced the write_metrics method to process and register tag values, improving metrics tracking. Adjusted tests to validate new functionality.

scouter/activities/gates.py

@@ -284,6 +284,7 @@ class Gates(BaseActivity):
             metadata: dict[str, Any]
         """
         metadata = input_data['metadata']
+        tag_values = DataFrame(input_data['tag_values'])
 
         self.info(f'Writing metrics for {metadata["model_name"]}', metadata=metadata)
 
@@ -293,4 +294,13 @@ class Gates(BaseActivity):
             pipeline_name=metadata['workflow_name'],
         ).inc()
 
+        # Register metrics
+        for _, row in tag_values.iterrows():
+            metrics.TAG_CHANGES_MONITOR.labels(
+                pod_id=self.pod_id,
+                model_name=metadata['model_name'],
+                pipeline_name=metadata['workflow_name'],
+                tag_name=row['variable'],
+            ).set(row['value'])
+
         self.info(f'Metrics written for {metadata["model_name"]}', metadata=metadata)

scouter/activities/redis.py

@@ -13,7 +13,6 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.temporal.activities.redis_base import Redis as RedisBase
     from sientia_do.temporal.constants import DATETIME_FORMAT, now
 
-    from scouter import metrics
 
 
 class Redis(RedisBase):
@@ -220,12 +219,10 @@ class Redis(RedisBase):
             data_hold = {tag: content for tag, content in data_hold.items() if tag in tags}
             self.debug(f'Data hold after removing removed tags: {data_hold}', metadata=metadata)
 
-            to_register_metrics = []
             for _, row in data.iterrows():
                 value = row['value']
 
                 data_hold[row['name']] = value
-                to_register_metrics.append((row['name'], value))
 
             if fill_missing_tags:
                 self.debug('Filling missing tags in data package', metadata=metadata)
@@ -240,16 +237,6 @@ class Redis(RedisBase):
 
             self.set(key, data_hold, ttl=retention_time)
 
-            # Register metrics
-            self.debug(f'Metrics to register: {to_register_metrics}', metadata=metadata)
-            for metric in to_register_metrics:
-                metrics.TAG_CHANGES_MONITOR.labels(
-                    pod_id=self.pod_id,
-                    model_name=metadata['model_name'],
-                    pipeline_name=metadata['workflow_name'],
-                    tag_name=metric[0],
-                ).set(metric[1])
-
             data_hold_df = DataFrame(data_hold, index=[0])
             data_hold_melted = data_hold_df.melt(
                 id_vars='timestamp', var_name='variable', value_name='value'

scouter/workflow/sub_workflows/core_scouter.py

@@ -119,6 +119,7 @@ class CoreScouter:
             Activities.write_metrics,
             {
                 **metadata,
+                'tag_values': held_data,
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60),

tests/activities/test_gates.py

@@ -1,5 +1,5 @@
 from typing import Any
-from unittest.mock import ANY, MagicMock, Mock, patch
+from unittest.mock import ANY, MagicMock, Mock, call, patch
 
 import numpy as np
 import pandas as pd
@@ -359,10 +359,43 @@ async def test_aggregate_data_raise_exception(gates_fixture):
 @patch('scouter.activities.gates.metrics')
 async def test_write_metrics(mock_metrics, gates_fixture):
     """Test write_metrics method."""
-    input_data = {'metadata': metadata['metadata']}
+    input_data = {
+        'metadata': metadata['metadata'],
+        'tag_values': {
+            'variable': ['tag1', 'tag2'],
+            'value': [1.0, 2.0],
+        },
+    }
     await gates_fixture.write_metrics(input_data)
     mock_metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels.assert_called_once_with(
         pod_id=gates_fixture.pod_id,
         model_name=metadata['metadata']['model_name'],
         pipeline_name=metadata['metadata']['workflow_name'],
     )
+    mock_metrics.LABORIOUS_DATA_WRITTEN_COUNT.labels.return_value.inc.assert_called_once()
+
+    mock_metrics.TAG_CHANGES_MONITOR.labels.return_value.set.assert_has_calls(
+        [
+            call(1.0),
+            call(2.0),
+        ],
+        any_order=True,
+    )
+
+    mock_metrics.TAG_CHANGES_MONITOR.labels.assert_has_calls(
+        [
+            call(
+                pod_id=gates_fixture.pod_id,
+                model_name=metadata['metadata']['model_name'],
+                pipeline_name=metadata['metadata']['workflow_name'],
+                tag_name='tag1',
+            ),
+            call(
+                pod_id=gates_fixture.pod_id,
+                model_name=metadata['metadata']['model_name'],
+                pipeline_name=metadata['metadata']['workflow_name'],
+                tag_name='tag2',
+            ),
+        ],
+        any_order=True,
+    )