From a36d9912ac8a6bea0677fba56dc0b0681a3cdb90 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 29 Oct 2026 09:36:17 -0300 Subject: [PATCH] SIENTIAPDE-1974 SIENTIAPDE-1974-connectors-pi-web-api: Introduce config_hash for robust schedule redeployment. Pipelines now redeploy on any content change (e.g., tag database updates) even without an updated_at bump. orchestrated_schedules tracks the hash; updated documentation, E2E tests, and the SIENTIAPDE-1974 backfill script. --- CLAUDE.md | 3 +- README.md | 14 +- e2e/scenarios.md | 9 +- e2e/test_orchestrator_main_workflow.py | 45 +++- orchestrator/activities/formatters.py | 83 ++++-- orchestrator/activities/mongo_db.py | 45 ++-- orchestrator/activities/temporal_manager.py | 16 +- orchestrator/utils/tag_database.py | 37 +-- orchestrator/workflows/orchestrator.py | 8 +- .../sientiapde_1974_backfill_write_tags.js | 75 +++++- .../activities/test_formatters.py | 248 ++++++++++++------ .../orchestrator/activities/test_mongo_db.py | 69 +++-- .../activities/test_temporal_manager.py | 11 + tests/orchestrator/utils/test_tag_database.py | 33 --- .../workflows/test_orchestrator.py | 17 +- 15 files changed, 450 insertions(+), 263 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index f58bfe2..390409b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,7 +47,8 @@ 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` (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. +- **`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`; tag changes reach the schedules through `config_hash`, see below); `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. +- **Schedule redeploys** — `orchestrated_schedules` stores, per schedule, the last deploy time (`updated_at`) and the `config_hash` it was deployed with (`formatters.schedule_config_hash`: SHA-256 of the built schedule without `updated_at`). `compare_config_timestamps` redeploys when the pipeline's `updated_at` is newer (manual force) or the hash differs (any content change, including tag database edits/deletes made without bumping `updated_at`). `create_schedules`/`update_schedules` pop `config_hash` before sending the schedule to Temporal and return it in their report for `create_pipelines_timestamps`/`update_pipelines_timestamps`. ### Temporal-specific conventions (all workflow/activity files follow this) - Every workflow and activity-consuming module wraps non-deterministic/side-effecting imports in `with workflow.unsafe.imports_passed_through(): ...` right after `from temporalio import workflow`, because Temporal's sandbox restricts what workflow code can import directly. When adding a new workflow file, follow the existing pattern (see `orchestrator/workflows/orchestrator.py`, `orchestrator/activities/activities.py`) rather than importing at module top level. diff --git a/README.md b/README.md index 01219b9..c7b2fc6 100644 --- a/README.md +++ b/README.md @@ -224,15 +224,21 @@ The legacy write tags (`server_id`, `type`, `addr`, `data_type` in the pipeline) - `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 +#### Redeploy on changes (`updated_at` and `config_hash`) -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. +`orchestrated_schedules` keeps, per schedule, when it was last deployed (`updated_at`) and the `config_hash` it was deployed with: a SHA-256 of the schedule as the Orchestrator builds it (the workflow input, frequency and offset, with tags and connectors already resolved), without `updated_at`. A schedule is redeployed when either: + +- the pipeline's `updated_at` is newer than its last deploy. Bumping `updated_at` by hand forces a redeploy even when nothing else changed; or +- its `config_hash` differs from the stored one. Any change that reaches the schedule redeploys it, whether or not someone bumped an `updated_at`: a pipeline edited by hand, a tag renamed, moved, retyped, deactivated or deleted in `connector_tags`, a connector changed, or a new Orchestrator version that builds the schedule differently. + +A change that does not reach the schedule (e.g. a tag's `description`) does not redeploy. The slot is compared by content and follows on its own. A pipeline without `updated_at` is still redeployed on every run. Schedules deployed before `config_hash` existed have none stored, so the first run of this version redeploys all of them once and records it. #### 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. +2. Paste `scripts/sientiapde_1974_backfill_write_tags.js` into the MongoDB Compass shell (or mongosh), with `ACTION = 'backfill'`: first with `DRY_RUN = true` (the default) to see the plan, then with `DRY_RUN = false`. 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`. `ACTION = 'revert'` undoes it. +3. Deploy this version. Its first run redeploys every schedule once, to record their `config_hash` (see above). +4. Once the write tags are confirmed, run the script with `ACTION = 'cleanup'` to remove `legacy_write_tags`. After that, `revert` can no longer restore the pipelines. #### Architecture diff --git a/e2e/scenarios.md b/e2e/scenarios.md index 7c5e2a0..d60b6b0 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -83,9 +83,12 @@ from `connector_tags` (`float` when absent), `Write` → `prediction_tags`, `Pre #### 1.9.2 A tag change redeploys its pipelines -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. +A second run with nothing changed redeploys nothing (`orchestrated_schedules` is not stamped +again). Then renaming the read tag, moving the write tag and deleting the confidence tag in +`connector_tags` by hand (no `updated_at` bumped anywhere, pipelines untouched) → the next run +redeploys both through their `config_hash`: the scouter `model_tags` and the slot `names` take +the new name, the `opc_output_config` the new address without the confidence tag, and +`orchestrated_schedules` records the new `config_hash`. #### 1.9.3 OPC slot from the tag database diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index 2f00f8f..9db20dc 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -402,6 +402,21 @@ def _opc_slots(redis_client: Redis) -> dict[str, dict]: } +def _orchestrated_schedules(mongo_uri: str) -> dict[str, dict]: + """orchestrated_schedules by schedule_name: updated_at (last deploy) and config_hash.""" + mongo = MongoClient(mongo_uri) + try: + return { + doc['schedule_name']: { + 'updated_at': doc['updated_at'], + 'config_hash': doc.get('config_hash'), + } + for doc in mongo[E2E_DATABASE]['orchestrated_schedules'].find() + } + finally: + mongo.close() + + @pytest.mark.e2e @pytest.mark.asyncio async def test_scenario_1_9_1_write_tags_through_tag_database( @@ -452,7 +467,10 @@ async def test_scenario_1_9_2_tag_change_redeploys_pipelines( temporal_host: str, redis_client: Redis, ): - """1.9.2: changing a tag in connector_tags redeploys the pipelines that reference it.""" + """ + 1.9.2: changing or deleting a tag in connector_tags, without touching any updated_at, + redeploys the pipelines that reference it (config_hash); an unchanged run does not. + """ _apply_orchestrator_seeds( load_scenario_input('orchestrator_tag_database'), mongo_uri, redis_client ) @@ -460,16 +478,21 @@ async def test_scenario_1_9_2_tag_change_redeploys_pipelines( scouter = await _schedule_input(temporal_host, 'opc-tagdb', namespace='scouter') assert set(scouter['model_tags']) == {'Pressure'} + deployed = _orchestrated_schedules(mongo_uri) + assert all(doc.get('config_hash') for doc in deployed.values()), deployed - # Rename the read tag and move the write tag, without touching the pipelines - changed_at = datetime.now(UTC) + timedelta(seconds=1) + # Nothing changed: no redeploy, so orchestrated_schedules is not stamped again + await _run_tag_database_scenario(temporal_env, 'unchanged') + assert _orchestrated_schedules(mongo_uri) == deployed + + # As an edit by hand in Compass: rename the read tag, move a write tag and delete the + # confidence tag, with no updated_at bumped anywhere 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}} - ) + tags.update_one({'tag_id': 7}, {'$set': {'tag_name': 'Pressao'}}) + tags.update_one({'tag_id': 101}, {'$set': {'address': 'ns=3;i=3001'}}) + tags.delete_one({'tag_id': 102}) finally: mongo.close() @@ -478,9 +501,13 @@ async def test_scenario_1_9_2_tag_change_redeploys_pipelines( 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'} + assert pred['opc_output_config'] == { + 'plant': {'prediction_tags': {'ns=3;i=3001': {'data_type': 'double'}}} } + redeployed = _orchestrated_schedules(mongo_uri) + for name in ('opc-tagdb', 'pred-tagdb'): + assert redeployed[name]['config_hash'] != deployed[name]['config_hash'] + assert redeployed[name]['updated_at'] > deployed[name]['updated_at'] # 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'] diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 164e972..b8c00fc 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -1,6 +1,7 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): + import hashlib import json from collections.abc import Callable from logging import Logger @@ -33,15 +34,15 @@ with workflow.unsafe.imports_passed_through(): WRITE_WORKFLOW_TYPES, index_connector_tags, index_connectors, - index_tag_documents, - latest_updated_at, resolve_tag_names, resolve_write_tags, - tag_documents_updated_at, ) topic_separator = '\n ========== \n' +# Schedule fields left out of config_hash: updated_at is compared on its own +CONFIG_HASH_EXCLUDED_FIELDS = ('updated_at', 'config_hash') + # Read tags left out of the slots, by gather_read_tags issue kind SLOT_ISSUES = ( ( @@ -168,6 +169,20 @@ Each entry specifies: """ +def schedule_config_hash(schedule: dict[str, Any]) -> str: + """ + Fingerprint of a schedule as process_schedules builds it, without updated_at. + + Stored in orchestrated_schedules when the schedule is deployed: any change to what + reaches Temporal (a pipeline edited without bumping updated_at, a tag renamed, moved, + retyped, deactivated or deleted, a connector changed, or a new Orchestrator version + building it differently) changes the hash and redeploys the schedule. + """ + content = {k: v for k, v in schedule.items() if k not in CONFIG_HASH_EXCLUDED_FIELDS} + serialized = json.dumps(content, sort_keys=True, separators=(',', ':'), default=str) + return hashlib.sha256(serialized.encode()).hexdigest() + + class Formatters(SientiaMonitoring): """ Schedule and slot configuration formatting and notification filtering activity. @@ -248,8 +263,8 @@ class Formatters(SientiaMonitoring): 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. + out and notified once per run. A change to a tag reaches the schedule through + create_schedule_config, which compares the resolved schedule (config_hash). Returns: - dict[str, Any]: The schedule configuration dictionary keyed by namespace @@ -263,7 +278,6 @@ class Formatters(SientiaMonitoring): 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]] = {} @@ -298,13 +312,11 @@ class Formatters(SientiaMonitoring): _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': latest_updated_at( - updated_at, tag_documents_updated_at(pipeline, tag_documents) - ), + # 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), } self._notify_issues(write_issues, WRITE_TAG_ISSUES, metadata) @@ -401,6 +413,8 @@ class Formatters(SientiaMonitoring): Transforms a list of schedule configurations into a nested dictionary structure organized by namespace and schedule name for efficient lookup and comparison. + Each schedule keeps when it was last deployed (updated_at) and the config_hash + it was deployed with (None for schedules deployed before config_hash existed). Args: input_data (dict[str, Any]): The input data containing: @@ -408,7 +422,7 @@ class Formatters(SientiaMonitoring): - metadata (dict): Metadata for logging purposes Returns: - dict[str, Any]: The formatted schedule config organized by namespace and schedule name + dict[str, Any]: {namespace: {schedule_name: {'updated_at', 'config_hash'}}} """ metadata = input_data['metadata'] @@ -416,16 +430,18 @@ class Formatters(SientiaMonitoring): schedule_config = input_data['schedule_config'] - config: dict[str, dict[str, str]] = {} + config: dict[str, dict[str, dict[str, Any]]] = {} for schedule in schedule_config: namespace = schedule['namespace'] schedule_name = schedule['schedule_name'] - updated_at = schedule['updated_at'] if namespace not in config: config[namespace] = {} - config[namespace][schedule_name] = updated_at + config[namespace][schedule_name] = { + 'updated_at': schedule['updated_at'], + 'config_hash': schedule.get('config_hash'), + } self.info('Formatted schedule config', metadata=metadata) self.debug(json.dumps(config, indent=4, sort_keys=True), metadata=metadata) @@ -444,32 +460,46 @@ class Formatters(SientiaMonitoring): """ Compare new and current schedules to determine which should be updated or created. - Compares schedule timestamps to identify schedules that need updating (newer timestamp) - or creating (schedule doesn't exist). Results are accumulated in the provided dictionaries. + A deployed schedule is updated when its updated_at is newer than its last deploy + (bumping updated_at forces a redeploy) or when its config_hash differs from the one + it was deployed with (any change to the resolved schedule, including tag database + changes and deleted tags). A schedule not deployed yet is created. Each schedule + sent to update or create carries its config_hash, stored once it is deployed. + Results are accumulated in the provided dictionaries. Args: schedules (dict[str, Any]): New schedules to compare - current_schedules (dict[str, Any]): Existing schedules to compare against + current_schedules (dict[str, Any]): Existing schedules to compare against, as + format_schedule_config gives them to_update (dict[str, Any]): Output accumulator for schedules that need updating to_create (dict[str, Any]): Output accumulator for schedules that need creating namespace (str): Namespace for the schedules being compared metadata (dict[str, Any]): Metadata for logging purposes """ for schedule_name, schedule in schedules.items(): + config_hash = schedule_config_hash(schedule) + deployed = {**schedule, 'config_hash': config_hash} + if schedule_name in current_schedules: update_timestamp = schedule.get('updated_at', now()) - old_timestamp = current_schedules[schedule_name] + current = current_schedules[schedule_name] + if not isinstance(current, dict): + # A run started before config_hash replays format_schedule_config's + # result, which held only the timestamp + current = {'updated_at': current, 'config_hash': None} + old_timestamp = current['updated_at'] self.debug( - f'Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}', + f'Comparing schedule {schedule_name}:{update_timestamp} vs {old_timestamp}' + f' (config_hash {config_hash} vs {current.get("config_hash")})', metadata=metadata, ) - if update_timestamp > old_timestamp: - to_update[namespace][schedule_name] = schedule + if update_timestamp > old_timestamp or config_hash != current.get('config_hash'): + to_update[namespace][schedule_name] = deployed else: - to_create[namespace][schedule_name] = schedule + to_create[namespace][schedule_name] = deployed @activity.defn(name='create_schedule_config') def create_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: @@ -478,7 +508,8 @@ class Formatters(SientiaMonitoring): Compares the current schedule configuration in Temporal with the new schedule configuration to determine which schedules need to be created, updated, or deleted. - Uses timestamp comparison to identify schedules that have changed. + Uses the updated_at timestamps and the config_hash (compare_config_timestamps) to + identify schedules that have changed. Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored), and a DRIFT_CHUNK_PERIOD_INVALID WARNING for drift schedules whose configured @@ -523,8 +554,8 @@ class Formatters(SientiaMonitoring): schedules, current_schedules, to_update, to_create, namespace, metadata ) - # Fires whenever the schedule is created or updated (each `updated_at` change, - # including escalation flips). + # Fires whenever the schedule is created or updated (each `updated_at` or + # config_hash change, including escalation flips). changed = {**to_create[self.laborious_namespace], **to_update[self.laborious_namespace]} for schedule_name, schedule in changed.items(): warning = frequency_alert_warning(schedule.get('frequency_alert')) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 4da0074..2197c84 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -224,12 +224,14 @@ class MongoDB(SientiaMonitoring): """ Update `updated_at` timestamps for successfully updated pipelines. - Updates the `updated_at` field in the `orchestrated_schedules` collection - for all pipelines that were successfully updated in Temporal. + Updates the `updated_at` and `config_hash` fields in the `orchestrated_schedules` + collection for all pipelines that were successfully updated in Temporal, so the + next run compares against what was deployed. Args: input_data (dict[str, Any]): Input data containing: - - updated_pipelines (list[dict]): Pipelines with success flags to consider + - updated_pipelines (list[dict]): Pipelines with success flags (and + config_hash) to consider - metadata (dict): Metadata for logging purposes """ updated_pipelines = input_data.get('updated_pipelines', []) @@ -238,22 +240,30 @@ class MongoDB(SientiaMonitoring): self.info('Updating pipelines timestamps...', metadata=metadata) - argument = [ - {'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']} - for pipeline in updated_pipelines - if pipeline['success'] - ] - if not argument: + succeeded = [pipeline for pipeline in updated_pipelines if pipeline['success']] + if not succeeded: # Every update failed: stamp nothing so they are retried next tick (same guard as - # create_pipelines_timestamps). An empty filter would match every tracked schedule. + # create_pipelines_timestamps). self.info('No successful pipeline updates to timestamp', metadata=metadata) return - data_filter = {'$or': argument} try: - self.mongo_db_repository.update_many( - 'orchestrated_schedules', data_filter, {'$set': {'updated_at': date_now}}, metadata - ) + # One update per schedule: each has its own config_hash + for pipeline in succeeded: + self.mongo_db_repository.update_one( + 'orchestrated_schedules', + { + 'schedule_name': pipeline['schedule_name'], + 'namespace': pipeline['namespace'], + }, + { + '$set': { + 'updated_at': date_now, + 'config_hash': pipeline.get('config_hash'), + } + }, + metadata, + ) except Exception as e: trace = traceback.format_exc() self.send_notification( @@ -278,11 +288,13 @@ class MongoDB(SientiaMonitoring): Insert `updated_at` timestamps for newly created pipelines. Inserts new documents into the `orchestrated_schedules` collection - for all pipelines that were successfully created in Temporal. + for all pipelines that were successfully created in Temporal, with the + `config_hash` they were created with. Args: input_data (dict[str, Any]): Input data containing: - - created_pipelines (list[dict]): Pipelines with success flags to consider + - created_pipelines (list[dict]): Pipelines with success flags (and + config_hash) to consider - metadata (dict): Metadata for logging purposes """ created_pipelines = input_data.get('created_pipelines', []) @@ -297,6 +309,7 @@ class MongoDB(SientiaMonitoring): 'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace'], 'updated_at': date_now, + 'config_hash': pipeline.get('config_hash'), } for pipeline in created_pipelines if pipeline['success'] diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index ec622e4..ea2ce28 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -177,7 +177,9 @@ class TemporalManager(SientiaMonitoring): - schedules (dict[str, Any]): The schedules to create. Returns: - - list[dict[str, Any]]: Report entries for each attempted schedule creation. + - list[dict[str, Any]]: Report entries for each attempted schedule creation, + with the schedule's config_hash (create_schedule_config) for + create_pipelines_timestamps. """ schedules_to_create = input_data['schedules'] @@ -198,6 +200,8 @@ class TemporalManager(SientiaMonitoring): ) for schedule_name, schedule in schedules.items(): + # Stored in orchestrated_schedules, not sent to the workflow + config_hash = schedule.pop('config_hash', None) search_attributes = TypedSearchAttributes( [ SearchAttributePair(key=self.model_id_id_key, value=schedule['model_id']), @@ -258,6 +262,7 @@ class TemporalManager(SientiaMonitoring): 'schedule_name': schedule_name, 'success': True, 'message': 'Schedule created successfully', + 'config_hash': config_hash, } ) @@ -272,6 +277,7 @@ class TemporalManager(SientiaMonitoring): 'schedule_name': schedule_name, 'success': False, 'message': str(e), + 'config_hash': config_hash, } ) @@ -360,7 +366,9 @@ class TemporalManager(SientiaMonitoring): - schedules (dict[str, Any]): The schedules to update. Returns: - - list[dict[str, Any]]: Report entries for each attempted schedule update. + - list[dict[str, Any]]: Report entries for each attempted schedule update, + with the schedule's config_hash (create_schedule_config) for + update_pipelines_timestamps. """ schedules_to_update = input_data['schedules'] @@ -380,6 +388,8 @@ class TemporalManager(SientiaMonitoring): ) for schedule_name, schedule in schedules.items(): + # Stored in orchestrated_schedules, not sent to the workflow + config_hash = schedule.pop('config_hash', None) try: await self._update_single_schedule(client, schedule_name, schedule, metadata) @@ -389,6 +399,7 @@ class TemporalManager(SientiaMonitoring): 'schedule_name': schedule_name, 'success': True, 'message': 'Schedule updated successfully', + 'config_hash': config_hash, } ) @@ -403,6 +414,7 @@ class TemporalManager(SientiaMonitoring): 'schedule_name': schedule_name, 'success': False, 'message': str(e), + 'config_hash': config_hash, } ) diff --git a/orchestrator/utils/tag_database.py b/orchestrator/utils/tag_database.py index 38485d5..d929a1e 100644 --- a/orchestrator/utils/tag_database.py +++ b/orchestrator/utils/tag_database.py @@ -123,7 +123,7 @@ def filter_by_tag_ids(query: dict[str, Any], tag_ids: list[Any]) -> dict[str, An def index_tag_documents(connector_tags: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """ 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. + category or completeness. """ indexed = {} for tag in connector_tags: @@ -181,41 +181,6 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An } -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]]: diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index 86527ef..99f62f1 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -21,9 +21,8 @@ PROJECT = '$project' # 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). +# Confidence ones (status and category are filtered in the code). Only the fields used +# are projected: other attributes may hold BSON types a Temporal payload cannot carry. DEFAULT_CONNECTORS_QUERY = { 'collection': 'connectors', 'aggregation': [ @@ -54,7 +53,6 @@ DEFAULT_CONNECTOR_TAGS_QUERY = { 'category': 1, 'data_type': 1, 'status': 1, - 'updated_at': 1, } }, ], @@ -125,7 +123,6 @@ 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), @@ -136,7 +133,6 @@ class Orchestrator: { **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), diff --git a/scripts/sientiapde_1974_backfill_write_tags.js b/scripts/sientiapde_1974_backfill_write_tags.js index ac37af6..6f40bee 100644 --- a/scripts/sientiapde_1974_backfill_write_tags.js +++ b/scripts/sientiapde_1974_backfill_write_tags.js @@ -19,18 +19,31 @@ * 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 + * Usage: paste the whole file into the MongoDB Compass shell (_MONGOSH) or mongosh, after + * setting ACTION and DRY_RUN below. It runs against DB_NAME whatever database the shell + * is on, and can be pasted again in the same shell. + * 1. ACTION 'backfill', DRY_RUN true: prints the plan, writes nothing. + * 2. ACTION 'backfill', DRY_RUN false: converts the pipelines. + * 3. Once the new write tags are confirmed in production, ACTION 'cleanup', DRY_RUN + * false: removes legacy_write_tags from the pipelines. After it, 'revert' can no + * longer restore them. + * ACTION 'revert', DRY_RUN false: restores legacy_write_tags into write_tags and deletes + * the connector_tags this script created that no other pipeline references. It only + * restores pipelines that still have legacy_write_tags (not cleaned up). */ +(() => { +// ---- settings ---------------------------------------------------------------------- +const ACTION = 'backfill'; // 'backfill' | 'revert' | 'cleanup' +const DRY_RUN = true; // false writes +const DB_NAME = 'sientia'; +// ------------------------------------------------------------------------------------- + +const db = globalThis.db.getSiblingDB(DB_NAME); 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 !== ''; @@ -158,10 +171,10 @@ const backfill = () => { ); if (plan.errors.length > 0) { print(`Aborted, nothing was written:\n- ${plan.errors.join('\n- ')}`); - quit(1); + return; } if (DRY_RUN) { - print('DRY_RUN: nothing was written (run with DRY_RUN=0 to apply)'); + print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)'); return; } @@ -216,10 +229,24 @@ const backfill = () => { 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}`); + // A created tag still referenced by a pipeline that is not restored (cleaned up, or + // pointed at it later) is kept + const inUse = new Set( + db.pipelines + .find({ legacy_write_tags: { $exists: false } }, { read_tags: 1, write_tags: 1 }) + .toArray() + .flatMap((p) => [...(Array.isArray(p.read_tags) ? p.read_tags : []), ...(Array.isArray(p.write_tags) ? p.write_tags : [])]) + .filter((tag) => hasValue(tag?.tag_id)) + .map((tag) => String(tag.tag_id)) + ); + const created = db.connector_tags.find({ source: SOURCE }, { tag_id: 1 }).toArray(); + const toDelete = created.filter((tag) => !inUse.has(String(tag.tag_id))).map((tag) => tag.tag_id); + print( + `Pipelines to restore: ${pipelines.length}; connector_tags to delete: ${toDelete.length}` + + `; kept (still referenced): ${created.length - toDelete.length}` + ); if (DRY_RUN) { - print('DRY_RUN: nothing was written (run with DRY_RUN=0 to apply)'); + print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)'); return; } const now = new Date(); @@ -229,9 +256,29 @@ const revert = () => { { $set: { write_tags: pipeline.legacy_write_tags, updated_at: now }, $unset: { legacy_write_tags: '' } } ); } - const { deletedCount } = db.connector_tags.deleteMany({ source: SOURCE }); + const { deletedCount } = db.connector_tags.deleteMany({ source: SOURCE, tag_id: { $in: toDelete } }); print(`pipelines restored: ${pipelines.length}; connector_tags deleted: ${deletedCount}`); }; -if (REVERT) revert(); -else backfill(); +// Run once the new write tags are confirmed: drops the copy that revert restores from +const cleanup = () => { + const filter = { legacy_write_tags: { $exists: true } }; + const count = db.pipelines.countDocuments(filter); + print(`Pipelines with legacy_write_tags to clean: ${count}`); + if (DRY_RUN) { + print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)'); + return; + } + // updated_at is not bumped: the Orchestrator does not read legacy_write_tags + const { modifiedCount } = db.pipelines.updateMany(filter, { $unset: { legacy_write_tags: '' } }); + print(`legacy_write_tags removed from ${modifiedCount} pipeline(s)`); +}; + +const actions = { backfill, revert, cleanup }; +if (!actions[ACTION]) { + print(`Unknown ACTION ${ACTION}: use backfill, revert or cleanup`); + return; +} +print(`${ACTION}${DRY_RUN ? ' (DRY_RUN)' : ''} on ${DB_NAME}`); +actions[ACTION](); +})(); diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index ab1694a..a5afa0f 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -5,7 +5,7 @@ from pandas import DataFrame from pytest import fixture from sientia_do.notifications.models import NotificationLevel -from orchestrator.activities.formatters import Formatters +from orchestrator.activities.formatters import Formatters, schedule_config_hash from orchestrator.utils.pipeline_configs import predictions_batch @@ -616,7 +616,13 @@ def test_process_schedules_names_scouter_tags_like_the_slot(formatters): def test_format_schedule_config(formatters): input_data = { 'schedule_config': [ - {'namespace': 'test_namespace1', 'schedule_name': 'test1', 'updated_at': '2021-01-01'}, + { + 'namespace': 'test_namespace1', + 'schedule_name': 'test1', + 'updated_at': '2021-01-01', + 'config_hash': 'abc', + }, + # Deployed before config_hash existed {'namespace': 'test_namespace2', 'schedule_name': 'test2', 'updated_at': '2021-01-02'}, ], **metadata, @@ -625,33 +631,34 @@ def test_format_schedule_config(formatters): result = formatters.format_schedule_config(input_data) assert result == { - 'test_namespace1': {'test1': '2021-01-01'}, - 'test_namespace2': {'test2': '2021-01-02'}, + 'test_namespace1': {'test1': {'updated_at': '2021-01-01', 'config_hash': 'abc'}}, + 'test_namespace2': {'test2': {'updated_at': '2021-01-02', 'config_hash': None}}, } def test_create_schedule_config(formatters): + to_create = {'frequency': 60, 'data': {'test': 'test'}, 'updated_at': '2021-01-03'} + to_update = {'frequency': 60, 'data': {'test': 'test2'}, 'updated_at': '2021-01-04'} + unchanged = {'frequency': 60, 'data': {'test': 'test3'}, 'updated_at': '2021-01-01'} input_data = { 'current_schedule_config': { 'scouter': { - 'test_schedule_name_to_delete': '2021-01-01', - 'test_schedule_name_to_update': '2021-01-02', + 'test_schedule_name_to_delete': {'updated_at': '2021-01-01', 'config_hash': 'x'}, + 'test_schedule_name_to_update': { + 'updated_at': '2021-01-02', + 'config_hash': schedule_config_hash(to_update), + }, + 'test_schedule_name_unchanged': { + 'updated_at': '2021-01-02', + 'config_hash': schedule_config_hash(unchanged), + }, } }, 'schedule_config': { - 'laborious': { - 'test_schedule_name_to_create': { - 'frequency': 60, - 'data': {'test': 'test'}, - 'updated_at': '2021-01-03', - } - }, + 'laborious': {'test_schedule_name_to_create': to_create}, 'scouter': { - 'test_schedule_name_to_update': { - 'frequency': 60, - 'data': {'test': 'test2'}, - 'updated_at': '2021-01-04', - } + 'test_schedule_name_to_update': to_update, + 'test_schedule_name_unchanged': unchanged, }, }, } @@ -662,9 +669,8 @@ def test_create_schedule_config(formatters): 'to_create': { 'laborious': { 'test_schedule_name_to_create': { - 'frequency': 60, - 'data': {'test': 'test'}, - 'updated_at': '2021-01-03', + **to_create, + 'config_hash': schedule_config_hash(to_create), } }, 'scouter': {}, @@ -672,9 +678,8 @@ def test_create_schedule_config(formatters): 'to_update': { 'scouter': { 'test_schedule_name_to_update': { - 'frequency': 60, - 'data': {'test': 'test2'}, - 'updated_at': '2021-01-04', + **to_update, + 'config_hash': schedule_config_hash(to_update), } }, 'laborious': {}, @@ -683,6 +688,75 @@ def test_create_schedule_config(formatters): } +def test_schedule_config_hash_ignores_updated_at_and_key_order(): + schedule = {'frequency': 60, 'data': {'a': 1, 'b': 2}, 'updated_at': '2021-01-01'} + same = {'data': {'b': 2, 'a': 1}, 'updated_at': '2021-02-02', 'frequency': 60} + + assert schedule_config_hash(schedule) == schedule_config_hash(same) + assert schedule_config_hash(schedule) == schedule_config_hash( + {**schedule, 'config_hash': 'stale'} + ) + assert schedule_config_hash(schedule) != schedule_config_hash({**schedule, 'frequency': 30}) + + +def test_create_schedule_config_updates_on_config_hash_change(formatters): + """Same updated_at, different content (e.g. a pipeline edited by hand): redeploy""" + schedule = {'frequency': 60, 'data': {'test': 'new'}, 'updated_at': '2021-01-01'} + + result = formatters.create_schedule_config( + { + 'current_schedule_config': { + 'scouter': { + 's': { + 'updated_at': '2021-01-02', + 'config_hash': schedule_config_hash({**schedule, 'data': {'test': 'old'}}), + } + } + }, + 'schedule_config': {'scouter': {'s': schedule}, 'laborious': {}}, + } + ) + + assert set(result['to_update']['scouter']) == {'s'} + + +def test_create_schedule_config_updates_on_newer_updated_at(formatters): + """Bumping updated_at forces a redeploy even when nothing else changed""" + schedule = {'frequency': 60, 'updated_at': '2021-01-05'} + + result = formatters.create_schedule_config( + { + 'current_schedule_config': { + 'scouter': { + 's': {'updated_at': '2021-01-02', 'config_hash': schedule_config_hash(schedule)} + } + }, + 'schedule_config': {'scouter': {'s': schedule}, 'laborious': {}}, + } + ) + + assert set(result['to_update']['scouter']) == {'s'} + + +def test_create_schedule_config_accepts_timestamp_only_current_schedules(formatters): + """ + A run started before config_hash replays format_schedule_config's timestamp-only + result: with no stored hash, the schedule is redeployed once to record it. + """ + schedule = {'frequency': 60, 'updated_at': '2021-01-01'} + + result = formatters.create_schedule_config( + { + 'current_schedule_config': {'scouter': {'s': '2021-01-02'}}, + 'schedule_config': {'scouter': {'s': schedule}, 'laborious': {}}, + } + ) + + assert result['to_update']['scouter'] == { + 's': {**schedule, 'config_hash': schedule_config_hash(schedule)} + } + + def test_create_slot_config(formatters): input_data = { 'current_slot_config': { @@ -1426,14 +1500,22 @@ def test_create_schedule_config_warns_on_invalid_chunk_period(formatters): def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatters): input_data = _drift_schedule_config('hour') - input_data['current_schedule_config'] = {'laborious': {'m1-drift-1m': '2026-10-27'}} + schedule = input_data['schedule_config']['laborious']['m1-drift-1m'] + input_data['current_schedule_config'] = { + 'laborious': { + 'm1-drift-1m': { + 'updated_at': '2026-10-27', + 'config_hash': schedule_config_hash(schedule), + } + } + } formatters.create_schedule_config(input_data) formatters.send_notification.assert_not_called() -# PI Web API through the tag database (connectors-pi-web-api) +# Write tags through the tag database (SIENTIAPDE-1974) PLANT_CONNECTOR = { 'connector_id': 'plant', @@ -1551,60 +1633,71 @@ def test_process_schedules_notifies_unresolved_write_tags_once(formatters): assert formatters.send_notification.call_count == 3 -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', +def test_tag_database_changes_redeploy_through_config_hash(formatters): + """ + A tag changed or deleted in the tag database (without bumping any updated_at) + changes the resolved schedule, so its config_hash, and redeploys the pipelines that + use it. A change that does not reach the schedule (e.g. a description) does not. + """ + pipelines = [ + _batch_pipeline('moved', [{'tag_id': 77}]), + _batch_pipeline('deleted', [{'tag_id': 78}]), + _batch_pipeline('same', [{'tag_id': 79}]), + { + 'workflow_type': 'scouter', + 'schedule_name': 'renamed', + 'model_id': 'm1', + 'model': {'name': 'model m1'}, + 'read_tags': [{'tag_id': 7, 'frequency': 1000}], + 'updated_at': '2026-10-01 00:00:00.000+0000', + }, + ] + before = [ + _write_tag(77, 'ns=2;s=P1'), + _write_tag(78, 'ns=2;s=P2'), + _write_tag(79, 'ns=2;s=P3', description='old'), + _write_tag(7, 'ns=3;i=1001', category='Read'), + ] + after = [ + _write_tag(77, 'ns=2;s=P1-moved'), + _write_tag(79, 'ns=2;s=P3', description='new'), + {**_write_tag(7, 'ns=3;i=1001', category='Read'), 'tag_name': 'renamed-tag'}, + ] + + def process(connector_tags): + return formatters.process_schedules( + { + 'pipelines': pipelines, + 'connectors': [PLANT_CONNECTOR], + 'connector_tags': connector_tags, + **metadata, + } + ) + + deployed = process(before) + # Last orchestrated after every pipeline's updated_at, with the hash of `before` + current = { + namespace: { + name: { + 'updated_at': '2026-10-05 00:00:00.000+0000', + 'config_hash': schedule_config_hash(schedule), + } + for name, schedule in schedules.items() + } + for namespace, schedules in deployed.items() } - untouched = _batch_pipeline('old', [{'tag_id': 78}], updated_at='2026-10-20 00:00:00.000+0000') - schedule_config = formatters.process_schedules( - { - '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, - } + unchanged = formatters.create_schedule_config( + {'current_schedule_config': current, 'schedule_config': process(before), **metadata} ) - - 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' - - # Last orchestrated on 2026-10-05: only the pipelines whose tags changed since update actions = formatters.create_schedule_config( - { - '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, - } + {'current_schedule_config': current, 'schedule_config': process(after), **metadata} ) - assert set(actions['to_update']['scouter']) == {'sin'} - assert set(actions['to_update']['laborious']) == {'batch'} + assert unchanged['to_update'] == {'scouter': {}, 'laborious': {}} + assert set(actions['to_update']['laborious']) == {'moved', 'deleted'} + assert set(actions['to_update']['scouter']) == {'renamed'} + assert actions['to_update']['laborious']['deleted']['opc_output_config'] == {} def test_process_schedules_without_updated_at_redeploys_every_run(formatters): @@ -1662,7 +1755,12 @@ def test_format_schedule_config_groups_schedules_of_one_namespace(formatters): } ) - assert result == {'laborious': {'a': '2021-01-01', 'b': '2021-01-02'}} + assert result == { + 'laborious': { + 'a': {'updated_at': '2021-01-01', 'config_hash': None}, + 'b': {'updated_at': '2021-01-02', 'config_hash': None}, + } + } def test_report_schedule_orchestration_success_only_and_empty_types(formatters): diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 5ca0769..bca7958 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -1,5 +1,5 @@ from datetime import UTC, datetime -from unittest.mock import ANY, AsyncMock, MagicMock, patch +from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel @@ -270,23 +270,27 @@ def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db): def test_update_pipelines_timestamps_success(now_mock, mongo_db): input_data = { 'updated_pipelines': [ - {'schedule_name': 'test1', 'namespace': 'test1', 'success': True}, + {'schedule_name': 'test1', 'namespace': 'test1', 'success': True, 'config_hash': 'h1'}, + # Updated by a run started before config_hash {'schedule_name': 'test2', 'namespace': 'test2', 'success': True}, ] } - mongo_db.mongo_db_repository.update_many = MagicMock(return_value=MagicMock()) + mongo_db.mongo_db_repository.update_one = MagicMock(return_value=1) mongo_db.update_pipelines_timestamps(input_data) - mongo_db.mongo_db_repository.update_many.assert_called_once_with( - 'orchestrated_schedules', - { - '$or': [ - {'schedule_name': 'test1', 'namespace': 'test1'}, - {'schedule_name': 'test2', 'namespace': 'test2'}, - ] - }, - {'$set': {'updated_at': now_mock.return_value}}, - {}, - ) + assert mongo_db.mongo_db_repository.update_one.call_args_list == [ + call( + 'orchestrated_schedules', + {'schedule_name': 'test1', 'namespace': 'test1'}, + {'$set': {'updated_at': now_mock.return_value, 'config_hash': 'h1'}}, + {}, + ), + call( + 'orchestrated_schedules', + {'schedule_name': 'test2', 'namespace': 'test2'}, + {'$set': {'updated_at': now_mock.return_value, 'config_hash': None}}, + {}, + ), + ] @patch('orchestrator.activities.mongo_db.now') @@ -295,31 +299,30 @@ def test_update_pipelines_timestamps_success(now_mock, mongo_db): ) def test_update_pipelines_timestamps_all_failed_stamps_nothing(now_mock, mongo_db, pipelines): input_data = {'updated_pipelines': pipelines} - mongo_db.mongo_db_repository.update_many = MagicMock() + mongo_db.mongo_db_repository.update_one = MagicMock() mongo_db.update_pipelines_timestamps(input_data) - # An empty filter would match and stamp every tracked schedule, so the failed one - # would never be retried. - mongo_db.mongo_db_repository.update_many.assert_not_called() + # A failed update is not stamped, so it is retried on the next tick + mongo_db.mongo_db_repository.update_one.assert_not_called() @patch('orchestrator.activities.mongo_db.now') def test_update_pipelines_timestamps_mixed_stamps_only_successes(now_mock, mongo_db): input_data = { 'updated_pipelines': [ - {'schedule_name': 'test1', 'namespace': 'test1', 'success': True}, - {'schedule_name': 'test2', 'namespace': 'test2', 'success': False}, + {'schedule_name': 'test1', 'namespace': 'test1', 'success': True, 'config_hash': 'h1'}, + {'schedule_name': 'test2', 'namespace': 'test2', 'success': False, 'config_hash': 'h2'}, ] } - mongo_db.mongo_db_repository.update_many = MagicMock() + mongo_db.mongo_db_repository.update_one = MagicMock() mongo_db.update_pipelines_timestamps(input_data) - mongo_db.mongo_db_repository.update_many.assert_called_once_with( + mongo_db.mongo_db_repository.update_one.assert_called_once_with( 'orchestrated_schedules', - {'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]}, - {'$set': {'updated_at': now_mock.return_value}}, + {'schedule_name': 'test1', 'namespace': 'test1'}, + {'$set': {'updated_at': now_mock.return_value, 'config_hash': 'h1'}}, {}, ) @@ -334,7 +337,7 @@ def test_update_pipelines_timestamps_failure(now_mock, mongo_db): **metadata, } - mongo_db.mongo_db_repository.update_many.side_effect = Exception('Error') + mongo_db.mongo_db_repository.update_one.side_effect = Exception('Error') try: mongo_db.update_pipelines_timestamps(input_data) except Exception as e: @@ -356,7 +359,7 @@ def test_update_pipelines_timestamps_failure(now_mock, mongo_db): def test_create_pipelines_timestamps_success(now_mock, mongo_db): input_data = { 'created_pipelines': [ - {'schedule_name': 'test1', 'namespace': 'test1', 'success': True}, + {'schedule_name': 'test1', 'namespace': 'test1', 'success': True, 'config_hash': 'h1'}, {'schedule_name': 'test2', 'namespace': 'test2', 'success': True}, ] } @@ -365,8 +368,18 @@ def test_create_pipelines_timestamps_success(now_mock, mongo_db): mongo_db.mongo_db_repository.insert_many.assert_called_once_with( 'orchestrated_schedules', [ - {'schedule_name': 'test1', 'namespace': 'test1', 'updated_at': now_mock.return_value}, - {'schedule_name': 'test2', 'namespace': 'test2', 'updated_at': now_mock.return_value}, + { + 'schedule_name': 'test1', + 'namespace': 'test1', + 'updated_at': now_mock.return_value, + 'config_hash': 'h1', + }, + { + 'schedule_name': 'test2', + 'namespace': 'test2', + 'updated_at': now_mock.return_value, + 'config_hash': None, + }, ], {}, ) diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index 51dd084..7e53e3d 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -145,6 +145,7 @@ async def test_create_schedule( 'data': {'test': 'test'}, 'execution_timeout_seconds': 100, 'task_timeout_seconds': 100, + 'config_hash': 'h1', }, 'test-schedule-invalid-frequency': { 'model_id': 2, @@ -291,24 +292,29 @@ async def test_create_schedule( ] ) + # config_hash goes to the report (orchestrated_schedules), not to the workflow + assert 'config_hash' not in input_data['schedules']['scouter']['test-schedule'] assert report == [ { 'schedule_name': 'test-schedule', 'namespace': 'scouter', 'success': True, 'message': 'Schedule created successfully', + 'config_hash': 'h1', }, { 'schedule_name': 'test-schedule-invalid-frequency', 'namespace': 'scouter', 'success': False, 'message': 'Invalid frequency', + 'config_hash': None, }, { 'schedule_name': 'test-schedule-laborious', 'namespace': 'laborious', 'success': True, 'message': 'Schedule created successfully', + 'config_hash': None, }, ] @@ -355,6 +361,7 @@ async def test_update_schedules( 'workflow_type': 'scouter', 'frequency': '1m', 'data': {'test': 'test'}, + 'config_hash': 'h1', }, 'test-schedule_no_handler': { 'workflow_type': 'scouter', @@ -400,24 +407,28 @@ async def test_update_schedules( handler_scouter.update.assert_called_once() handler_laborious.update.assert_called_once() + assert 'config_hash' not in input_data['schedules']['scouter']['test-schedule'] assert report == [ { 'schedule_name': 'test-schedule', 'namespace': 'scouter', 'success': True, 'message': 'Schedule updated successfully', + 'config_hash': 'h1', }, { 'schedule_name': 'test-schedule_no_handler', 'namespace': 'scouter', 'success': False, 'message': 'Schedule test-schedule_no_handler not found', + 'config_hash': None, }, { 'schedule_name': 'test-schedule-laborious', 'namespace': 'laborious', 'success': True, 'message': 'Schedule updated successfully', + 'config_hash': None, }, ] diff --git a/tests/orchestrator/utils/test_tag_database.py b/tests/orchestrator/utils/test_tag_database.py index 647d959..e0eab7f 100644 --- a/tests/orchestrator/utils/test_tag_database.py +++ b/tests/orchestrator/utils/test_tag_database.py @@ -4,11 +4,9 @@ from orchestrator.utils.tag_database import ( index_connector_tags, index_connectors, index_tag_documents, - latest_updated_at, referenced_tag_ids, resolve_tag_names, resolve_write_tags, - tag_documents_updated_at, tag_id_text, ) from tests.orchestrator.utils.tag_database_fixtures import ( @@ -216,34 +214,3 @@ def test_resolve_write_tags_issues(): "{'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'), - ] - ) - - 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 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 diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index 4eea5c2..0e400f1 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -125,11 +125,7 @@ async def test_run(workflow_mock, referenced_tag_ids_mock, orchestrator): [ call( Activities.aggregate_documents_in_mongodb, - { - **metadata, - 'query': DEFAULT_CONNECTORS_QUERY, - 'timestamp_fields': ['updated_at'], - }, + {**metadata, 'query': DEFAULT_CONNECTORS_QUERY}, retry_policy=ANY, start_to_close_timeout=ANY, ), @@ -144,7 +140,6 @@ 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, @@ -508,12 +503,16 @@ def _process_schedules_input(workflow_mock): def test_default_connector_tags_query_loads_every_category_and_status(): - """The status / category filters are applied in the code, so changes still redeploy""" + """ + The status / category filters are applied in the code. updated_at is not loaded: a + tag change reaches the schedules through their config_hash. + """ 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'): + for field in ('category', 'data_type', 'status'): assert projected[field] == 1 + assert 'updated_at' not in projected @mark.asyncio @@ -535,8 +534,6 @@ async def test_run_hands_the_tag_database_to_process_schedules(workflow_mock, _i 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