Merge pull request #48 from Aignosi/feature/SIENTIAPDE-2193-parallel-windows

test(simple_metrics): parallel windows convention + adaptive frequency e2e (SIENTIAPDE-2193)

README.md

@@ -612,6 +612,37 @@ Temporal input configuration sample:
 }
 ```
 
+#### Simple metrics: parallel windows & adaptive frequency
+
+One `pipelines` doc per window, all with the same `model_id`. Name them `<model>-simple_metrics-<interval_minutes>m`; each doc becomes its own Temporal schedule. Thresholds are shared through `model.model_config`. The adaptive state is per doc.
+
+```json
+{
+    "schedule_name": "model-x-simple_metrics-60m",
+    "workflow_type": "simple_metrics",
+    "model_id": "model-x",
+    "active": true,
+    "interval_minutes": 60,
+    "frequency": "1h",
+    "frequency_alert": "5m"
+}
+```
+
+| Field | Meaning | Default |
+|---|---|---|
+| `frequency_alert` | Frequency while escalated, same format as `frequency`. Absent = adaptive mode off. Below `1m` (or unparseable), the orchestrator sends a WARNING each time it creates or updates the schedule. A value below `1m` is kept; an unparseable value is ignored (the schedule keeps `frequency`) | — |
+| `adaptive_state` | `{escalated, healthy_streak}`, written **only** by laborious (with `updated_at` as a BSON datetime when `escalated` flips) | `{false, 0}` |
+| `model.model_config.simple_metrics_thresholds` | `{<metric>_max, <metric>_min}` shared by all windows | — |
+| `model.model_config.simple_metrics_min_samples` | Windows with fewer rows neither alert nor escalate, and count toward recovery | — |
+
+Escalation, recovery and the per-model alert are implemented in laborious (SIENTIAPDE-2189, SIENTIAPDE-2191); the orchestrator only picks `frequency_alert` while `adaptive_state.escalated` is true.
+
+Rules:
+- **Escalation:** laborious escalates on the first breach.
+- **Recovery:** after `min(12, ceil(interval_minutes / frequency_alert))` consecutive runs without one. Empty windows count as runs without a breach.
+- **Alerts** are per model (trigger `<model_id>-simple_metrics`), so several windows seeing the same breach produce one alert.
+- **Delay:** the schedule switches on the orchestrator tick after laborious flips the state, at most one tick later.
+
 #### Alerts Workflow
 
 ```json

e2e/scenario_inputs/orchestrator_simple_metrics_windows.json

@@ -0,0 +1,45 @@
+{
+  "workflow_input": {
+    "schedule_name": "orchestrator-e2e-sm-windows",
+    "pipelines_query": {
+      "collection": "pipelines",
+      "aggregation": [{"$match": {"active": true}}]
+    },
+    "opc_servers_query": {
+      "collection": "opc_servers",
+      "filters": {"active": true}
+    }
+  },
+  "pipelines": [
+    {
+      "schedule_name": "model-sm-simple_metrics-60m",
+      "workflow_type": "simple_metrics",
+      "model_id": "model-sm",
+      "model": {
+        "name": "Model model-sm",
+        "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}}
+      },
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1h",
+      "frequency_alert": "5m",
+      "interval_minutes": 60
+    },
+    {
+      "schedule_name": "model-sm-simple_metrics-1440m",
+      "workflow_type": "simple_metrics",
+      "model_id": "model-sm",
+      "model": {
+        "name": "Model model-sm",
+        "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}}
+      },
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1d",
+      "frequency_alert": "1h",
+      "interval_minutes": 1440
+    }
+  ],
+  "opc_servers": [],
+  "active_ingestors": ["heartbeat:ingestor:1"]
+}

e2e/scenarios.md

@@ -45,6 +45,17 @@ unrelated Temporal schedule present.
 
 First run creates TTL index on notification collection used by scouter pipelines.
 
