From 4bd29ae0d768c7f26e6e7849cadc8c5e38da4fa1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 20 Oct 2025 16:24:36 -0300 Subject: [PATCH 1/4] SIENTIAPDE-1318 SIENTIAPDE-1318: Add support for filling missing tags in Redis data processing. Enhanced the Redis class to include a new parameter for handling missing tags, ensuring that absent tags are filled with None in the data package. --- scouter/activities/redis.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index d724772..013fdcc 100644 --- a/scouter/activities/redis.py +++ b/scouter/activities/redis.py @@ -185,6 +185,7 @@ class Redis(RedisBase): data = DataFrame(input_data['data']) model_tags = input_data['model_tags'] retention_time = input_data['retention_time'] + fill_missing_tags = input_data['fill_missing_tags'] key = f'held_data_{input_data["workflow_name"]}_{input_data["schedule_name"]}' @@ -219,6 +220,7 @@ 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'] @@ -226,6 +228,13 @@ class Redis(RedisBase): 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) + missing_tags = [tag for tag in tags if tag not in list(data_hold.keys())] + + for tag in missing_tags: + data_hold[tag] = None + data_hold['timestamp'] = ( data['timestamp'].max() if not data.empty else data_hold['timestamp'] ) From bd6474a975f7ce51d2db758651014eaecb07d67c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 21 Oct 2025 09:01:08 -0300 Subject: [PATCH 2/4] SIENTIAPDE-1312 SIENTIAPDE-1318: Add 'fill_missing_tags' parameter to CoreScouter for enhanced tag handling in data processing. This update allows for the specification of missing tags during workflow execution. --- scouter/workflow/sub_workflows/core_scouter.py | 1 + 1 file changed, 1 insertion(+) diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 1531fb9..d4cbbc0 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -93,6 +93,7 @@ class CoreScouter: 'model_id': input_data['model_id'], 'model_tags': input_data['model_tags'], 'retention_time': input_data['retention_time'], + 'fill_missing_tags': input_data.get['fill_missing_tags'] }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60), From 9cc55ea5e73de161bbfd1802df1497b6e2c98931 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 21 Oct 2025 09:14:41 -0300 Subject: [PATCH 3/4] SIENTIAPDE-1312 Update Redis and CoreScouter to incorporate 'fill_missing_tags' parameter in data processing. Adjusted related tests to ensure proper handling of missing tags, enhancing overall functionality and consistency across workflows. --- scouter/activities/redis.py | 3 +- .../workflow/sub_workflows/core_scouter.py | 2 +- tests/activities/test_redis.py | 29 +++++++++++++++---- .../sub_workflows/test_core_scouter.py | 4 +++ 4 files changed, 29 insertions(+), 9 deletions(-) diff --git a/scouter/activities/redis.py b/scouter/activities/redis.py index 013fdcc..7619857 100644 --- a/scouter/activities/redis.py +++ b/scouter/activities/redis.py @@ -220,7 +220,6 @@ 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'] @@ -229,7 +228,7 @@ class Redis(RedisBase): to_register_metrics.append((row['name'], value)) if fill_missing_tags: - self.debug("Filling missing tags in data package", metadata=metadata) + self.debug('Filling missing tags in data package', metadata=metadata) missing_tags = [tag for tag in tags if tag not in list(data_hold.keys())] for tag in missing_tags: diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index d4cbbc0..85634cf 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -93,7 +93,7 @@ class CoreScouter: 'model_id': input_data['model_id'], 'model_tags': input_data['model_tags'], 'retention_time': input_data['retention_time'], - 'fill_missing_tags': input_data.get['fill_missing_tags'] + 'fill_missing_tags': input_data['fill_missing_tags'], }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60), diff --git a/tests/activities/test_redis.py b/tests/activities/test_redis.py index 2340ed6..b220424 100644 --- a/tests/activities/test_redis.py +++ b/tests/activities/test_redis.py @@ -216,6 +216,7 @@ async def test_group_and_hold_data_new_key(redis_activity): } ).to_dict('records'), 'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'}, + 'fill_missing_tags': False, } # Mock get to return None for new key @@ -243,7 +244,7 @@ async def test_group_and_hold_data_new_key(redis_activity): @pytest.mark.asyncio -async def test_group_and_hold_data_update_existing(redis_activity): +async def test_group_and_hold_data_update_existing_fill_missing(redis_activity): """Test updating existing data with group_and_hold_data""" # Setup initial data in Redis existing_data = {'sensor1': 20.0, 'sensor2': 28.0, 'timestamp': '2023-01-01 11:00:00'} @@ -262,7 +263,13 @@ async def test_group_and_hold_data_update_existing(redis_activity): 'timestamp': ['2023-01-01 12:00:00'] * 2, } ).to_dict('records'), - 'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2', 'sensor3': 'sensor3'}, + 'model_tags': { + 'sensor1': 'sensor1', + 'sensor2': 'sensor2', + 'sensor3': 'sensor3', + 'sensor4': 'sensor4', + }, + 'fill_missing_tags': True, } # Mock get to return existing data @@ -274,10 +281,15 @@ async def test_group_and_hold_data_update_existing(redis_activity): # Verify the result expected_result = { - 'timestamp': {0: '2023-01-01 12:00:00', 1: '2023-01-01 12:00:00', 2: '2023-01-01 12:00:00'}, - 'variable': {0: 'sensor1', 1: 'sensor2', 2: 'sensor3'}, - 'value': {0: 25.5, 1: 28.0, 2: 42.0}, - 'model_id': {0: 1, 1: 1, 2: 1}, + 'timestamp': { + 0: '2023-01-01 12:00:00', + 1: '2023-01-01 12:00:00', + 2: '2023-01-01 12:00:00', + 3: '2023-01-01 12:00:00', + }, + 'variable': {0: 'sensor1', 1: 'sensor2', 2: 'sensor3', 3: 'sensor4'}, + 'value': {0: 25.5, 1: 28.0, 2: 42.0, 3: None}, + 'model_id': {0: 1, 1: 1, 2: 1, 3: 1}, } assert result == expected_result @@ -289,6 +301,7 @@ async def test_group_and_hold_data_update_existing(redis_activity): 'sensor1': 25.5, 'sensor2': 28.0, 'sensor3': 42.0, + 'sensor4': None, 'timestamp': '2023-01-01 12:00:00', } assert kwargs['ttl'] == 3600 @@ -312,6 +325,7 @@ async def test_group_and_hold_data_with_none_values(redis_activity): } ).to_dict('records'), 'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'}, + 'fill_missing_tags': False, } # Mock get to return None for new key @@ -337,6 +351,7 @@ async def test_group_and_hold_data_empty_dataframe(redis_activity): 'retention_time': 3600, 'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'), 'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'}, + 'fill_missing_tags': False, } redis_activity.get = MagicMock(return_value=None) @@ -358,6 +373,7 @@ async def test_group_and_hold_data_error_get(redis_activity): 'model_id': 1, 'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'), 'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'}, + 'fill_missing_tags': False, } redis_activity.get = MagicMock(side_effect=Exception('test')) @@ -393,6 +409,7 @@ async def test_group_and_hold_data_error_set(redis_activity): 'model_id': 1, 'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'), 'model_tags': {'sensor1': 'sensor1', 'sensor2': 'sensor2'}, + 'fill_missing_tags': False, } existing_data = {'sensor1': 20.0, 'sensor2': 28.0, 'timestamp': '2023-01-01 11:00:00'} diff --git a/tests/workflow/sub_workflows/test_core_scouter.py b/tests/workflow/sub_workflows/test_core_scouter.py index b60e19a..259b721 100644 --- a/tests/workflow/sub_workflows/test_core_scouter.py +++ b/tests/workflow/sub_workflows/test_core_scouter.py @@ -42,6 +42,7 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter): 'retention_time': 3600, 'model_tags': {}, 'debug_data_package': True, + 'fill_missing_tags': False, } ) @@ -91,6 +92,7 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter): 'model_id': 'test_model_id', 'retention_time': 3600, 'model_tags': {}, + 'fill_missing_tags': False, }, retry_policy=ANY, start_to_close_timeout=ANY, @@ -161,6 +163,7 @@ async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter 'table_name': 'test_table', 'retention_time': 3600, 'model_tags': {}, + 'fill_missing_tags': False, } ) @@ -210,6 +213,7 @@ async def test_core_scouter_workflow_with_empty_data(mock_workflow, core_scouter 'model_id': 'test_model_id', 'retention_time': 3600, 'model_tags': {}, + 'fill_missing_tags': False, }, retry_policy=ANY, start_to_close_timeout=ANY, From 507ee8cc188c6d710cd699943c9f8215c3b03bed Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 21 Oct 2025 14:40:07 -0300 Subject: [PATCH 4/4] SIENTIAPDE-1312 Update image tag to 0.4.9 and change GITHUB_BRANCH to SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados in values.yaml --- values.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/values.yaml b/values.yaml index 26916ab..a44982e 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.4.5" + tag: "0.4.9" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -150,7 +150,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" - name: GITHUB_BRANCH - value: "main" + value: "SIENTIAPDE-1312-melhorias-e-correcoes-nas-pipelines-de-dados" - name: PYTHON_APP value: "scouter.worker.worker"