SIENTIAPDE-1974

SIENTIAPDE-1974  Write OPC tags through the tag database and drop the PI Web API path

- predictions_batch write_tags are now [{tag_id}] only: resolve_write_tags builds
  the opc_output_config from connector_tags + connectors (server = connector_id,
  Write / Prediction Confidence -> prediction_tags / confidence_tags, address and
  data_type from the tag, float by default). The legacy server_id/type/addr write
  tags are no longer read and are notified as invalid.
- A pipeline's effective updated_at is the latest of its own and of the
  connector_tags it references, so renaming, moving or deactivating a tag
  redeploys it (removes the rename limitation).
- The connector_tags query loads the referenced tags of every category and
  status; the code filters them. aggregate_documents_in_mongodb formats only
  datetime timestamp fields.
- Remove the PI Web API tag database path (connectors-pi-web-api patch, PI loads,
  resolution, connection and held schedules): PI pipelines behave as on main.
- tag_id is canonical text and the workflow-side id collection never raises;
  a malformed pipeline no longer fails the whole OPC slot.
- Replay test against histories recorded on the e2e stack (main and this
  version), e2e scenarios for write tags and tag changes, and the runbook
  script scripts/sientiapde_1974_backfill_write_tags.js.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-29 08:25:01 -03:00
parent 9eaec54fc8
commit 2c331b5f16
28 changed files with 4451 additions and 1385 deletions

View File

