diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 27e7087..1d070b3 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -524,7 +524,7 @@ class ModelMetrics(SientiaMonitoring): @activity.defn(name='update_simple_metrics_adaptive_state') def update_simple_metrics_adaptive_state(self, input_data: dict[str, Any]) -> dict | None: """ - Best-effort update of the adaptive frequency state on the schedule's `pipelines` doc. + Update the adaptive frequency state on the schedule's `pipelines` doc. The orchestrator derives the schedule frequency from `adaptive_state.escalated` and `frequency_alert`. State is read from Mongo, not from the workflow input, because the @@ -541,48 +541,62 @@ class ModelMetrics(SientiaMonitoring): - interval_minutes (int): Window size, used to compute the recovery runs. Return: - The new state, or `None` when adaptive mode is off, Mongo is not configured, or the - read/parse/write failed. + The new state, or `None` when adaptive mode is off, Mongo is not configured, the doc + is missing or `frequency_alert` is invalid. + + Raises: + Exception: Mongo read/write failures, so the workflow run fails explicitly. """ metadata = input_data['metadata'] if self.mongo_repository is None: return None + # Mongo errors propagate on purpose: the run must fail explicitly so the problem shows + # up in the Temporal metrics (PR #59 review). Metrics were already exported by then. query = {'schedule_name': metadata['schedule_name']} - try: - pipeline = self.mongo_repository.find_one(self._PIPELINES_COLLECTION, query, metadata) - if not pipeline: - # Likely misconfig: MONGODB_DATABASE_NAME differs from the orchestrator's, or the - # schedule_name does not match its doc. - self.warning( - f'No pipelines doc for schedule {metadata["schedule_name"]}: ' - 'adaptive state skipped', - metadata, - ) - return None - if not pipeline.get('frequency_alert'): - return None - - runs = recovery_runs(input_data['interval_minutes'], pipeline['frequency_alert']) - breached = bool( - threshold_breaches(input_data['simple_metrics'], input_data.get('thresholds')) + pipeline = self.mongo_repository.find_one(self._PIPELINES_COLLECTION, query, metadata) + if not pipeline: + # Likely misconfig: MONGODB_DATABASE_NAME differs from the orchestrator's, or the + # schedule_name does not match its doc. + self.warning( + f'No pipelines doc for schedule {metadata["schedule_name"]}: ' + 'adaptive state skipped', + metadata, ) - state = pipeline.get('adaptive_state') or {'escalated': False, 'healthy_streak': 0} - new_state = next_adaptive_state(state, breached, runs) - if new_state == state: - return new_state - - update: dict[str, Any] = {'adaptive_state': new_state} - if new_state['escalated'] != bool(state.get('escalated')): - # BSON datetime, never a string: the orchestrator normalizes `updated_at` with - # `.replace(tzinfo=UTC).strftime(...)` and a string would fail its whole tick. - update['updated_at'] = now() - self.mongo_repository.update_one( - self._PIPELINES_COLLECTION, query, {'$set': update}, metadata - ) - except Exception as e: - self.error(f'Failed to update simple metrics adaptive state: {e}', metadata) return None + frequency_alert = pipeline.get('frequency_alert') + if not frequency_alert: + return None + + 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. + self.warning( + f'Invalid frequency_alert={frequency_alert!r}: adaptive state skipped', metadata + ) + return None + + breached = bool( + threshold_breaches(input_data['simple_metrics'], input_data.get('thresholds')) + ) + state = pipeline.get('adaptive_state') + if not isinstance(state, dict): + # Missing or hand-edited garbage: not escalated, same as the orchestrator reads it. + state = {'escalated': False, 'healthy_streak': 0} + new_state = next_adaptive_state(state, breached, runs) + if new_state == state: + return new_state + + update: dict[str, Any] = {'adaptive_state': new_state} + if new_state['escalated'] != bool(state.get('escalated')): + # BSON datetime, never a string: the orchestrator normalizes `updated_at` with + # `.replace(tzinfo=UTC).strftime(...)` and a string would fail its whole tick. + update['updated_at'] = now() + self.mongo_repository.update_one( + self._PIPELINES_COLLECTION, query, {'$set': update}, metadata + ) self.info(f'Adaptive state of {query["schedule_name"]}: {state} -> {new_state}', metadata) return new_state diff --git a/laborious/workflows/simple_metrics.py b/laborious/workflows/simple_metrics.py index 142f59d..61531c1 100644 --- a/laborious/workflows/simple_metrics.py +++ b/laborious/workflows/simple_metrics.py @@ -114,23 +114,20 @@ class SimpleMetrics: start_to_close_timeout=timedelta(seconds=300), ) - try: - await workflow.execute_activity_method( - Activities.update_simple_metrics_adaptive_state, - { - **metadata, - # Only a valid breach escalates: an empty or under-sampled window is sent as - # no rows, which counts as a healthy run (spec decision 8). Always called, even - # without thresholds (no breach), so an escalated doc recovers after they are - # removed. - 'simple_metrics': (simple_metrics or []) if enough_samples else [], - 'thresholds': thresholds, - 'interval_minutes': interval_minutes, - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(minutes=1), - ) - except Exception as e: - # Best-effort: metrics are already persisted. retry_policy has maximum_attempts=1, so - # a dispatch failure would otherwise fail the workflow (same as prediction_process). - workflow.logger.warning(f'Failed to update adaptive state, continuing anyway: {e}') + # No try/except: a failure here must fail the run so it shows up in the Temporal metrics + # (metrics are already persisted above). + await workflow.execute_activity_method( + Activities.update_simple_metrics_adaptive_state, + { + **metadata, + # Only a valid breach escalates: an empty or under-sampled window is sent as no + # rows, which counts as a healthy run (spec decision 8). Always called, even + # without thresholds (no breach), so an escalated doc recovers after they are + # removed. + 'simple_metrics': (simple_metrics or []) if enough_samples else [], + 'thresholds': thresholds, + 'interval_minutes': interval_minutes, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(minutes=1), + ) diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index 1c8cf8b..79128b0 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/tests/laborious/activities/test_model_metrics.py @@ -1424,16 +1424,33 @@ def test_adaptive_state_unchanged_does_not_write(model_metrics_activity, mongo): mongo.update_one.assert_not_called() -def test_adaptive_state_invalid_frequency_alert_is_swallowed(model_metrics_activity, mongo): - mongo.find_one.return_value = _pipeline(frequency_alert='5min') +@mark.parametrize('frequency_alert', ['5min', '0m', 5]) +def test_adaptive_state_invalid_frequency_alert_is_skipped( + model_metrics_activity, mongo, frequency_alert +): + mongo.find_one.return_value = _pipeline(frequency_alert=frequency_alert) assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None mongo.update_one.assert_not_called() - model_metrics_activity.error.assert_called_once() + model_metrics_activity.warning.assert_called_once() -def test_adaptive_state_mongo_error_is_swallowed(model_metrics_activity, mongo): - mongo.find_one.side_effect = RuntimeError('mongo down') +@mark.parametrize('adaptive_state', ['garbage', ['escalated']]) +def test_adaptive_state_malformed_state_is_not_escalated( + model_metrics_activity, mongo, adaptive_state +): + mongo.find_one.return_value = _pipeline(frequency_alert='5m', adaptive_state=adaptive_state) - assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None - model_metrics_activity.error.assert_called_once() + result = model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) + + assert result == {'escalated': True, 'healthy_streak': 0} + mongo.update_one.assert_called_once() + + +@mark.parametrize('failing_call', ['find_one', 'update_one']) +def test_adaptive_state_mongo_error_propagates(model_metrics_activity, mongo, failing_call): + mongo.find_one.return_value = _pipeline(frequency_alert='5m') + getattr(mongo, failing_call).side_effect = RuntimeError('mongo down') + + with raises(RuntimeError, match='mongo down'): + model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) diff --git a/tests/laborious/workflows/test_simple_metrics.py b/tests/laborious/workflows/test_simple_metrics.py index dfb6833..065ccec 100644 --- a/tests/laborious/workflows/test_simple_metrics.py +++ b/tests/laborious/workflows/test_simple_metrics.py @@ -1,6 +1,6 @@ from unittest.mock import ANY, AsyncMock, MagicMock, call, patch -from pytest import fixture, mark +from pytest import fixture, mark, raises from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ from laborious.activities.activities import Activities @@ -218,7 +218,7 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim # Act await simple_metrics.run(input_data) - # Assert - all three activities ran (a swallowed exhausted side_effect would warn) + # Assert - all three activities ran assert workflow_mock.execute_activity_method.call_count == 3 workflow_mock.logger.warning.assert_not_called() @@ -402,13 +402,16 @@ async def test_run_below_min_samples_sends_no_rows_to_adaptive_state(workflow_mo @mark.asyncio @patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) -async def test_run_adaptive_state_failure_does_not_fail_workflow(workflow_mock, simple_metrics): +async def test_run_adaptive_state_failure_fails_workflow(workflow_mock, simple_metrics): workflow_mock.logger = MagicMock() workflow_mock.execute_activity_method = AsyncMock( side_effect=[[_ROW] * 3, None, RuntimeError('dispatch failed')] ) workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS) - await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS)) + with raises(RuntimeError, match='dispatch failed'): + await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS)) - workflow_mock.logger.warning.assert_called_once() + # Metrics were exported before the failing call + calls = workflow_mock.execute_activity_method.call_args_list + assert calls[1].args[0] is Activities.export_data_to_postgres