fix(simple_metrics): skip invalid interval_minutes and restart a malformed healthy_streak (SIENTIAPDE-2191)

- interval_minutes is untyped Mongo config: a string or null now warns and
  skips the adaptive state instead of failing every run (TypeError).
- A hand-edited healthy_streak (null, text, inf) restarts the count and keeps
  escalated, instead of raising on every run.
- README: step 4 runs on every schedule, so a MongoDB failure fails the run.
This commit is contained in:
PedroHMCosme
2026-10-09 10:06:05 -03:00
parent a812acd4e9
commit e1b94b1da4
5 changed files with 31 additions and 6 deletions

View File

@@ -489,7 +489,7 @@ The **SimpleMetrics** workflow calculates regression metrics (RMSE, MSE, MAE, R
1. **Data Loading**: Loads prediction vs target data via a JOIN query 1. **Data Loading**: Loads prediction vs target data via a JOIN query
2. **Metrics Calculation**: Calculates configured regression metrics; alerts on `simple_metrics_thresholds` breaches (one alert per model, the message names the window) unless the window has fewer rows than `simple_metrics_min_samples` 2. **Metrics Calculation**: Calculates configured regression metrics; alerts on `simple_metrics_thresholds` breaches (one alert per model, the message names the window) unless the window has fewer rows than `simple_metrics_min_samples`
3. **Data Export**: Exports metrics to PostgreSQL 3. **Data Export**: Exports metrics to PostgreSQL
4. **Adaptive State**: when the schedule's `pipelines` doc has `frequency_alert`, escalates `adaptive_state` on a breach and recovers after the window is renewed without one (`min(12, ceil(interval_minutes / frequency_alert))` runs, at least 1; empty, under-sampled or threshold-less windows count as healthy). The orchestrator runs the schedule at `frequency_alert` while escalated 4. **Adaptive State**: when the schedule's `pipelines` doc has `frequency_alert`, escalates `adaptive_state` on a breach and recovers after the window is renewed without one (`min(12, ceil(interval_minutes / frequency_alert))` runs, at least 1; empty, under-sampled or threshold-less windows count as healthy). The orchestrator runs the schedule at `frequency_alert` while escalated. Runs on every schedule (it reads the `pipelines` doc to check for `frequency_alert`), so a MongoDB failure here fails the run; the metrics are already exported by then
#### Architecture Diagram #### Architecture Diagram
```mermaid ```mermaid

View File

@@ -85,7 +85,10 @@ def next_adaptive_state(state: dict[str, Any], breached: bool, runs: int) -> dic
return {'escalated': True, 'healthy_streak': 0} return {'escalated': True, 'healthy_streak': 0}
if not state.get('escalated'): if not state.get('escalated'):
return {'escalated': False, 'healthy_streak': 0} return {'escalated': False, 'healthy_streak': 0}
streak = int(state.get('healthy_streak', 0)) + 1 try:
streak = int(state.get('healthy_streak', 0)) + 1
except (TypeError, ValueError, OverflowError):
streak = 1 # hand-edited streak (null, text, inf): restart the count, stay escalated
if streak >= runs: if streak >= runs:
return {'escalated': False, 'healthy_streak': 0} return {'escalated': False, 'healthy_streak': 0}
return {'escalated': True, 'healthy_streak': streak} return {'escalated': True, 'healthy_streak': streak}

View File

@@ -192,11 +192,14 @@ class MongoActivities(SientiaMonitoring):
try: try:
runs = recovery_runs(input_data['interval_minutes'], str(frequency_alert)) runs = recovery_runs(input_data['interval_minutes'], str(frequency_alert))
except ValueError: except (TypeError, ValueError):
# Config error, not a failure: the orchestrator ignores this value too (keeps # Config error, not a failure: the orchestrator ignores an invalid `frequency_alert`
# `frequency`) and already sends SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR for it. # too (keeps `frequency`) and already sends SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR
# for it. `interval_minutes` is untyped Mongo config as well (e.g. "60" or null).
self.warning( self.warning(
f'Invalid frequency_alert={frequency_alert!r}: adaptive state skipped', metadata f'Invalid frequency_alert={frequency_alert!r} or '
f'interval_minutes={input_data["interval_minutes"]!r}: adaptive state skipped',
metadata,
) )
return None return None

View File

@@ -1267,6 +1267,9 @@ def _state(escalated: bool, streak: int) -> dict:
(_state(True, 2), False, 3, _state(False, 0)), # N-th healthy run recovers (_state(True, 2), False, 3, _state(False, 0)), # N-th healthy run recovers
(_state(True, 2), True, 3, _state(True, 0)), # breach resets the streak (_state(True, 2), True, 3, _state(True, 0)), # breach resets the streak
(_state(True, 0), False, 1, _state(False, 0)), # N=1 recovers on first healthy run (_state(True, 0), False, 1, _state(False, 0)), # N=1 recovers on first healthy run
(_state(True, None), False, 3, _state(True, 1)), # hand-edited streak restarts
(_state(True, 'x'), False, 3, _state(True, 1)),
(_state(True, float('inf')), False, 3, _state(True, 1)),
], ],
) )
def test_next_adaptive_state(state, breached, runs, expected): def test_next_adaptive_state(state, breached, runs, expected):

View File

@@ -322,6 +322,22 @@ def test_adaptive_state_invalid_frequency_alert_is_skipped(
mongo_activities.warning.assert_called_once() mongo_activities.warning.assert_called_once()
@mark.parametrize('interval_minutes', ['60', None])
def test_adaptive_state_invalid_interval_minutes_is_skipped(
mongo_activities, mongo, interval_minutes
):
# Untyped Mongo config: "60" * 60 is a string, None * 60 a TypeError.
mongo.find_one.return_value = _pipeline(frequency_alert='5m')
result = mongo_activities.update_simple_metrics_adaptive_state(
{**_BREACH_INPUT, 'interval_minutes': interval_minutes}
)
assert result is None
mongo.update_one.assert_not_called()
mongo_activities.warning.assert_called_once()
@mark.parametrize('adaptive_state', ['garbage', ['escalated']]) @mark.parametrize('adaptive_state', ['garbage', ['escalated']])
def test_adaptive_state_malformed_state_is_not_escalated(mongo_activities, mongo, adaptive_state): def test_adaptive_state_malformed_state_is_not_escalated(mongo_activities, mongo, adaptive_state):
mongo.find_one.return_value = _pipeline(frequency_alert='5m', adaptive_state=adaptive_state) mongo.find_one.return_value = _pipeline(frequency_alert='5m', adaptive_state=adaptive_state)