@@ -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`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (OPC UA names, PI Web API WebIDs and connection); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list; see the README section "OPC slots"). `resolve_tag_names` (in `tag_database.py`) 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.
- **`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`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (scouter read tag names and the `predictions_batch` write tags → `opc_output_config` keyed by `connector_id`, via `resolve_write_tags`; a pipeline's effective `updated_at` also follows its tags' `updated_at`, so tag changes redeploy it); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list; see the README section "OPC slots"). `resolve_tag_names` (in `tag_database.py`) 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.

View File

@@ -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, active ingestors, and the tag database (`connectors` and `connector_tags`, OPC UA for the slot and PI Web API for the schedules)
1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, active ingestors, and the tag database (`connectors` and `connector_tags`, for the slot and the schedules)
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
@@ -136,14 +136,16 @@ Optional: `connectors_query` and `connector_tags_query` override the tag databas
"connector_tags_query": {
"collection": "connector_tags",
"aggregation": [
{"$match": {"category": "Read", "status": "Active"}},
{"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}}
{"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1,
"category": 1, "data_type": 1, "status": 1, "updated_at": 1}}
]
}
}
```
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`).
The tag database is read only when a scouter read tag or a `predictions_batch` write 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`). `updated_at` is turned into text by the load.
The referenced tags are loaded whatever their category and status: the code uses only the active ones (`Read` for the slot and the scouter names, `Write` / `Prediction Confidence` for the write tags), and every referenced tag's `updated_at` redeploys its pipelines (see "Redeploy on tag database changes" below). A document without `status` counts as active.
#### OPC slots (slot v2, contract C1)
@@ -189,56 +191,48 @@ Rules:
- `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`.
- `ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION`: the read tag has no integer `frequency`, has an enabled `active_read` without an integer `cycle_multiple` or, without `tag_id`, lacks `server_id`, `tag_address` or `tag_name`; or its pipeline lacks `schedule_name`, `model_id` or `model.name` (the value is then the `schedule_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`.
#### Write tags (predictions_batch, through the tag database)
#### PI Web API through the tag database
The tag database also holds PI Web API servers: `connectors` documents with `type: "pi_web_api"`, whose `url` is the PI Web API base URL and whose `pi_web_api` block carries `auth` (`type`: `basic` | `bearer`, `secret`: `{name, key}` of a Kubernetes secret, never the credential) and optional connection defaults (`period`, `max_count`, `api_timeout`, `data_server_webid`). Their tags are ordinary `connector_tags` whose `address` is the PI WebID. The slot never reads them: they only feed `process_schedules`.
The workflow loads them behind the `connectors-pi-web-api` patch, with two more optional overrides (defaults in `orchestrator/workflows/orchestrator.py`):
A `predictions_batch` (or `xgboost_predictions_batch`) references its write tags by `tag_id` only, exactly like the read tags. Everything else comes from the tag database, so the pipeline does not say which server, address or type a tag has:
```json
{
"pi_connectors_query": {
"collection": "connectors",
"aggregation": [
{"$match": {"type": "pi_web_api", "status": "Active"}},
{"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "pi_web_api": 1}}
]
},
"pi_connector_tags_query": {
"collection": "connector_tags",
"aggregation": [
{"$match": {"status": "Active"}},
{"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1, "category": 1}}
]
}
}
{"workflow_type": "predictions_batch", "write_tags": [{"tag_id": 101}, {"tag_id": 102}]}
```
As for OPC, they are read only when a PI Web API tag carries a `tag_id`, and the workflow puts the referenced `tag_id`s in front of `pi_connector_tags_query`. Tags of every category are loaded, since `predictions_batch` writes to `Write` and `Prediction Confidence` tags.
`process_schedules` turns them into the `opc_output_config` the Laborious already reads:
How the PI Web API pipelines are resolved:
```json
{"plant": {
"prediction_tags": {"ns=3;i=2001": {"data_type": "double"}},
"confidence_tags": {"ns=3;i=2002": {"data_type": "float"}}}}
```
| Pipeline field | With `tag_id` | Without `tag_id` |
|---|---|---|
| `pi_web_api_scouter` `read_tags` (`{model_tag: {...}}`) | `webid` = the `Read` tag's `address`; the key, `aggr_func` and `data_range` stay the pipeline's | `webid` from the pipeline, as before |
| `predictions_batch` `pi_web_api_output_config.prediction_tags` / `confidence_tags` given as a list | `{tag_name: address}` of the `Write` / `Prediction Confidence` tag | A mapping already given as `{tag_name: webid}` passes through unchanged |
| Field | Source |
|---|---|
| Server key | `connector_tags.connector_id`. The Laborious publishes it as `server_id` on NATS, and the OPC Gateway resolves it in `connectors` |
| `prediction_tags` / `confidence_tags` | `connector_tags.category`: `Write` / `Prediction Confidence` |
| Address | `connector_tags.address` |
| `data_type` | `connector_tags.data_type` (`float`, `double`, `real`, `int`, `integer`, `bool`, `boolean`, `str`, `string`), else `float` |
Rules:
- **Connection.** When a `pi_web_api_scouter` resolves at least one tag through the tag database, its schedule input gets `connection: {connector_id, base_url, auth: {type, secret: {name, key}}}` (plus `data_server_webid` when set), and the connector's `period`, `max_count` and `api_timeout` become defaults under the pipeline's `pi_web_api_config`: pipeline values win, and `api_timeout` is still capped at the frequency. `connection` is additive: the scouter keeps reading `pi_web_api_query` until its own card. Only the secret reference is copied; the orchestrator never resolves the secret.
- **Legacy pipelines are unchanged.** Without any `tag_id`, the schedule input is byte-identical to before.
- **One connector per pipeline.** A `pi_web_api_scouter` whose tags belong to two connectors is held: its existing schedule is neither updated nor deleted (and none is created), and the conflict is notified.
- **Unresolved tags** are left out, the rest of the pipeline is still scheduled, and they are notified once per run (values are `schedule_name/tag_id`):
- `ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`: the `tag_id` is not an active tag, or its document lacks `connector_id`, `address` or `tag_name`;
- `ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION`: the tag's category does not fit (`Read` for read tags, `Write` for `prediction_tags`, `Prediction Confidence` for `confidence_tags`);
- `ORCHESTRATOR_PI_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`: the tag's connector is not an active `pi_web_api` connector with `url` and `pi_web_api.auth` (`type`, `secret.name`, `secret.key`);
- `ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION`: the held pipelines above.
The legacy write tags (`server_id`, `type`, `addr`, `data_type` in the pipeline) are no longer read: migrate them to `tag_id` before deploying this version (see "Deploying the tag database write tags" below). A write tag that cannot be resolved is left out, the rest of the pipeline is still scheduled, and it is notified once per run (values are `schedule_name/tag_id`):
- `ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`: the `tag_id` is not an active `Write` / `Prediction Confidence` tag, or its document lacks `connector_id`, `address`, `tag_name` or `category`;
- `ORCHESTRATOR_WRITE_TAG_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`: the tag's connector is missing, inactive or incomplete in `connectors`;
- `ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION`: the entry has no `tag_id` (a legacy write tag), or the tag's `data_type` is not one the gateway accepts.
#### Redeploy on tag database changes
A schedule is redeployed when its `updated_at` is newer than the last time it was orchestrated. That `updated_at` is the latest of the pipeline's own and of the `updated_at` of every `connector_tags` document it references (read and write tags, active or not). So renaming a read tag, moving or retyping a write tag, or deactivating a tag redeploys the pipelines that use it on the next run, without touching them. The slot is compared by content and follows on its own. A pipeline without `updated_at` is still redeployed on every run. A deleted tag document (as opposed to a deactivated one) is not detected: deactivate tags instead of deleting them.
#### Deploying the tag database write tags
1. Run the migration that adds `data_type` to the `connector_tags` validator (`sientia-dataops-database-migrations`).
2. Run `scripts/sientiapde_1974_backfill_write_tags.js`, first with `DRY_RUN=1` (the default). It creates the `connector_tags` of the legacy write tags, rewrites `write_tags` as `[{tag_id}]` (keeping the original in `legacy_write_tags`) and bumps the pipelines' `updated_at`.
3. Deploy this version.
#### Architecture
@@ -739,14 +733,14 @@ Temporal input configuration sample:
"connector_tags_query": {
"collection": "connector_tags",
"aggregation": [
{"$match": {"category": "Read", "status": "Active"}},
{"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1}}
{"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1,
"category": 1, "data_type": 1, "status": 1, "updated_at": 1}}
]
}
}
```
`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code (see "OPC slots" above).
`connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code (see "OPC slots" and "Write tags" above).
#### Simple metrics: parallel windows & adaptive frequency
@@ -960,7 +954,7 @@ orchestrator/
│ ├── connectors_config.py # Database configuration
│ ├── email_builder.py # Email template generation
│ ├── pipeline_configs.py # Schedule input builders per pipeline type
│ ├── tag_database.py # Tag database indexes, OPC names and PI Web API resolution
│ ├── tag_database.py # Tag database indexes, OPC read names and write tags
│ ├── opc_slots.py # OPC slot v2 (contract C1)
│ ├── converters.py # Data type conversion utilities
│ └── templates/ # Email HTML templates

View File

@@ -22,11 +22,17 @@
"frequency": "1m",
"offset": "0m",
"query": "SELECT 1",
"write_tags": [
{"server_id": "srv-1", "type": "prediction", "addr": "ns=2;s=Tag1"}
]
"write_tags": [{"tag_id": 101}]
}
],
"connectors": [
{"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840",
"server_uri": "urn:opc-1", "use_security": false, "status": "Active"}
],
"connector_tags": [
{"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1",
"category": "Write", "status": "Active", "updated_at": "@now-1d"}
],
"opc_servers": [
{"id": "srv-1", "active": true}
],

View File

@@ -22,11 +22,17 @@
"frequency": "1m",
"offset": "0m",
"query": "SELECT 1",
"write_tags": [
{"server_id": "srv-1", "type": "prediction", "addr": "ns=2;s=Tag1"}
]
"write_tags": [{"tag_id": 101}]
}
],
"connectors": [
{"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840",
"server_uri": "urn:opc-1", "use_security": false, "status": "Active"}
],
"connector_tags": [
{"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1",
"category": "Write", "status": "Active", "updated_at": "@now-1d"}
],
"opc_servers": [
{"id": "srv-1", "active": true}
],

View File

@@ -22,9 +22,7 @@
"frequency": "1m",
"offset": "0m",
"query": "SELECT 1",
"write_tags": [
{"server_id": "srv-1", "type": "prediction", "addr": "ns=2;s=Tag1"}
]
"write_tags": [{"tag_id": 101}]
},
{
"schedule_name": "drift-gpu",
@@ -39,6 +37,14 @@
"interval_minutes": 60
}
],
"connectors": [
{"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840",
"server_uri": "urn:opc-1", "use_security": false, "status": "Active"}
],
"connector_tags": [
{"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1",
"category": "Write", "status": "Active", "updated_at": "@now-1d"}
],
"opc_servers": [
{"id": "srv-1", "active": true, "name": "opc-1"}
],

View File

@@ -22,11 +22,17 @@
"frequency": "1m",
"offset": "0m",
"query": "SELECT 1",
"write_tags": [
{"server_id": "srv-1", "type": "prediction", "addr": "ns=2;s=Tag1"}
]
"write_tags": [{"tag_id": 101}]
}
],
"connectors": [
{"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840",
"server_uri": "urn:opc-1", "use_security": false, "status": "Active"}
],
"connector_tags": [
{"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1",
"category": "Write", "status": "Active", "updated_at": "@now-1d"}
],
"opc_servers": [
{"id": "srv-1", "active": true}
],

View File

@@ -1,118 +0,0 @@
{
"workflow_input": {
"schedule_name": "orchestrator-e2e-pi-web-api",
"pipelines_query": {
"collection": "pipelines",
"aggregation": [{"$match": {"active": true}}]
},
"opc_servers_query": {
"collection": "opc_servers",
"filters": {"active": true}
}
},
"connectors": [
{
"connector_id": "pi-usina",
"type": "pi_web_api",
"name": "PI Usina",
"url": "https://pi-usina.example/piwebapi",
"status": "Active",
"pi_web_api": {
"auth": {"type": "bearer", "secret": {"name": "pi-web-api-auth-token", "key": "token"}},
"period": "*-2h",
"max_count": 10
}
},
{
"connector_id": "pi-other",
"type": "pi_web_api",
"name": "PI Other",
"url": "https://pi-other.example/piwebapi",
"status": "Active",
"pi_web_api": {
"auth": {"type": "basic", "secret": {"name": "pi-other-auth", "key": "credentials"}}
}
},
{
"connector_id": "plant",
"type": "opc_ua",
"name": "plant",
"url": "opc.tcp://plant:4840",
"server_uri": "urn:plant",
"use_security": false,
"subscription_period_ms": 500,
"status": "Active"
}
],
"connector_tags": [
{"tag_id": 42, "connector_id": "pi-usina", "address": "F1DP-A", "tag_name": "TI-101-db", "category": "Read", "status": "Active"},
{"tag_id": 43, "connector_id": "pi-other", "address": "F1DP-B", "tag_name": "TI-102-db", "category": "Read", "status": "Active"},
{"tag_id": 77, "connector_id": "pi-usina", "address": "F1DP-P", "tag_name": "PRED-101", "category": "Write", "status": "Active"},
{"tag_id": 78, "connector_id": "pi-usina", "address": "F1DP-C", "tag_name": "CONF-101", "category": "Prediction Confidence", "status": "Active"},
{"tag_id": 7, "connector_id": "plant", "address": "ns=3;i=1001", "tag_name": "Pressure", "category": "Read", "status": "Active"}
],
"pipelines": [
{
"schedule_name": "pi-tagdb",
"workflow_type": "pi_web_api_scouter",
"model_id": "model-pi",
"model": {"name": "Model model-pi"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"read_tags": {
"TI-101": {"tag_id": 42, "aggr_func": "avg"},
"TI-404": {"tag_id": 404}
},
"pi_web_api_config": {"endpoint": "/streamsets/recorded", "max_count": 5}
},
{
"schedule_name": "pi-legacy",
"workflow_type": "pi_web_api_scouter",
"model_id": "model-pi-legacy",
"model": {"name": "Model model-pi-legacy"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"read_tags": {"TI-900": {"webid": "F1DP-X"}},
"pi_web_api_config": {"endpoint": "/streamsets/recorded", "api_timeout": 10}
},
{
"schedule_name": "pred-pi",
"workflow_type": "predictions_batch",
"model_id": "model-pi",
"model": {"name": "Model model-pi"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"query": "SELECT 1",
"pi_web_api_output_config": {
"prediction_tags": [{"tag_id": 77}],
"confidence_tags": [{"tag_id": 78}, {"tag_id": 42}]
}
},
{
"schedule_name": "pred-pi-legacy",
"workflow_type": "predictions_batch",
"model_id": "model-pi-legacy",
"model": {"name": "Model model-pi-legacy"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"query": "SELECT 1",
"pi_web_api_output_config": {"prediction_tags": {"PRED-900": "F1DP-L"}}
},
{
"schedule_name": "opc-tagdb",
"workflow_type": "scouter",
"model_id": "model-opc",
"model": {"name": "Model model-opc"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"read_tags": [{"tag_id": 7, "frequency": 1000, "aggr_func": "avg"}]
}
],
"opc_servers": [],
"active_ingestors": ["heartbeat:ingestor:1"]
}

View File

@@ -0,0 +1,64 @@
{
"workflow_input": {
"schedule_name": "orchestrator-e2e-tag-database",
"pipelines_query": {
"collection": "pipelines",
"aggregation": [{"$match": {"active": true}}]
},
"opc_servers_query": {
"collection": "opc_servers",
"filters": {"active": true}
}
},
"connectors": [
{
"connector_id": "plant",
"type": "opc_ua",
"name": "plant",
"url": "opc.tcp://plant:4840",
"server_uri": "urn:plant",
"use_security": false,
"subscription_period_ms": 500,
"status": "Active",
"updated_at": "@now-1d"
}
],
"connector_tags": [
{"tag_id": 7, "connector_id": "plant", "address": "ns=3;i=1001", "tag_name": "Pressure", "category": "Read", "status": "Active", "updated_at": "@now-1d"},
{"tag_id": 101, "connector_id": "plant", "address": "ns=3;i=2001", "tag_name": "PRED-101", "category": "Write", "data_type": "double", "status": "Active", "updated_at": "@now-1d"},
{"tag_id": 102, "connector_id": "plant", "address": "ns=3;i=2002", "tag_name": "CONF-101", "category": "Prediction Confidence", "status": "Active", "updated_at": "@now-1d"},
{"tag_id": 103, "connector_id": "plant", "address": "ns=3;i=2003", "tag_name": "PRED-OLD", "category": "Write", "status": "Inactive", "updated_at": "@now-1d"}
],
"pipelines": [
{
"schedule_name": "opc-tagdb",
"workflow_type": "scouter",
"model_id": "model-opc",
"model": {"name": "Model model-opc"},
"active": true,
"updated_at": "@now-2d",
"frequency": "1m",
"read_tags": [{"tag_id": 7, "frequency": 1000, "aggr_func": "avg"}]
},
{
"schedule_name": "pred-tagdb",
"workflow_type": "predictions_batch",
"model_id": "model-opc",
"model": {"name": "Model model-opc"},
"active": true,
"updated_at": "@now-2d",
"frequency": "1m",
"query": "SELECT 1",
"write_tags": [
{"tag_id": 101},
{"tag_id": 102},
{"tag_id": 7},
{"tag_id": 103},
{"tag_id": 404},
{"server_id": "plant", "type": "prediction", "addr": "ns=3;i=9999"}
]
}
],
"opc_servers": [],
"active_ingestors": ["heartbeat:ingestor:1"]
}

View File

@@ -22,11 +22,17 @@
"frequency": "5m",
"offset": "0m",
"query": "SELECT 1",
"write_tags": [
{"server_id": "srv-1", "type": "prediction", "addr": "ns=2;s=Tag1"}
]
"write_tags": [{"tag_id": 101}]
}
],
"connectors": [
{"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840",
"server_uri": "urn:opc-1", "use_security": false, "status": "Active"}
],
"connector_tags": [
{"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1",
"category": "Write", "status": "Active", "updated_at": "@now-1d"}
],
"opc_servers": [
{"id": "srv-1", "active": true}
],

View File

@@ -27,8 +27,8 @@ Source: `e2e/test_orchestrator_main_workflow.py`
### 1.1.1 Happy path
Pipelines in Mongo → both Temporal schedules created in the `laborious` namespace, one
`orchestrated_schedules` entry per pipeline, and no PI Web API output config when no pipeline uses
the tag database.
`orchestrated_schedules` entry per pipeline, and the `predictions_batch` write tag (`tag_id`)
resolved into `opc_output_config` through the tag database.
### 1.3.1 Create-only
@@ -66,37 +66,30 @@ Two `drift` pipelines on `frequency='1m'`: one without `interval_minutes` and wi
`interval_minutes=60` and `chunk_period='s'` → both forwarded unchanged, no warning
(SIENTIAPDE-2212).
### 1.9 PI Web API through the tag database (connectors-pi-web-api)
### 1.9 OPC tag database: read and write tags by tag_id
Scenario input: `orchestrator_pi_web_api.json` (two `pi_web_api` connectors, one `opc_ua`
connector, their `connector_tags`, PI Web API pipelines by `tag_id` and legacy ones, and an OPC
scouter by `tag_id`).
Scenario input: `orchestrator_tag_database.json` (one `opc_ua` connector, its `connector_tags` of
every category, one inactive, an OPC scouter reading by `tag_id` and a `predictions_batch` writing by
`tag_id`, with a Read tag, an inactive tag, an unknown `tag_id` and a legacy entry among its write
tags).
#### 1.9.1 PI scouter through the tag database
#### 1.9.1 Write tags through the tag database
`pi_web_api_scouter` with `tag_id` read tags → `model_tags` WebID from `connector_tags.address`,
`pi_web_api_query` with the connector `period` under the pipeline `max_count` (`api_timeout` capped
at the frequency), and `connection` with base URL and the secret reference only. An unknown
`tag_id` is left out and notified (`ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`).
A legacy pipeline (inline `webid`) gets the pre-change input, without `connection`.
`write_tags: [{tag_id}]` → `opc_output_config` keyed by `connector_id`, address and `data_type`
from `connector_tags` (`float` when absent), `Write` → `prediction_tags`, `Prediction Confidence` →
`confidence_tags`. The Read, inactive and unknown tags are left out and notified once
(`ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`); the legacy entry is notified
(`ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION`).
#### 1.9.2 PI output tags through the tag database
#### 1.9.2 A tag change redeploys its pipelines
`predictions_batch` `pi_web_api_output_config` given as `tag_id` lists → `{tag_name: WebID}` from
`Write` / `Prediction Confidence` tags; a `Read` tag given as a confidence tag is left out and
notified (`ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION`). A legacy
`{tag_name: webid}` mapping passes through unchanged.
Renaming the read tag and moving the write tag in `connector_tags` (with a newer `updated_at`),
without touching the pipelines → the next run redeploys both: the scouter `model_tags` and the
slot `names` take the new name, the `opc_output_config` the new address.
#### 1.9.3 Mixed PI connectors hold the schedule
#### 1.9.3 OPC slot from the tag database
A pipeline updated to read tags of two PI connectors → its existing scouter schedule is neither
updated nor deleted, and the conflict is notified
(`ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION`).
#### 1.9.4 OPC slot unchanged by PI connectors
The OPC slot v2 written to Redis for an OPC scouter by `tag_id` is the expected one with the PI
connectors and tags present, and identical after removing them.
The slot v2 written to Redis holds only the active Read tag, with the connector's server fields.
---

View File

@@ -100,9 +100,11 @@ async def test_scenario_1_1_1_happy_path(
assert {'pred-legacy', 'drift-gpu'} <= set(laborious_schedules), (
f'Expected Temporal schedules in laborious, got {laborious_schedules}'
)
# No pipeline uses the tag database: the PI Web API loads stay empty
# The write tag comes from the tag database, keyed by the connector the gateway writes to
pred = await _schedule_input(temporal_host, 'pred-legacy')
assert pred['pi_web_api_output_config'] == {}
assert pred['opc_output_config'] == {
'srv-1': {'prediction_tags': {'ns=2;s=Tag1': {'data_type': 'float'}}}
}
@pytest.mark.e2e
@@ -377,17 +379,15 @@ async def test_scenario_1_8_1_drift_window_and_chunk_period(
assert warnings[0]['model_id'] == 'model-dr'
# 1.9 PI Web API through the tag database (connectors-pi-web-api)
PI_SECRET = {'name': 'pi-web-api-auth-token', 'key': 'token'}
# 1.9 OPC tag database: read and write tags by tag_id (SIENTIAPDE-1974)
async def _run_pi_scenario(temporal_env: WorkflowEnvironment, scenario: dict, tag: str) -> None:
async def _run_tag_database_scenario(temporal_env: WorkflowEnvironment, tag: str) -> None:
await start_and_await_workflow(
temporal_env.client,
Orchestrator.run,
scenario['workflow_input'],
make_workflow_id(f'orchestrator-pi-{tag}'),
load_scenario_input('orchestrator_tag_database')['workflow_input'],
make_workflow_id(f'orchestrator-tagdb-{tag}'),
)
@@ -395,132 +395,6 @@ def _notifications(notification_handler, notification_id: str) -> list[dict]:
return list(notification_handler.mongo_collection.find({'notification_id': notification_id}))
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_1_pi_scouter_through_tag_database(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.1: tag_id read tags get WebID + connection from the connector; legacy unchanged."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'scouter')
resolved = await _schedule_input(temporal_host, 'pi-tagdb', namespace='scouter')
# The unknown tag_id is left out; the rest of the pipeline is still scheduled
assert resolved['model_tags'] == {
'TI-101': {'webid': 'F1DP-A', 'aggr_func': 'avg', 'data_range': [-100, 100]}
}
assert resolved['pi_web_api_query'] == {
'endpoint': '/streamsets/recorded',
'period': '*-2h', # connector default
'max_count': 5, # pipeline wins over the connector
'api_timeout': 60, # capped at the frequency
}
assert resolved['connection'] == {
'connector_id': 'pi-usina',
'base_url': 'https://pi-usina.example/piwebapi',
'auth': {'type': 'bearer', 'secret': PI_SECRET},
}
legacy = await _schedule_input(temporal_host, 'pi-legacy', namespace='scouter')
assert 'connection' not in legacy
assert legacy['model_tags'] == {
'TI-900': {'webid': 'F1DP-X', 'aggr_func': 'lts', 'data_range': [-100, 100]}
}
assert legacy['pi_web_api_query'] == {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 1,
'api_timeout': 10,
}
[not_found] = _notifications(
notification_handler, 'ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pi-tagdb/404' in not_found['message']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_2_pi_output_tags_through_tag_database(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.2: predictions_batch output tags by tag_id become {tag_name: WebID}; legacy unchanged."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'output')
resolved = await _schedule_input(temporal_host, 'pred-pi')
# The Read tag given as a confidence tag is left out
assert resolved['pi_web_api_output_config'] == {
'prediction_tags': {'PRED-101': 'F1DP-P'},
'confidence_tags': {'CONF-101': 'F1DP-C'},
}
legacy = await _schedule_input(temporal_host, 'pred-pi-legacy')
assert legacy['pi_web_api_output_config'] == {'prediction_tags': {'PRED-900': 'F1DP-L'}}
[wrong_category] = _notifications(
notification_handler, 'ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pred-pi/42' in wrong_category['message']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_3_pi_mixed_connectors_hold_schedule(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.3: tags of two PI connectors in one pipeline leave its schedule untouched."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'hold-create')
before = await _schedule_input(temporal_host, 'pi-tagdb', namespace='scouter')
mongo = MongoClient(mongo_uri)
try:
mongo[E2E_DATABASE]['pipelines'].update_one(
{'schedule_name': 'pi-tagdb'},
{
'$set': {
'read_tags.TI-102': {'tag_id': 43},
'updated_at': datetime.now(UTC) + timedelta(minutes=1),
}
},
)
finally:
mongo.close()
await _run_pi_scenario(temporal_env, scenario, 'hold-mixed')
assert 'pi-tagdb' in await _schedule_ids(temporal_host, 'scouter')
assert await _schedule_input(temporal_host, 'pi-tagdb', namespace='scouter') == before
[mixed] = _notifications(
notification_handler, 'ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pi-tagdb/pi-other, pi-usina' in mixed['message']
def _opc_slots(redis_client: Redis) -> dict[str, dict]:
return {
key if isinstance(key, str) else key.decode(): json.loads(redis_client.get(key))
@@ -530,21 +404,106 @@ def _opc_slots(redis_client: Redis) -> dict[str, dict]:
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_4_opc_slot_unchanged_by_pi_connectors(
async def test_scenario_1_9_1_write_tags_through_tag_database(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.1: write_tags by tag_id become the opc_output_config; the rest is notified."""
_apply_orchestrator_seeds(
load_scenario_input('orchestrator_tag_database'), mongo_uri, redis_client
)
await _run_tag_database_scenario(temporal_env, 'write')
pred = await _schedule_input(temporal_host, 'pred-tagdb')
assert pred['opc_output_config'] == {
'plant': {
'prediction_tags': {'ns=3;i=2001': {'data_type': 'double'}},
'confidence_tags': {'ns=3;i=2002': {'data_type': 'float'}},
}
}
not_found = _notifications(
notification_handler, 'ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION'
)
assert len(not_found) == 1
# A Read tag, an inactive tag and an unknown tag_id
for value in ('pred-tagdb/7', 'pred-tagdb/103', 'pred-tagdb/404'):
assert value in not_found[0]['message']
invalid = _notifications(
notification_handler, 'ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION'
)
assert len(invalid) == 1
assert 'ns=3;i=9999' in invalid[0]['message']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_2_tag_change_redeploys_pipelines(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
):
"""1.9.2: changing a tag in connector_tags redeploys the pipelines that reference it."""
_apply_orchestrator_seeds(
load_scenario_input('orchestrator_tag_database'), mongo_uri, redis_client
)
await _run_tag_database_scenario(temporal_env, 'before-change')
scouter = await _schedule_input(temporal_host, 'opc-tagdb', namespace='scouter')
assert set(scouter['model_tags']) == {'Pressure'}
# Rename the read tag and move the write tag, without touching the pipelines
changed_at = datetime.now(UTC) + timedelta(seconds=1)
mongo = MongoClient(mongo_uri)
try:
tags = mongo[E2E_DATABASE]['connector_tags']
tags.update_one({'tag_id': 7}, {'$set': {'tag_name': 'Pressao', 'updated_at': changed_at}})
tags.update_one(
{'tag_id': 101}, {'$set': {'address': 'ns=3;i=3001', 'updated_at': changed_at}}
)
finally:
mongo.close()
await _run_tag_database_scenario(temporal_env, 'after-change')
scouter = await _schedule_input(temporal_host, 'opc-tagdb', namespace='scouter')
assert set(scouter['model_tags']) == {'Pressao'}
pred = await _schedule_input(temporal_host, 'pred-tagdb')
assert pred['opc_output_config']['plant']['prediction_tags'] == {
'ns=3;i=3001': {'data_type': 'double'}
}
# The slot names the read tag like the scouter does (SC1)
[slot] = _opc_slots(redis_client).values()
[name] = slot['plant']['tags']['ns=3;i=1001']['names']
assert name['tag_name'] == 'Pressao'
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_3_opc_slot_from_tag_database(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
redis_client: Redis,
):
"""1.9.4: the OPC slot built from the tag database ignores the PI connectors and tags."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
"""1.9.3: the slot v2 holds only the active Read tags of the tag database."""
_apply_orchestrator_seeds(
load_scenario_input('orchestrator_tag_database'), mongo_uri, redis_client
)
await _run_pi_scenario(temporal_env, scenario, 'slot-with-pi')
with_pi = _opc_slots(redis_client)
await _run_tag_database_scenario(temporal_env, 'slot')
assert with_pi == {
assert _opc_slots(redis_client) == {
'slot:opc_tags:1': {
'plant': {
'server_id': 'plant',
@@ -570,15 +529,3 @@ async def test_scenario_1_9_4_opc_slot_unchanged_by_pi_connectors(
}
}
}
mongo = MongoClient(mongo_uri)
try:
mongo[E2E_DATABASE]['connectors'].delete_many({'type': 'pi_web_api'})
mongo[E2E_DATABASE]['connector_tags'].delete_many({'connector_id': {'$ne': 'plant'}})
finally:
mongo.close()
redis_client.delete(*redis_client.keys('slot:opc_tags:*'))
await _run_pi_scenario(temporal_env, scenario, 'slot-without-pi')
assert _opc_slots(redis_client) == with_pi

View File

@@ -29,13 +29,15 @@ with workflow.unsafe.imports_passed_through():
simple_metrics,
)
from orchestrator.utils.tag_database import (
PI_OUTPUT_WORKFLOW_TYPES,
WRITE_CATEGORIES,
WRITE_WORKFLOW_TYPES,
index_connector_tags,
index_connectors,
index_pi_connectors,
resolve_pi_output_tags,
resolve_pi_read_tags,
index_tag_documents,
latest_updated_at,
resolve_tag_names,
resolve_write_tags,
tag_documents_updated_at,
)
topic_separator = '\n ========== \n'
@@ -60,34 +62,30 @@ SLOT_ISSUES = (
(
'invalid',
'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION',
'Read tags {} have no integer frequency or lack server_id, tag_address or tag_name',
'Read tags {} have no integer frequency, an active_read without an integer '
'cycle_multiple, or lack server_id, tag_address or tag_name; or their pipeline lacks '
'schedule_name, model_id or model.name',
),
)
# PI Web API tags left out of the schedules, by resolve_pi_read_tags / resolve_pi_output_tags
# Write tags left out of the predictions_batch opc_output_config, by resolve_write_tags
# issue kind; values are schedule_name/<tag_id or connector_id>
PI_SCHEDULE_ISSUES = (
WRITE_TAG_ISSUES = (
(
'tags',
'ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'PI Web API tags {} not found among the active, complete tags of connector_tags',
),
(
'categories',
'ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION',
'PI Web API tags {} have a category the pipeline cannot use (Read for read_tags, '
'Write for prediction_tags, Prediction Confidence for confidence_tags)',
'ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'Write tags {} not found among the active, complete Write and Prediction Confidence '
'tags of connector_tags',
),
(
'connectors',
'ORCHESTRATOR_PI_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'PI Web API connectors {} not found, not active or incomplete in connectors',
'ORCHESTRATOR_WRITE_TAG_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'Connectors {} of write tags not found, not active or incomplete in connectors',
),
(
'mixed',
'ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION',
'Pipelines {} read tags of several PI Web API connectors; their schedules were left '
'untouched',
'invalid',
'ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION',
'Write tags {} have no tag_id or a data_type the OPC Gateway does not accept',
),
)
@@ -100,46 +98,6 @@ def _merge_issues(
issues.setdefault(kind, set()).update(f'{schedule_name}/{value}' for value in values)
def _resolve_pi_pipeline(
pipeline: dict[str, Any],
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
issues: dict[str, set[str]],
) -> dict[str, Any] | None:
"""
The pipeline with its PI Web API tags resolved through the tag database.
pi_web_api_scouter read tags get their WebIDs (and the pipeline its connector);
predictions_batch output tags get {tag_name: WebID}. Other pipelines come back as
they are. What cannot be resolved is added to issues. None means the pipeline must
be held: its read tags span several connectors.
"""
workflow_type = pipeline['workflow_type']
schedule_name = pipeline['schedule_name']
if workflow_type == 'pi_web_api_scouter' and isinstance(pipeline.get('read_tags'), dict):
read_tags, connector, read_issues = resolve_pi_read_tags(
pipeline['read_tags'], pi_tags, pi_connectors
)
_merge_issues(issues, read_issues, schedule_name)
if read_issues['mixed']:
return None
pipeline = {**pipeline, 'read_tags': read_tags}
if connector is not None:
pipeline['pi_web_api_connector'] = connector
if workflow_type in PI_OUTPUT_WORKFLOW_TYPES and isinstance(
pipeline.get('pi_web_api_output_config'), dict
):
output_config, output_issues = resolve_pi_output_tags(
pipeline['pi_web_api_output_config'], pi_tags, pi_connectors
)
_merge_issues(issues, output_issues, schedule_name)
pipeline = {**pipeline, 'pi_web_api_output_config': output_config}
return pipeline
class ScheduleType(TypedDict):
"""
Type definition for schedule type configuration entries.
@@ -282,18 +240,16 @@ 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.
- pi_connectors (list[dict[str, Any]], optional): Active pi_web_api
connectors of the tag database.
- pi_connector_tags (list[dict[str, Any]], optional): Active tags of the
tag database referenced by the PI Web API pipelines (all categories).
- connectors (list[dict[str, Any]], optional): OPC UA connectors of the
tag database, for the write tags.
- connector_tags (list[dict[str, Any]], optional): The connector_tags the
pipelines reference, of every category and status.
PI Web API tags given by tag_id (pi_web_api_scouter read tags, predictions_batch
pi_web_api_output_config) are resolved through the tag database; what cannot be
resolved is left out and notified once per run. A pi_web_api_scouter whose tags
span several connectors is held: it gets {'held': True} instead of a config, so
create_schedule_config neither creates, updates nor deletes its schedule.
Scouter read tags that carry only a tag_id are named from the tag database (the
same names as the slot). A predictions_batch gets its opc_output_config from its
write_tags ([{tag_id}]) through the tag database; what cannot be resolved is left
out and notified once per run. A pipeline's updated_at is the latest of its own and
of the tag database documents it references, so changing a tag redeploys it.
Returns:
- dict[str, Any]: The schedule configuration dictionary keyed by namespace
@@ -304,10 +260,12 @@ class Formatters(SientiaMonitoring):
self.info('Processing schedules...', metadata=metadata)
pipelines = input_data['pipelines']
tags_by_id = index_connector_tags(input_data.get('connector_tags', []))
pi_connectors = index_pi_connectors(input_data.get('pi_connectors', []))
pi_tags = index_connector_tags(input_data.get('pi_connector_tags', []))
pi_issues: dict[str, set[str]] = {}
connector_tags = input_data.get('connector_tags', [])
read_tags_by_id = index_connector_tags(connector_tags)
write_tags_by_id = index_connector_tags(connector_tags, WRITE_CATEGORIES)
tag_documents = index_tag_documents(connector_tags)
connectors = index_connectors(input_data.get('connectors', []))
write_issues: dict[str, set[str]] = {}
schedule_config: dict[str, dict[str, Any]] = {
self.scouter_namespace: {},
@@ -330,23 +288,26 @@ class Formatters(SientiaMonitoring):
# 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),
'read_tags': resolve_tag_names(pipeline['read_tags'], read_tags_by_id),
}
resolved = _resolve_pi_pipeline(pipeline, pi_tags, pi_connectors, pi_issues)
if resolved is None:
schedule_config[namespace][schedule_name] = {'held': True}
continue
pipeline = resolved
if workflow_type in WRITE_WORKFLOW_TYPES:
opc_output_config, issues = resolve_write_tags(
pipeline.get('write_tags') or [], write_tags_by_id, connectors
)
_merge_issues(write_issues, issues, schedule_name)
pipeline = {**pipeline, 'opc_output_config': opc_output_config}
# A pipeline without updated_at is redeployed on every run, as before
updated_at = pipeline.get('updated_at') or now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
schedule_config[namespace][schedule_name] = {
**function(pipeline),
'updated_at': pipeline.get(
'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
'updated_at': latest_updated_at(
updated_at, tag_documents_updated_at(pipeline, tag_documents)
),
}
self._notify_issues(pi_issues, PI_SCHEDULE_ISSUES, metadata)
self._notify_issues(write_issues, WRITE_TAG_ISSUES, metadata)
self.info('Processed schedules', metadata=metadata)
self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata)
@@ -374,8 +335,8 @@ class Formatters(SientiaMonitoring):
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.
- connector_tags (list[dict[str, Any]], optional): The connector_tags the
pipelines reference; only the active Read tags are used.
- active_ingestors (list[str]): The active ingestors to divide into slots.
Returns:
@@ -420,7 +381,7 @@ class Formatters(SientiaMonitoring):
Args:
issues (dict[str, set[str]]): What was left out, by kind
kinds (tuple[tuple[str, str, str], ...]): (kind, notification_id, message with
one {} for the joined values), e.g. SLOT_ISSUES or PI_SCHEDULE_ISSUES
one {} for the joined values), e.g. SLOT_ISSUES or WRITE_TAG_ISSUES
metadata (dict[str, Any]): Metadata for the notifications
"""
for kind, notification_id, message in kinds:
@@ -526,8 +487,7 @@ class Formatters(SientiaMonitoring):
Args:
input_data (dict[str, Any]): The input data containing:
- current_schedule_config (dict[str, Any]): The current schedule config in Temporal server
- schedule_config (dict[str, Any]): The schedule config to process; an
entry {'held': True} keeps that schedule exactly as it is
- schedule_config (dict[str, Any]): The schedule config to process
- metadata (dict): Metadata for logging purposes
Returns:
@@ -559,9 +519,6 @@ class Formatters(SientiaMonitoring):
for namespace, schedules in schedule_config.items():
current_schedules = current_schedule_config.get(namespace, {})
# A held schedule (process_schedules) is neither created nor updated; it stays
# in schedule_config, so it is not deleted either
schedules = {name: s for name, s in schedules.items() if not s.get('held')}
self.compare_config_timestamps(
schedules, current_schedules, to_update, to_create, namespace, metadata
)

View File

@@ -193,7 +193,8 @@ class MongoDB(SientiaMonitoring):
for document in aggregated_documents:
for timestamp_field in timestamp_fields:
if timestamp_field in document:
# Only BSON dates: a field already stored as text is left as it is
if isinstance(document.get(timestamp_field), datetime):
document[timestamp_field] = (
document[timestamp_field]
.replace(tzinfo=UTC)

View File

@@ -81,23 +81,45 @@ def _locate_read_tag(
return _opc_server(server_id, opc_server), tag['tag_address']
def _cycle_multiple(active_read: Any) -> int | None:
"""The cycle_multiple of an enabled active_read as int (default 1), or None when invalid."""
try:
return int(active_read.get('cycle_multiple', 1))
except (AttributeError, TypeError, ValueError):
return None
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.
A read tag the slot cannot use: no integer frequency, an enabled active_read without
an integer cycle_multiple, or (without tag_id) no server_id, tag_address or tag_name.
"""
if _frequency(tag) is None:
return True
active_read = tag.get('active_read')
if isinstance(active_read, dict) and active_read.get('enabled'):
if _cycle_multiple(active_read) is None:
return True
if tag_id_text(tag) is not None:
return False
return not is_complete(tag, ('server_id', 'tag_address', 'tag_name'))
def _is_slot_pipeline(pipeline: dict[str, Any]) -> bool:
"""A scouter pipeline whose read tags can be named in the slot (schedule, model id and name)."""
model = pipeline.get('model')
return (
is_complete(pipeline, ('schedule_name', 'model_id'))
and isinstance(model, dict)
and is_complete(model, ('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'):
if not isinstance(requested, dict) or not requested.get('enabled'):
return current
cycle_multiple = int(requested.get('cycle_multiple', 1))
cycle_multiple = cast(int, _cycle_multiple(requested))
if current.get('enabled'):
cycle_multiple = min(cycle_multiple, current['cycle_multiple'])
return {'enabled': True, 'cycle_multiple': cycle_multiple}
@@ -198,7 +220,8 @@ def gather_read_tags(
model_id and model_name, one per pipeline name, sorted
- 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)
usable frequency, cycle_multiple or legacy fields, or the schedule_name of a
pipeline without schedule_name, model_id or model.name)
"""
issues: dict[str, set[str]] = {
'tags': set(),
@@ -210,7 +233,10 @@ def gather_read_tags(
entries: dict[tuple[str, str], dict[str, Any]] = {}
for pipeline in pipelines:
if pipeline['workflow_type'] != 'scouter':
if pipeline.get('workflow_type') != 'scouter':
continue
if not _is_slot_pipeline(pipeline):
issues['invalid'].add(str(pipeline.get('schedule_name')))
continue
for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id):

View File

@@ -4,7 +4,6 @@ from math import ceil
from typing import Any
from orchestrator.utils.converters import parse_frequency
from orchestrator.utils.tag_database import PI_QUERY_DEFAULTS, pi_connection
def common_config(config: dict[str, Any]):
@@ -307,26 +306,21 @@ def pi_web_api_scouter(config: dict[str, Any]):
Args:
config (dict[str, Any]): Pipeline configuration containing:
- read_tags (dict[str, dict]): Tag configurations keyed by model tag name, already
passed through resolve_pi_read_tags (process_schedules does it), with:
- read_tags (list[dict]): List of tag configurations with:
- tag_name (str): Name of the tag
- webid (str): PI Web API WebID for the tag
- aggr_func (str, optional): Aggregation function (default: 'lts')
- data_range (list[int], optional): Valid data range (default: [-100, 100])
- pi_web_api_config (dict): PI Web API query settings with:
- pi_web_api_config (dict): PI Web API connection settings with:
- endpoint (str): PI Web API endpoint URL
- period (str, optional): Time period for data retrieval (default: '*-1d')
- max_count (int, optional): Maximum number of values to retrieve (default: 1)
- api_timeout (int, optional): API request timeout in seconds
- pi_web_api_connector (dict, optional): The tag database connector of the
pipeline's resolved tags (set by process_schedules). Its pi_web_api period,
max_count and api_timeout are defaults under pi_web_api_config
- Additional fields from base_scouter
Returns:
dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings.
API timeout is automatically adjusted to not exceed workflow frequency. With a
connector, it also carries connection: {connector_id, base_url, auth} (auth holds
the secret reference only, never the credential).
API timeout is automatically adjusted to not exceed workflow frequency.
"""
tags = {}
for tag, tag_config in config['read_tags'].items():
@@ -338,18 +332,7 @@ def pi_web_api_scouter(config: dict[str, Any]):
base_config = base_scouter(config)
connector = config.get('pi_web_api_connector')
pi_web_api_config = config['pi_web_api_config']
if connector:
connector_defaults = connector['pi_web_api']
pi_web_api_config = {
**{
key: connector_defaults[key]
for key in PI_QUERY_DEFAULTS
if key in connector_defaults
},
**pi_web_api_config,
}
config_timeout = pi_web_api_config.get('api_timeout', None)
frequency = parse_frequency(base_config['frequency'])
@@ -357,7 +340,7 @@ def pi_web_api_scouter(config: dict[str, Any]):
if config_timeout is None or config_timeout > frequency:
config_timeout = frequency
output = {
return {
**base_config,
'model_tags': tags,
'pi_web_api_query': {
@@ -367,9 +350,6 @@ def pi_web_api_scouter(config: dict[str, Any]):
'api_timeout': config_timeout,
},
}
if connector:
output['connection'] = pi_connection(connector)
return output
def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]):
@@ -441,11 +421,10 @@ def predictions_batch(config: dict[str, Any]):
Args:
config (dict[str, Any]): Pipeline configuration containing:
- query (str): SQL query to retrieve input data for predictions
- write_tags (list[dict]): List of OPC tag configurations for write-back with:
- server_id (str): ID of the target OPC server
- type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence)
- addr (str): OPC tag address/path
- data_type (str, optional): OPC data type (default: 'float')
- opc_output_config (dict, optional): The OPC write-back configuration, built by
process_schedules from the pipeline's write_tags ([{tag_id}]) through the tag
database (resolve_write_tags): {connector_id: {prediction_tags |
confidence_tags: {address: {data_type}}}} (default: {})
- datetime_columns (list[str], optional): Column names to parse as datetime (default: [])
- path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"])
- input_filters (list[dict], optional): Input data validation filters
@@ -454,31 +433,13 @@ def predictions_batch(config: dict[str, Any]):
- model_retention_minutes (int, optional): Data retention time in minutes (default: 60)
- save_transform (bool, optional): Save transformed data to database (default: True)
- predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1')
- pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {}),
already passed through resolve_pi_output_tags (process_schedules does it)
- pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {})
- Additional fields from common_config
Returns:
dict[str, Any]: Complete predictions batch configuration with OPC output mappings,
PI Web API output configuration, multi-stage filters, SQL query, and retention policies
"""
tags: dict[str, Any] = {}
for tag in config.get('write_tags', []):
if tag['server_id'] not in tags:
tags[tag['server_id']] = {}
tag_type = tag['type']
if tag_type == 'prediction' or tag_type == 'confidence':
tag_type_str = f'{tag_type}_tags'
if tag_type_str not in tags[tag['server_id']]:
tags[tag['server_id']][tag_type_str] = {}
tags[tag['server_id']][tag_type_str][tag['addr']] = {
'data_type': tag.get('data_type', 'float'),
}
path_priority = process_path_priority(
config.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT'])
)
@@ -492,7 +453,7 @@ def predictions_batch(config: dict[str, Any]):
'save_transform': config.get('save_transform', True),
'transform_table_name': 'transformed_data',
'retention_time': config.get('model_retention_minutes', 60) * 60,
'opc_output_config': tags,
'opc_output_config': config.get('opc_output_config', {}),
'pi_web_api_output_config': config.get('pi_web_api_output_config', {}),
'input_filters': overlap_filter_config(
{'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config.get('input_filters', [])

View File

@@ -1,103 +1,104 @@
"""
Tag database (connectors + connector_tags): which tags the pipelines reference, the
indexes of what was loaded, and the resolution of OPC UA names and PI Web API tags.
indexes of what was loaded, and the resolution of the OPC UA read and write tags.
"""
from typing import Any, cast
from typing import 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.
# Fields a tag database document needs to take part in a slot or a write config.
# Documents missing any of them are left out of the index, so their 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')
# connector_tags categories each use accepts
READ_CATEGORIES = ('Read',)
WRITE_CATEGORIES = ('Write', 'Prediction Confidence')
# opc_output_config block each write category goes to
WRITE_TAG_GROUPS = {'Write': 'prediction_tags', 'Prediction Confidence': 'confidence_tags'}
# data_type values the OPC Gateway write engine accepts (contract C2)
OPC_DATA_TYPES = ('float', 'double', 'real', 'int', 'integer', 'bool', 'boolean', 'str', 'string')
DEFAULT_DATA_TYPE = 'float'
# Workflow types whose write_tags reference the tag database
WRITE_WORKFLOW_TYPES = ('predictions_batch', 'xgboost_predictions_batch')
def is_complete(document: dict[str, Any], fields: tuple[str, ...]) -> bool:
return all(document.get(field) not in (None, '') for field in fields)
def _is_integer_text(text: str) -> bool:
"""True for ASCII decimal text with an optional sign ('7', '007', '-3'), never raising."""
digits = text[1:] if text.startswith('-') else text
return digits.isascii() and digits.isdecimal()
def tag_id_text(tag: dict[str, Any]) -> str | None:
"""The read tag's tag_id as text, or None when it has none (missing, null or empty)."""
"""
The tag's tag_id as text, or None when it has none (missing, null or empty).
Integer ids are canonical ('007' and 7 both give '7'), so a pipeline and the tag
database always agree on the key whether they store the id as int or text.
"""
tag_id = tag.get('tag_id')
return None if tag_id in (None, '') else str(tag_id)
if tag_id in (None, ''):
return None
text = str(tag_id).strip()
return str(int(text)) if _is_integer_text(text) else text
def _is_active(document: dict[str, Any]) -> bool:
"""A document without status (an override that does not project it) counts as active."""
return document.get('status') in (None, 'Active')
def pipeline_tag_refs(pipeline: dict[str, Any]) -> list[Any]:
"""
The tag references of a pipeline: the read_tags of a scouter and the write_tags of a
predictions_batch. Entries are returned as they are (they may lack a tag_id).
"""
workflow_type = pipeline.get('workflow_type')
if workflow_type == 'scouter':
return list(pipeline.get('read_tags') or [])
if workflow_type in WRITE_WORKFLOW_TYPES:
return list(pipeline.get('write_tags') or [])
return []
def referenced_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
"""
The tag_ids the scouter pipelines reference, for the connector_tags query.
The tag_ids the pipelines reference (scouter read_tags and predictions_batch
write_tags), 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.
pipelines may hold either. Sorted, so the query is the same on every run. Runs in
the workflow, so it never raises on malformed references.
Args:
pipelines (list[dict[str, Any]]): Pipeline configurations
Returns:
list[Any]: The referenced tag_ids (empty when no read tag carries a tag_id)
list[Any]: The referenced tag_ids (empty when no tag carries a tag_id)
"""
tags = [
tag
for pipeline in pipelines
if pipeline.get('workflow_type') == 'scouter'
for tag in pipeline.get('read_tags') or []
]
return _query_tag_ids(tags)
def _query_tag_ids(tags: list[Any]) -> list[Any]:
"""The tag_ids of the tags in both forms (int and text), sorted for a stable query."""
ids: set[Any] = set()
for tag in tags:
tag_id = tag_id_text(tag) if isinstance(tag, dict) else None
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)))
# Workflow types whose pi_web_api_output_config may reference the tag database
PI_OUTPUT_WORKFLOW_TYPES = ('predictions_batch', 'xgboost_predictions_batch')
# connector_tags category each PI Web API output mapping accepts
PI_OUTPUT_CATEGORIES = {'prediction_tags': 'Write', 'confidence_tags': 'Prediction Confidence'}
# Connection defaults a pi_web_api connector may carry for the scouter query
PI_QUERY_DEFAULTS = ('period', 'max_count', 'api_timeout')
def referenced_pi_tag_ids(pipelines: list[dict[str, Any]]) -> list[Any]:
"""
The tag_ids the PI Web API pipelines reference, for the PI connector_tags query.
Read tags of pi_web_api_scouter pipelines and the prediction_tags / confidence_tags
entries of a predictions_batch pi_web_api_output_config given as a list. Same forms
and order as referenced_tag_ids.
Args:
pipelines (list[dict[str, Any]]): Pipeline configurations
Returns:
list[Any]: The referenced tag_ids (empty when no PI Web API tag carries a tag_id)
"""
tags: list[Any] = []
for pipeline in pipelines:
workflow_type = pipeline.get('workflow_type')
if workflow_type == 'pi_web_api_scouter':
read_tags = pipeline.get('read_tags')
if isinstance(read_tags, dict):
tags.extend(read_tags.values())
elif workflow_type in PI_OUTPUT_WORKFLOW_TYPES:
output_config = pipeline.get('pi_web_api_output_config') or {}
for key in PI_OUTPUT_CATEGORIES:
if isinstance(output_config.get(key), list):
tags.extend(output_config[key])
return _query_tag_ids(tags)
for tag in pipeline_tag_refs(pipeline):
tag_id = tag_id_text(tag) if isinstance(tag, dict) else None
if tag_id is None:
continue
ids.add(tag_id)
if _is_integer_text(tag_id):
ids.add(int(tag_id))
return sorted(ids, key=lambda value: (isinstance(value, str), str(value)))
def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, Any]:
@@ -119,22 +120,44 @@ def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, An
}
def index_connector_tags(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
def index_tag_documents(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""
Index the tag database's connector_tags by tag_id.
Every loaded connector_tags document keyed by tag_id text, whatever its status,
category or completeness: the documents whose updated_at a pipeline depends on.
"""
indexed = {}
for tag in connector_tags:
tag_id = tag_id_text(tag)
if tag_id is not None:
indexed[tag_id] = tag
return indexed
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.
def index_connector_tags(
connector_tags: list[dict[str, Any]], categories: tuple[str, ...] = READ_CATEGORIES
) -> dict[str, dict[str, Any]]:
"""
Index the tag database's usable connector_tags by tag_id.
The key is the tag_id as text (tag_id_text), because pipelines may store it as int
or str. Documents that are not active, have another category, or lack tag_id,
connector_id, address or tag_name are left out.
Args:
connector_tags (list[dict[str, Any]]): connector_tags documents (SIENTIAPDE-1569),
each with tag_id, connector_id, address and tag_name
connector_tags (list[dict[str, Any]]): connector_tags documents (SIENTIAPDE-1569)
categories (tuple[str, ...]): Accepted categories (READ_CATEGORIES for the slot
and the scouter names, WRITE_CATEGORIES for the write tags). A document
without category is accepted.
Returns:
dict[str, dict[str, Any]]: connector_tags documents keyed by str(tag_id)
dict[str, dict[str, Any]]: connector_tags documents keyed by tag_id text
"""
return {
str(tag['tag_id']): tag for tag in connector_tags if is_complete(tag, CONNECTOR_TAG_FIELDS)
tag_id: tag
for tag_id, tag in index_tag_documents(connector_tags).items()
if is_complete(tag, CONNECTOR_TAG_FIELDS)
and _is_active(tag)
and tag.get('category') in (None, *categories)
}
@@ -142,7 +165,8 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An
"""
Index the tag database's connectors by connector_id (as text).
Documents without connector_id, name, url or server_uri are left out.
Documents that are not active or lack connector_id, name, url or server_uri are
left out.
Args:
connectors (list[dict[str, Any]]): connectors documents (SIENTIAPDE-1569)
@@ -153,10 +177,45 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An
return {
str(connector['connector_id']): connector
for connector in connectors
if is_complete(connector, CONNECTOR_FIELDS)
if is_complete(connector, CONNECTOR_FIELDS) and _is_active(connector)
}
def latest_updated_at(*values: Any) -> str | None:
"""
The latest of the given updated_at values, or None when none is set.
The values are text in DATETIME_FORMAT_MS_WITH_TZ (UTC), the format
aggregate_documents_in_mongodb gives with timestamp_fields, so they compare as text.
"""
timestamps = [value for value in values if isinstance(value, str) and value]
return max(timestamps) if timestamps else None
def tag_documents_updated_at(
pipeline: dict[str, Any], tag_documents: dict[str, dict[str, Any]]
) -> str | None:
"""
The latest updated_at among the tag database documents a pipeline references.
Every referenced document counts, inactive or of another category too, so renaming,
moving, retyping or deactivating a tag redeploys the pipelines that use it.
Args:
pipeline (dict[str, Any]): A pipeline configuration
tag_documents (dict[str, dict[str, Any]]): From index_tag_documents
Returns:
str | None: The latest updated_at, or None when no referenced document has one
"""
values = []
for tag in pipeline_tag_refs(pipeline):
tag_id = tag_id_text(tag) if isinstance(tag, dict) else None
if tag_id is not None and tag_id in tag_documents:
values.append(tag_documents[tag_id].get('updated_at'))
return latest_updated_at(*values)
def resolve_tag_names(
read_tags: list[dict[str, Any]], tags_by_id: dict[str, dict[str, Any]]
) -> list[dict[str, Any]]:
@@ -186,174 +245,56 @@ def resolve_tag_names(
return resolved
def _pi_auth(connector: dict[str, Any]) -> dict[str, Any] | None:
"""The connector's pi_web_api.auth when it names a type and a secret {name, key}."""
block = connector.get('pi_web_api')
auth = block.get('auth') if isinstance(block, dict) else None
if not isinstance(auth, dict) or auth.get('type') in (None, ''):
return None
secret = auth.get('secret')
if not isinstance(secret, dict) or not is_complete(secret, ('name', 'key')):
return None
return auth
def index_pi_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""
Index the tag database's pi_web_api connectors by connector_id (as text).
Documents without connector_id, url or a pi_web_api.auth with type and secret
{name, key} are left out, so their tags are notified as connector not found.
Args:
connectors (list[dict[str, Any]]): Active pi_web_api connectors documents
Returns:
dict[str, dict[str, Any]]: connectors documents keyed by str(connector_id)
"""
return {
str(connector['connector_id']): connector
for connector in connectors
if is_complete(connector, ('connector_id', 'url')) and _pi_auth(connector) is not None
}
def pi_connection(connector: dict[str, Any]) -> dict[str, Any]:
"""
Connection block of a pi_web_api connector for the schedule input.
Only the secret reference is copied, field by field, so nothing else the document
may hold under auth ever reaches the schedule input or the Temporal history (D3).
"""
auth = cast(dict[str, Any], _pi_auth(connector))
connection = {
'connector_id': str(connector['connector_id']),
'base_url': connector['url'],
'auth': {
'type': auth['type'],
'secret': {'name': auth['secret']['name'], 'key': auth['secret']['key']},
},
}
data_server_webid = connector['pi_web_api'].get('data_server_webid')
if data_server_webid:
connection['data_server_webid'] = data_server_webid
return connection
def _new_pi_issues() -> dict[str, set[str]]:
return {'tags': set(), 'categories': set(), 'connectors': set(), 'mixed': set()}
def _locate_pi_tag(
tag_id: str,
category: str,
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
issues: dict[str, set[str]],
) -> dict[str, Any] | None:
"""
The connector_tags document of a PI Web API tag_id, or None when it cannot be used:
unknown tag_id ('tags'), another category ('categories'), or a connector that is not
an active, complete pi_web_api connector ('connectors').
"""
connector_tag = pi_tags.get(tag_id)
if connector_tag is None:
issues['tags'].add(tag_id)
return None
if connector_tag.get('category') != category:
issues['categories'].add(tag_id)
return None
connector_id = str(connector_tag['connector_id'])
if connector_id not in pi_connectors:
issues['connectors'].add(connector_id)
return None
return connector_tag
def resolve_pi_read_tags(
read_tags: dict[str, dict[str, Any]],
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
) -> tuple[dict[str, dict[str, Any]], dict[str, Any] | None, dict[str, set[str]]]:
"""
Resolve the read tags of a pi_web_api_scouter pipeline through the tag database.
An entry with tag_id gets its webid from connector_tags.address (Read category); its
key (the model tag name), aggr_func and data_range stay the pipeline's. An entry
without tag_id is kept as is (legacy webid). An entry that cannot be resolved is left
out. All the resolved tags must belong to one connector (D6).
Args:
read_tags (dict[str, dict[str, Any]]): The pipeline's read_tags
pi_tags (dict[str, dict[str, Any]]): PI connector_tags indexed by index_connector_tags
pi_connectors (dict[str, dict[str, Any]]): Indexed by index_pi_connectors
Returns:
tuple: the resolved read_tags; the connector of the resolved tags (None when no tag
was resolved through the tag database or they span several connectors); and
what was left out by kind: 'tags' / 'categories' (tag_id), 'connectors'
(connector_id) and 'mixed' (the connector_ids, joined, when there are several)
"""
issues = _new_pi_issues()
resolved: dict[str, dict[str, Any]] = {}
connector_ids: set[str] = set()
for name, tag in read_tags.items():
tag_id = tag_id_text(tag)
if tag_id is None:
resolved[name] = tag
continue
connector_tag = _locate_pi_tag(tag_id, 'Read', pi_tags, pi_connectors, issues)
if connector_tag is None:
continue
connector_ids.add(str(connector_tag['connector_id']))
resolved[name] = {**tag, 'webid': connector_tag['address']}
if len(connector_ids) > 1:
issues['mixed'].add(', '.join(sorted(connector_ids)))
return resolved, None, issues
connector = pi_connectors[connector_ids.pop()] if connector_ids else None
return resolved, connector, issues
def resolve_pi_output_tags(
output_config: dict[str, Any],
pi_tags: dict[str, dict[str, Any]],
pi_connectors: dict[str, dict[str, Any]],
def resolve_write_tags(
write_tags: list[Any],
tags_by_id: dict[str, dict[str, Any]],
connectors: dict[str, dict[str, Any]],
) -> tuple[dict[str, Any], dict[str, set[str]]]:
"""
Resolve the prediction_tags / confidence_tags of a pi_web_api_output_config.
Build the opc_output_config of a predictions_batch from its write_tags.
A mapping given as a list is turned into {tag_name: WebID}: an entry with tag_id takes
tag_name and address from connector_tags ('Write' for prediction_tags, 'Prediction
Confidence' for confidence_tags); an entry without tag_id is read as a legacy
{tag_name: webid} fragment. A mapping already given as a dict passes through unchanged,
so the laborious input keeps today's shape.
Each write tag is only a tag_id: the tag database gives the rest. The category says
where the value goes ('Write' → prediction_tags, 'Prediction Confidence' →
confidence_tags), the connector_id is the server the OPC Gateway writes to (it
resolves it in connectors), and the address and data_type come from connector_tags
(data_type defaults to 'float').
Args:
output_config (dict[str, Any]): The pipeline's pi_web_api_output_config
pi_tags (dict[str, dict[str, Any]]): PI connector_tags indexed by index_connector_tags
pi_connectors (dict[str, dict[str, Any]]): Indexed by index_pi_connectors
write_tags (list[Any]): The pipeline's write_tags ([{tag_id}])
tags_by_id (dict[str, dict[str, Any]]): Write connector_tags indexed by
index_connector_tags(..., WRITE_CATEGORIES)
connectors (dict[str, dict[str, Any]]): Active OPC UA connectors indexed by
index_connectors
Returns:
tuple: the resolved output config and what was left out, by kind ('tags',
'categories', 'connectors'), as in resolve_pi_read_tags
tuple: the opc_output_config ({connector_id: {prediction_tags | confidence_tags:
{address: {data_type}}}}) and what was left out, by kind: 'tags' (tag_id not
an active, complete Write / Prediction Confidence tag), 'connectors'
(connector_id) and 'invalid' (an entry without tag_id, or a tag whose
data_type the gateway does not accept)
"""
issues = _new_pi_issues()
resolved = dict(output_config)
for key, category in PI_OUTPUT_CATEGORIES.items():
entries = output_config.get(key)
if not isinstance(entries, list):
issues: dict[str, set[str]] = {'tags': set(), 'connectors': set(), 'invalid': set()}
output_config: dict[str, Any] = {}
for entry in write_tags:
tag_id = tag_id_text(entry) if isinstance(entry, dict) else None
if tag_id is None:
issues['invalid'].add(str(entry))
continue
mapping: dict[str, Any] = {}
for entry in entries:
if not isinstance(entry, dict):
continue
tag_id = tag_id_text(entry)
if tag_id is None:
mapping.update(entry)
continue
connector_tag = _locate_pi_tag(tag_id, category, pi_tags, pi_connectors, issues)
if connector_tag is not None:
mapping[connector_tag['tag_name']] = connector_tag['address']
resolved[key] = mapping
return resolved, issues
connector_tag = tags_by_id.get(tag_id)
# The category is what tells a prediction from a confidence: it cannot be missing
if connector_tag is None or connector_tag.get('category') not in WRITE_TAG_GROUPS:
issues['tags'].add(tag_id)
continue
connector_id = str(connector_tag['connector_id'])
if connector_id not in connectors:
issues['connectors'].add(connector_id)
continue
data_type = connector_tag.get('data_type') or DEFAULT_DATA_TYPE
if data_type not in OPC_DATA_TYPES:
issues['invalid'].add(tag_id)
continue
group = WRITE_TAG_GROUPS[connector_tag['category']]
output_config.setdefault(connector_id, {}).setdefault(group, {})[
connector_tag['address']
] = {'data_type': data_type}
return output_config, issues

View File

@@ -1,7 +1,6 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
import asyncio
from datetime import timedelta
from typing import Any
@@ -10,7 +9,6 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from orchestrator.utils.tag_database import (
filter_by_tag_ids,
referenced_pi_tag_ids,
referenced_tag_ids,
)
@@ -18,11 +16,14 @@ with workflow.unsafe.imports_passed_through():
MATCH = '$match'
PROJECT = '$project'
# Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow
# 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.
# Tag database (SIENTIAPDE-1569) read by the slot v2 and the schedules (SIENTIAPDE-1974).
# The workflow input may override these aggregations (connectors_query /
# connector_tags_query); the workflow puts a tag_id $match in front of
# connector_tags_query. Every category and status of the referenced tags is loaded: the
# slot uses the active Read tags, predictions_batch the active Write / Prediction
# Confidence ones, and every referenced tag's updated_at redeploys its pipelines. Only
# the fields used are projected: other attributes may hold BSON types a Temporal payload
# cannot carry (updated_at is turned into text by timestamp_fields).
DEFAULT_CONNECTORS_QUERY = {
'collection': 'connectors',
'aggregation': [
@@ -43,28 +44,6 @@ DEFAULT_CONNECTORS_QUERY = {
DEFAULT_CONNECTOR_TAGS_QUERY = {
'collection': 'connector_tags',
'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'
# PI Web API side of the tag database, read by process_schedules only (never by the
# slot). The input may override them (pi_connectors_query / pi_connector_tags_query);
# the workflow puts a tag_id $match in front of pi_connector_tags_query. Tags of every
# category are loaded: predictions_batch writes to Write / Prediction Confidence tags.
DEFAULT_PI_CONNECTORS_QUERY = {
'collection': 'connectors',
'aggregation': [
{MATCH: {'type': 'pi_web_api', 'status': 'Active'}},
{PROJECT: {'_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, 'pi_web_api': 1}},
],
}
DEFAULT_PI_CONNECTOR_TAGS_QUERY = {
'collection': 'connector_tags',
'aggregation': [
{MATCH: {'status': 'Active'}},
{
PROJECT: {
'_id': 0,
@@ -73,12 +52,15 @@ DEFAULT_PI_CONNECTOR_TAGS_QUERY = {
'address': 1,
'tag_name': 1,
'category': 1,
'data_type': 1,
'status': 1,
'updated_at': 1,
}
},
],
}
# Runs started before the PI Web API loads existed replay without them
PI_WEB_API_PATCH = 'connectors-pi-web-api'
# Runs started before the tag database loads existed replay without them
TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database'
@workflow.defn(name='orchestrator')
@@ -99,12 +81,14 @@ class Orchestrator:
pipelines: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""
Load the tag database (SIENTIAPDE-1974) for the read tags that carry a tag_id.
Load the tag database (SIENTIAPDE-1974) for the read and write tags that carry a
tag_id.
Only the referenced tags are loaded, with the fields the slot uses, so an
environment without a tag database never depends on its content (P4) and the
payload stays bounded by the pipelines. Nothing is loaded when no scouter read
tag has a tag_id, or when replaying a run started before TAG_DATABASE_PATCH.
Only the referenced tags are loaded, with the fields used, so an environment
without a tag database never depends on its content (P4) and the payload stays
bounded by the pipelines. Nothing is loaded when no scouter read tag nor
predictions_batch write tag has a tag_id, or when replaying a run started before
TAG_DATABASE_PATCH.
Args:
input_data (dict[str, Any]): Workflow input (optional connectors_query and
@@ -127,39 +111,6 @@ class Orchestrator:
tag_ids,
)
async def _load_pi_tag_database(
self,
input_data: dict[str, Any],
metadata: dict[str, Any],
pipelines: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""
Load the PI Web API connectors and the tags the PI Web API pipelines reference.
Same bounds as _load_tag_database: nothing is loaded when no PI Web API tag has a
tag_id, or when replaying a run started before PI_WEB_API_PATCH.
Args:
input_data (dict[str, Any]): Workflow input (optional pi_connectors_query and
pi_connector_tags_query)
metadata (dict[str, Any]): Activity metadata ({'metadata': {...}})
pipelines (list[dict[str, Any]]): The loaded pipeline configurations
Returns:
tuple[list[dict[str, Any]], list[dict[str, Any]]]: pi connectors and their tags
"""
tag_ids = referenced_pi_tag_ids(pipelines)
# patched() first: it must run on every execution to record its marker
if not workflow.patched(PI_WEB_API_PATCH) or not tag_ids:
return [], []
return await self._load_connectors(
metadata,
input_data.get('pi_connectors_query', DEFAULT_PI_CONNECTORS_QUERY),
input_data.get('pi_connector_tags_query', DEFAULT_PI_CONNECTOR_TAGS_QUERY),
tag_ids,
)
async def _load_connectors(
self,
metadata: dict[str, Any],
@@ -174,6 +125,7 @@ class Orchestrator:
**metadata,
# A fresh aggregation list: the activity appends a stage to the one it gets
'query': {**connectors_query, 'aggregation': list(connectors_query['aggregation'])},
'timestamp_fields': ['updated_at'],
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),
@@ -181,7 +133,11 @@ class Orchestrator:
connector_tags_handler = workflow.start_local_activity_method(
Activities.aggregate_documents_in_mongodb,
{**metadata, 'query': filter_by_tag_ids(connector_tags_query, tag_ids)},
{
**metadata,
'query': filter_by_tag_ids(connector_tags_query, tag_ids),
'timestamp_fields': ['updated_at'],
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),
)
@@ -204,16 +160,12 @@ 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 (used only when a scouter read tag carries a tag_id):
Optional fields (used only when a scouter read tag or a predictions_batch
write 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)
Optional fields (used only when a PI Web API tag carries a tag_id):
- pi_connectors_query (dict[str, Any]): MongoDB aggregation for the
pi_web_api connectors (default: DEFAULT_PI_CONNECTORS_QUERY)
- pi_connector_tags_query (dict[str, Any]): MongoDB aggregation for their
tags (default: DEFAULT_PI_CONNECTOR_TAGS_QUERY)
Returns:
None: Workflow completes without return value
@@ -282,10 +234,8 @@ class Orchestrator:
pipeline_config = await pipeline_config_handler
# Started in this order on every run, so commands stay deterministic
(connectors, connector_tags), (pi_connectors, pi_connector_tags) = await asyncio.gather(
self._load_tag_database(input_data, metadata, pipeline_config),
self._load_pi_tag_database(input_data, metadata, pipeline_config),
connectors, connector_tags = await self._load_tag_database(
input_data, metadata, pipeline_config
)
orchestrated_schedules = await orchestrated_schedules_handler
@@ -305,9 +255,8 @@ class Orchestrator:
{
**metadata,
'pipelines': pipeline_config,
'connectors': connectors,
'connector_tags': connector_tags,
'pi_connectors': pi_connectors,
'pi_connector_tags': pi_connector_tags,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),

View File

@@ -0,0 +1,237 @@
/**
* SIENTIAPDE-1974 runbook: move the predictions_batch write tags to the tag database.
*
* The Orchestrator reads write_tags as [{tag_id}] only: server, address, category and
* data_type come from connector_tags + connectors. This script converts the legacy write
* tags ({server_id, type, addr, data_type}):
* - server_id (opc-servers id) -> connector: the connector whose legacy_opc_server_id is
* the server_id, else the one whose connector_id is the slug of the server name (D-D),
* else the one whose connector_id is the server_id itself;
* - each tag -> the connector_tags document with the same connector, address and category
* (type prediction -> Write, confidence -> Prediction Confidence), created when missing
* with the name <schedule>:<type>[:n] (D-A) and the tag's data_type;
* - write_tags -> [{tag_id}], the original array kept in legacy_write_tags, and
* updated_at bumped (D-C) so the Orchestrator redeploys the schedule.
* Entries that already carry a tag_id are reduced to {tag_id}; their data_type, when the
* pipeline sets one, is copied to the connector_tags document that has none (a different
* one aborts).
*
* Every check runs before the first write: any conflict aborts with nothing written.
* Idempotent: a converted pipeline has no legacy entry left and is skipped.
*
* Usage (mongosh, against the sientia database):
* DRY_RUN=1 mongosh "$MONGO_URI" scripts/sientiapde_1974_backfill_write_tags.js # plan only (default)
* DRY_RUN=0 mongosh "$MONGO_URI" scripts/sientiapde_1974_backfill_write_tags.js # write
* REVERT=1 DRY_RUN=0 mongosh "$MONGO_URI" scripts/sientiapde_1974_backfill_write_tags.js
*/
const SOURCE = 'backfill-SIENTIAPDE-1974-write-tags';
const WRITE_TYPES = ['predictions_batch', 'xgboost_predictions_batch'];
const WRITE_CATEGORY = { prediction: 'Write', confidence: 'Prediction Confidence' };
const DATA_TYPES = ['float', 'double', 'real', 'int', 'integer', 'bool', 'boolean', 'str', 'string'];
const DRY_RUN = (process.env.DRY_RUN ?? '1') !== '0';
const REVERT = process.env.REVERT === '1';
const hasValue = (value) => value !== undefined && value !== null && value !== '';
// D-D: connector_id is the slug of the server name (same rule as the tag database backfill)
const slugify = (name) =>
String(name)
.normalize('NFKD')
.replace(/[^\x00-\x7f]/g, '')
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '');
// D-A: write tags have no name: <schedule>:<type>, plus :<n> when the type repeats
const writeTagName = (schedule, tags, index) => {
const same = tags.flatMap((tag, i) => (tag.type === tags[index].type ? [i] : []));
const base = `${schedule}:${tags[index].type}`;
return same.length === 1 ? base : `${base}:${same.indexOf(index) + 1}`;
};
const connectorResolver = () => {
const connectors = db.connectors.find({ type: 'opc_ua' }).toArray();
const byLegacyId = new Map(
connectors.filter((c) => hasValue(c.legacy_opc_server_id)).map((c) => [String(c.legacy_opc_server_id), c])
);
const byId = new Map(connectors.map((c) => [String(c.connector_id), c]));
const servers = new Map(db.getCollection('opc-servers').find({}).toArray().map((s) => [String(s.id), s]));
return (serverId) => {
const sid = String(serverId);
if (byLegacyId.has(sid)) return byLegacyId.get(sid);
const server = servers.get(sid);
if (server && hasValue(server.server_name) && byId.has(slugify(server.server_name))) {
return byId.get(slugify(server.server_name));
}
return byId.get(sid);
};
};
const buildPlan = () => {
const errors = [];
const connectorOf = connectorResolver();
const tags = db.connector_tags.find({}).toArray();
const byAddress = new Map(tags.map((t) => [`${t.connector_id}|${t.address}|${t.category}`, t]));
const byName = new Map(tags.map((t) => [`${t.connector_id}|${t.tag_name}`, t]));
const byTagId = new Map(tags.map((t) => [String(t.tag_id), t]));
const newTags = [];
const dataTypes = new Map(); // tag_id -> data_type to set on an existing connector_tags
const updates = [];
const pipelines = db.pipelines.find({ workflow_type: { $in: WRITE_TYPES } }).toArray();
for (const pipeline of pipelines) {
const legacy = pipeline.write_tags;
if (!Array.isArray(legacy) || legacy.every((t) => hasValue(t?.tag_id) && Object.keys(t).length === 1)) {
continue; // nothing to convert
}
const refs = [];
legacy.forEach((tag, index) => {
const where = `${pipeline.schedule_name} write_tags[${index}]`;
if (hasValue(tag?.tag_id)) {
const existing = byTagId.get(String(tag.tag_id));
if (!existing) return errors.push(`${where}: tag_id ${tag.tag_id} not in connector_tags`);
if (hasValue(tag.data_type)) {
const current = existing.data_type ?? dataTypes.get(String(tag.tag_id));
if (!DATA_TYPES.includes(tag.data_type)) {
return errors.push(`${where}: data_type ${tag.data_type} not accepted`);
}
if (hasValue(current) && current !== tag.data_type) {
return errors.push(`${where}: tag_id ${tag.tag_id} has data_type ${current}, the pipeline ${tag.data_type}`);
}
if (!hasValue(existing.data_type)) dataTypes.set(String(tag.tag_id), tag.data_type);
}
refs.push({ tag_id: existing.tag_id });
return;
}
const connector = connectorOf(tag?.server_id);
const category = WRITE_CATEGORY[tag?.type];
if (!connector) return errors.push(`${where}: no opc_ua connector for server_id ${tag?.server_id}`);
if (!category) return errors.push(`${where}: type ${tag?.type} is neither prediction nor confidence`);
if (!hasValue(tag.addr)) return errors.push(`${where}: no addr`);
const dataType = tag.data_type ?? 'float';
if (!DATA_TYPES.includes(dataType)) return errors.push(`${where}: data_type ${dataType} not accepted`);
const connectorId = String(connector.connector_id);
const addressKey = `${connectorId}|${tag.addr}|${category}`;
let entry = byAddress.get(addressKey);
if (!entry) {
const name = writeTagName(pipeline.schedule_name, legacy, index);
const taken = byName.get(`${connectorId}|${name}`);
if (taken) {
return errors.push(`${where}: name ${name} already used by tag_id ${taken.tag_id} (${taken.address})`);
}
entry = { connector_id: connectorId, address: tag.addr, tag_name: name, category, data_type: dataType, tag_id: null };
byAddress.set(addressKey, entry);
byName.set(`${connectorId}|${name}`, entry);
newTags.push(entry);
} else if (hasValue(entry.data_type) && entry.data_type !== dataType) {
errors.push(`${where}: tag_id ${entry.tag_id} has data_type ${entry.data_type}, the pipeline ${dataType}`);
}
refs.push({ entry });
});
updates.push({ pipeline, refs });
}
return { errors, newTags, updates, dataTypes };
};
const reserveIds = (count) => {
if (!db.connector_id_counters.findOne({ _id: 'connector_tags' })) {
const max = db.connector_tags.find({}, { tag_id: 1 }).sort({ tag_id: -1 }).limit(1).toArray()[0];
db.connector_id_counters.insertOne({ _id: 'connector_tags', seq: max ? Number(max.tag_id) : 0 });
}
const doc = db.connector_id_counters.findOneAndUpdate(
{ _id: 'connector_tags' },
{ $inc: { seq: count } },
{ returnDocument: 'after' }
);
return Number(doc.seq) - count + 1;
};
const backfill = () => {
const plan = buildPlan();
print(`Pipelines to convert: ${plan.updates.length}; connector_tags to create: ${plan.newTags.length}`);
plan.newTags.forEach((t) => print(` + ${t.connector_id} ${t.category} ${t.address} as ${t.tag_name} (${t.data_type})`));
plan.dataTypes.forEach((dataType, tagId) => print(` * connector_tags ${tagId}: data_type ${dataType}`));
plan.updates.forEach(({ pipeline, refs }) =>
print(` ~ ${pipeline.schedule_name}: ${refs.length} write tag(s)`)
);
if (plan.errors.length > 0) {
print(`Aborted, nothing was written:\n- ${plan.errors.join('\n- ')}`);
quit(1);
}
if (DRY_RUN) {
print('DRY_RUN: nothing was written (run with DRY_RUN=0 to apply)');
return;
}
const now = new Date();
if (plan.newTags.length > 0) {
let next = reserveIds(plan.newTags.length);
plan.newTags.forEach((tag) => {
tag.tag_id = next;
next += 1;
});
db.connector_tags.insertMany(
plan.newTags.map((tag) => ({
tag_id: tag.tag_id,
connector_id: tag.connector_id,
address: tag.address,
tag_name: tag.tag_name,
category: tag.category,
data_type: tag.data_type,
unit: '',
status: 'Active',
description: '',
metadata: {},
source: SOURCE,
created_at: now,
updated_at: now,
}))
);
}
plan.dataTypes.forEach((dataType, tagId) => {
db.connector_tags.updateOne(
{ tag_id: Number(tagId), data_type: { $in: [null, ''] } },
{ $set: { data_type: dataType, updated_at: now } }
);
});
let changed = 0;
const raced = [];
for (const { pipeline, refs } of plan.updates) {
const writeTags = refs.map((ref) => ({ tag_id: ref.tag_id ?? ref.entry.tag_id }));
// Only if write_tags is still the array read above: a pipeline edited meanwhile is skipped
const result = db.pipelines.updateOne(
{ _id: pipeline._id, write_tags: pipeline.write_tags },
{ $set: { write_tags: writeTags, legacy_write_tags: pipeline.write_tags, updated_at: now } }
);
if (result.modifiedCount === 1) changed += 1;
else raced.push(pipeline.schedule_name);
}
print(`connector_tags created: ${plan.newTags.length}; pipelines converted: ${changed}`);
if (raced.length > 0) print(`Changed during the backfill, run it again: ${raced.join(', ')}`);
};
const revert = () => {
const pipelines = db.pipelines.find({ legacy_write_tags: { $exists: true } }).toArray();
const created = db.connector_tags.countDocuments({ source: SOURCE });
print(`Pipelines to restore: ${pipelines.length}; connector_tags to delete: ${created}`);
if (DRY_RUN) {
print('DRY_RUN: nothing was written (run with DRY_RUN=0 to apply)');
return;
}
const now = new Date();
for (const pipeline of pipelines) {
db.pipelines.updateOne(
{ _id: pipeline._id },
{ $set: { write_tags: pipeline.legacy_write_tags, updated_at: now }, $unset: { legacy_write_tags: '' } }
);
}
const { deletedCount } = db.connector_tags.deleteMany({ source: SOURCE });
print(`pipelines restored: ${pipelines.length}; connector_tags deleted: ${deletedCount}`);
};
if (REVERT) revert();
else backfill();

View File

@@ -6,7 +6,7 @@ from pytest import fixture
from sientia_do.notifications.models import NotificationLevel
from orchestrator.activities.formatters import Formatters
from orchestrator.utils.pipeline_configs import pi_web_api_scouter, predictions_batch
from orchestrator.utils.pipeline_configs import predictions_batch
@fixture
@@ -141,7 +141,10 @@ def test_process_schedules(formatters):
}
mock_scouter.assert_called_once_with(input_data['pipelines'][0])
mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1])
# A predictions_batch gets its opc_output_config from its (here absent) write_tags
mock_predictions_batch.assert_called_once_with(
{**input_data['pipelines'][1], 'opc_output_config': {}}
)
mock_minimal_retrain.assert_called_once_with(input_data['pipelines'][2])
mock_drift.assert_called_once_with(input_data['pipelines'][3])
mock_simple_metrics.assert_called_once_with(input_data['pipelines'][4])
@@ -492,8 +495,9 @@ def test_process_slots_notifies_unresolved_tags_once(formatters):
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'
'Read tags sin/ns=2;s=C have no integer frequency, an active_read without '
'an integer cycle_multiple, or lack server_id, tag_address or tag_name; or '
'their pipeline lacks schedule_name, model_id or model.name'
),
block='orchestrator',
level=NotificationLevel.ERROR,
@@ -1431,224 +1435,220 @@ def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatter
# PI Web API through the tag database (connectors-pi-web-api)
PI_SECRET = {'name': 'pi-web-api-auth-token', 'key': 'token'}
PI_CONNECTOR = {
'connector_id': 'pi-usina',
'name': 'PI Usina',
'url': 'https://pi.example/piwebapi',
'pi_web_api': {'auth': {'type': 'bearer', 'secret': PI_SECRET}, 'period': '*-2h'},
PLANT_CONNECTOR = {
'connector_id': 'plant',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'status': 'Active',
}
def _pi_tag(tag_id, address, tag_name, category='Read', connector_id='pi-usina'):
def _write_tag(tag_id, address, category='Write', **extra):
return {
'tag_id': tag_id,
'connector_id': connector_id,
'connector_id': 'plant',
'address': address,
'tag_name': tag_name,
'tag_name': f'tag-{tag_id}',
'category': category,
'status': 'Active',
**extra,
}
def _pi_pipeline(schedule_name, read_tags):
return {
'workflow_type': 'pi_web_api_scouter',
'schedule_name': schedule_name,
'model_id': 'm1',
'model': {'name': 'model'},
'frequency': '1m',
'read_tags': read_tags,
'pi_web_api_config': {'endpoint': '/streamsets/recorded', 'max_count': 5},
'updated_at': '2021-01-01',
}
def _predictions_pipeline(schedule_name, output_config):
def _batch_pipeline(schedule_name, write_tags, updated_at='2026-10-01 00:00:00.000+0000'):
return {
'workflow_type': 'predictions_batch',
'schedule_name': schedule_name,
'model_id': 'm2',
'model': {'name': 'model'},
'model_id': 'model_id',
'model': {'name': 'model_name'},
'query': 'SELECT 1',
'pi_web_api_output_config': output_config,
'updated_at': '2021-01-01',
'write_tags': write_tags,
'updated_at': updated_at,
}
def test_process_schedules_legacy_pi_pipelines_unchanged(formatters):
"""Legacy PI Web API pipelines produce byte-identical schedule input"""
scouter = _pi_pipeline('pi', {'TI-101': {'webid': 'F1DP-X', 'aggr_func': 'avg'}})
batch = _predictions_pipeline(
'batch', {'prediction_tags': {'PRED-101': 'F1DP-P'}, 'confidence_tags': {}}
)
def test_process_schedules_resolves_write_tags_through_tag_database(formatters):
batch = _batch_pipeline('batch', [{'tag_id': 77}, {'tag_id': 78}])
with_tag_database = formatters.process_schedules(
result = formatters.process_schedules(
{
'pipelines': [scouter, batch],
'pi_connectors': [PI_CONNECTOR],
'pi_connector_tags': [_pi_tag(42, 'F1DP-A', 'TI-101')],
'pipelines': [batch],
'connectors': [PLANT_CONNECTOR],
'connector_tags': [
_write_tag(77, 'ns=2;s=P1'),
_write_tag(78, 'ns=2;s=C1', category='Prediction Confidence', data_type='int'),
],
**metadata,
}
)
before_change = {
'scouter': {'pi': {**pi_web_api_scouter(scouter), 'updated_at': '2021-01-01'}},
'laborious': {'batch': {**predictions_batch(batch), 'updated_at': '2021-01-01'}},
}
assert json.dumps(with_tag_database, sort_keys=True) == json.dumps(
before_change, sort_keys=True
)
assert 'connection' not in with_tag_database['scouter']['pi']
assert with_tag_database['laborious']['batch']['pi_web_api_output_config'] == {
'prediction_tags': {'PRED-101': 'F1DP-P'},
'confidence_tags': {},
schedule = result['laborious']['batch']
assert schedule['opc_output_config'] == {
'plant': {
'prediction_tags': {'ns=2;s=P1': {'data_type': 'float'}},
'confidence_tags': {'ns=2;s=C1': {'data_type': 'int'}},
}
}
assert 'write_tags' not in schedule
formatters.send_notification.assert_not_called()
def test_process_schedules_resolves_pi_tags_through_tag_database(formatters):
scouter = _pi_pipeline('pi', {'TI-101': {'tag_id': 42, 'aggr_func': 'avg'}})
batch = _predictions_pipeline('batch', {'prediction_tags': [{'tag_id': 77}]})
def test_process_schedules_notifies_unresolved_write_tags_once(formatters):
pipelines = [
_batch_pipeline('a', [{'tag_id': 404}, {'tag_id': 7}, {'tag_id': 9}]),
_batch_pipeline('b', [{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=L'}]),
]
result = formatters.process_schedules(
{
'pipelines': [scouter, batch],
'pi_connectors': [PI_CONNECTOR],
'pi_connector_tags': [
_pi_tag(42, 'F1DP-A', 'TI-101'),
_pi_tag(77, 'F1DP-P', 'PRED-101', category='Write'),
'pipelines': pipelines,
'connectors': [PLANT_CONNECTOR],
'connector_tags': [
_write_tag(7, 'ns=2;s=R', category='Read'),
{**_write_tag(9, 'ns=2;s=X'), 'connector_id': 'gone'},
],
**metadata,
}
)
pi = result['scouter']['pi']
assert pi['model_tags'] == {
'TI-101': {'webid': 'F1DP-A', 'aggr_func': 'avg', 'data_range': [-100, 100]}
}
assert pi['pi_web_api_query'] == {
'endpoint': '/streamsets/recorded',
'period': '*-2h',
'max_count': 5,
'api_timeout': 60,
}
assert pi['connection'] == {
'connector_id': 'pi-usina',
'base_url': 'https://pi.example/piwebapi',
'auth': {'type': 'bearer', 'secret': PI_SECRET},
}
assert result['laborious']['batch']['pi_web_api_output_config'] == {
'prediction_tags': {'PRED-101': 'F1DP-P'}
}
# The builder input is never the pipeline document itself
assert 'pi_web_api_connector' not in scouter
formatters.send_notification.assert_not_called()
def test_process_schedules_secret_value_never_in_schedule_input(formatters):
connector = {
**PI_CONNECTOR,
'pi_web_api': {
'auth': {
'type': 'basic',
'secret': {**PI_SECRET, 'value': 'leaked-credential'},
'password': 'leaked-credential',
}
},
}
result = formatters.process_schedules(
{
'pipelines': [_pi_pipeline('pi', {'TI-101': {'tag_id': 42}})],
'pi_connectors': [connector],
'pi_connector_tags': [_pi_tag(42, 'F1DP-A', 'TI-101')],
**metadata,
}
# Unresolved tags are left out; the pipelines are still scheduled
assert result['laborious']['a']['opc_output_config'] == {}
assert result['laborious']['b']['opc_output_config'] == {}
formatters.send_notification.assert_has_calls(
[
call(
metadata=metadata['metadata'],
notification_id='ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
message=(
'Write tags a/404, a/7 not found among the active, complete Write and '
'Prediction Confidence tags of connector_tags'
),
block='orchestrator',
level=NotificationLevel.ERROR,
),
call(
metadata=metadata['metadata'],
notification_id=(
'ORCHESTRATOR_WRITE_TAG_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION'
),
message='Connectors a/gone of write tags not found, not active or incomplete in connectors',
block='orchestrator',
level=NotificationLevel.ERROR,
),
call(
metadata=metadata['metadata'],
notification_id='ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION',
message=(
"Write tags b/{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=L'} "
'have no tag_id or a data_type the OPC Gateway does not accept'
),
block='orchestrator',
level=NotificationLevel.ERROR,
),
]
)
assert result['scouter']['pi']['connection']['auth'] == {'type': 'basic', 'secret': PI_SECRET}
assert 'leaked-credential' not in json.dumps(result)
assert formatters.send_notification.call_count == 3
def test_process_schedules_notifies_unresolved_pi_tags(formatters):
scouter = _pi_pipeline('pi', {'TI-101': {'tag_id': 42}, 'TI-404': {'tag_id': 404}})
batch = _predictions_pipeline('batch', {'prediction_tags': [{'tag_id': 42}]})
def test_process_schedules_updated_at_follows_the_tag_database(formatters):
"""A tag changed after the pipeline redeploys the pipelines that reference it"""
batch = _batch_pipeline('batch', [{'tag_id': 77}], updated_at='2026-10-01 00:00:00.000+0000')
scouter = {
'workflow_type': 'scouter',
'schedule_name': 'sin',
'model_id': 'm1',
'model': {'name': 'model m1'},
'read_tags': [{'tag_id': 7, 'frequency': 1000}],
'updated_at': '2026-10-01 00:00:00.000+0000',
}
untouched = _batch_pipeline('old', [{'tag_id': 78}], updated_at='2026-10-20 00:00:00.000+0000')
result = formatters.process_schedules(
schedule_config = formatters.process_schedules(
{
'pipelines': [scouter, batch],
'pi_connectors': [PI_CONNECTOR],
'pi_connector_tags': [
_pi_tag(42, 'F1DP-A', 'TI-101'),
_pi_tag(43, 'F1DP-B', 'TI-OFF', connector_id='pi-off'),
'pipelines': [batch, scouter, untouched],
'connectors': [PLANT_CONNECTOR],
'connector_tags': [
_write_tag(77, 'ns=2;s=P2', updated_at='2026-10-10 00:00:00.000+0000'),
# Deactivated: no longer usable, but its change still redeploys
_write_tag(
7,
'ns=3;i=1001',
category='Read',
status='Inactive',
updated_at='2026-10-11 00:00:00.000+0000',
),
_write_tag(78, 'ns=2;s=C', updated_at='2026-10-02 00:00:00.000+0000'),
],
**metadata,
}
)
# Still scheduled, without the unresolved tags
assert list(result['scouter']['pi']['model_tags']) == ['TI-101']
assert result['laborious']['batch']['pi_web_api_output_config'] == {'prediction_tags': {}}
assert schedule_config['laborious']['batch']['updated_at'] == '2026-10-10 00:00:00.000+0000'
assert schedule_config['scouter']['sin']['updated_at'] == '2026-10-11 00:00:00.000+0000'
assert schedule_config['laborious']['old']['updated_at'] == '2026-10-20 00:00:00.000+0000'
notified = {
c.kwargs['notification_id']: c.kwargs['message']
for c in formatters.send_notification.call_args_list
}
assert set(notified) == {
'ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION',
'ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION',
}
assert 'pi/404' in notified['ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION']
assert (
'batch/42' in notified['ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION']
)
for c in formatters.send_notification.call_args_list:
assert c.kwargs['level'] == NotificationLevel.ERROR
def test_process_schedules_holds_mixed_connector_pipeline(formatters):
second = {**PI_CONNECTOR, 'connector_id': 'pi-other'}
pipeline = _pi_pipeline('pi', {'A': {'tag_id': 1}, 'B': {'tag_id': 2}})
result = formatters.process_schedules(
# Last orchestrated on 2026-10-05: only the pipelines whose tags changed since update
actions = formatters.create_schedule_config(
{
'pipelines': [pipeline],
'pi_connectors': [PI_CONNECTOR, second],
'pi_connector_tags': [
_pi_tag(1, 'F1DP-1', 'A'),
_pi_tag(2, 'F1DP-2', 'B', connector_id='pi-other'),
],
**metadata,
}
)
assert result['scouter'] == {'pi': {'held': True}}
formatters.send_notification.assert_called_once()
kwargs = formatters.send_notification.call_args.kwargs
assert (
kwargs['notification_id']
== 'ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pi/pi-other, pi-usina' in kwargs['message']
def test_create_schedule_config_leaves_held_schedule_untouched(formatters):
result = formatters.create_schedule_config(
{
'current_schedule_config': {'scouter': {'pi': '2021-01-01'}, 'laborious': {}},
'schedule_config': {
'scouter': {'pi': {'held': True}, 'new_held': {'held': True}},
'laborious': {},
'current_schedule_config': {
'scouter': {'sin': '2026-10-05 00:00:00.000+0000'},
'laborious': {
'batch': '2026-10-05 00:00:00.000+0000',
'old': '2026-10-21 00:00:00.000+0000',
},
},
'schedule_config': schedule_config,
**metadata,
}
)
assert result == {
'to_update': {'scouter': {}, 'laborious': {}},
'to_create': {'scouter': {}, 'laborious': {}},
'to_delete': {'scouter': [], 'laborious': []},
assert set(actions['to_update']['scouter']) == {'sin'}
assert set(actions['to_update']['laborious']) == {'batch'}
def test_process_schedules_without_updated_at_redeploys_every_run(formatters):
batch = _batch_pipeline('batch', [{'tag_id': 77}])
del batch['updated_at']
with patch(
'orchestrator.activities.formatters.now',
return_value=MagicMock(strftime=MagicMock(return_value='2026-10-28 12:00:00.000+0000')),
):
result = formatters.process_schedules(
{
'pipelines': [batch],
'connectors': [PLANT_CONNECTOR],
'connector_tags': [
_write_tag(77, 'ns=2;s=P1', updated_at='2026-10-10 00:00:00.000+0000')
],
**metadata,
}
)
assert result['laborious']['batch']['updated_at'] == '2026-10-28 12:00:00.000+0000'
def test_predictions_batch_schedule_matches_the_builder(formatters):
"""process_schedules only adds the resolved opc_output_config to the builder's input"""
batch = _batch_pipeline('batch', [{'tag_id': 77}])
result = formatters.process_schedules(
{
'pipelines': [batch],
'connectors': [PLANT_CONNECTOR],
'connector_tags': [_write_tag(77, 'ns=2;s=P1')],
**metadata,
}
)
resolved = {
**batch,
'opc_output_config': {'plant': {'prediction_tags': {'ns=2;s=P1': {'data_type': 'float'}}}},
}
assert json.dumps(result['laborious']['batch'], sort_keys=True) == json.dumps(
{**predictions_batch(resolved), 'updated_at': batch['updated_at']}, sort_keys=True
)
def test_format_schedule_config_groups_schedules_of_one_namespace(formatters):

View File

@@ -192,6 +192,30 @@ def test_aggregate_documents_in_mongodb_success(mongo_db):
)
def test_aggregate_documents_in_mongodb_formats_only_datetimes(mongo_db):
"""A timestamp field stored as text, or missing, is left as it is"""
mongo_db.mongo_db_repository.aggregate = MagicMock(
return_value=[
{'tag_id': 1, 'updated_at': datetime(2026, 10, 1, 12, 0, 0)},
{'tag_id': 2, 'updated_at': '2026-10-02 00:00:00.000+0000'},
{'tag_id': 3},
]
)
result = mongo_db.aggregate_documents_in_mongodb(
{
'query': {'collection': 'connector_tags', 'aggregation': [{'$match': {}}]},
'timestamp_fields': ['updated_at'],
}
)
assert result == [
{'tag_id': 1, 'updated_at': '2026-10-01 12:00:00.000000+0000'},
{'tag_id': 2, 'updated_at': '2026-10-02 00:00:00.000+0000'},
{'tag_id': 3},
]
def test_aggregate_documents_in_mongodb_failure(mongo_db):
input_data = {
'collection': 'test_collection',

View File

@@ -27,38 +27,37 @@ TAGS_BY_ID = {
}
# PI Web API through the tag database (connectors-pi-web-api)
# Write tags (predictions_batch write_tags through the tag database)
PI_SECRET = {'name': 'pi-web-api-auth-token', 'key': 'token'}
OPC_CONNECTORS = {
'plant': {
'connector_id': 'plant',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
},
}
def pi_connector_doc(connector_id='pi-usina', **pi_web_api):
return {
'connector_id': connector_id,
'name': 'PI Usina',
'url': f'https://{connector_id}.example/piwebapi',
'pi_web_api': {'auth': {'type': 'bearer', 'secret': dict(PI_SECRET)}, **pi_web_api},
}
def pi_tag_doc(tag_id, address, tag_name, category='Read', connector_id='pi-usina'):
def write_tag_doc(tag_id, address, category='Write', connector_id='plant', **extra):
return {
'tag_id': tag_id,
'connector_id': connector_id,
'address': address,
'tag_name': tag_name,
'tag_name': f'tag-{tag_id}',
'category': category,
'status': 'Active',
**extra,
}
def pi_scouter_pipeline_doc(read_tags, **fields):
def batch_pipeline_doc(schedule_name, write_tags, **fields):
return {
'workflow_type': 'pi_web_api_scouter',
'schedule_name': 'pi_schedule',
'workflow_type': 'predictions_batch',
'schedule_name': schedule_name,
'model_id': 'model_id',
'model': {'name': 'model_name'},
'frequency': '1m',
'read_tags': read_tags,
'pi_web_api_config': {'endpoint': '/streamsets/recorded'},
'query': 'SELECT 1',
'write_tags': write_tags,
**fields,
}

View File

@@ -472,3 +472,42 @@ def test_gather_read_tags_server_key_never_collides():
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_gather_read_tags_invalid_cycle_multiple_does_not_fail_the_slot():
"""A bad active_read is reported as invalid instead of failing every pipeline's slot"""
pipelines = [
scouter_pipeline_doc(
'sin',
'm1',
[
legacy_tag('ns=2;s=A', 'a'),
legacy_tag('ns=2;s=B', 'b', active_read={'enabled': True, 'cycle_multiple': 'x'}),
legacy_tag('ns=2;s=C', 'c', active_read={'enabled': True, 'cycle_multiple': None}),
# Not a dict: treated as no active read
legacy_tag('ns=2;s=D', 'd', active_read=True),
],
)
]
entries, issues = _gather(pipelines)
assert [e['address'] for e in entries] == ['ns=2;s=A', 'ns=2;s=D']
assert entries[1]['active_read'] == {'enabled': False}
assert issues['invalid'] == {'sin/ns=2;s=B', 'sin/ns=2;s=C'}
def test_gather_read_tags_skips_incomplete_pipelines():
"""A pipeline without schedule_name, model_id or model.name is reported, not raised"""
no_model = scouter_pipeline_doc('no-model', 'm1', [legacy_tag('ns=2;s=A', 'a')])
del no_model['model']
no_name = scouter_pipeline_doc('no-name', 'm2', [legacy_tag('ns=2;s=B', 'b')])
no_name['model'] = {}
no_model_id = scouter_pipeline_doc('no-id', None, [legacy_tag('ns=2;s=C', 'c')])
no_schedule = scouter_pipeline_doc(None, 'm4', [legacy_tag('ns=2;s=D', 'd')])
ok = scouter_pipeline_doc('ok', 'm5', [legacy_tag('ns=2;s=E', 'e')])
entries, issues = _gather([no_model, no_name, no_model_id, no_schedule, ok])
assert [e['address'] for e in entries] == ['ns=2;s=E']
assert issues['invalid'] == {'no-model', 'no-name', 'no-id', 'None'}

View File

@@ -16,11 +16,6 @@ from orchestrator.utils.pipeline_configs import (
scouter,
simple_metrics,
)
from tests.orchestrator.utils.tag_database_fixtures import (
PI_SECRET,
pi_connector_doc,
pi_scouter_pipeline_doc,
)
def test_common_config():
@@ -362,10 +357,14 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
'model_id': 'test_model_id',
'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
'query': 'test_query',
'write_tags': [
{'server_id': 'test_server_id', 'type': 'prediction', 'addr': 'test_addr'},
{'server_id': 'test_server_id', 'type': 'confidence', 'addr': 'test_addr'},
],
'write_tags': [{'tag_id': 77}, {'tag_id': 78}],
# Resolved from write_tags by process_schedules
'opc_output_config': {
'test_server_id': {
'prediction_tags': {'test_addr': {'data_type': 'float'}},
'confidence_tags': {'test_addr': {'data_type': 'float'}},
}
},
'pi_web_api_output_config': {
'endpoint': 'test_endpoint',
'prediction_tags': {
@@ -685,16 +684,23 @@ def test_frequency_alert_warning(frequency_alert, warns):
assert (frequency_alert_warning(frequency_alert) is not None) is warns
def test_pi_web_api_scouter_legacy_shape():
"""Without a connector the output is exactly the pre-tag-database one"""
config = pi_scouter_pipeline_doc(
{'TI-101': {'webid': 'F1DP-X'}},
pi_web_api_config={'endpoint': '/streamsets/recorded', 'max_count': 5, 'api_timeout': 10},
)
def test_pi_web_api_scouter_caps_api_timeout_at_frequency_only_when_larger():
config = {
'workflow_type': 'pi_web_api_scouter',
'schedule_name': 'pi_schedule',
'model_id': 'model_id',
'model': {'name': 'model_name'},
'frequency': '1m',
'read_tags': {'TI-101': {'webid': 'F1DP-X'}},
'pi_web_api_config': {
'endpoint': '/streamsets/recorded',
'max_count': 5,
'api_timeout': 10,
},
}
result = pi_web_api_scouter(config)
assert 'connection' not in result
assert result['model_tags'] == {
'TI-101': {'webid': 'F1DP-X', 'aggr_func': 'lts', 'data_range': [-100, 100]}
}
@@ -706,73 +712,31 @@ def test_pi_web_api_scouter_legacy_shape():
}
def test_pi_web_api_scouter_connector_defaults_with_pipeline_override():
connector = pi_connector_doc(
period='*-2h', max_count=10, api_timeout=600, data_server_webid='DS1'
)
config = pi_scouter_pipeline_doc(
{'TI-101': {'tag_id': 42, 'webid': 'F1DP-A', 'aggr_func': 'avg'}},
pi_web_api_config={'endpoint': '/streamsets/recorded', 'max_count': 5},
pi_web_api_connector=connector,
)
result = pi_web_api_scouter(config)
assert result['model_tags'] == {
'TI-101': {'webid': 'F1DP-A', 'aggr_func': 'avg', 'data_range': [-100, 100]}
}
assert result['pi_web_api_query'] == {
'endpoint': '/streamsets/recorded',
'period': '*-2h',
'max_count': 5,
# The connector default is still capped at the 1m frequency
'api_timeout': 60,
}
assert result['connection'] == {
'connector_id': 'pi-usina',
'base_url': 'https://pi-usina.example/piwebapi',
'auth': {'type': 'bearer', 'secret': PI_SECRET},
'data_server_webid': 'DS1',
}
def test_pi_web_api_scouter_connection_carries_only_the_secret_reference():
connector = pi_connector_doc()
connector['pi_web_api']['auth']['secret']['value'] = 'leaked-credential'
connector['pi_web_api']['auth']['token'] = 'leaked-credential'
config = pi_scouter_pipeline_doc(
{'TI-101': {'tag_id': 42, 'webid': 'F1DP-A'}}, pi_web_api_connector=connector
)
result = pi_web_api_scouter(config)
assert result['connection']['auth'] == {'type': 'bearer', 'secret': PI_SECRET}
assert 'leaked-credential' not in str(result)
def test_predictions_batch_groups_write_tags_by_server_and_type():
def test_predictions_batch_copies_the_resolved_opc_output_config():
opc_output_config = {'plant': {'prediction_tags': {'ns=2;s=P1': {'data_type': 'int'}}}}
config = {
'workflow_type': 'predictions_batch',
'schedule_name': 'batch',
'model_id': 'model_id',
'model': {'name': 'model_name'},
'query': 'SELECT 1',
'write_tags': [
{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=P1'},
{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=P2', 'data_type': 'int'},
# Neither prediction nor confidence: the server is known, the tag is ignored
{'server_id': '2', 'type': 'other', 'addr': 'ns=2;s=O1'},
],
'write_tags': [{'tag_id': 77}],
'opc_output_config': opc_output_config,
}
result = predictions_batch(config)
assert result['opc_output_config'] == {
'1': {
'prediction_tags': {
'ns=2;s=P1': {'data_type': 'float'},
'ns=2;s=P2': {'data_type': 'int'},
}
},
'2': {},
assert result['opc_output_config'] == opc_output_config
assert 'write_tags' not in result
def test_predictions_batch_without_write_tags_has_no_opc_output():
config = {
'workflow_type': 'predictions_batch',
'schedule_name': 'batch',
'model_id': 'model_id',
'model': {'name': 'model_name'},
'query': 'SELECT 1',
}
assert predictions_batch(config)['opc_output_config'] == {}

View File

@@ -1,21 +1,23 @@
from orchestrator.utils.tag_database import (
WRITE_CATEGORIES,
filter_by_tag_ids,
index_connector_tags,
index_connectors,
index_pi_connectors,
referenced_pi_tag_ids,
index_tag_documents,
latest_updated_at,
referenced_tag_ids,
resolve_pi_output_tags,
resolve_pi_read_tags,
resolve_tag_names,
resolve_write_tags,
tag_documents_updated_at,
tag_id_text,
)
from tests.orchestrator.utils.tag_database_fixtures import (
OPC_CONNECTORS,
TAGS_BY_ID,
batch_pipeline_doc,
legacy_tag,
pi_connector_doc,
pi_scouter_pipeline_doc,
pi_tag_doc,
scouter_pipeline_doc,
write_tag_doc,
)
@@ -97,195 +99,151 @@ def test_index_skips_incomplete_documents():
assert list(index_connectors(connector_list)) == ['3']
def test_index_pi_connectors_leaves_out_incomplete():
complete = pi_connector_doc()
no_url = {**pi_connector_doc('pi-a'), 'url': ''}
no_auth = {**pi_connector_doc('pi-b'), 'pi_web_api': {}}
inline_token = {
**pi_connector_doc('pi-c'),
'pi_web_api': {'auth': {'type': 'bearer', 'token': 'secret-value'}},
}
no_secret_key = {
**pi_connector_doc('pi-d'),
'pi_web_api': {'auth': {'type': 'basic', 'secret': {'name': 'x'}}},
}
index = index_pi_connectors([complete, no_url, no_auth, inline_token, no_secret_key])
assert index == {'pi-usina': complete}
def test_tag_id_text_is_canonical_for_integers():
assert tag_id_text({'tag_id': 7}) == '7'
assert tag_id_text({'tag_id': '007'}) == '7'
assert tag_id_text({'tag_id': ' 12 '}) == '12'
assert tag_id_text({'tag_id': '-3'}) == '-3'
assert tag_id_text({'tag_id': 'abc'}) == 'abc'
assert tag_id_text({'tag_id': ''}) is None
assert tag_id_text({'tag_id': None}) is None
assert tag_id_text({}) is None
def test_referenced_pi_tag_ids():
def test_referenced_tag_ids_never_raises_on_non_ascii_digits():
"""'²'.isdigit() is True but int('²') raises: it must stay text (runs in the workflow)"""
pipelines = [scouter_pipeline_doc('sin', 'm1', [{'tag_id': '²'}, {'tag_id': '007'}])]
assert referenced_tag_ids(pipelines) == [7, '7', '²']
def test_referenced_tag_ids_includes_write_tags():
pipelines = [
pi_scouter_pipeline_doc({'TI-101': {'tag_id': 42}, 'TI-102': {'webid': 'F1DP-X'}}),
{
'workflow_type': 'predictions_batch',
'pi_web_api_output_config': {
'prediction_tags': [{'tag_id': '77'}],
'confidence_tags': {'CONF-1': 'F1DP-C'},
},
},
{
'workflow_type': 'xgboost_predictions_batch',
'pi_web_api_output_config': {'confidence_tags': [{'tag_id': 78}]},
},
# OPC scouter tag_ids belong to the OPC loads
{'workflow_type': 'scouter', 'read_tags': [{'tag_id': 7}]},
{'workflow_type': 'drift'},
scouter_pipeline_doc('sin', 'm1', [{'tag_id': 7}]),
batch_pipeline_doc('pred', [{'tag_id': 77}, {'tag_id': 78}, {'server_id': '1'}]),
batch_pipeline_doc('xgb', [{'tag_id': 79}], workflow_type='xgboost_predictions_batch'),
batch_pipeline_doc('retrain', [{'tag_id': 99}], workflow_type='minimal_retrain'),
]
assert referenced_pi_tag_ids(pipelines) == [42, 77, 78, '42', '77', '78']
assert referenced_tag_ids(pipelines) == [7, 77, 78, 79, '7', '77', '78', '79']
def test_referenced_pi_tag_ids_none():
assert referenced_pi_tag_ids([pi_scouter_pipeline_doc({'TI-101': {'webid': 'F1DP-X'}})]) == []
def test_index_connector_tags_filters_status_and_category():
tags = [
write_tag_doc(1, 'ns=2;s=R', category='Read'),
write_tag_doc(2, 'ns=2;s=W'),
write_tag_doc(3, 'ns=2;s=C', category='Prediction Confidence'),
write_tag_doc(4, 'ns=2;s=I', status='Inactive'),
# An override that projects neither status nor category
{'tag_id': 5, 'connector_id': 'plant', 'address': 'ns=2;s=N', 'tag_name': 'n'},
]
assert list(index_connector_tags(tags)) == ['1', '5']
assert list(index_connector_tags(tags, WRITE_CATEGORIES)) == ['2', '3', '5']
# Every document, whatever its status or category, for the updated_at
assert list(index_tag_documents(tags)) == ['1', '2', '3', '4', '5']
def test_resolve_pi_read_tags_by_tag_id():
connectors = index_pi_connectors([pi_connector_doc()])
tags = index_connector_tags([pi_tag_doc(42, 'F1DP-A', 'TI-101-db')])
def test_index_connectors_leaves_out_inactive():
connectors = [
{**OPC_CONNECTORS['plant'], 'status': 'Active'},
{**OPC_CONNECTORS['plant'], 'connector_id': 'old', 'status': 'Inactive'},
]
resolved, connector, issues = resolve_pi_read_tags(
{'TI-101': {'tag_id': 42, 'aggr_func': 'avg'}}, tags, connectors
)
# The key stays the model tag name; only the WebID comes from the tag database
assert resolved == {'TI-101': {'tag_id': 42, 'aggr_func': 'avg', 'webid': 'F1DP-A'}}
assert connector == connectors['pi-usina']
assert not any(issues.values())
assert list(index_connectors(connectors)) == ['plant']
def test_resolve_pi_read_tags_legacy_untouched():
legacy = {'TI-101': {'webid': 'F1DP-X'}}
resolved, connector, issues = resolve_pi_read_tags(legacy, {}, {})
assert resolved == legacy
assert resolved['TI-101'] is legacy['TI-101']
assert connector is None
assert not any(issues.values())
def test_resolve_pi_read_tags_issues():
connectors = index_pi_connectors([pi_connector_doc()])
def test_resolve_write_tags():
tags = index_connector_tags(
[
pi_tag_doc(42, 'F1DP-A', 'TI-101'),
pi_tag_doc(43, 'F1DP-W', 'PRED-1', category='Write'),
pi_tag_doc(44, 'F1DP-O', 'TI-OFF', connector_id='pi-inactive'),
]
write_tag_doc(77, 'ns=2;s=P1'),
write_tag_doc(78, 'ns=2;s=C1', category='Prediction Confidence', data_type='double'),
write_tag_doc(79, 'ns=2;s=P2', connector_id='other', data_type='int'),
],
WRITE_CATEGORIES,
)
connectors = {**OPC_CONNECTORS, 'other': {**OPC_CONNECTORS['plant'], 'connector_id': 'other'}}
output_config, issues = resolve_write_tags(
[{'tag_id': 77}, {'tag_id': '78'}, {'tag_id': 79}], tags, connectors
)
resolved, connector, issues = resolve_pi_read_tags(
{
'TI-101': {'tag_id': 42},
'UNKNOWN': {'tag_id': 99},
'WRITE': {'tag_id': 43},
'OFF': {'tag_id': 44},
'LEGACY': {'webid': 'F1DP-X'},
assert output_config == {
'plant': {
'prediction_tags': {'ns=2;s=P1': {'data_type': 'float'}},
'confidence_tags': {'ns=2;s=C1': {'data_type': 'double'}},
},
tags,
connectors,
)
# Unresolvable tags are left out; the rest of the pipeline is still scheduled
assert set(resolved) == {'TI-101', 'LEGACY'}
assert connector == connectors['pi-usina']
assert issues == {
'tags': {'99'},
'categories': {'43'},
'connectors': {'pi-inactive'},
'mixed': set(),
}
def test_resolve_pi_read_tags_mixed_connectors():
connectors = index_pi_connectors([pi_connector_doc('pi-a'), pi_connector_doc('pi-b')])
tags = index_connector_tags(
[
pi_tag_doc(1, 'F1DP-1', 'A', connector_id='pi-a'),
pi_tag_doc(2, 'F1DP-2', 'B', connector_id='pi-b'),
]
)
_resolved, connector, issues = resolve_pi_read_tags(
{'A': {'tag_id': 1}, 'B': {'tag_id': 2}}, tags, connectors
)
assert connector is None
assert issues['mixed'] == {'pi-a, pi-b'}
def test_resolve_pi_output_tags_by_tag_id():
connectors = index_pi_connectors([pi_connector_doc()])
tags = index_connector_tags(
[
pi_tag_doc(77, 'F1DP-P', 'PRED-101', category='Write'),
pi_tag_doc(78, 'F1DP-C', 'CONF-101', category='Prediction Confidence'),
]
)
resolved, issues = resolve_pi_output_tags(
{
'prediction_tags': [{'tag_id': 77}],
'confidence_tags': [{'tag_id': 78}],
'other': 'kept',
},
tags,
connectors,
)
assert resolved == {
'prediction_tags': {'PRED-101': 'F1DP-P'},
'confidence_tags': {'CONF-101': 'F1DP-C'},
'other': 'kept',
'other': {'prediction_tags': {'ns=2;s=P2': {'data_type': 'int'}}},
}
assert not any(issues.values())
def test_resolve_pi_output_tags_legacy_untouched():
legacy = {'prediction_tags': {'PRED-101': 'F1DP-P'}, 'confidence_tags': {}}
resolved, issues = resolve_pi_output_tags(legacy, {}, {})
assert resolved == legacy
assert resolved['prediction_tags'] is legacy['prediction_tags']
assert not any(issues.values())
def test_resolve_pi_output_tags_issues():
connectors = index_pi_connectors([pi_connector_doc()])
def test_resolve_write_tags_issues():
tags = index_connector_tags(
[
pi_tag_doc(42, 'F1DP-R', 'TI-101', category='Read'),
pi_tag_doc(77, 'F1DP-P', 'PRED-101', category='Write'),
write_tag_doc(1, 'ns=2;s=R', category='Read'),
write_tag_doc(2, 'ns=2;s=X', connector_id='gone'),
write_tag_doc(3, 'ns=2;s=B', data_type='Double'),
write_tag_doc(4, 'ns=2;s=I', status='Inactive'),
# No category: a write tag cannot tell prediction from confidence
{'tag_id': 5, 'connector_id': 'plant', 'address': 'ns=2;s=N', 'tag_name': 'n'},
write_tag_doc(6, 'ns=2;s=OK'),
],
WRITE_CATEGORIES,
)
write_tags = [
{'tag_id': 1},
{'tag_id': 2},
{'tag_id': 3},
{'tag_id': 4},
{'tag_id': 5},
{'tag_id': 404},
# Legacy entries are no longer accepted
{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=L'},
{'tag_id': None},
'ns=2;s=raw',
{'tag_id': 6},
]
output_config, issues = resolve_write_tags(write_tags, tags, OPC_CONNECTORS)
assert output_config == {'plant': {'prediction_tags': {'ns=2;s=OK': {'data_type': 'float'}}}}
assert issues['tags'] == {'1', '4', '5', '404'}
assert issues['connectors'] == {'gone'}
assert issues['invalid'] == {
'3',
"{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=L'}",
"{'tag_id': None}",
'ns=2;s=raw',
}
def test_latest_updated_at():
assert latest_updated_at(
'2026-10-01 00:00:00.000+0000', None, '', '2026-10-02 00:00:00.000+0000'
) == ('2026-10-02 00:00:00.000+0000')
assert latest_updated_at(None, '') is None
def test_tag_documents_updated_at():
documents = index_tag_documents(
[
write_tag_doc(
7, 'ns=2;s=R', category='Read', updated_at='2026-10-05 00:00:00.000+0000'
),
# Inactive documents count: deactivating a tag redeploys its pipelines
write_tag_doc(
77, 'ns=2;s=P', status='Inactive', updated_at='2026-10-09 00:00:00.000+0000'
),
write_tag_doc(78, 'ns=2;s=C', updated_at='2026-10-03 00:00:00.000+0000'),
]
)
resolved, issues = resolve_pi_output_tags(
{
# A Read tag as prediction, a Write tag as confidence, an unknown tag_id
'prediction_tags': [{'tag_id': 42}, {'tag_id': 99}, {'PRED-LEGACY': 'F1DP-L'}],
'confidence_tags': [{'tag_id': 77}],
},
tags,
connectors,
scouter_pipeline = scouter_pipeline_doc(
'sin', 'm1', [{'tag_id': 7}, legacy_tag('ns=2;s=A', 'a')]
)
batch_pipeline = batch_pipeline_doc('pred', [{'tag_id': '77'}, {'tag_id': 78}, {'tag_id': 404}])
assert resolved['prediction_tags'] == {'PRED-LEGACY': 'F1DP-L'}
assert resolved['confidence_tags'] == {}
assert issues['tags'] == {'99'}
assert issues['categories'] == {'42', '77'}
def test_referenced_pi_tag_ids_ignores_malformed_read_tags():
assert referenced_pi_tag_ids([pi_scouter_pipeline_doc([{'tag_id': 42}])]) == []
def test_resolve_pi_output_tags_ignores_non_dict_entries():
resolved, issues = resolve_pi_output_tags(
{'prediction_tags': ['F1DP-P', None, {'PRED-101': 'F1DP-P'}]}, {}, {}
)
assert resolved == {'prediction_tags': {'PRED-101': 'F1DP-P'}}
assert not any(issues.values())
assert tag_documents_updated_at(scouter_pipeline, documents) == '2026-10-05 00:00:00.000+0000'
assert tag_documents_updated_at(batch_pipeline, documents) == '2026-10-09 00:00:00.000+0000'
assert tag_documents_updated_at(batch_pipeline_doc('none', []), documents) is None

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -6,9 +6,6 @@ from orchestrator.activities.activities import Activities
from orchestrator.workflows.orchestrator import (
DEFAULT_CONNECTOR_TAGS_QUERY,
DEFAULT_CONNECTORS_QUERY,
DEFAULT_PI_CONNECTOR_TAGS_QUERY,
DEFAULT_PI_CONNECTORS_QUERY,
PI_WEB_API_PATCH,
TAG_DATABASE_PATCH,
Orchestrator,
)
@@ -119,10 +116,7 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
]
)
assert workflow_mock.patched.call_args_list == [
call(TAG_DATABASE_PATCH),
call(PI_WEB_API_PATCH),
]
assert workflow_mock.patched.call_args_list == [call(TAG_DATABASE_PATCH)]
referenced_tag_ids_mock.assert_called_once_with(
workflow_mock.start_local_activity_method.return_value
)
@@ -131,7 +125,11 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
[
call(
Activities.aggregate_documents_in_mongodb,
{**metadata, 'query': DEFAULT_CONNECTORS_QUERY},
{
**metadata,
'query': DEFAULT_CONNECTORS_QUERY,
'timestamp_fields': ['updated_at'],
},
retry_policy=ANY,
start_to_close_timeout=ANY,
),
@@ -146,6 +144,7 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
*DEFAULT_CONNECTOR_TAGS_QUERY['aggregation'],
],
},
'timestamp_fields': ['updated_at'],
},
retry_policy=ANY,
start_to_close_timeout=ANY,
@@ -168,10 +167,8 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator):
{
**metadata,
'pipelines': 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,
# No PI Web API tag with tag_id: nothing loaded
'pi_connectors': [],
'pi_connector_tags': [],
},
retry_policy=ANY,
start_to_close_timeout=ANY,
@@ -475,10 +472,7 @@ async def test_run_without_tag_ids_skips_tag_database(workflow_mock, _ids, orche
await orchestrator.run(input_data)
assert workflow_mock.patched.call_args_list == [
call(TAG_DATABASE_PATCH),
call(PI_WEB_API_PATCH),
]
assert workflow_mock.patched.call_args_list == [call(TAG_DATABASE_PATCH)]
assert _tag_database_calls(workflow_mock) == []
slots_input = _process_slots_input(workflow_mock)
assert slots_input['connectors'] == []
@@ -513,25 +507,20 @@ def _process_schedules_input(workflow_mock):
)
def _pi_tag_database_calls(workflow_mock):
"""The PI Web API loads: the ones whose aggregation is not the OPC one"""
return [
c
for c in _tag_database_calls(workflow_mock)
if c.args[1]['query']['aggregation'][-1]
in (
DEFAULT_PI_CONNECTORS_QUERY['aggregation'][-1],
DEFAULT_PI_CONNECTOR_TAGS_QUERY['aggregation'][-1],
)
]
def test_default_connector_tags_query_loads_every_category_and_status():
"""The status / category filters are applied in the code, so changes still redeploy"""
stages = DEFAULT_CONNECTOR_TAGS_QUERY['aggregation']
assert all('$match' not in stage for stage in stages)
projected = stages[-1]['$project']
for field in ('category', 'data_type', 'status', 'updated_at'):
assert projected[field] == 1
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.referenced_pi_tag_ids', return_value=[42, '42'])
@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[])
@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[77])
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run_loads_pi_tag_database(workflow_mock, _ids, _pi_ids, orchestrator):
"""A new run loads the PI connectors and tags and hands them to process_schedules only"""
async def test_run_hands_the_tag_database_to_process_schedules(workflow_mock, _ids, orchestrator):
"""The write tags of process_schedules need the connectors as well as the tags"""
workflow_mock.patched = MagicMock(return_value=True)
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
@@ -541,97 +530,13 @@ async def test_run_loads_pi_tag_database(workflow_mock, _ids, _pi_ids, orchestra
await orchestrator.run(input_data)
assert workflow_mock.patched.call_args_list == [
call(TAG_DATABASE_PATCH),
call(PI_WEB_API_PATCH),
]
queries = [c.args[1]['query'] for c in _tag_database_calls(workflow_mock)]
assert queries == [
DEFAULT_PI_CONNECTORS_QUERY,
{
**DEFAULT_PI_CONNECTOR_TAGS_QUERY,
'aggregation': [
{'$match': {'tag_id': {'$in': [42, '42']}}},
*DEFAULT_PI_CONNECTOR_TAGS_QUERY['aggregation'],
],
},
]
for c in _tag_database_calls(workflow_mock):
assert c.args[1]['query']['aggregation'] is not DEFAULT_PI_CONNECTORS_QUERY['aggregation']
loaded = workflow_mock.start_local_activity_method.return_value
schedules_input = _process_schedules_input(workflow_mock)
assert schedules_input['pi_connectors'] is loaded
assert schedules_input['pi_connector_tags'] is loaded
# process_slots input unchanged: no PI key, OPC loads skipped (no OPC tag_id)
slots_input = _process_slots_input(workflow_mock)
assert 'pi_connectors' not in slots_input
assert 'pi_connector_tags' not in slots_input
assert slots_input['connectors'] == []
assert slots_input['connector_tags'] == []
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.referenced_pi_tag_ids', return_value=[42])
@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[])
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run_with_pi_tag_database_queries_from_input(
workflow_mock, _ids, _pi_ids, orchestrator
):
"""The input overrides the default PI Web API aggregations"""
workflow_mock.patched = MagicMock(return_value=True)
pi_connectors_query = {
'collection': 'connectors',
'aggregation': [{'$match': {'type': 'pi_web_api'}}],
}
pi_connector_tags_query = {
'collection': 'connector_tags',
'aggregation': [{'$match': {'status': 'Active'}}],
}
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
'opc_servers_query': 'SELECT * FROM servers',
'pi_connectors_query': pi_connectors_query,
'pi_connector_tags_query': pi_connector_tags_query,
'schedule_name': 'test-schedule-name',
}
await orchestrator.run(input_data)
queries = [c.args[1]['query'] for c in _tag_database_calls(workflow_mock)]
assert queries == [
pi_connectors_query,
{
'collection': 'connector_tags',
'aggregation': [
{'$match': {'tag_id': {'$in': [42]}}},
{'$match': {'status': 'Active'}},
],
},
]
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.referenced_pi_tag_ids', return_value=[42])
@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[7])
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run_replay_before_pi_web_api_patch(workflow_mock, _ids, _pi_ids, orchestrator):
"""A run started after the tag database patch but before the PI one replays without PI loads"""
workflow_mock.patched = MagicMock(side_effect=lambda patch_id: patch_id == TAG_DATABASE_PATCH)
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
'opc_servers_query': 'SELECT * FROM servers',
'schedule_name': 'test-schedule-name',
}
await orchestrator.run(input_data)
assert _pi_tag_database_calls(workflow_mock) == []
assert len(_tag_database_calls(workflow_mock)) == 2
schedules_input = _process_schedules_input(workflow_mock)
assert schedules_input['pi_connectors'] == []
assert schedules_input['pi_connector_tags'] == []
assert schedules_input['connectors'] is loaded
assert schedules_input['connector_tags'] is loaded
assert 'pi_connectors' not in schedules_input
for c in _tag_database_calls(workflow_mock):
assert c.args[1]['timestamp_fields'] == ['updated_at']
@mark.asyncio

View File

@@ -0,0 +1,33 @@
"""
Replay real Orchestrator histories against the current workflow code.
The histories were recorded on the e2e stack (real Temporal, Mongo and Redis):
- orchestrator_before_tag_database.json: main before SIENTIAPDE-1974 (no patch marker,
no tag database load), as a run in flight during the deploy would replay;
- orchestrator_tag_database.json: this version, with the SIENTIAPDE-1974-tag-database
patch and the connectors / connector_tags loads of read and write tags by tag_id.
A NondeterminismError here means the change breaks runs started on those versions.
"""
from pathlib import Path
from pytest import mark
from temporalio.client import WorkflowHistory
from temporalio.worker import Replayer
from orchestrator.workflows.orchestrator import Orchestrator
HISTORIES = Path(__file__).parent / 'histories'
@mark.asyncio
@mark.parametrize(
'history_file', ['orchestrator_before_tag_database.json', 'orchestrator_tag_database.json']
)
async def test_replay_recorded_history(history_file):
history = WorkflowHistory.from_json(
'orchestrator-history', (HISTORIES / history_file).read_text()
)
await Replayer(workflows=[Orchestrator]).replay_workflow(history)