15 Commits

Author SHA1 Message Date
github-actions[bot]
c96d94a045 chore: bump version to 0.7.0 2026-10-08 14:03:23 +00:00
bec855d0b7 Merge branch 'main' into feature/SIENTIAPDE-2190-adaptive-frequency 2026-10-08 11:03:12 -03:00
d41bef39bd Merge pull request #48 from Aignosi/feature/SIENTIAPDE-2193-parallel-windows
test(simple_metrics): parallel windows convention + adaptive frequency e2e (SIENTIAPDE-2193)
2026-10-08 11:02:57 -03:00
github-actions[bot]
da27e2ba91 chore: bump version to 0.6.8 2026-10-08 13:52:43 +00:00
3f42cdea74 Merge pull request #46 from Aignosi/fix/SIENTIAPDE-2192-empty-timestamp-filter
fix(orchestrator): do not stamp or untrack every schedule when all updates/deletes fail (SIENTIAPDE-2192)
2026-10-08 10:52:19 -03:00
PedroHMCosme
9f21372ab3 docs: floor warning fires on each schedule create/update; unparseable frequency_alert is ignored (SIENTIAPDE-2193) 2026-10-08 08:48:25 -03:00
PedroHMCosme
b5512498b9 fix(simple_metrics): ignore unparseable frequency_alert and malformed adaptive_state (SIENTIAPDE-2190) 2026-10-08 08:48:24 -03:00
PedroHMCosme
33d99827b1 docs: clarify floor warning repeats and laborious owns escalation (SIENTIAPDE-2193) 2026-10-08 08:48:24 -03:00
PedroHMCosme
2c3ce07f5c docs: simple metrics parallel windows and adaptive frequency convention (SIENTIAPDE-2193) 2026-10-08 08:48:24 -03:00
PedroHMCosme
7b3618d274 test(e2e): adaptive frequency escalation and recovery round trip (SIENTIAPDE-2193) 2026-10-08 08:48:24 -03:00
PedroHMCosme
0f902b53a9 test(e2e): one simple_metrics schedule per window for the same model (SIENTIAPDE-2193) 2026-10-08 08:48:24 -03:00
PedroHMCosme
d963cd3e8a test(orchestrator): cover mixed and empty timestamp updates/deletes (SIENTIAPDE-2192) 2026-10-08 08:46:05 -03:00
PedroHMCosme
7222b5d1ef feat(simple_metrics): warn when frequency_alert is below the 1m floor (SIENTIAPDE-2190) 2026-10-07 15:28:45 -03:00
PedroHMCosme
0d3ff73557 fix(orchestrator): do not stamp or untrack every schedule when all updates/deletes fail (SIENTIAPDE-2192) 2026-10-07 15:24:38 -03:00
PedroHMCosme
40f8d8174b feat(simple_metrics): use frequency_alert while adaptive_state is escalated (SIENTIAPDE-2190) 2026-10-07 15:17:53 -03:00
11 changed files with 448 additions and 5 deletions

View File

@@ -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 #### Alerts Workflow
```json ```json

View File

@@ -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"]
}

View File

@@ -45,6 +45,17 @@ unrelated Temporal schedule present.
First run creates TTL index on notification collection used by scouter pipelines. 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 ## 2. Alerts workflow

View File

@@ -1,5 +1,7 @@
"""E2E tests for the Orchestrator main workflow.""" """E2E tests for the Orchestrator main workflow."""
from datetime import UTC, datetime, timedelta
import pytest import pytest
from pymongo import MongoClient from pymongo import MongoClient
from redis import Redis 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()] 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( def _apply_orchestrator_seeds(
scenario: dict, scenario: dict,
mongo_uri: str, 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()) assert any('expireAfterSeconds' in info for info in indexes.values())
finally: finally:
mongo.close() 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)

View File

@@ -20,6 +20,7 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.utils.orchestrator_functions import ( from orchestrator.utils.orchestrator_functions import (
build_tag_config, build_tag_config,
drift, drift,
frequency_alert_warning,
gather_read_tags, gather_read_tags,
minimal_retrain, minimal_retrain,
pi_web_api_scouter, pi_web_api_scouter,
@@ -369,6 +370,8 @@ class Formatters(SientiaMonitoring):
Compares the current schedule configuration in Temporal with the new schedule Compares the current schedule configuration in Temporal with the new schedule
configuration to determine which schedules need to be created, updated, or deleted. configuration to determine which schedules need to be created, updated, or deleted.
Uses timestamp comparison to identify schedules that have changed. 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: Args:
input_data (dict[str, Any]): The input data containing: input_data (dict[str, Any]): The input data containing:
@@ -409,6 +412,25 @@ class Formatters(SientiaMonitoring):
schedules, current_schedules, to_update, to_create, namespace, metadata schedules, current_schedules, to_update, to_create, namespace, metadata
) )
# 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'))
if warning:
self.send_notification(
metadata={
**metadata,
'schedule_name': schedule_name,
'model_id': schedule.get('model_id'),
'model_name': schedule.get('model_name'),
},
notification_id='SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR',
message=warning,
block='create_schedule_config',
level=NotificationLevel.WARNING,
)
for namespace, schedules in current_schedule_config.items(): for namespace, schedules in current_schedule_config.items():
for schedule_name in schedules: for schedule_name in schedules:
if schedule_name not in schedule_config[namespace]: if schedule_name not in schedule_config[namespace]:

