From e1b94b1da4b390d8fa2f3a42c8a2d0d23e1814d0 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Fri, 9 Oct 2026 10:06:05 -0300 Subject: [PATCH] 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. --- README.md | 2 +- laborious/activities/model_metrics.py | 5 ++++- laborious/activities/mongo.py | 11 +++++++---- tests/laborious/activities/test_model_metrics.py | 3 +++ tests/laborious/activities/test_mongo.py | 16 ++++++++++++++++ 5 files changed, 31 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 8c9d0dc..6f3172c 100644 --- a/README.md +++ b/README.md @@ -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 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 -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 ```mermaid diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index d2271b9..650ed94 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -85,7 +85,10 @@ def next_adaptive_state(state: dict[str, Any], breached: bool, runs: int) -> dic return {'escalated': True, 'healthy_streak': 0} if not state.get('escalated'): 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: return {'escalated': False, 'healthy_streak': 0} return {'escalated': True, 'healthy_streak': streak} diff --git a/laborious/activities/mongo.py b/laborious/activities/mongo.py index 2e060e8..fc4d92f 100644 --- a/laborious/activities/mongo.py +++ b/laborious/activities/mongo.py @@ -192,11 +192,14 @@ class MongoActivities(SientiaMonitoring): try: runs = recovery_runs(input_data['interval_minutes'], str(frequency_alert)) - except ValueError: - # Config error, not a failure: the orchestrator ignores this value too (keeps - # `frequency`) and already sends SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR for it. + except (TypeError, ValueError): + # Config error, not a failure: the orchestrator ignores an invalid `frequency_alert` + # 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( - 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 diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index c6490ad..ed7ce56 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/tests/laborious/activities/test_model_metrics.py @@ -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), 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, 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): diff --git a/tests/laborious/activities/test_mongo.py b/tests/laborious/activities/test_mongo.py index 415cafd..837e410 100644 --- a/tests/laborious/activities/test_mongo.py +++ b/tests/laborious/activities/test_mongo.py @@ -322,6 +322,22 @@ def test_adaptive_state_invalid_frequency_alert_is_skipped( 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']]) 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)