diff --git a/e2e/scenario_inputs/orchestrator_simple_metrics_windows.json b/e2e/scenario_inputs/orchestrator_simple_metrics_windows.json new file mode 100644 index 0000000..8f93c8b --- /dev/null +++ b/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"] +} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index bd66ac4..6f439a0 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -45,6 +45,11 @@ 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. + --- ## 2. Alerts workflow diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index f2cb635..3f2d0ce 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -1,5 +1,7 @@ """E2E tests for the Orchestrator main workflow.""" +from datetime import 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,29 @@ 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)