+### 1.7.1 Simple metrics parallel windows
+
+Two `simple_metrics` pipelines for the same `model_id` (60 min and 1440 min windows) → one
+Temporal schedule per window, each with its own `frequency` as interval.
+
+### 1.7.2 Adaptive frequency escalation/recovery round trip
+
+Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_at`) on one
+window → next run switches only that schedule to `frequency_alert`; the recovery write switches it
+back to `frequency`.
+
 ---
 
 ## 2. Alerts workflow

e2e/test_orchestrator_main_workflow.py

@@ -1,5 +1,7 @@
 """E2E tests for the Orchestrator main workflow."""
 
+from datetime import UTC, datetime, timedelta
+
 import pytest
 from pymongo import MongoClient
 from redis import Redis
@@ -26,6 +28,13 @@ async def _schedule_ids(host: str, namespace: str) -> list[str]:
     return [schedule.id async for schedule in await client.list_schedules()]
 
 
+async def _schedule_every(host: str, schedule_id: str) -> timedelta:
+    """Return the interval of a laborious-namespace schedule."""
+    client = await Client.connect(host, namespace='laborious')
+    description = await client.get_schedule_handle(schedule_id).describe()
+    return description.schedule.spec.intervals[0].every
+
+
 def _apply_orchestrator_seeds(
     scenario: dict,
     mongo_uri: str,
@@ -224,3 +233,83 @@ async def test_scenario_1_6_1_ttl_index_bootstrap(
         assert any('expireAfterSeconds' in info for info in indexes.values())
     finally:
         mongo.close()
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_7_1_simple_metrics_parallel_windows(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+):
+    """1.7.1: One pipelines doc per window gives one schedule per window, each with its own interval."""
+    scenario = load_scenario_input('orchestrator_simple_metrics_windows')
+    _apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        Orchestrator.run,
+        scenario['workflow_input'],
+        make_workflow_id('orchestrator-sm-windows'),
+    )
+
+    short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m'
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+    assert await _schedule_every(temporal_host, long) == timedelta(days=1)
+
+
+def _laborious_adaptive_write(mongo_uri: str, schedule_name: str, escalated: bool) -> None:
+    """Replicates laborious `update_simple_metrics_adaptive_state` on an escalated flip."""
+    mongo = MongoClient(mongo_uri)
+    try:
+        mongo[E2E_DATABASE]['pipelines'].update_one(
+            {'schedule_name': schedule_name},
+            {
+                '$set': {
+                    'adaptive_state': {'escalated': escalated, 'healthy_streak': 0},
+                    # BSON datetime, as laborious writes it; a string would break the tick.
+                    'updated_at': datetime.now(UTC),
+                }
+            },
+        )
+    finally:
+        mongo.close()
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+):
+    """1.7.2: Escalation written by laborious switches only that window to frequency_alert; recovery switches it back."""
+    scenario = load_scenario_input('orchestrator_simple_metrics_windows')
+    _apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
+    short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m'
+
+    async def run(tag: str) -> None:
+        await start_and_await_workflow(
+            temporal_env.client,
+            Orchestrator.run,
+            scenario['workflow_input'],
+            make_workflow_id(f'orchestrator-sm-adaptive-{tag}'),
+        )
+
+    await run('create')
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+
+    _laborious_adaptive_write(mongo_uri, short, escalated=True)
+    await run('escalate')
+    assert await _schedule_every(temporal_host, short) == timedelta(minutes=5)
+    assert await _schedule_every(temporal_host, long) == timedelta(days=1)
+
+    _laborious_adaptive_write(mongo_uri, short, escalated=False)
+    await run('recover')
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)

orchestrator/activities/formatters.py

@@ -370,6 +370,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.
+        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).
 
         Args:
             input_data (dict[str, Any]): The input data containing:
@@ -410,7 +412,8 @@ class Formatters(SientiaMonitoring):
                 schedules, current_schedules, to_update, to_create, namespace, metadata
             )
 
-        # Only schedules being created/updated, so this fires once per config change.
+        # Fires whenever the schedule is created or updated (each `updated_at` 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'))