diff --git a/CLAUDE.md b/CLAUDE.md index cfa191c..2ee7305 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,7 +47,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`). - **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. Receiver groups also support a `model_ids` allow-list (SIENTIAPDE-2149), independent of the pre-existing `ignore_models_id` deny-list — both apply together (AND) when a group configures both, and `model_ids` follows the same union-at-query-time / precise-per-group-slice split as `levels`. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run. - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections. -- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing. +- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`) and builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list, tags resolved through the tag database when they carry a `tag_id`; see the README section "OPC slots"). `resolve_tag_names` is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing. ### Temporal-specific conventions (all workflow/activity files follow this) - Every workflow and activity-consuming module wraps non-deterministic/side-effecting imports in `with workflow.unsafe.imports_passed_through(): ...` right after `from temporalio import workflow`, because Temporal's sandbox restricts what workflow code can import directly. When adding a new workflow file, follow the existing pattern (see `orchestrator/workflows/orchestrator.py`, `orchestrator/activities/activities.py`) rather than importing at module top level. diff --git a/README.md b/README.md index cd37937..136b7d0 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,7 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip - **Infrastructure Management**: Creates, updates, and deletes workflow schedules **Execution Flow**: -1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, and active ingestors +1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, active ingestors, and the tag database (`connectors` and `connector_tags`) 2. **Parallel Processing**: Formats orchestrated schedules and processes new schedules and slots 3. **Parallel Config Creation**: Creates schedule and slot action configurations, normalizes schedules, and creates collections with TTL indexes 4. **Parallel Operations**: Executes slot deletions, slot updates, schedule deletions, schedule creations, and schedule updates concurrently @@ -121,6 +121,70 @@ 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: + +```json +{ + "connectors_query": { + "collection": "connectors", + "filters": {"type": "opc_ua", "status": "Active"} + }, + "connector_tags_query": { + "collection": "connector_tags", + "filters": {"category": "Read", "status": "Active"} + } +} +``` + +An empty or missing collection loads nothing, and the pipelines keep working through `opc-servers`. + +#### OPC slots (slot v2, contract C1) + +`process_slots` writes one slot per active ingestor to Redis (`slot:opc_tags:{n}`). The OPC Gateway reads it (SIENTIAPDE-1974). Each server carries its connection fields, and each tag address carries the list of names it is written under: + +```json +{ + "kepserver": { + "server_id": "1", "name": "kepserver", "url": "opc.tcp://host:49320", + "server_uri": "urn:...", "use_security": true, "subscription_period_ms": 7500.0, + "tags": { + "ns=2;s=Channel1.Device1.Tag1": { + "frequency": 15000, + "active_read": {"enabled": false}, + "names": [ + {"collection": "raw_model_a", "tag_name": "Counter", "model_id": "1", "model_name": "model a"}, + {"collection": "raw_model_b", "tag_name": "Contador", "model_id": "2", "model_name": "model b"} + ] + } + } + } +} +``` + +How each scouter `read_tags` entry is resolved: + +| Read tag | Address and server | Name | +|---|---|---| +| With `tag_id` | `connector_tags` (`address`, `connector_id`) and `connectors` (`url`, `server_uri`, `use_security`, `subscription_period_ms`) | The read tag's `tag_name`, else the `connector_tags` one | +| Without `tag_id` | `server_id` + `tag_address`, server from `opc-servers` | The read tag's `tag_name` | + +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`. +- **`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`. +- **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`. +- **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. + +**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 ```mermaid @@ -608,10 +672,20 @@ Temporal input configuration sample: "filters": { } + }, + "connectors_query": { + "collection": "connectors", + "filters": {"type": "opc_ua", "status": "Active"} + }, + "connector_tags_query": { + "collection": "connector_tags", + "filters": {"category": "Read", "status": "Active"} } } ``` +`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code. + #### Alerts Workflow ```json diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index f33b4db..3ba4247 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -21,9 +21,11 @@ with workflow.unsafe.imports_passed_through(): build_tag_config, drift, gather_read_tags, + index_connector_tags, minimal_retrain, pi_web_api_scouter, predictions_batch, + resolve_tag_names, scouter, simple_metrics, ) @@ -173,6 +175,8 @@ class Formatters(SientiaMonitoring): - input_data (dict[str, Any]): The input data containing the schedules to process. - pipelines (list[dict[str, Any]]): The schedules to process. + - connector_tags (list[dict[str, Any]], optional): Active Read tags of + the tag database, to name the scouter read tags that carry only a tag_id. Returns: - dict[str, Any]: The schedule configuration dictionary keyed by namespace @@ -183,6 +187,7 @@ class Formatters(SientiaMonitoring): self.info('Processing schedules...', metadata=metadata) pipelines = input_data['pipelines'] + tags_by_id = index_connector_tags(input_data.get('connector_tags', [])) schedule_config: dict[str, dict[str, Any]] = { self.scouter_namespace: {}, @@ -196,6 +201,13 @@ class Formatters(SientiaMonitoring): self.error(f'Workflow type {workflow_type} not supported', metadata=metadata) continue + if workflow_type == 'scouter' and 'read_tags' in pipeline: + # Same names as the slot's names[] (process_slots): SIENTIAPDE-1974 SC1 + pipeline = { + **pipeline, + 'read_tags': resolve_tag_names(pipeline['read_tags'], tags_by_id), + } + schedule_type = schedule_types[workflow_type] namespace = schedule_type['namespace'] function = schedule_type['function'] @@ -216,14 +228,24 @@ class Formatters(SientiaMonitoring): def process_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Extracts all read tags from input pipelines, divides them into slots and - returns a slot config dictionary. If no ingestor is available, only one slot - is created. + returns a slot config dictionary in the slot v2 format (contract C1). If no + ingestor is available, only one slot is created. + + 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. Args: - input_data (dict[str, Any]): The input data containing the schedules to process. - pipelines (list[dict[str, Any]]): The schedules to process. - - opc_servers (list[str]): The OPC servers to create ingestor config. + - opc_servers (list[dict[str, Any]]): The OPC servers (opc-servers), for + read tags without tag_id. + - connectors (list[dict[str, Any]], optional): Active OPC UA connectors of + the tag database, for read tags with tag_id. + - connector_tags (list[dict[str, Any]], optional): Active Read tags of the + tag database, for read tags with tag_id. - active_ingestors (list[str]): The active ingestors to divide into slots. Returns: @@ -235,55 +257,66 @@ class Formatters(SientiaMonitoring): self.info('Processing slots...', metadata=metadata) pipelines = input_data['pipelines'] - opc_servers_list = input_data['opc_servers'] 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', []) + } + tags_by_id = index_connector_tags(input_data.get('connector_tags', [])) - opc_servers = {} - for server in opc_servers_list: - opc_servers[server['id']] = { - **server, - } + entries, issues = gather_read_tags(pipelines, opc_servers, connectors, tags_by_id) + self._notify_slot_issues(issues, metadata) - tags = list(gather_read_tags(pipelines).values()) - - number_of_tags = len(tags) number_of_slots = len(active_ingestors) if active_ingestors else 1 - tags_per_slot = ceil(number_of_tags / number_of_slots) + entries_per_slot = ceil(len(entries) / number_of_slots) slot_config: dict[str, Any] = {} - last_index = 0 - - for i in range(1, number_of_slots): - slot_tags = tags[last_index : last_index + tags_per_slot] - slot_config[f'{i}'], notifications = build_tag_config(slot_tags, opc_servers) - - if notifications: - self.send_notification( - metadata=metadata, - notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION', - message=f'Servers {", ".join(notifications)} not found in opc_servers', - block='orchestrator', - level=NotificationLevel.ERROR, - ) - - last_index += tags_per_slot - - slot_tags = tags[last_index:] - slot_config[f'{number_of_slots}'], notifications = build_tag_config(slot_tags, opc_servers) - if notifications: - self.send_notification( - metadata=metadata, - notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION', - message=f'Servers {", ".join(notifications)} not found in opc_servers', - block='orchestrator', - level=NotificationLevel.ERROR, - ) + for i in range(number_of_slots): + slot_entries = entries[i * entries_per_slot : (i + 1) * entries_per_slot] + slot_config[f'{i + 1}'] = build_tag_config(slot_entries) self.info('Processed slots', metadata=metadata) self.debug(json.dumps(slot_config, indent=4, sort_keys=True), metadata=metadata) return slot_config + def _notify_slot_issues(self, issues: dict[str, set[str]], metadata: dict[str, Any]) -> None: + """ + 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) + metadata (dict[str, Any]): Metadata for the notifications + """ + kinds = ( + ( + 'servers', + 'ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION', + 'Servers {} not found in opc_servers', + ), + ( + 'tags', + 'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION', + 'Tags {} not found among the active Read tags of connector_tags', + ), + ( + 'connectors', + 'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION', + 'Connectors {} not found or not active in connectors', + ), + ) + for kind, notification_id, message in kinds: + if issues.get(kind): + self.send_notification( + metadata=metadata, + notification_id=notification_id, + message=message.format(', '.join(sorted(issues[kind]))), + block='orchestrator', + level=NotificationLevel.ERROR, + ) + @activity.defn(name='format_schedule_config') def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ @@ -430,15 +463,21 @@ class Formatters(SientiaMonitoring): to determine which slots need to be inserted or deleted. Slots are identified by numeric IDs, and excess slots are marked for deletion. + Only slots whose content changed are inserted, so an unchanged slot is not + rewritten (nor reported) on every run. gather_read_tags builds the slot in a + stable order, so the same pipelines always give the same slot (SIENTIAPDE-1974, O6). + A slot in the v1 format differs from its v2 rewrite and is replaced once. + Args: input_data (dict[str, Any]): The input data containing: - - current_slot_config (dict[str, Any]): The current slot config in Redis + - current_slot_config (dict[str, Any]): The current slot config in Redis, + keyed by Redis key (slot:opc_tags:{id}) or by slot id - slot_config (dict[str, Any]): The slot config to process - metadata (dict): Metadata for logging purposes Returns: dict[str, Any]: A dictionary containing: - - to_insert (dict): Slots that need to be inserted/updated + - to_insert (dict): Slots that are new or changed - to_delete (list[str]): Slot IDs that need to be deleted """ @@ -456,7 +495,16 @@ class Formatters(SientiaMonitoring): if number_of_current_slots > number_of_slots: to_delete = [str(i) for i in range(number_of_slots + 1, number_of_current_slots + 1)] - output = {'to_delete': to_delete, 'to_insert': slot_config} + current_by_id = { + str(key).rsplit(':', 1)[-1]: config for key, config in current_slot_config.items() + } + to_insert = { + slot: config + for slot, config in slot_config.items() + if slot not in current_by_id or current_by_id[slot] != config + } + + output = {'to_delete': to_delete, 'to_insert': to_insert} self.info('Created slot config', metadata=metadata) self.debug(json.dumps(output, indent=4, sort_keys=True), metadata=metadata) diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 39ee0aa..615ef00 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -174,8 +174,10 @@ def scouter(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - schedule_name (str): Name of the schedule (used for topic generation) - - read_tags (list[dict]): List of tag configurations with: - - tag_name (str): Name of the tag to read + - read_tags (list[dict]): List of tag configurations, already passed through + resolve_tag_names (process_schedules does it), with: + - tag_name (str): Name of the tag to read. A tag without one (a tag_id the + tag database cannot resolve) is left out; process_slots notifies it - aggr_func (str, optional): Aggregation function for data collection (default: 'lts') Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum' - data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100]) @@ -188,6 +190,8 @@ def scouter(config: dict[str, Any]): tags = {} for tag in config['read_tags']: + if not tag.get('tag_name'): + continue tags[tag['tag_name']] = { 'aggr_func': tag.get('aggr_func', 'lts'), 'data_range': tag.get('data_range', [-100, 100]), @@ -396,110 +400,278 @@ def predictions_batch(config: dict[str, Any]): } -def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: +def index_connector_tags(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """ - Gather all read tags from scouter pipeline configurations. + Index the tag database's connector_tags by tag_id. - Collects all read tags from scouter-type pipelines and organizes them by - server_id and tag_address, tracking which Kafka topics each tag is associated with. - This function is used during slot configuration to aggregate tags across multiple - scouter pipelines for efficient OPC server slot allocation. + The key is the tag_id as text, because pipelines may store it as int or str. Args: - pipelines (list[dict[str, Any]]): List of pipeline configurations to process. - Only pipelines with workflow_type 'scouter' are processed. Each scouter - pipeline should contain a 'read_tags' list with tag configurations. + connector_tags (list[dict[str, Any]]): connector_tags documents (SIENTIAPDE-1569), + each with tag_id, connector_id, address and tag_name Returns: - dict[str, Any]: Dictionary of read tags keyed by "server_id:tag_address", - where each entry contains: - - All original tag configuration fields - - topics (list[str]): List of Kafka topic names associated with this tag - (format: 'raw_{schedule_name}') + dict[str, dict[str, Any]]: connector_tags documents keyed by str(tag_id) """ + return {str(tag['tag_id']): tag for tag in connector_tags} - tags = {} - # Get all read tags from pipelines +def resolve_tag_names( + read_tags: list[dict[str, Any]], tags_by_id: dict[str, dict[str, Any]] +) -> list[dict[str, Any]]: + """ + Fill in the tag_name of each read tag of a scouter pipeline. + + The name is the pipeline's own tag_name when it has one, else the connector_tags + name of its tag_id. The scouter's model_tags (scouter()) and the slot's names[] + (gather_read_tags()) both take the name from here, so the name the gateway writes + is always the one the scouter looks for (SIENTIAPDE-1974, SC1). + + Args: + read_tags (list[dict[str, Any]]): The pipeline's read_tags + tags_by_id (dict[str, dict[str, Any]]): connector_tags indexed by index_connector_tags + + Returns: + list[dict[str, Any]]: The read tags, each with tag_name set when it could be + resolved. A tag_id missing from tags_by_id leaves the tag without a name. + """ + 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') + resolved.append({**tag, 'tag_name': name} if name else tag) + return resolved + + +def _connector_server(connector: dict[str, Any]) -> dict[str, Any]: + """Slot server fields of a tag database connector (connectors collection).""" + return { + 'server_id': str(connector['connector_id']), + 'name': connector['name'], + 'url': connector['url'], + 'server_uri': connector['server_uri'], + 'use_security': bool(connector.get('use_security', False)), + 'subscription_period_ms': connector.get('subscription_period_ms'), + 'from_tag_database': True, + } + + +def _opc_server(server_id: str, opc_server: dict[str, Any]) -> dict[str, Any]: + """ + Slot server fields of a legacy opc-servers document. + + The certificate now always comes from MinIO on the gateway side, so the file paths + are not carried: a cert_path only tells that the server uses security (RF4). The + subscription period is left to build_tag_config (half the lowest tag frequency). + """ + return { + 'server_id': server_id, + 'name': opc_server['server_name'], + 'url': opc_server['url'], + 'server_uri': opc_server['uri'], + 'use_security': bool(opc_server.get('cert_path')), + 'subscription_period_ms': None, + 'from_tag_database': False, + } + + +def _locate_read_tag( + tag: dict[str, Any], + opc_servers: dict[str, dict[str, Any]], + connectors: dict[str, dict[str, Any]], + tags_by_id: dict[str, dict[str, Any]], + issues: dict[str, set[str]], +) -> tuple[dict[str, Any], str] | None: + """ + Server and address of a read tag, or None when they cannot be resolved. + + A tag with tag_id takes its address and connector from connector_tags and its server + from connectors (RF3). A tag without one keeps the current fields: server_id + + 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']) + connector_tag = tags_by_id.get(tag_id) + if connector_tag is None: + issues['tags'].add(tag_id) + return None + connector_id = str(connector_tag['connector_id']) + connector = connectors.get(connector_id) + if connector is None: + issues['connectors'].add(connector_id) + return None + return _connector_server(connector), connector_tag['address'] + + server_id = str(tag['server_id']) + opc_server = opc_servers.get(server_id) + if opc_server is None: + issues['servers'].add(server_id) + return None + return _opc_server(server_id, opc_server), tag['tag_address'] + + +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'): + return current + cycle_multiple = int(requested.get('cycle_multiple', 1)) + if current.get('enabled'): + cycle_multiple = min(cycle_multiple, current['cycle_multiple']) + return {'enabled': True, 'cycle_multiple': cycle_multiple} + + +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. + """ + 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}' + server['key'] = key + taken.add(key) + + +def gather_read_tags( + pipelines: list[dict[str, Any]], + opc_servers: dict[str, dict[str, Any]], + connectors: dict[str, dict[str, Any]], + tags_by_id: dict[str, dict[str, Any]], +) -> tuple[list[dict[str, Any]], dict[str, set[str]]]: + """ + Group the read tags of all scouter pipelines by (server, address) for the slot v2. + + An address read by N pipelines becomes ONE entry with N items in names[]: the + gateway reads it once and writes it to N collections (SIENTIAPDE-1974, RF2). Each + read tag is resolved through the tag database when it has a tag_id (RF3), or through + the current server_id + tag_address + opc-servers fields when it has not (RF4). + + A server known through both connectors and opc-servers under the same server_id is + the same server (the gateway keys its connections by server_id); the connectors + fields win. + + Args: + pipelines (list[dict[str, Any]]): Pipeline configurations. Only workflow_type + '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) + tags_by_id (dict[str, dict[str, Any]]): Active Read connector_tags indexed by + index_connector_tags + + Returns: + tuple[list[dict[str, Any]], dict[str, set[str]]]: + - One entry per (server, address), sorted by server_id then address, with: + - server (dict): server_id, name, url, server_uri, use_security, + 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) + - 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) + """ + issues: dict[str, set[str]] = {'tags': set(), 'connectors': set(), 'servers': set()} + servers: dict[str, dict[str, Any]] = {} + entries: dict[tuple[str, str], dict[str, Any]] = {} + 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: - tags[tag_string] = {**tag, 'topics': []} + for tag in resolve_tag_names(pipeline.get('read_tags', []), tags_by_id): + located = _locate_read_tag(tag, opc_servers, connectors, tags_by_id, issues) + if located is None: + continue + server, address = located - tags[tag_string]['topics'].append(f'raw_{pipeline["schedule_name"]}') + 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 - return tags + entry = entries.setdefault( + (server_id, address), + { + 'address': address, + 'frequency': tag['frequency'], + 'active_read': {'enabled': False}, + 'names': [], + }, + ) + if int(tag['frequency']) < int(entry['frequency']): + entry['frequency'] = tag['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) + + _assign_server_keys(servers) + + gathered = [] + for (server_id, _address), entry in sorted(entries.items()): + # A stable order keeps the slot identical between runs, so the gateway does + # not re-subscribe an unchanged slot (O6) + entry['names'].sort(key=lambda n: (n['collection'], n['tag_name'], str(n['model_id']))) + gathered.append({'server': servers[server_id], **entry}) + + return gathered, issues -def build_tag_config( - tags: list[dict[str, Any]], opc_servers: dict[str, Any] -) -> tuple[dict[str, Any], list]: +def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]: """ - Build tag configuration for a specific slot and OPC server. + Build one slot in the slot v2 format (contract C1) from address entries. - Organizes tags by OPC server name and calculates the minimum subscription period - based on tag frequencies. Validates that all server IDs exist in the OPC - servers configuration. The subscription period is set to half of the minimum - tag frequency to ensure efficient data collection. + The slot no longer carries cert_path, private_key_path or server_cert_path: the + gateway takes the certificate from MinIO (SIENTIAPDE-1974, O4). Args: - tags (list[dict[str, Any]]): List of tag configurations containing: - - server_id (str): ID of the OPC server - - tag_address (str): Address/path of the OPC tag - - frequency (int): Tag read frequency in milliseconds - - Additional tag-specific configuration fields - opc_servers (dict[str, Any]): Dictionary of OPC server configurations keyed by server_id. - Each server configuration should contain: - - server_name (str): Human-readable server name - - url (str): OPC server URL - - uri (str): OPC server URI - - cert_path (str, optional): Certificate file path - - private_key_path (str, optional): Private key file path - - server_cert_path (str, optional): Server certificate file path + entries (list[dict[str, Any]]): Address entries from gather_read_tags for this slot Returns: - tuple[dict[str, Any], list]: A tuple containing: - - Slot configuration dictionary organized by server_name, where each server - contains connection details, tags dictionary, and subscription_period_ms - - List of server IDs (str) that were not found in opc_servers configuration + dict[str, Any]: The slot keyed by server key, each server with server_id, name, + url, server_uri, use_security, subscription_period_ms and tags. Each tag (keyed + by address) has frequency, active_read and names. A server without a + subscription period of its own (opc-servers) gets half its lowest tag + frequency in this slot, as before. """ + slot_config: dict[str, Any] = {} - slot_config = {} - notifications = [] - - for tag in tags: - server_id = tag['server_id'] - - if server_id not in opc_servers: - notifications.append(server_id) - continue - - server_name = opc_servers[server_id]['server_name'] - if server_name not in slot_config: - slot_config[server_name] = { - 'server_id': server_id, - 'name': server_name, - 'url': opc_servers[server_id]['url'], - 'server_uri': opc_servers[server_id]['uri'], - 'cert_path': opc_servers[server_id].get('cert_path', None), - 'private_key_path': opc_servers[server_id].get('private_key_path', None), - 'server_cert_path': opc_servers[server_id].get('server_cert_path', None), + for entry in entries: + server = entry['server'] + if server['key'] not in slot_config: + slot_config[server['key']] = { + 'server_id': server['server_id'], + 'name': server['name'], + 'url': server['url'], + 'server_uri': server['server_uri'], + 'use_security': server['use_security'], + 'subscription_period_ms': server['subscription_period_ms'], 'tags': {}, } - slot_config[server_name]['tags'][tag['tag_address']] = { - **tag, + slot_config[server['key']]['tags'][entry['address']] = { + 'frequency': entry['frequency'], + 'active_read': entry['active_read'], + 'names': entry['names'], } - for server_name in slot_config: - frequencies = [int(x['frequency']) for x in slot_config[server_name]['tags'].values()] + 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()] + server_config['subscription_period_ms'] = min(frequencies) / 2 - min_frequency = min(frequencies) if frequencies else 1000 - - slot_config[server_name]['subscription_period_ms'] = min_frequency / 2 - - return slot_config, notifications + return slot_config diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index a0fe690..565b388 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -8,6 +8,19 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.activities.activities import Activities +# Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow +# input may override these queries (connectors_query / connector_tags_query). +DEFAULT_CONNECTORS_QUERY = { + 'collection': 'connectors', + 'filters': {'type': 'opc_ua', 'status': 'Active'}, +} +DEFAULT_CONNECTOR_TAGS_QUERY = { + 'collection': 'connector_tags', + 'filters': {'category': 'Read', 'status': 'Active'}, +} +# Runs started before the tag database loads existed replay without them +TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database' + @workflow.defn(name='orchestrator') class Orchestrator: @@ -36,6 +49,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) Returns: None: Workflow completes without return value @@ -102,11 +120,38 @@ 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): + connectors_handler = workflow.start_local_activity_method( + Activities.find_documents_in_mongodb, + { + **metadata, + 'query': input_data.get('connectors_query', DEFAULT_CONNECTORS_QUERY), + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60), + ) + + connector_tags_handler = workflow.start_local_activity_method( + Activities.find_documents_in_mongodb, + { + **metadata, + 'query': input_data.get('connector_tags_query', DEFAULT_CONNECTOR_TAGS_QUERY), + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60), + ) + pipeline_config = await pipeline_config_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, @@ -117,7 +162,7 @@ class Orchestrator: schedules_config_handler = workflow.start_local_activity_method( Activities.process_schedules, - {**metadata, 'pipelines': pipeline_config}, + {**metadata, 'pipelines': pipeline_config, 'connector_tags': connector_tags}, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60), ) @@ -127,6 +172,8 @@ class Orchestrator: { **metadata, 'opc_servers': opc_servers, + 'connectors': connectors, + 'connector_tags': connector_tags, 'active_ingestors': active_ingestors, 'pipelines': pipeline_config, }, diff --git a/tests/orchestrator/activities/test_email.py b/tests/orchestrator/activities/test_email.py index 23ec052..99e0fd8 100644 --- a/tests/orchestrator/activities/test_email.py +++ b/tests/orchestrator/activities/test_email.py @@ -85,7 +85,9 @@ def test_close(sientia_monitoring_mock, email): email.close() email.server.quit.assert_called_once() - sientia_monitoring_mock.shutdown.assert_called_once() + # Count only this instance: an Email left by another test may be garbage-collected + # (Email.__del__ -> close) while the patch is active + assert sientia_monitoring_mock.shutdown.call_args_list.count(call(email)) == 1 def test_build_email_html(email): diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index a7aba4a..d5d028b 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -218,109 +218,348 @@ def test_process_schedules_with_invalid_workflow_type(formatters): mock_scouter.assert_any_call(pipelines[2]) -@patch( - 'orchestrator.activities.formatters.gather_read_tags', - return_value={ - '1:test_tag_address': { - 'server_id': '1', - 'server_name': 'test_server_name', - 'tag_address': 'test_tag_address', - 'topics': ['raw_test_schedule'], - 'frequency': 1000, - }, - '2:test_tag_address2': { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address2', - 'topics': ['raw_test_schedule2'], - 'frequency': 1000, - }, - '2:test_tag_address3': { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address3', - 'topics': ['raw_test_schedule2'], - 'frequency': 1000, - }, - }, -) -@patch('orchestrator.activities.formatters.build_tag_config') -def test_process_slots(mock_build_tag_config, mock_gather_read_tags, formatters): +def _scouter_pipeline(schedule_name, model_id, read_tags): + return { + 'workflow_type': 'scouter', + 'schedule_name': schedule_name, + 'model_id': model_id, + 'model': {'name': f'model {model_id}'}, + 'read_tags': read_tags, + } + + +slot_opc_servers = [ + {'id': '1', 'server_name': 'kepserver', 'url': 'opc.tcp://kep:49320', 'uri': 'urn:kep'}, +] + + +def test_process_slots(formatters): + """Slot v2 split by address: the names of an address never land in different slots""" + pipelines = [ + _scouter_pipeline( + 'sin', + 'm1', + [ + {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}, + {'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000}, + ], + ), + _scouter_pipeline( + 'cos', + 'm2', + [ + {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'x', 'frequency': 500}, + {'server_id': '1', 'tag_address': 'ns=2;s=C', 'tag_name': 'c', 'frequency': 4000}, + ], + ), + ] input_data = { - 'opc_servers': [ - {'id': '1', 'server_name': 'test_server_name', 'url': 'test_url', 'uri': 'test_uri'}, - {'id': '2', 'server_name': 'test_server_name2', 'url': 'test_url2', 'uri': 'test_uri2'}, - ], - 'active_ingestors': ['test_active_ingestor1', 'test_active_ingestor2'], - 'pipelines': 'test_gather_read_tags', + 'opc_servers': slot_opc_servers, + 'active_ingestors': ['heartbeat:ingestor:1', 'heartbeat:ingestor:2'], + 'pipelines': pipelines, **metadata, } - expected_opc_servers = { + result = formatters.process_slots(input_data) + + server = { + 'server_id': '1', + 'name': 'kepserver', + 'url': 'opc.tcp://kep:49320', + 'server_uri': 'urn:kep', + 'use_security': False, + } + assert result == { '1': { - 'id': '1', - 'server_name': 'test_server_name', - 'url': 'test_url', - 'uri': 'test_uri', + 'kepserver': { + **server, + 'subscription_period_ms': 250, + 'tags': { + 'ns=2;s=A': { + 'frequency': 500, + 'active_read': {'enabled': False}, + 'names': [ + { + 'collection': 'raw_cos', + 'tag_name': 'x', + 'model_id': 'm2', + 'model_name': 'model m2', + }, + { + 'collection': 'raw_sin', + 'tag_name': 'a', + 'model_id': 'm1', + 'model_name': 'model m1', + }, + ], + }, + 'ns=2;s=B': { + 'frequency': 1000, + 'active_read': {'enabled': False}, + 'names': [ + { + 'collection': 'raw_sin', + 'tag_name': 'b', + 'model_id': 'm1', + 'model_name': 'model m1', + }, + ], + }, + }, + } }, '2': { - 'id': '2', - 'server_name': 'test_server_name2', - 'url': 'test_url2', - 'uri': 'test_uri2', + 'kepserver': { + **server, + 'subscription_period_ms': 2000, + 'tags': { + 'ns=2;s=C': { + 'frequency': 4000, + 'active_read': {'enabled': False}, + 'names': [ + { + 'collection': 'raw_cos', + 'tag_name': 'c', + 'model_id': 'm2', + 'model_name': 'model m2', + }, + ], + }, + }, + } }, } + formatters.send_notification.assert_not_called() - slot_mock = { - 'test_server_name': { - 'server_id': '1', - 'name': 'test_server_name', - 'url': 'test_url', - 'server_uri': 'test_uri', - 'cert_path': None, - 'private_key_path': None, - 'server_cert_path': None, - } + +def test_process_slots_without_ingestors_builds_one_slot(formatters): + pipelines = [ + _scouter_pipeline( + 'sin', + 'm1', + [{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}], + ) + ] + input_data = { + 'opc_servers': slot_opc_servers, + 'active_ingestors': [], + 'pipelines': pipelines, + **metadata, } - mock_build_tag_config.return_value = (slot_mock, ['2']) - result = formatters.process_slots(input_data) - tags = list(mock_gather_read_tags.return_value.values()) + assert list(result) == ['1'] + assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A'] - mock_gather_read_tags.assert_called_once_with(input_data['pipelines']) - mock_build_tag_config.assert_has_calls( - [ - call(tags[:2], expected_opc_servers), - call(tags[2:], expected_opc_servers), - ] - ) +def test_process_slots_more_ingestors_than_addresses(formatters): + """Extra ingestors get empty slots, as before""" + pipelines = [ + _scouter_pipeline( + 'sin', + 'm1', + [{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}], + ) + ] + input_data = { + 'opc_servers': slot_opc_servers, + 'active_ingestors': ['i1', 'i2', 'i3'], + 'pipelines': pipelines, + **metadata, + } + result = formatters.process_slots(input_data) + + assert list(result) == ['1', '2', '3'] + assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A'] + assert result['2'] == {} + assert result['3'] == {} + + +def test_process_slots_with_tag_database(formatters): + """A read tag with tag_id takes address, name and server from the tag database""" + pipelines = [ + _scouter_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}]), + ] + input_data = { + 'opc_servers': slot_opc_servers, + 'connectors': [ + { + 'connector_id': 3, + 'type': 'opc_ua', + 'name': 'plant', + 'url': 'opc.tcp://plant:4840', + 'server_uri': 'urn:plant', + 'use_security': True, + 'subscription_period_ms': 1500, + 'status': 'Active', + } + ], + 'connector_tags': [ + {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'} + ], + 'active_ingestors': ['i1'], + 'pipelines': pipelines, + **metadata, + } + + result = formatters.process_slots(input_data) + + assert result == { + '1': { + 'plant': { + 'server_id': '3', + 'name': 'plant', + 'url': 'opc.tcp://plant:4840', + 'server_uri': 'urn:plant', + 'use_security': True, + 'subscription_period_ms': 1500, + 'tags': { + 'ns=3;i=1001': { + 'frequency': 1000, + 'active_read': {'enabled': False}, + 'names': [ + { + 'collection': 'raw_sin', + 'tag_name': 'Pressure', + 'model_id': 'm1', + 'model_name': 'model m1', + } + ], + } + }, + } + } + } + + +def test_process_slots_notifies_unresolved_tags_once(formatters): + """Each kind of unresolved read tag is notified once per run, not once per slot""" + pipelines = [ + _scouter_pipeline( + 'sin', + 'm1', + [ + {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}, + {'server_id': '9', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000}, + {'tag_id': 404, 'frequency': 1000}, + {'tag_id': 8, 'frequency': 1000}, + ], + ), + ] + input_data = { + 'opc_servers': slot_opc_servers, + 'connectors': [], + 'connector_tags': [ + {'tag_id': 8, 'connector_id': 5, 'address': 'ns=3;i=8', 'tag_name': 'Level'} + ], + 'active_ingestors': ['i1', 'i2'], + 'pipelines': pipelines, + **metadata, + } + + result = formatters.process_slots(input_data) + + assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A'] + assert result['2'] == {} formatters.send_notification.assert_has_calls( [ call( metadata=metadata['metadata'], notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION', - message='Servers 2 not found in opc_servers', + message='Servers 9 not found in opc_servers', block='orchestrator', level=NotificationLevel.ERROR, ), call( metadata=metadata['metadata'], - notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION', - message='Servers 2 not found in opc_servers', + notification_id='ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION', + message='Tags 404 not found among the active 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', block='orchestrator', level=NotificationLevel.ERROR, ), ] ) + assert formatters.send_notification.call_count == 3 - assert result == { - '1': slot_mock, - '2': slot_mock, + +def test_process_slots_is_stable_across_pipeline_order(formatters): + """The same pipelines in another order give the same slots (O6)""" + pipelines = [ + _scouter_pipeline( + 'sin', + 'm1', + [{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}], + ), + _scouter_pipeline( + 'cos', + 'm2', + [ + {'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000}, + {'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'x', 'frequency': 1000}, + ], + ), + ] + + def run(order): + return formatters.process_slots( + { + 'opc_servers': slot_opc_servers, + 'active_ingestors': ['i1', 'i2'], + 'pipelines': order, + **metadata, + } + ) + + assert json.dumps(run(pipelines)) == json.dumps(run(list(reversed(pipelines)))) + + +def test_process_schedules_names_scouter_tags_like_the_slot(formatters): + """model_tags and the slot's names[] take the tag name from the same source (SC1)""" + pipeline = { + **_scouter_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000, 'aggr_func': 'avg'}]), + 'updated_at': '2021-01-01', } + connector_tags = [ + {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'} + ] + connectors = [ + { + 'connector_id': 3, + 'name': 'plant', + 'url': 'opc.tcp://plant:4840', + 'server_uri': 'urn:plant', + } + ] + + schedules = formatters.process_schedules( + {'pipelines': [pipeline], 'connector_tags': connector_tags, **metadata} + ) + slots = formatters.process_slots( + { + 'opc_servers': [], + 'connectors': connectors, + 'connector_tags': connector_tags, + 'active_ingestors': [], + 'pipelines': [pipeline], + **metadata, + } + ) + + model_tags = schedules['scouter']['sin']['model_tags'] + names = slots['1']['plant']['tags']['ns=3;i=1001']['names'] + assert list(model_tags) == ['Pressure'] + assert model_tags['Pressure']['aggr_func'] == 'avg' + assert [name['tag_name'] for name in names] == ['Pressure'] + assert names[0]['collection'] == schedules['scouter']['sin']['topic'] def test_format_schedule_config(formatters): @@ -410,6 +649,31 @@ def test_create_slot_config(formatters): } +def test_create_slot_config_skips_unchanged_slots(formatters): + """Only new or changed slots are written (O6); current slots come keyed by Redis key""" + slot_v2 = {'kepserver': {'server_id': '1', 'tags': {'ns=2;s=A': {'names': []}}}} + slot_v1 = {'kepserver': {'server_id': '1', 'tags': {'ns=2;s=A': {'topics': ['raw_x']}}}} + input_data = { + 'current_slot_config': { + 'slot:opc_tags:1': slot_v2, + 'slot:opc_tags:2': slot_v1, + }, + 'slot_config': {'1': slot_v2, '2': slot_v2, '3': {}}, + } + + result = formatters.create_slot_config(input_data) + + assert result == {'to_delete': [], 'to_insert': {'2': slot_v2, '3': {}}} + + +def test_create_slot_config_without_current_slots(formatters): + slot_config = {'1': {'kepserver': {'tags': {}}}, '2': {}} + + result = formatters.create_slot_config({'current_slot_config': {}, 'slot_config': slot_config}) + + assert result == {'to_delete': [], 'to_insert': slot_config} + + def test_send_success_report(formatters): formatters.send_success_report( metadata=metadata, diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 54a9771..e8c3db3 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -6,11 +6,13 @@ from orchestrator.utils.orchestrator_functions import ( common_config, drift, gather_read_tags, + index_connector_tags, minimal_retrain, overlap_filter_config, pi_web_api_scouter, predictions_batch, process_path_priority, + resolve_tag_names, scouter, simple_metrics, ) @@ -311,249 +313,427 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi assert result == expected -def test_gather_read_tags(): - pipelines = [ - { - 'workflow_type': 'scouter', - 'schedule_name': 'test_schedule', - 'read_tags': [ - { - 'server_id': '1', - 'server_name': 'test_server_name', - 'tag_address': 'test_tag_address', - } - ], - }, - { - 'workflow_type': 'scouter', - 'schedule_name': 'test_schedule2', - 'read_tags': [ - { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address2', - }, - { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address3', - }, - ], - }, - ] - - result = gather_read_tags(pipelines) - - expected = { - '1:test_tag_address': { - 'server_id': '1', - 'server_name': 'test_server_name', - 'tag_address': 'test_tag_address', - 'topics': ['raw_test_schedule'], - }, - '2:test_tag_address2': { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address2', - 'topics': ['raw_test_schedule2'], - }, - '2:test_tag_address3': { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address3', - 'topics': ['raw_test_schedule2'], - }, +def _pipeline(schedule_name, model_id, read_tags, workflow_type='scouter'): + return { + 'workflow_type': workflow_type, + 'schedule_name': schedule_name, + 'model_id': model_id, + 'model': {'name': f'model {model_id}'}, + 'read_tags': read_tags, } - assert result == expected + +def _legacy_tag(address, name, frequency=1000, server_id='1', **extra): + return { + 'server_id': server_id, + 'tag_address': address, + 'tag_name': name, + 'frequency': frequency, + **extra, + } + + +def _name(schedule_name, tag_name, model_id): + return { + 'collection': f'raw_{schedule_name}', + 'tag_name': tag_name, + 'model_id': model_id, + 'model_name': f'model {model_id}', + } + + +opc_servers = { + '1': { + 'id': '1', + 'server_name': 'kepserver', + 'url': 'opc.tcp://kep:49320', + 'uri': 'urn:kep', + 'cert_path': 'certs/client.pem', + }, + '2': {'id': '2', 'server_name': 'simulator', 'url': 'opc.tcp://sim:4840', 'uri': 'urn:sim'}, +} + +connectors = { + '3': { + 'connector_id': 3, + 'name': 'plant', + 'url': 'opc.tcp://plant:4840', + 'server_uri': 'urn:plant', + 'use_security': True, + 'subscription_period_ms': 1500, + } +} + +tags_by_id = { + '7': {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'}, + '8': {'tag_id': 8, 'connector_id': 5, 'address': 'ns=3;i=1002', 'tag_name': 'Level'}, +} + + +def _gather(pipelines, servers=None, connector_index=None, tag_index=None): + return gather_read_tags( + pipelines, + opc_servers if servers is None else servers, + connectors if connector_index is None else connector_index, + tags_by_id if tag_index is None else tag_index, + ) + + +def test_index_connector_tags(): + tags = [{'tag_id': 7, 'tag_name': 'a'}, {'tag_id': '8', 'tag_name': 'b'}] + + assert index_connector_tags(tags) == {'7': tags[0], '8': tags[1]} + + +def test_resolve_tag_names(): + read_tags = [ + {'tag_id': 7}, + {'tag_id': '7', 'tag_name': 'own name'}, + {'tag_id': 404}, + {'tag_name': 'legacy'}, + ] + + assert resolve_tag_names(read_tags, tags_by_id) == [ + {'tag_id': 7, 'tag_name': 'Pressure'}, + {'tag_id': '7', 'tag_name': 'own name'}, + {'tag_id': 404}, + {'tag_name': 'legacy'}, + ] + + +def test_gather_read_tags_multi_name(): + """One address read by two pipelines with different names: one entry, two names""" + pipelines = [ + _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'Counter', frequency=15000)]), + _pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'Contador', frequency=5000)]), + ] + + entries, issues = _gather(pipelines) + + assert len(entries) == 1 + 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()} + + +def test_gather_read_tags_same_name_in_two_pipelines(): + """The same name read by two pipelines still goes to both collections""" + pipelines = [ + _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'Counter')]), + _pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'Counter')]), + ] + + entries, _issues = _gather(pipelines) + + assert entries[0]['names'] == [_name('cos', 'Counter', 'm2'), _name('sin', 'Counter', 'm1')] + + +def test_gather_read_tags_duplicate_tag_in_one_pipeline(): + pipelines = [ + _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a'), _legacy_tag('ns=2;s=A', 'a')]), + ] + + entries, _issues = _gather(pipelines) + + assert entries[0]['names'] == [_name('sin', 'a', 'm1')] + + +def test_gather_read_tags_same_address_on_two_servers(): + pipelines = [ + _pipeline( + 'sin', + 'm1', + [_legacy_tag('ns=2;s=A', 'a', server_id='2'), _legacy_tag('ns=2;s=A', 'a')], + ), + ] + + entries, _issues = _gather(pipelines) + + assert [(e['server']['server_id'], e['address']) for e in entries] == [ + ('1', 'ns=2;s=A'), + ('2', 'ns=2;s=A'), + ] + + +def test_gather_read_tags_legacy_server(): + """Without tag_id: opc-servers fields, no certificate paths, security from cert_path""" + entries, _issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')])]) + + assert entries[0]['server'] == { + 'server_id': '1', + 'name': 'kepserver', + 'url': 'opc.tcp://kep:49320', + 'server_uri': 'urn:kep', + 'use_security': True, + 'subscription_period_ms': None, + 'from_tag_database': False, + 'key': 'kepserver', + } + + +def test_gather_read_tags_numeric_server_id(): + """A numeric server_id in the pipeline still finds its opc-servers document""" + entries, issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', server_id=1)])]) + + assert entries[0]['server']['server_id'] == '1' + assert not issues['servers'] + + +def test_gather_read_tags_with_tag_database(): + """With tag_id: address, name and server come from connector_tags + connectors""" + entries, _issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}])]) + + assert entries == [ + { + 'server': { + 'server_id': '3', + 'name': 'plant', + 'url': 'opc.tcp://plant:4840', + 'server_uri': 'urn:plant', + 'use_security': True, + 'subscription_period_ms': 1500, + 'from_tag_database': True, + 'key': 'plant', + }, + 'address': 'ns=3;i=1001', + 'frequency': 1000, + 'active_read': {'enabled': False}, + 'names': [_name('sin', 'Pressure', 'm1')], + } + ] + + +def test_gather_read_tags_pipeline_name_wins_over_tag_database(): + entries, _issues = _gather( + [_pipeline('sin', 'm1', [{'tag_id': 7, 'tag_name': 'P1', 'frequency': 1000}])] + ) + + assert entries[0]['names'] == [_name('sin', 'P1', 'm1')] + + +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()} + + +def test_gather_read_tags_inactive_connector(): + """Connector 5 is not among the active connectors: its tag is left out""" + entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 8, 'frequency': 1000}])]) + + assert entries == [] + assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set()} + + +def test_gather_read_tags_unknown_server(): + entries, issues = _gather( + [_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', server_id='9')])] + ) + + assert entries == [] + assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}} + + +def test_gather_read_tags_mixed_with_and_without_tag_id(): + """Both paths in one run; the same node via tag_id and via tag_address merges""" + connector_1 = { + 'connector_id': 1, + 'name': 'kepserver', + 'url': 'opc.tcp://kep:49320', + 'server_uri': 'urn:kep', + 'use_security': True, + 'subscription_period_ms': 2000, + } + tag_index = { + '20': {'tag_id': 20, 'connector_id': 1, 'address': 'ns=2;s=A', 'tag_name': 'A from db'}, + } + pipelines = [ + _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a'), _legacy_tag('ns=2;s=B', 'b')]), + _pipeline( + 'cos', 'm2', [{'tag_id': 20, 'frequency': 500}, {'tag_id': 7, 'frequency': 1000}] + ), + ] + + entries, issues = _gather( + pipelines, + connector_index={**connectors, '1': connector_1}, + tag_index={**tags_by_id, **tag_index}, + ) + + assert [(e['server']['server_id'], e['address']) for e in entries] == [ + ('1', 'ns=2;s=A'), + ('1', 'ns=2;s=B'), + ('3', 'ns=3;i=1001'), + ] + assert entries[0]['names'] == [_name('cos', 'A from db', 'm2'), _name('sin', 'a', 'm1')] + assert entries[0]['frequency'] == 500 + # 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()} + + +def test_gather_read_tags_server_name_clash(): + """Two servers with the same name get distinct slot keys, stable by server_id""" + servers = { + '1': opc_servers['1'], + '2': {**opc_servers['2'], 'server_name': 'kepserver'}, + } + pipelines = [ + _pipeline( + 'sin', + 'm1', + [_legacy_tag('ns=2;s=A', 'a', server_id='2'), _legacy_tag('ns=2;s=B', 'b')], + ), + ] + + entries, _issues = _gather(pipelines, servers=servers) + + assert [e['server']['key'] for e in entries] == ['kepserver', 'kepserver:2'] + assert [e['server']['name'] for e in entries] == ['kepserver', 'kepserver'] + + +def test_gather_read_tags_active_read(): + """Active read: on if any pipeline asks, with the lowest cycle_multiple (D3)""" + pipelines = [ + _pipeline('p1', 'm1', [_legacy_tag('ns=2;s=A', 'a')]), + _pipeline( + 'p2', + 'm2', + [_legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 4})], + ), + _pipeline( + 'p3', + 'm3', + [_legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 2})], + ), + _pipeline( + 'p4', + 'm4', + [_legacy_tag('ns=2;s=B', 'b', active_read={'enabled': False, 'cycle_multiple': 1})], + ), + _pipeline('p5', 'm5', [_legacy_tag('ns=2;s=C', 'c', active_read={'enabled': True})]), + ] + + entries, _issues = _gather(pipelines) + + assert [e['active_read'] for e in entries] == [ + {'enabled': True, 'cycle_multiple': 2}, + {'enabled': False}, + {'enabled': True, 'cycle_multiple': 1}, + ] def test_gather_read_tags_filters_non_scouter_workflows(): - """Test that gather_read_tags only processes scouter workflow types and ignores others""" + """Only scouter pipelines feed the slots""" 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', - } - ], - }, + _pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')]), + _pipeline('pred', 'm2', [_legacy_tag('ns=2;s=P', 'p')], workflow_type='predictions_batch'), + _pipeline('retrain', 'm3', [_legacy_tag('ns=2;s=R', 'r')], workflow_type='minimal_retrain'), + _pipeline('drift', 'm4', [_legacy_tag('ns=2;s=D', 'd')], workflow_type='drift'), ] - result = gather_read_tags(pipelines) + entries, _issues = _gather(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 [e['address'] for e in entries] == ['ns=2;s=A'] - 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_gather_read_tags_without_read_tags(): + pipeline = _pipeline('sin', 'm1', []) + del pipeline['read_tags'] + + assert _gather([pipeline]) == ([], {'tags': set(), 'connectors': set(), 'servers': set()}) def test_build_tag_config(): - tags = [ - { - 'server_id': '1', - 'server_name': 'test_server_name', - 'tag_address': 'test_tag_address', - 'frequency': 1000, - }, - { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address2', - 'frequency': 1000, - }, - { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address3', - 'frequency': 300, - }, - { - 'server_id': '3', - 'server_name': 'test_server_name3', - 'tag_address': 'test_tag_address4', - 'frequency': 200, - }, + """Slot v2 (contract C1): no certificate paths, subscription period per server""" + pipelines = [ + _pipeline( + 'sin', + 'm1', + [ + _legacy_tag('ns=2;s=A', 'a', frequency=1000), + _legacy_tag('ns=2;s=B', 'b', frequency='300'), + _legacy_tag('ns=2;s=C', 'c', frequency=200, server_id='2'), + {'tag_id': 7, 'frequency': 100}, + ], + ) ] + entries, _issues = _gather(pipelines) - opc_servers = { - '1': { - 'server_name': 'test_server_name', - 'url': 'test_url', - 'uri': 'test_uri', - }, - '2': { - 'server_name': 'test_server_name2', - 'url': 'test_url2', - 'uri': 'test_uri2', - }, - } + result = build_tag_config(entries) - result = build_tag_config(tags, opc_servers) - - expected = { - 'test_server_name': { + assert result == { + 'kepserver': { 'server_id': '1', - 'name': 'test_server_name', - 'url': 'test_url', - 'server_uri': 'test_uri', - 'cert_path': None, - 'private_key_path': None, - 'server_cert_path': None, - 'subscription_period_ms': 500, - 'tags': { - 'test_tag_address': { - 'server_id': '1', - 'server_name': 'test_server_name', - 'tag_address': 'test_tag_address', - 'frequency': 1000, - }, - }, - }, - 'test_server_name2': { - 'server_id': '2', - 'name': 'test_server_name2', - 'url': 'test_url2', - 'server_uri': 'test_uri2', - 'cert_path': None, - 'private_key_path': None, - 'server_cert_path': None, + 'name': 'kepserver', + 'url': 'opc.tcp://kep:49320', + 'server_uri': 'urn:kep', + 'use_security': True, 'subscription_period_ms': 150, 'tags': { - 'test_tag_address2': { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address2', + 'ns=2;s=A': { 'frequency': 1000, + 'active_read': {'enabled': False}, + 'names': [_name('sin', 'a', 'm1')], }, - 'test_tag_address3': { - 'server_id': '2', - 'server_name': 'test_server_name2', - 'tag_address': 'test_tag_address3', - 'frequency': 300, + 'ns=2;s=B': { + 'frequency': '300', + 'active_read': {'enabled': False}, + 'names': [_name('sin', 'b', 'm1')], + }, + }, + }, + 'simulator': { + 'server_id': '2', + 'name': 'simulator', + 'url': 'opc.tcp://sim:4840', + 'server_uri': 'urn:sim', + 'use_security': False, + 'subscription_period_ms': 100, + 'tags': { + 'ns=2;s=C': { + 'frequency': 200, + 'active_read': {'enabled': False}, + 'names': [_name('sin', 'c', 'm1')], + }, + }, + }, + 'plant': { + 'server_id': '3', + 'name': 'plant', + 'url': 'opc.tcp://plant:4840', + 'server_uri': 'urn:plant', + 'use_security': True, + 'subscription_period_ms': 1500, + 'tags': { + 'ns=3;i=1001': { + 'frequency': 100, + 'active_read': {'enabled': False}, + 'names': [_name('sin', 'Pressure', 'm1')], }, }, }, } - assert result == (expected, ['3']) + +def test_build_tag_config_empty(): + assert build_tag_config([]) == {} + + +def test_scouter_skips_tags_without_name(): + """A tag_id the tag database cannot resolve has no name and stays out of model_tags""" + config = { + 'workflow_type': 'scouter', + 'schedule_name': 'sin', + 'model_id': 'm1', + 'model': {'name': 'model m1'}, + 'read_tags': [{'tag_name': 'a'}, {'tag_id': 404}], + } + + assert list(scouter(config)['model_tags']) == ['a'] def test_base_scouter(): diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index 61243b0..c68e1d4 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -1,9 +1,14 @@ -from unittest.mock import ANY, AsyncMock, call, patch +from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from orchestrator.activities.activities import Activities -from orchestrator.workflows.orchestrator import Orchestrator +from orchestrator.workflows.orchestrator import ( + DEFAULT_CONNECTOR_TAGS_QUERY, + DEFAULT_CONNECTORS_QUERY, + TAG_DATABASE_PATCH, + Orchestrator, +) @fixture @@ -24,6 +29,7 @@ metadata = { @mark.asyncio @patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock) async def test_run(workflow_mock, orchestrator): + workflow_mock.patched = MagicMock(return_value=True) input_data = { 'pipelines_query': 'SELECT * FROM bucket', 'opc_servers_query': 'SELECT * FROM servers', @@ -109,11 +115,34 @@ async def test_run(workflow_mock, orchestrator): ] ) + workflow_mock.patched.assert_called_once_with(TAG_DATABASE_PATCH) + + workflow_mock.start_local_activity_method.assert_has_calls( + [ + call( + Activities.find_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}, + retry_policy=ANY, + start_to_close_timeout=ANY, + ), + ] + ) + workflow_mock.start_local_activity_method.assert_has_calls( [ call( Activities.process_schedules, - {**metadata, 'pipelines': workflow_mock.start_local_activity_method.return_value}, + { + **metadata, + 'pipelines': workflow_mock.start_local_activity_method.return_value, + 'connector_tags': workflow_mock.start_local_activity_method.return_value, + }, retry_policy=ANY, start_to_close_timeout=ANY, ) @@ -127,6 +156,8 @@ async def test_run(workflow_mock, orchestrator): { **metadata, 'opc_servers': workflow_mock.start_local_activity_method.return_value, + 'connectors': workflow_mock.start_local_activity_method.return_value, + 'connector_tags': workflow_mock.start_local_activity_method.return_value, 'active_ingestors': workflow_mock.start_local_activity_method.return_value, 'pipelines': workflow_mock.start_local_activity_method.return_value, }, @@ -346,3 +377,78 @@ async def test_run(workflow_mock, orchestrator): ) ] ) + + +@mark.asyncio +@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)""" + workflow_mock.patched = MagicMock(return_value=True) + connectors_query = {'collection': 'connectors', 'filters': {'type': 'opc_ua'}} + connector_tags_query = {'collection': 'connector_tags', 'filters': {'category': 'Read'}} + input_data = { + 'pipelines_query': 'SELECT * FROM bucket', + 'opc_servers_query': 'SELECT * FROM servers', + 'connectors_query': connectors_query, + 'connector_tags_query': connector_tags_query, + 'schedule_name': 'test-schedule-name', + } + + 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, + ), + ] + ) + + +@mark.asyncio +@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock) +async def test_run_replay_before_tag_database_patch(workflow_mock, orchestrator): + """A run started before the patch replays without the tag database loads""" + workflow_mock.patched = MagicMock(return_value=False) + input_data = { + 'pipelines_query': 'SELECT * FROM bucket', + 'opc_servers_query': 'SELECT * FROM servers', + 'schedule_name': 'test-schedule-name', + } + + 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, + ) + ] + )