diff --git a/scouter/activities/gates.py b/scouter/activities/gates.py index b5da7dd..ca9e2ce 100644 --- a/scouter/activities/gates.py +++ b/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) diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index 7619857..f0a31a6 100644 --- a/scouter/activities/redis.py +++ b/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' diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 85634cf..04d0984 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/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), diff --git a/tests/activities/test_gates.py b/tests/activities/test_gates.py index 7f0f026..c916ace 100644 --- a/tests/activities/test_gates.py +++ b/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, + )