View File

@@ -242,7 +242,12 @@ class MongoDB(SientiaMonitoring):
for pipeline in updated_pipelines for pipeline in updated_pipelines
if pipeline['success'] if pipeline['success']
] ]
data_filter = {'$or': argument} if argument else {} if not argument:
# 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.
self.info('No successful pipeline updates to timestamp', metadata=metadata)
return
data_filter = {'$or': argument}
try: try:
self.mongo_db_repository.update_many( self.mongo_db_repository.update_many(
@@ -343,7 +348,13 @@ class MongoDB(SientiaMonitoring):
for pipeline in deleted_pipelines for pipeline in deleted_pipelines
if pipeline['success'] if pipeline['success']
] ]
data_filter = {'$or': argument} if argument else {} if not argument:
# Every delete failed: remove nothing (same guard as create_pipelines_timestamps).
# An empty filter would wipe orchestrated_schedules and make the next tick delete
# and recreate every orchestrated schedule.
self.info('No successful pipeline deletions to untrack', metadata=metadata)
return
data_filter = {'$or': argument}
try: try:
self.mongo_db_repository.delete_many('orchestrated_schedules', data_filter, metadata) self.mongo_db_repository.delete_many('orchestrated_schedules', data_filter, metadata)

View File

@@ -85,6 +85,9 @@ def simple_metrics(config: dict[str, Any]):
- interval_minutes (int, optional): Computation interval in minutes (default: 60) - interval_minutes (int, optional): Computation interval in minutes (default: 60)
- metrics (list[str], optional): List of metrics to compute - metrics (list[str], optional): List of metrics to compute
(default: ['rmse', 'mse', 'mae', 'r2']) (default: ['rmse', 'mse', 'mae', 'r2'])
- frequency_alert (str, optional): Frequency used while `adaptive_state.escalated`
is true; same format as `frequency`. Absent = adaptive frequency off.
- adaptive_state (dict, optional): `{escalated, healthy_streak}`, written by laborious.
- Additional fields from common_config - Additional fields from common_config
Returns: Returns:
@@ -92,17 +95,56 @@ def simple_metrics(config: dict[str, Any]):
actual data), target table, time interval, and metrics list. Results stored in actual data), target table, time interval, and metrics list. Results stored in
'simple_metrics' table 'simple_metrics' table
""" """
base = common_config(config)
frequency_alert = config.get('frequency_alert')
# SIENTIAPDE-1995: laborious flips `adaptive_state.escalated` on a threshold breach (and back
# after the window is renewed without one) and bumps `updated_at`, which makes this tick
# rewrite the schedule.
state = config.get('adaptive_state')
# Hand-written docs: a malformed state must not raise and kill the tick for every model.
if frequency_alert and isinstance(state, dict) and state.get('escalated') is True:
try:
# Temporal parses the schedule frequency on create/update; an unparseable value
# would fail it, so keep `frequency` (create_schedule_config still warns).
parse_frequency(str(frequency_alert))
base['frequency'] = frequency_alert
except ValueError:
pass
return { return {
**common_config(config), **base,
'schema': 'sientia_data', 'schema': 'sientia_data',
'predictions_table_name': 'predictions', 'predictions_table_name': 'predictions',
'data_table_name': 'laborious_data', 'data_table_name': 'laborious_data',
'target_table_name': 'simple_metrics', 'target_table_name': 'simple_metrics',
'interval_minutes': config.get('interval_minutes', 60), 'interval_minutes': config.get('interval_minutes', 60),
'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']), 'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
# Forwarded so create_schedule_config can warn on a value below the floor.
'frequency_alert': frequency_alert,
} }
MIN_FREQUENCY_ALERT_SECONDS = 60
def frequency_alert_warning(frequency_alert: str | None) -> str | None:
"""
Warning message when `frequency_alert` is below the 1-minute floor or does not parse,
otherwise None. Warn only (SIENTIAPDE-1995 decision 7): a value below the floor is kept;
an unparseable one is ignored by `simple_metrics()`, which keeps `frequency`.
"""
if not frequency_alert:
return None
try:
# str(): a non-string value (e.g. a number typed into the doc) must warn, not raise
# AttributeError and fail the whole create_schedule_config activity.
seconds = parse_frequency(str(frequency_alert))
except ValueError:
return f'frequency_alert={frequency_alert!r} is not a valid frequency'
if seconds < MIN_FREQUENCY_ALERT_SECONDS:
return f'frequency_alert={frequency_alert} is below the 1m floor; kept as configured'
return None
def minimal_retrain(config: dict[str, Any]): def minimal_retrain(config: dict[str, Any]):
""" """
Build minimal retrain configuration from pipeline config. Build minimal retrain configuration from pipeline config.

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "orchestrator" name = "orchestrator"
version = "0.6.7" version = "0.7.0"
description = "Sientia DataOps Orchestrator - ML Model Orchestration System" description = "Sientia DataOps Orchestrator - ML Model Orchestration System"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"

