From 60f4e3e4211f940e52fc49477b84a615b1671e63 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 6 Oct 2026 15:59:36 -0300 Subject: [PATCH] SIENTIAPDE-1974: fix(slots): harden the tag database path after code review Code review (code-review skill, xhigh) of efa0d94: 8 findings, 6 fixed here, 2 documented (decisions for the team). Fixed: - The tag database is read only when a scouter read tag carries a tag_id, and only the referenced tags ($match tag_id $in, int and text forms), through aggregations that project only the fields the slot uses. Environments without a tag database, or whose pipelines do not use tag_id, never query it (P4), and other attributes (e.g. an ObjectId, which the Temporal JSON converter rejects) never reach a payload. Fresh aggregation lists are passed, because aggregate_documents_in_mongodb appends a stage to the list it gets. - Incomplete connectors / connector_tags documents are left out of the indexes (their read tags are notified as not found) instead of raising KeyError and failing the whole run. - frequency is stored as int, so 15000 vs '15000' across pipelines no longer makes the slot depend on the pipeline order (O6). - A missing, null or empty tag_id means "no tag database": an empty string no longer drops a legacy read tag. - A read tag without an integer frequency (or, without tag_id, without server_id / tag_address / tag_name) is skipped and notified (ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION) instead of failing the slot build. - Server slot keys can no longer collide (name:server_id is extended until free). Documented in the README: - The gateway's plan B (READ_ENGINE=legacy) reads v1 only: switching to it after this deploy also requires rolling the Orchestrator back. - Renaming a tag in connector_tags does not redeploy the scouter (schedules update on the pipeline's updated_at). Validation: 193 unit tests; the real workflow on a local Temporal test server (sandbox on) with the real Formatters activities and the dev data: slot v2 written without querying the tag database, a second run inserts nothing, and a tag_id pipeline queries both aggregations with the $in filter; the dev dry run is still equivalent to the current v1 slot through the gateway's parse_slot. Co-Authored-By: Claude Opus 5.5 --- README.md | 44 ++++-- orchestrator/activities/formatters.py | 23 +-- orchestrator/utils/orchestrator_functions.py | 143 +++++++++++++++--- orchestrator/workflows/orchestrator.py | 80 +++++++--- .../activities/test_formatters.py | 52 ++++++- .../utils/test_orchestrator_functions.py | 139 ++++++++++++++++- .../workflows/test_orchestrator.py | 139 +++++++++++------ 7 files changed, 496 insertions(+), 124 deletions(-) diff --git a/README.md b/README.md index 136b7d0..159a3d9 100644 --- a/README.md +++ b/README.md @@ -121,22 +121,29 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip } ``` -Optional: `connectors_query` and `connector_tags_query` override the tag database queries. The defaults live in the code (`orchestrator/workflows/orchestrator.py`), so the schedule input does not need them: +Optional: `connectors_query` and `connector_tags_query` override the tag database aggregations. The defaults live in the code (`orchestrator/workflows/orchestrator.py`), so the schedule input does not need them: ```json { "connectors_query": { "collection": "connectors", - "filters": {"type": "opc_ua", "status": "Active"} + "aggregation": [ + {"$match": {"type": "opc_ua", "status": "Active"}}, + {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "server_uri": 1, + "use_security": 1, "subscription_period_ms": 1}} + ] }, "connector_tags_query": { "collection": "connector_tags", - "filters": {"category": "Read", "status": "Active"} + "aggregation": [ + {"$match": {"category": "Read", "status": "Active"}}, + {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}} + ] } } ``` -An empty or missing collection loads nothing, and the pipelines keep working through `opc-servers`. +The tag database is read only when a scouter read tag carries a `tag_id`, and the workflow puts `{"$match": {"tag_id": {"$in": [...]}}}` in front of `connector_tags_query`, so it loads only the referenced tags. Environments without a tag database, or whose pipelines do not use `tag_id`, never query it. An override should keep a `$project` of these fields: any other attribute may hold a BSON type a Temporal payload cannot carry (such as `ObjectId`). #### OPC slots (slot v2, contract C1) @@ -171,18 +178,22 @@ How each scouter `read_tags` entry is resolved: Rules: - **One entry per address.** An address read by N pipelines is read once and written to N collections: one `names` item per pipeline (`collection` = `raw_{schedule_name}`). - **Same name source as the scouter.** `resolve_tag_names` names both the slot's `names` and the scouter's `model_tags`, so the gateway always writes the name the scouter looks for. -- **`frequency`:** the lowest among the pipelines of the address. Every read tag needs one, with or without `tag_id`. +- **`frequency`:** the lowest among the pipelines of the address, as an integer. Every read tag needs one, with or without `tag_id`. +- **`tag_id`:** a missing, `null` or empty `tag_id` means the read tag does not use the tag database. - **`subscription_period_ms`:** the connector's value. For `opc-servers`, it is half the server's lowest tag frequency in the slot, as before. - **`use_security`:** the connector's value. For `opc-servers`, it is `true` when the server has a `cert_path`. The certificate itself comes from MinIO on the gateway, so the slot carries no certificate path. - **`active_read`:** off by default. It is on when any pipeline sets `read_tags[].active_read.enabled`, with the lowest `cycle_multiple`. -- **Same `server_id` in both sources:** a server known through `connectors` and `opc-servers` under the same id is one server, and the `connectors` fields win. Two different servers with the same name get the slot keys `name` and `name:server_id`. +- **Same `server_id` in both sources:** a server known through `connectors` and `opc-servers` under the same id is one server, and the `connectors` fields win. Two different servers with the same name get distinct slot keys (`name`, then `name:server_id`). - **Slot division:** the addresses are split across the slots, sorted by server and address, and the names of an address always stay in the same slot. -- **Unresolved tags:** a read tag that cannot be resolved is left out of the slot. It is notified once per run: - - `ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `tag_id` is not an active Read tag; - - `ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the connector is missing or inactive; - - `ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `server_id` is not in `opc-servers`. +- **Unresolved tags:** a read tag that cannot be resolved is left out of the slot, and the rest of the slot is still written. It is notified once per run: + - `ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `tag_id` is not an active Read tag, or its document lacks `connector_id`, `address` or `tag_name`; + - `ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the connector is missing or inactive, or lacks `name`, `url` or `server_uri`; + - `ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `server_id` is not in `opc-servers`; + - `ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION`: the read tag has no integer `frequency` or, without `tag_id`, lacks `server_id`, `tag_address` or `tag_name`. - **Only changed slots are written.** The slot is built in a stable order, so unchanged pipelines produce an identical slot, and the gateway does not re-subscribe. A slot still in the v1 format (`topics`, `cert_path`) is replaced once. +**Gateway plan B:** the gateway's legacy read engine (`READ_ENGINE=legacy`, kept until the end of SIENTIAPDE-1974) reads the slot v1 only. Once this version writes the slot v2, switching the gateway to the legacy engine also requires rolling the Orchestrator back to a version that writes v1. + **Known limitation:** scouter schedules are only redeployed when the pipeline's `updated_at` changes. If a tag is renamed in `connector_tags`, the slot picks up the new name on the next run, but the scouter keeps the old `model_tags` until the pipeline is updated. When renaming, also update the pipeline, or keep the name in `read_tags[].tag_name`. #### Architecture @@ -675,16 +686,23 @@ Temporal input configuration sample: }, "connectors_query": { "collection": "connectors", - "filters": {"type": "opc_ua", "status": "Active"} + "aggregation": [ + {"$match": {"type": "opc_ua", "status": "Active"}}, + {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "server_uri": 1, + "use_security": 1, "subscription_period_ms": 1}} + ] }, "connector_tags_query": { "collection": "connector_tags", - "filters": {"category": "Read", "status": "Active"} + "aggregation": [ + {"$match": {"category": "Read", "status": "Active"}}, + {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}} + ] } } ``` -`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code. +`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code (see "OPC slots" above). #### Alerts Workflow diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 3ba4247..9a86592 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -22,6 +22,7 @@ with workflow.unsafe.imports_passed_through(): drift, gather_read_tags, index_connector_tags, + index_connectors, minimal_retrain, pi_web_api_scouter, predictions_batch, @@ -234,7 +235,8 @@ class Formatters(SientiaMonitoring): The division is by address: all the names of an address stay in the same slot (SIENTIAPDE-1974, O5). Read tags that cannot be resolved are left out and notified once per run: a tag_id missing from connector_tags, a connector missing - or inactive, or a server_id missing from opc_servers. + or inactive, a server_id missing from opc_servers, or a read tag without a usable + frequency (or, without tag_id, without server_id, tag_address or tag_name). Args: - input_data (dict[str, Any]): The input data containing @@ -259,10 +261,7 @@ class Formatters(SientiaMonitoring): pipelines = input_data['pipelines'] active_ingestors = input_data['active_ingestors'] opc_servers = {str(server['id']): server for server in input_data['opc_servers']} - connectors = { - str(connector['connector_id']): connector - for connector in input_data.get('connectors', []) - } + connectors = index_connectors(input_data.get('connectors', [])) tags_by_id = index_connector_tags(input_data.get('connector_tags', [])) entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id) @@ -286,8 +285,9 @@ class Formatters(SientiaMonitoring): Notify the read tags left out of the slots, one notification per kind of issue. Args: - issues (dict[str, set[str]]): Unresolved ids from gather_read_tags, by kind: - 'servers' (opc-servers server_id), 'tags' (tag_id), 'connectors' (connector_id) + issues (dict[str, set[str]]): What gather_read_tags left out, by kind: + 'servers' (opc-servers server_id), 'tags' (tag_id), 'connectors' + (connector_id) and 'invalid' (schedule_name/read tag) metadata (dict[str, Any]): Metadata for the notifications """ kinds = ( @@ -299,12 +299,17 @@ class Formatters(SientiaMonitoring): ( 'tags', 'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION', - 'Tags {} not found among the active Read tags of connector_tags', + 'Tags {} not found among the active, complete Read tags of connector_tags', ), ( 'connectors', 'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION', - 'Connectors {} not found or not active in connectors', + 'Connectors {} not found, not active or incomplete in connectors', + ), + ( + 'invalid', + 'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION', + 'Read tags {} have no integer frequency or lack server_id, tag_address or tag_name', ), ) for kind, notification_id, message in kinds: diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 615ef00..a34fbdb 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -1,4 +1,4 @@ -from typing import Any +from typing import Any, cast from orchestrator.utils.converters import parse_frequency @@ -400,11 +400,56 @@ def predictions_batch(config: dict[str, Any]): } +# Fields a tag database document needs to take part in a slot. Documents missing any +# of them are left out of the index, so their read tags are notified as not found +# instead of failing the whole run. +CONNECTOR_TAG_FIELDS = ('tag_id', 'connector_id', 'address', 'tag_name') +CONNECTOR_FIELDS = ('connector_id', 'name', 'url', 'server_uri') + + +def _complete(document: dict[str, Any], fields: tuple[str, ...]) -> bool: + return all(document.get(field) not in (None, '') for field in fields) + + +def _tag_id(tag: dict[str, Any]) -> str | None: + """The read tag's tag_id as text, or None when it has none (missing, null or empty).""" + tag_id = tag.get('tag_id') + return None if tag_id in (None, '') else str(tag_id) + + +def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]: + """ + The tag_ids the scouter pipelines reference, for the connector_tags query. + + Each id goes in both forms, int and text, because the tag database stores int and + pipelines may hold either. Sorted, so the query is the same on every run. + + Args: + pipelines (list[dict[str, Any]]): Pipeline configurations + + Returns: + list[Any]: The referenced tag_ids (empty when no read tag carries a tag_id) + """ + ids: set[Any] = set() + for pipeline in pipelines: + if pipeline.get('workflow_type') != 'scouter': + continue + for tag in pipeline.get('read_tags') or []: + tag_id = _tag_id(tag) + if tag_id is None: + continue + ids.add(tag_id) + if tag_id.lstrip('-').isdigit(): + ids.add(int(tag_id)) + return sorted(ids, key=lambda value: (isinstance(value, str), str(value))) + + 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. The key is the tag_id as text, because pipelines may store it as int or str. + Documents without tag_id, connector_id, address or tag_name are left out. Args: connector_tags (list[dict[str, Any]]): connector_tags documents (SIENTIAPDE-1569), @@ -413,7 +458,28 @@ def index_connector_tags(connector_tags: list[dict[str, Any]]) -> dict[str, dict Returns: dict[str, dict[str, Any]]: connector_tags documents keyed by str(tag_id) """ - return {str(tag['tag_id']): tag for tag in connector_tags} + return { + str(tag['tag_id']): tag for tag in connector_tags if _complete(tag, CONNECTOR_TAG_FIELDS) + } + + +def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + """ + Index the tag database's connectors by connector_id (as text). + + Documents without connector_id, name, url or server_uri are left out. + + Args: + connectors (list[dict[str, Any]]): connectors documents (SIENTIAPDE-1569) + + Returns: + dict[str, dict[str, Any]]: connectors documents keyed by str(connector_id) + """ + return { + str(connector['connector_id']): connector + for connector in connectors + if _complete(connector, CONNECTOR_FIELDS) + } def resolve_tag_names( @@ -438,8 +504,9 @@ def resolve_tag_names( resolved = [] for tag in read_tags: name = tag.get('tag_name') - if not name and tag.get('tag_id') is not None: - name = tags_by_id.get(str(tag['tag_id']), {}).get('tag_name') + tag_id = _tag_id(tag) + if not name and tag_id is not None: + name = tags_by_id.get(tag_id, {}).get('tag_name') resolved.append({**tag, 'tag_name': name} if name else tag) return resolved @@ -476,6 +543,14 @@ def _opc_server(server_id: str, opc_server: dict[str, Any]) -> dict[str, Any]: } +def _frequency(tag: dict[str, Any]) -> int | None: + """The read tag's frequency as int, or None when it is missing or not an integer.""" + try: + return int(tag['frequency']) + except (KeyError, TypeError, ValueError): + return None + + def _locate_read_tag( tag: dict[str, Any], opc_servers: dict[str, dict[str, Any]], @@ -491,8 +566,8 @@ def _locate_read_tag( tag_address, with the server from opc-servers (RF4). What cannot be resolved is recorded in issues ('tags', 'connectors' or 'servers'). """ - if tag.get('tag_id') is not None: - tag_id = str(tag['tag_id']) + tag_id = _tag_id(tag) + if tag_id is not None: connector_tag = tags_by_id.get(tag_id) if connector_tag is None: issues['tags'].add(tag_id) @@ -512,6 +587,18 @@ def _locate_read_tag( return _opc_server(server_id, opc_server), tag['tag_address'] +def _invalid_read_tag(tag: dict[str, Any]) -> bool: + """ + A read tag the slot cannot use: no integer frequency, or (without tag_id) no + server_id, tag_address or tag_name. + """ + if _frequency(tag) is None: + return True + if _tag_id(tag) is not None: + return False + return not _complete(tag, ('server_id', 'tag_address', 'tag_name')) + + def _merge_active_read(current: dict[str, Any], requested: dict[str, Any] | None) -> dict[str, Any]: """Active read of an address: on if any pipeline asks, with the lowest cycle_multiple (D3).""" if not requested or not requested.get('enabled'): @@ -524,14 +611,16 @@ def _merge_active_read(current: dict[str, Any], requested: dict[str, Any] | None def _assign_server_keys(servers: dict[str, dict[str, Any]]) -> None: """ - Give each server a unique slot key: its name, or name:server_id when another - server_id already took that name. Assigned in server_id order, so the keys are the - same on every run and in every slot. + Give each server a unique slot key: its name, or name:server_id (extended until + free) when another server already took it. Assigned in server_id order, so the keys + are the same on every run and in every slot. """ taken: set[str] = set() for server_id in sorted(servers): server = servers[server_id] - key = server['name'] if server['name'] not in taken else f'{server["name"]}:{server_id}' + key = server['name'] + while key in taken: + key = f'{key}:{server_id}' server['key'] = key taken.add(key) @@ -559,8 +648,8 @@ def gather_read_tags( 'scouter' is read; each needs schedule_name, model_id, model.name and read_tags (with frequency, and optionally tag_id and active_read) opc_servers (dict[str, dict[str, Any]]): opc-servers documents keyed by str(id) - connectors (dict[str, dict[str, Any]]): Active OPC UA connectors keyed by - str(connector_id) + connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by + index_connectors tags_by_id (dict[str, dict[str, Any]]): Active Read connector_tags indexed by index_connector_tags @@ -571,14 +660,20 @@ def gather_read_tags( subscription_period_ms (None = derive from the frequencies) and key (the server's unique slot key) - address (str): The OPC node address - - frequency: The lowest frequency among the pipelines (RF6) + - frequency (int): The lowest frequency among the pipelines (RF6) - active_read (dict): {'enabled': False} unless a pipeline enables it (D3) - names (list[dict]): collection ('raw_{schedule_name}'), tag_name, model_id and model_name, one per pipeline name, sorted - - Ids that could not be resolved: 'tags' (tag_id), 'connectors' - (connector_id) and 'servers' (opc-servers server_id) + - What was left out: 'tags' (tag_id), 'connectors' (connector_id), 'servers' + (opc-servers server_id) and 'invalid' (schedule_name/read tag without a + usable frequency or legacy fields) """ - issues: dict[str, set[str]] = {'tags': set(), 'connectors': set(), 'servers': set()} + issues: dict[str, set[str]] = { + 'tags': set(), + 'connectors': set(), + 'servers': set(), + 'invalid': set(), + } servers: dict[str, dict[str, Any]] = {} entries: dict[tuple[str, str], dict[str, Any]] = {} @@ -586,7 +681,12 @@ def gather_read_tags( if pipeline['workflow_type'] != 'scouter': continue - for tag in resolve_tag_names(pipeline.get('read_tags', []), tags_by_id): + for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id): + if _invalid_read_tag(tag): + ref = _tag_id(tag) or tag.get('tag_address') or tag.get('tag_name') + issues['invalid'].add(f'{pipeline["schedule_name"]}/{ref}') + continue + located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues) if located is None: continue @@ -597,17 +697,18 @@ def gather_read_tags( 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': tag['frequency'], + 'frequency': frequency, 'active_read': {'enabled': False}, 'names': [], }, ) - if int(tag['frequency']) < int(entry['frequency']): - entry['frequency'] = tag['frequency'] + entry['frequency'] = min(entry['frequency'], frequency) entry['active_read'] = _merge_active_read(entry['active_read'], tag.get('active_read')) name = { @@ -671,7 +772,7 @@ def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]: for server_config in slot_config.values(): if server_config['subscription_period_ms'] is None: - frequencies = [int(tag['frequency']) for tag in server_config['tags'].values()] + frequencies = [tag['frequency'] for tag in server_config['tags'].values()] server_config['subscription_period_ms'] = min(frequencies) / 2 return slot_config diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index 565b388..fc0af9a 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -7,16 +7,36 @@ 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 # Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow -# input may override these queries (connectors_query / connector_tags_query). +# input may override these aggregations (connectors_query / connector_tags_query); the +# workflow puts a tag_id $match in front of connector_tags_query. Only the fields the +# slot uses are projected: other attributes may hold BSON types a Temporal payload +# cannot carry. DEFAULT_CONNECTORS_QUERY = { 'collection': 'connectors', - 'filters': {'type': 'opc_ua', 'status': 'Active'}, + 'aggregation': [ + {'$match': {'type': 'opc_ua', 'status': 'Active'}}, + { + '$project': { + '_id': 0, + 'connector_id': 1, + 'name': 1, + 'url': 1, + 'server_uri': 1, + 'use_security': 1, + 'subscription_period_ms': 1, + } + }, + ], } DEFAULT_CONNECTOR_TAGS_QUERY = { 'collection': 'connector_tags', - 'filters': {'category': 'Read', 'status': 'Active'}, + 'aggregation': [ + {'$match': {'category': 'Read', 'status': 'Active'}}, + {'$project': {'_id': 0, 'tag_id': 1, 'connector_id': 1, 'address': 1, 'tag_name': 1}}, + ], } # Runs started before the tag database loads existed replay without them TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database' @@ -49,11 +69,11 @@ class Orchestrator: - schedule_name (str): Name of the orchestration schedule - pipelines_query (dict[str, Any]): MongoDB query for pipeline configurations - opc_servers_query (dict[str, Any]): MongoDB query for OPC server data - Optional fields: - - connectors_query (dict[str, Any]): MongoDB query for the tag database - connectors (default: DEFAULT_CONNECTORS_QUERY) - - connector_tags_query (dict[str, Any]): MongoDB query for the tag database - tags (default: DEFAULT_CONNECTOR_TAGS_QUERY) + Optional fields (used only when a scouter read tag carries a tag_id): + - connectors_query (dict[str, Any]): MongoDB aggregation for the tag + database connectors (default: DEFAULT_CONNECTORS_QUERY) + - connector_tags_query (dict[str, Any]): MongoDB aggregation for the tag + database tags (default: DEFAULT_CONNECTOR_TAGS_QUERY) Returns: None: Workflow completes without return value @@ -120,38 +140,58 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60), ) - # Tag database: empty or missing collections just load nothing, and the read - # tags without tag_id keep using opc-servers - connectors_handler = None - connector_tags_handler = None - if workflow.patched(TAG_DATABASE_PATCH): + 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.find_documents_in_mongodb, + Activities.aggregate_documents_in_mongodb, { **metadata, - 'query': input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY), + '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.find_documents_in_mongodb, + Activities.aggregate_documents_in_mongodb, { **metadata, - 'query': input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY), + '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), ) - pipeline_config = await pipeline_config_handler + connectors = await connectors_handler + connector_tags = await connector_tags_handler + orchestrated_schedules = await orchestrated_schedules_handler current_slot_config = await current_slot_config_handler opc_servers = await opc_servers_handler active_ingestors = await active_ingestors_handler - connectors = await connectors_handler if connectors_handler else [] - connector_tags = await connector_tags_handler if connector_tags_handler else [] formatted_orchestrated_schedules_handler = workflow.start_local_activity_method( Activities.format_schedule_config, diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index d5d028b..9cf173f 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -445,6 +445,7 @@ def test_process_slots_notifies_unresolved_tags_once(formatters): {'server_id': '9', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000}, {'tag_id': 404, 'frequency': 1000}, {'tag_id': 8, 'frequency': 1000}, + {'server_id': '1', 'tag_address': 'ns=2;s=C', 'tag_name': 'c'}, ], ), ] @@ -475,20 +476,65 @@ def test_process_slots_notifies_unresolved_tags_once(formatters): call( metadata=metadata['metadata'], notification_id='ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION', - message='Tags 404 not found among the active Read tags of connector_tags', + message='Tags 404 not found among the active, complete Read tags of connector_tags', block='orchestrator', level=NotificationLevel.ERROR, ), call( metadata=metadata['metadata'], notification_id='ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION', - message='Connectors 5 not found or not active in connectors', + message='Connectors 5 not found, not active or incomplete in connectors', + block='orchestrator', + level=NotificationLevel.ERROR, + ), + call( + metadata=metadata['metadata'], + notification_id='ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION', + message=( + 'Read tags sin/ns=2;s=C have no integer frequency ' + 'or lack server_id, tag_address or tag_name' + ), block='orchestrator', level=NotificationLevel.ERROR, ), ] ) - assert formatters.send_notification.call_count == 3 + assert formatters.send_notification.call_count == 4 + + +def test_process_slots_survives_malformed_tag_database(formatters): + """An incomplete connector or tag leaves only its own read tags out of the slot""" + pipelines = [ + _scouter_pipeline( + 'sin', + 'm1', + [ + {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}, + {'tag_id': 7, 'frequency': 1000}, + {'tag_id': 8, 'frequency': 1000}, + ], + ), + ] + input_data = { + 'opc_servers': slot_opc_servers, + 'connectors': [{'connector_id': 3, 'name': 'plant', 'server_uri': 'urn:plant'}], + 'connector_tags': [ + {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'P'}, + {'tag_id': 8, 'connector_id': 3, 'tag_name': 'no address'}, + ], + 'active_ingestors': [], + 'pipelines': pipelines, + **metadata, + } + + result = formatters.process_slots(input_data) + + assert list(result['1']) == ['kepserver'] + notified = {c.kwargs['notification_id'] for c in formatters.send_notification.call_args_list} + assert notified == { + 'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION', + 'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION', + } def test_process_slots_is_stable_across_pipeline_order(formatters): diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index e8c3db3..bc3ab57 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -7,11 +7,13 @@ from orchestrator.utils.orchestrator_functions import ( drift, gather_read_tags, index_connector_tags, + index_connectors, minimal_retrain, overlap_filter_config, pi_web_api_scouter, predictions_batch, process_path_priority, + referenced_tag_ids, resolve_tag_names, scouter, simple_metrics, @@ -380,7 +382,10 @@ def _gather(pipelines, servers=None, connector_index=None, tag_index=None): def test_index_connector_tags(): - tags = [{'tag_id': 7, 'tag_name': 'a'}, {'tag_id': '8', 'tag_name': 'b'}] + tags = [ + {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'a'}, + {'tag_id': '8', 'connector_id': 3, 'address': 'ns=3;i=2', 'tag_name': 'b'}, + ] assert index_connector_tags(tags) == {'7': tags[0], '8': tags[1]} @@ -414,7 +419,7 @@ def test_gather_read_tags_multi_name(): assert entries[0]['address'] == 'ns=2;s=A' assert entries[0]['frequency'] == 5000 assert entries[0]['names'] == [_name('cos', 'Contador', 'm2'), _name('sin', 'Counter', 'm1')] - assert issues == {'tags': set(), 'connectors': set(), 'servers': set()} + assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()} def test_gather_read_tags_same_name_in_two_pipelines(): @@ -516,7 +521,7 @@ def test_gather_read_tags_unknown_tag_id(): entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 404, 'frequency': 1000}])]) assert entries == [] - assert issues == {'tags': {'404'}, 'connectors': set(), 'servers': set()} + assert issues == {'tags': {'404'}, 'connectors': set(), 'servers': set(), 'invalid': set()} def test_gather_read_tags_inactive_connector(): @@ -524,7 +529,7 @@ def test_gather_read_tags_inactive_connector(): entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 8, 'frequency': 1000}])]) assert entries == [] - assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set()} + assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set(), 'invalid': set()} def test_gather_read_tags_unknown_server(): @@ -533,7 +538,7 @@ def test_gather_read_tags_unknown_server(): ) assert entries == [] - assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}} + assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}, 'invalid': set()} def test_gather_read_tags_mixed_with_and_without_tag_id(): @@ -572,7 +577,7 @@ def test_gather_read_tags_mixed_with_and_without_tag_id(): # Same server_id in connectors and opc-servers: the connectors fields win assert entries[0]['server']['subscription_period_ms'] == 2000 assert entries[1]['server'] is entries[0]['server'] - assert issues == {'tags': set(), 'connectors': set(), 'servers': set()} + assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()} def test_gather_read_tags_server_name_clash(): @@ -644,7 +649,10 @@ def test_gather_read_tags_without_read_tags(): pipeline = _pipeline('sin', 'm1', []) del pipeline['read_tags'] - assert _gather([pipeline]) == ([], {'tags': set(), 'connectors': set(), 'servers': set()}) + assert _gather([pipeline]) == ( + [], + {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()}, + ) def test_build_tag_config(): @@ -680,7 +688,7 @@ def test_build_tag_config(): 'names': [_name('sin', 'a', 'm1')], }, 'ns=2;s=B': { - 'frequency': '300', + 'frequency': 300, 'active_read': {'enabled': False}, 'names': [_name('sin', 'b', 'm1')], }, @@ -736,6 +744,121 @@ def test_scouter_skips_tags_without_name(): assert list(scouter(config)['model_tags']) == ['a'] +def test_referenced_tag_ids(): + """Ids in int and text form, scouter pipelines only, empty or null tag_id ignored""" + pipelines = [ + _pipeline('sin', 'm1', [{'tag_id': 7}, {'tag_id': '12'}, {'tag_id': ''}, {'tag_id': None}]), + _pipeline('cos', 'm2', [{'tag_id': 'abc'}, _legacy_tag('ns=2;s=A', 'a')]), + _pipeline('pred', 'm3', [{'tag_id': 99}], workflow_type='predictions_batch'), + ] + + # Sorted by text (ints first), so the query is the same on every run + assert referenced_tag_ids(pipelines) == [12, 7, '12', '7', 'abc'] + + +def test_referenced_tag_ids_without_tag_ids(): + pipeline = _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')]) + no_read_tags = _pipeline('cos', 'm2', []) + del no_read_tags['read_tags'] + + assert referenced_tag_ids([pipeline, no_read_tags]) == [] + + +def test_index_skips_incomplete_documents(): + """A malformed tag database document is left out instead of failing the run""" + tags = [ + {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'a'}, + {'connector_id': 3, 'address': 'ns=3;i=2', 'tag_name': 'no id'}, + {'tag_id': 9, 'connector_id': 3, 'address': '', 'tag_name': 'no address'}, + ] + connector_list = [ + {'connector_id': 3, 'name': 'plant', 'url': 'opc.tcp://p:4840', 'server_uri': 'urn:p'}, + {'connector_id': 4, 'name': 'no url', 'server_uri': 'urn:x'}, + {'name': 'no id', 'url': 'opc.tcp://x:4840', 'server_uri': 'urn:x'}, + ] + + assert list(index_connector_tags(tags)) == ['7'] + assert list(index_connectors(connector_list)) == ['3'] + + +def test_gather_read_tags_empty_tag_id_keeps_legacy_fields(): + """tag_id '' is no tag database reference: the current fields are used (P4)""" + entries, issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', tag_id='')])]) + + assert [(e['server']['server_id'], e['address']) for e in entries] == [('1', 'ns=2;s=A')] + assert issues == {'tags': set(), 'connectors': set(), 'servers': set(), 'invalid': set()} + + +def test_gather_read_tags_invalid_read_tags(): + """Unusable read tags are reported and skipped; the others still make the slot""" + pipelines = [ + _pipeline( + 'sin', + 'm1', + [ + _legacy_tag('ns=2;s=A', 'a'), + {'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b'}, + _legacy_tag('ns=2;s=C', 'c', frequency='fast'), + {'server_id': '1', 'tag_address': 'ns=2;s=D', 'frequency': 1000}, + {'tag_address': 'ns=2;s=E', 'tag_name': 'e', 'frequency': 1000}, + {'tag_id': 7}, + ], + ) + ] + + entries, issues = _gather(pipelines) + + assert [e['address'] for e in entries] == ['ns=2;s=A'] + assert issues['invalid'] == { + 'sin/7', + 'sin/ns=2;s=B', + 'sin/ns=2;s=C', + 'sin/ns=2;s=D', + 'sin/ns=2;s=E', + } + + +def test_gather_read_tags_frequency_type_does_not_depend_on_order(): + """15000 and '15000' give the same int frequency in any pipeline order (O6)""" + as_int = _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', frequency=15000)]) + as_text = _pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'a', frequency='15000')]) + + first, _ = _gather([as_int, as_text]) + second, _ = _gather([as_text, as_int]) + + assert first[0]['frequency'] == second[0]['frequency'] == 15000 + assert isinstance(first[0]['frequency'], int) + assert isinstance(second[0]['frequency'], int) + + +def test_gather_read_tags_server_key_never_collides(): + """A server literally named 'x:3' does not share its key with server 3""" + servers = { + '1': {**opc_servers['1'], 'server_name': 'x:3'}, + '2': {**opc_servers['2'], 'server_name': 'x'}, + '3': {'id': '3', 'server_name': 'x', 'url': 'opc.tcp://three:4840', 'uri': 'urn:3'}, + } + pipelines = [ + _pipeline( + 'sin', + 'm1', + [ + _legacy_tag('ns=2;s=A', 'a', server_id='1'), + _legacy_tag('ns=2;s=B', 'b', server_id='2'), + _legacy_tag('ns=2;s=C', 'c', server_id='3'), + ], + ), + ] + + entries, _issues = _gather(pipelines, servers=servers) + slot = build_tag_config(entries) + + keys = [e['server']['key'] for e in entries] + assert len(set(keys)) == 3 + assert {server['server_id'] for server in slot.values()} == {'1', '2', '3'} + assert all(len(server['tags']) == 1 for server in slot.values()) + + def test_base_scouter(): config = { 'workflow_type': 'scouter', diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index c68e1d4..72f20c8 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -27,8 +27,9 @@ metadata = { @mark.asyncio +@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7, '7']) @patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock) -async def test_run(workflow_mock, orchestrator): +async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator): workflow_mock.patched = MagicMock(return_value=True) input_data = { 'pipelines_query': 'SELECT * FROM bucket', @@ -116,24 +117,44 @@ async def test_run(workflow_mock, orchestrator): ) workflow_mock.patched.assert_called_once_with(TAG_DATABASE_PATCH) + referenced_tag_ids_mock.assert_called_once_with( + workflow_mock.start_local_activity_method.return_value + ) workflow_mock.start_local_activity_method.assert_has_calls( [ call( - Activities.find_documents_in_mongodb, + Activities.aggregate_documents_in_mongodb, {**metadata, 'query': DEFAULT_CONNECTORS_QUERY}, retry_policy=ANY, start_to_close_timeout=ANY, ), call( - Activities.find_documents_in_mongodb, - {**metadata, 'query': DEFAULT_CONNECTOR_TAGS_QUERY}, + Activities.aggregate_documents_in_mongodb, + { + **metadata, + 'query': { + **DEFAULT_CONNECTOR_TAGS_QUERY, + 'aggregation': [ + {'$match': {'tag_id': {'$in': [7, '7']}}}, + *DEFAULT_CONNECTOR_TAGS_QUERY['aggregation'], + ], + }, + }, retry_policy=ANY, start_to_close_timeout=ANY, ), ] ) + # The activity appends a stage to the aggregation it gets: never the module default + tag_database_calls = _tag_database_calls(workflow_mock) + assert len(tag_database_calls) == 2 + for c in tag_database_calls: + aggregation = c.args[1]['query']['aggregation'] + assert aggregation is not DEFAULT_CONNECTORS_QUERY['aggregation'] + assert aggregation is not DEFAULT_CONNECTOR_TAGS_QUERY['aggregation'] + workflow_mock.start_local_activity_method.assert_has_calls( [ call( @@ -379,13 +400,35 @@ async def test_run(workflow_mock, orchestrator): ) +def _tag_database_calls(workflow_mock): + return [ + c + for c in workflow_mock.start_local_activity_method.call_args_list + if c.args[0] is Activities.aggregate_documents_in_mongodb + and isinstance(c.args[1]['query'], dict) + and c.args[1]['query'].get('collection') in ('connectors', 'connector_tags') + ] + + +def _process_slots_input(workflow_mock): + return next( + c.args[1] + for c in workflow_mock.start_local_activity_method.call_args_list + if c.args[0] is Activities.process_slots + ) + + @mark.asyncio +@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7]) @patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock) -async def test_run_with_tag_database_queries_from_input(workflow_mock, orchestrator): - """The input overrides the default tag database queries (SIENTIAPDE-1974, D9)""" +async def test_run_with_tag_database_queries_from_input(workflow_mock, _ids, orchestrator): + """The input overrides the default tag database aggregations (SIENTIAPDE-1974, D9)""" workflow_mock.patched = MagicMock(return_value=True) - connectors_query = {'collection': 'connectors', 'filters': {'type': 'opc_ua'}} - connector_tags_query = {'collection': 'connector_tags', 'filters': {'category': 'Read'}} + connectors_query = {'collection': 'connectors', 'aggregation': [{'$match': {'type': 'opc_ua'}}]} + connector_tags_query = { + 'collection': 'connector_tags', + 'aggregation': [{'$match': {'category': 'Read'}}], + } input_data = { 'pipelines_query': 'SELECT * FROM bucket', 'opc_servers_query': 'SELECT * FROM servers', @@ -396,27 +439,44 @@ async def test_run_with_tag_database_queries_from_input(workflow_mock, orchestra await orchestrator.run(input_data) - workflow_mock.start_local_activity_method.assert_has_calls( - [ - call( - Activities.find_documents_in_mongodb, - {**metadata, 'query': connectors_query}, - retry_policy=ANY, - start_to_close_timeout=ANY, - ), - call( - Activities.find_documents_in_mongodb, - {**metadata, 'query': connector_tags_query}, - retry_policy=ANY, - start_to_close_timeout=ANY, - ), - ] - ) + queries = [c.args[1]['query'] for c in _tag_database_calls(workflow_mock)] + assert queries == [ + connectors_query, + { + 'collection': 'connector_tags', + 'aggregation': [ + {'$match': {'tag_id': {'$in': [7]}}}, + {'$match': {'category': 'Read'}}, + ], + }, + ] @mark.asyncio +@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[]) @patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock) -async def test_run_replay_before_tag_database_patch(workflow_mock, orchestrator): +async def test_run_without_tag_ids_skips_tag_database(workflow_mock, _ids, orchestrator): + """No read tag with tag_id: the tag database is not queried at all (P4)""" + workflow_mock.patched = MagicMock(return_value=True) + input_data = { + 'pipelines_query': 'SELECT * FROM bucket', + 'opc_servers_query': 'SELECT * FROM servers', + 'schedule_name': 'test-schedule-name', + } + + await orchestrator.run(input_data) + + workflow_mock.patched.assert_called_once_with(TAG_DATABASE_PATCH) + assert _tag_database_calls(workflow_mock) == [] + slots_input = _process_slots_input(workflow_mock) + assert slots_input['connectors'] == [] + assert slots_input['connector_tags'] == [] + + +@mark.asyncio +@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7]) +@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock) +async def test_run_replay_before_tag_database_patch(workflow_mock, _ids, orchestrator): """A run started before the patch replays without the tag database loads""" workflow_mock.patched = MagicMock(return_value=False) input_data = { @@ -427,28 +487,7 @@ async def test_run_replay_before_tag_database_patch(workflow_mock, orchestrator) await orchestrator.run(input_data) - queries = [ - c.args[1].get('query') - for c in workflow_mock.start_local_activity_method.call_args_list - if c.args[0] is Activities.find_documents_in_mongodb - ] - assert DEFAULT_CONNECTORS_QUERY not in queries - assert DEFAULT_CONNECTOR_TAGS_QUERY not in queries - - workflow_mock.start_local_activity_method.assert_has_calls( - [ - call( - Activities.process_slots, - { - **metadata, - 'opc_servers': workflow_mock.start_local_activity_method.return_value, - 'connectors': [], - 'connector_tags': [], - 'active_ingestors': workflow_mock.start_local_activity_method.return_value, - 'pipelines': workflow_mock.start_local_activity_method.return_value, - }, - retry_policy=ANY, - start_to_close_timeout=ANY, - ) - ] - ) + assert _tag_database_calls(workflow_mock) == [] + slots_input = _process_slots_input(workflow_mock) + assert slots_input['connectors'] == [] + assert slots_input['connector_tags'] == []