diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index a2c5e16..f2d39b8 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -405,6 +405,9 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: # Get all read tags from pipelines for pipeline in pipelines: + if pipeline['workflow_type'] != 'scouter': + continue + for tag in pipeline.get('read_tags', []): tag_string = f'{tag["server_id"]}:{tag["tag_address"]}' if tag_string not in tags: diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index f225d45..f440f3a 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -264,6 +264,7 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi def test_gather_read_tags(): pipelines = [ { + 'workflow_type': 'scouter', 'schedule_name': 'test_schedule', 'read_tags': [ { @@ -274,6 +275,7 @@ def test_gather_read_tags(): ], }, { + 'workflow_type': 'scouter', 'schedule_name': 'test_schedule2', 'read_tags': [ { @@ -316,6 +318,103 @@ def test_gather_read_tags(): assert result == expected +def test_gather_read_tags_filters_non_scouter_workflows(): + """Test that gather_read_tags only processes scouter workflow types and ignores others""" + pipelines = [ + { + 'workflow_type': 'scouter', + 'schedule_name': 'test_scouter_schedule', + 'read_tags': [ + { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address', + } + ], + }, + { + 'workflow_type': 'predictions_batch', + 'schedule_name': 'test_predictions_schedule', + 'read_tags': [ + { + 'server_id': '2', + 'server_name': 'test_server_name2', + 'tag_address': 'test_tag_address_predictions', + } + ], + }, + { + 'workflow_type': 'minimal_retrain', + 'schedule_name': 'test_retrain_schedule', + 'read_tags': [ + { + 'server_id': '3', + 'server_name': 'test_server_name3', + 'tag_address': 'test_tag_address_retrain', + } + ], + }, + { + 'workflow_type': 'pi_web_api_scouter', + 'schedule_name': 'test_pi_web_api_schedule', + 'read_tags': [ + { + 'server_id': '4', + 'server_name': 'test_server_name4', + 'tag_address': 'test_tag_address_pi_web_api', + } + ], + }, + { + 'workflow_type': 'drift', + 'schedule_name': 'test_drift_schedule', + 'read_tags': [ + { + 'server_id': '5', + 'server_name': 'test_server_name5', + 'tag_address': 'test_tag_address_drift', + } + ], + }, + { + 'workflow_type': 'scouter', + 'schedule_name': 'test_scouter_schedule2', + 'read_tags': [ + { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address2', + } + ], + }, + ] + + result = gather_read_tags(pipelines) + + # Only scouter workflow types should be included + expected = { + '1:test_tag_address': { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address', + 'topics': ['raw_test_scouter_schedule'], + }, + '1:test_tag_address2': { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address2', + 'topics': ['raw_test_scouter_schedule2'], + }, + } + + assert result == expected + # Ensure non-scouter pipelines are not included + assert '2:test_tag_address_predictions' not in result + assert '3:test_tag_address_retrain' not in result + assert '4:test_tag_address_pi_web_api' not in result + assert '5:test_tag_address_drift' not in result + + def test_build_tag_config(): tags = [ {