View File

@@ -1023,3 +1023,53 @@ def test_filter_notification_reports_ignore_models_id_explicit_none_does_not_rai
response = formatters.filter_notification_reports(input_data) response = formatters.filter_notification_reports(input_data)
assert response['test_group_1']['notifications'] == input_data['notification_package'] assert response['test_group_1']['notifications'] == input_data['notification_package']
def _schedule_config_with_frequency_alert(frequency_alert):
return {
'metadata': {},
'current_schedule_config': {'laborious': {}},
'schedule_config': {
'laborious': {
'm1-simple_metrics-60m': {
'workflow_type': 'simple_metrics',
'model_id': 'm1',
'model_name': 'm1',
'frequency': '1h',
'frequency_alert': frequency_alert,
'updated_at': '2026-10-07',
}
},
'scouter': {},
},
}
def test_create_schedule_config_warns_on_frequency_alert_below_floor(formatters):
result = formatters.create_schedule_config(_schedule_config_with_frequency_alert('30s'))
# The configured value is kept.
assert result['to_create']['laborious']['m1-simple_metrics-60m']['frequency_alert'] == '30s'
formatters.send_notification.assert_called_once()
kwargs = formatters.send_notification.call_args.kwargs
assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
assert kwargs['level'] == NotificationLevel.WARNING
assert kwargs['metadata']['schedule_name'] == 'm1-simple_metrics-60m'
def test_create_schedule_config_warns_on_frequency_alert_below_floor_on_update(formatters):
input_data = _schedule_config_with_frequency_alert('30s')
input_data['current_schedule_config'] = {'laborious': {'m1-simple_metrics-60m': '2026-10-06'}}
result = formatters.create_schedule_config(input_data)
assert 'm1-simple_metrics-60m' in result['to_update']['laborious']
formatters.send_notification.assert_called_once()
kwargs = formatters.send_notification.call_args.kwargs
assert kwargs['notification_id'] == 'SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR'
def test_create_schedule_config_no_warning_at_or_above_floor(formatters):
formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m'))
formatters.send_notification.assert_not_called()

View File

