From 6d6d169b090f9f2efd0336781c195dc9b9d2b322 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Wed, 7 Oct 2026 15:19:36 -0300 Subject: [PATCH] SIENTIAPDE-1974: refactor(slots): fix the SonarQube quality gate findings SonarQube on PR #45 reported 3 new issues; all fixed without behavior change: - S3776 gather_read_tags (cognitive complexity 22): the merge of one resolved read tag into its (server, address) entry moves to _add_read_tag (now 14). - S3776 Orchestrator.run (16): the tag database load moves to Orchestrator._load_tag_database (run back to 14). workflow.patched() is still evaluated first on every execution, so its marker is always recorded. - S1192 '$match' duplicated in the workflow module: the connector_tags tag_id filter moves to orchestrator_functions.filter_by_tag_ids, which also returns a new aggregation list. 194 unit tests (new: filter_by_tag_ids); real workflow on a local Temporal test server still passes the 3 scenarios; the dev dry run is still equivalent. Co-Authored-By: Claude Opus 5.5 --- orchestrator/utils/orchestrator_functions.py | 87 ++++++++++----- orchestrator/workflows/orchestrator.py | 101 ++++++++++-------- .../utils/test_orchestrator_functions.py | 15 +++ 3 files changed, 128 insertions(+), 75 deletions(-) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index a34fbdb..122679f 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -444,6 +444,25 @@ def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]: return sorted(ids, key=lambda value: (isinstance(value, str), str(value))) +def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, Any]: + """ + Restrict a connector_tags aggregation to the referenced tag_ids. + + Args: + query (dict[str, Any]): The connector_tags query ({'collection', 'aggregation'}) + tag_ids (list[Any]): The ids from referenced_tag_ids + + Returns: + dict[str, Any]: A new query whose aggregation starts with a tag_id $match. The + aggregation list is new too: aggregate_documents_in_mongodb appends a stage + to the list it gets. + """ + return { + **query, + 'aggregation': [{'$match': {'tag_id': {'$in': tag_ids}}}, *query['aggregation']], + } + + def index_connector_tags(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """ Index the tag database's connector_tags by tag_id. @@ -625,6 +644,44 @@ def _assign_server_keys(servers: dict[str, dict[str, Any]]) -> None: taken.add(key) +def _add_read_tag( + entries: dict[tuple[str, str], dict[str, Any]], + servers: dict[str, dict[str, Any]], + pipeline: dict[str, Any], + tag: dict[str, Any], + server: dict[str, Any], + address: str, +) -> None: + """Merge one resolved read tag into its (server, address) entry and its server.""" + server_id = server['server_id'] + known = servers.get(server_id) + if known is None or (server['from_tag_database'] and not known['from_tag_database']): + servers[server_id] = server + + # int, so the slot does not depend on how each pipeline stored the number + frequency = cast(int, _frequency(tag)) + entry = entries.setdefault( + (server_id, address), + { + 'address': address, + 'frequency': frequency, + 'active_read': {'enabled': False}, + 'names': [], + }, + ) + entry['frequency'] = min(entry['frequency'], frequency) + entry['active_read'] = _merge_active_read(entry['active_read'], tag.get('active_read')) + + name = { + 'collection': f'raw_{pipeline["schedule_name"]}', + 'tag_name': tag['tag_name'], + 'model_id': pipeline['model_id'], + 'model_name': pipeline['model']['name'], + } + if name not in entry['names']: + entry['names'].append(name) + + def gather_read_tags( pipelines: list[dict[str, Any]], opc_servers: dict[str, dict[str, Any]], @@ -690,35 +747,7 @@ def gather_read_tags( located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues) if located is None: continue - server, address = located - - server_id = server['server_id'] - known = servers.get(server_id) - if known is None or (server['from_tag_database'] and not known['from_tag_database']): - servers[server_id] = server - - # int, so the slot does not depend on how each pipeline stored the number - frequency = cast(int, _frequency(tag)) - entry = entries.setdefault( - (server_id, address), - { - 'address': address, - 'frequency': frequency, - 'active_read': {'enabled': False}, - 'names': [], - }, - ) - entry['frequency'] = min(entry['frequency'], frequency) - entry['active_read'] = _merge_active_read(entry['active_read'], tag.get('active_read')) - - name = { - 'collection': f'raw_{pipeline["schedule_name"]}', - 'tag_name': tag['tag_name'], - 'model_id': pipeline['model_id'], - 'model_name': pipeline['model']['name'], - } - if name not in entry['names']: - entry['names'].append(name) + _add_read_tag(entries, servers, pipeline, tag, *located) _assign_server_keys(servers) diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index fc0af9a..ca46db5 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -7,7 +7,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.policies import retry_policy from orchestrator.activities.activities import Activities - from orchestrator.utils.orchestrator_functions import referenced_tag_ids + from orchestrator.utils.orchestrator_functions import filter_by_tag_ids, referenced_tag_ids # Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow # input may override these aggregations (connectors_query / connector_tags_query); the @@ -53,6 +53,57 @@ class Orchestrator: and Redis infrastructure. """ + async def _load_tag_database( + self, + input_data: dict[str, Any], + metadata: dict[str, Any], + pipelines: list[dict[str, Any]], + ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """ + Load the tag database (SIENTIAPDE-1974) for the read tags that carry a tag_id. + + Only the referenced tags are loaded, with the fields the slot uses, so an + environment without a tag database never depends on its content (P4) and the + payload stays bounded by the pipelines. Nothing is loaded when no scouter read + tag has a tag_id, or when replaying a run started before TAG_DATABASE_PATCH. + + Args: + input_data (dict[str, Any]): Workflow input (optional connectors_query and + connector_tags_query) + metadata (dict[str, Any]): Activity metadata ({'metadata': {...}}) + pipelines (list[dict[str, Any]]): The loaded pipeline configurations + + Returns: + tuple[list[dict[str, Any]], list[dict[str, Any]]]: connectors and connector_tags + """ + tag_ids = referenced_tag_ids(pipelines) + # patched() first: it must run on every execution to record its marker + if not workflow.patched(TAG_DATABASE_PATCH) or not tag_ids: + return [], [] + + connectors_query = input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY) + connector_tags_query = input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY) + + connectors_handler = workflow.start_local_activity_method( + Activities.aggregate_documents_in_mongodb, + { + **metadata, + # A fresh aggregation list: the activity appends a stage to the one it gets + 'query': {**connectors_query, 'aggregation': list(connectors_query['aggregation'])}, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60), + ) + + connector_tags_handler = workflow.start_local_activity_method( + Activities.aggregate_documents_in_mongodb, + {**metadata, 'query': filter_by_tag_ids(connector_tags_query, tag_ids)}, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60), + ) + + return await connectors_handler, await connector_tags_handler + @workflow.run async def run(self, input_data: dict[str, Any]): """ @@ -142,51 +193,9 @@ class Orchestrator: pipeline_config = await pipeline_config_handler - # Tag database: loaded only when a scouter read tag carries a tag_id, and only - # those tags, so an environment without it never depends on its content (P4) and - # the payload stays bounded by the pipelines. Read tags without tag_id keep using - # opc-servers. - connectors: list[dict[str, Any]] = [] - connector_tags: list[dict[str, Any]] = [] - tag_ids = referenced_tag_ids(pipeline_config) - if workflow.patched(TAG_DATABASE_PATCH) and tag_ids: - connectors_query = input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY) - connector_tags_query = input_data.get( - 'connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY - ) - - # Fresh aggregation lists: the activity appends a stage to the one it gets - connectors_handler = workflow.start_local_activity_method( - Activities.aggregate_documents_in_mongodb, - { - **metadata, - 'query': { - **connectors_query, - 'aggregation': list(connectors_query['aggregation']), - }, - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=60), - ) - - connector_tags_handler = workflow.start_local_activity_method( - Activities.aggregate_documents_in_mongodb, - { - **metadata, - 'query': { - **connector_tags_query, - 'aggregation': [ - {'$match': {'tag_id': {'$in': tag_ids}}}, - *connector_tags_query['aggregation'], - ], - }, - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=60), - ) - - connectors = await connectors_handler - connector_tags = await connector_tags_handler + connectors, connector_tags = await self._load_tag_database( + input_data, metadata, pipeline_config + ) orchestrated_schedules = await orchestrated_schedules_handler current_slot_config = await current_slot_config_handler diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index bc3ab57..5f3675e 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -5,6 +5,7 @@ from orchestrator.utils.orchestrator_functions import ( build_tag_config, common_config, drift, + filter_by_tag_ids, gather_read_tags, index_connector_tags, index_connectors, @@ -764,6 +765,20 @@ def test_referenced_tag_ids_without_tag_ids(): assert referenced_tag_ids([pipeline, no_read_tags]) == [] +def test_filter_by_tag_ids(): + """A tag_id $match goes in front, on a new aggregation list""" + query = {'collection': 'connector_tags', 'aggregation': [{'$project': {'_id': 0}}]} + + result = filter_by_tag_ids(query, [7, '7']) + + assert result == { + 'collection': 'connector_tags', + 'aggregation': [{'$match': {'tag_id': {'$in': [7, '7']}}}, {'$project': {'_id': 0}}], + } + assert result['aggregation'] is not query['aggregation'] + assert query['aggregation'] == [{'$project': {'_id': 0}}] + + def test_index_skips_incomplete_documents(): """A malformed tag database document is left out instead of failing the run""" tags = [