Compare commits
15 Commits
test-SIENT
...
0.7.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c96d94a045 | ||
| bec855d0b7 | |||
| d41bef39bd | |||
|
|
da27e2ba91 | ||
| 3f42cdea74 | |||
|
|
9f21372ab3 | ||
|
|
b5512498b9 | ||
|
|
33d99827b1 | ||
|
|
2c3ce07f5c | ||
|
|
7b3618d274 | ||
|
|
0f902b53a9 | ||
|
|
d963cd3e8a | ||
|
|
7222b5d1ef | ||
|
|
0d3ff73557 | ||
|
|
40f8d8174b |
31
README.md
31
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
|
#### Alerts Workflow
|
||||||
|
|
||||||
```json
|
```json
|
||||||
|
|||||||
45
e2e/scenario_inputs/orchestrator_simple_metrics_windows.json
Normal file
45
e2e/scenario_inputs/orchestrator_simple_metrics_windows.json
Normal 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"]
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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]:
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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 = {
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user