@@ -1,7 +1,7 @@
from datetime import UTC, datetime from datetime import UTC, datetime
from unittest.mock import ANY, AsyncMock, MagicMock, patch from unittest.mock import ANY, AsyncMock, MagicMock, patch
from pytest import fixture from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
@@ -265,6 +265,41 @@ def test_update_pipelines_timestamps_success(now_mock, mongo_db):
) )
@patch('orchestrator.activities.mongo_db.now')
@mark.parametrize(
'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]]
)
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.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()
@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},
]
}
mongo_db.mongo_db_repository.update_many = MagicMock()
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'}]},
{'$set': {'updated_at': now_mock.return_value}},
{},
)
@patch('orchestrator.activities.mongo_db.now') @patch('orchestrator.activities.mongo_db.now')
def test_update_pipelines_timestamps_failure(now_mock, mongo_db): def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
input_data = { input_data = {
@@ -362,6 +397,38 @@ def test_delete_pipelines_timestamps_success(now_mock, mongo_db):
) )
@mark.parametrize(
'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]]
)
def test_delete_pipelines_timestamps_all_failed_deletes_nothing(mongo_db, pipelines):
input_data = {'deleted_pipelines': pipelines}
mongo_db.mongo_db_repository.delete_many = MagicMock()
mongo_db.delete_pipelines_timestamps(input_data)
# An empty filter would wipe orchestrated_schedules; the next tick's normalize_schedules
# would then delete and recreate every orchestrated Temporal schedule.
mongo_db.mongo_db_repository.delete_many.assert_not_called()
def test_delete_pipelines_timestamps_mixed_deletes_only_successes(mongo_db):
input_data = {
'deleted_pipelines': [
{'schedule_name': 'test1', 'namespace': 'test1', 'success': True},
{'schedule_name': 'test2', 'namespace': 'test2', 'success': False},
]
}
mongo_db.mongo_db_repository.delete_many = MagicMock()
mongo_db.delete_pipelines_timestamps(input_data)
mongo_db.mongo_db_repository.delete_many.assert_called_once_with(
'orchestrated_schedules',
{'$or': [{'schedule_name': 'test1', 'namespace': 'test1'}]},
{},
)
@patch('orchestrator.activities.mongo_db.now') @patch('orchestrator.activities.mongo_db.now')
def test_delete_pipelines_timestamps_failure(now_mock, mongo_db): def test_delete_pipelines_timestamps_failure(now_mock, mongo_db):
input_data = { input_data = {

View File

@@ -1,10 +1,13 @@
from unittest.mock import call, patch from unittest.mock import call, patch
from pytest import mark
from orchestrator.utils.orchestrator_functions import ( from orchestrator.utils.orchestrator_functions import (
base_scouter, base_scouter,
build_tag_config, build_tag_config,
common_config, common_config,
drift, drift,
frequency_alert_warning,
gather_read_tags, gather_read_tags,
minimal_retrain, minimal_retrain,
overlap_filter_config, overlap_filter_config,
@@ -110,6 +113,7 @@ def test_simple_metrics():
'target_table_name': 'simple_metrics', 'target_table_name': 'simple_metrics',
'interval_minutes': 120, 'interval_minutes': 120,
'metrics': ['rmse', 'mse'], 'metrics': ['rmse', 'mse'],
'frequency_alert': None,
'execution_timeout_seconds': 300, 'execution_timeout_seconds': 300,
'task_timeout_seconds': 300, 'task_timeout_seconds': 300,
'runtime': 'legacy', 'runtime': 'legacy',
@@ -117,6 +121,61 @@ def test_simple_metrics():
assert result == expected assert result == expected
def _simple_metrics_config(**fields):
return {
'workflow_type': 'simple_metrics',
'schedule_name': 'm1-simple_metrics-60m',
'model_id': 'm1',
'model': {'name': 'm1'},
'frequency': '1h',
**fields,
}
def test_simple_metrics_escalated_uses_frequency_alert():
config = _simple_metrics_config(
frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
)
result = simple_metrics(config)
assert result['frequency'] == '5m'
assert result['frequency_alert'] == '5m'
def test_simple_metrics_not_escalated_keeps_frequency():
config = _simple_metrics_config(
frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 2}
)
assert simple_metrics(config)['frequency'] == '1h'
def test_simple_metrics_escalated_without_frequency_alert_keeps_frequency():
# Removing frequency_alert turns adaptive mode off even if a stale state says escalated.
config = _simple_metrics_config(adaptive_state={'escalated': True, 'healthy_streak': 0})
assert simple_metrics(config)['frequency'] == '1h'
def test_simple_metrics_frequency_alert_without_state_keeps_frequency():
assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h'
def test_simple_metrics_escalated_unparseable_frequency_alert_keeps_frequency():
# An unparseable frequency would make the Temporal schedule create/update fail.
config = _simple_metrics_config(frequency_alert='5min', adaptive_state={'escalated': True})
assert simple_metrics(config)['frequency'] == '1h'
def test_simple_metrics_escalated_below_floor_frequency_alert_is_applied():
# Decision 7: below the floor only warns, the value is kept.
config = _simple_metrics_config(frequency_alert='30s', adaptive_state={'escalated': True})
assert simple_metrics(config)['frequency'] == '30s'
@mark.parametrize('adaptive_state', ['yes', {'escalated': 'false'}])
def test_simple_metrics_malformed_adaptive_state_keeps_frequency(adaptive_state):
config = _simple_metrics_config(frequency_alert='5m', adaptive_state=adaptive_state)
assert simple_metrics(config)['frequency'] == '1h'
def test_minimal_retrain(): def test_minimal_retrain():
config = { config = {
'workflow_type': 'minimal_retrain', 'workflow_type': 'minimal_retrain',
@@ -762,3 +821,19 @@ def test_pi_web_api_scouter_with_no_timeout():
'runtime': 'legacy', 'runtime': 'legacy',
} }
assert result == expected assert result == expected
@mark.parametrize(
'frequency_alert, warns',
[
(None, False),
('5m', False),
('1m', False), # the floor itself is allowed
('60s', False),
('30s', True),
('5min', True), # unparseable
(5, True), # not a string: must warn, never raise
],
)
def test_frequency_alert_warning(frequency_alert, warns):
assert (frequency_alert_warning(frequency_alert) is not None) is warns