diff --git a/README.md b/README.md index 2c3be54..8e5c04d 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** (only with thresholds): 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; empty or under-sampled 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 #### Architecture Diagram ```mermaid diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 2aa8fa6..074b6fc 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -537,7 +537,7 @@ class ModelMetrics(SientiaMonitoring): - metadata (dict): Workflow metadata; `schedule_name` identifies the doc. - simple_metrics (list[dict]): Rows from `calculate_simple_metrics`; empty for an empty or under-sampled window, which counts as a non-breach run. - - thresholds (dict): `simple_metrics_thresholds` from `model_config`. + - thresholds (dict): `simple_metrics_thresholds` from `model_config` (may be None: no breach). - interval_minutes (int): Window size, used to compute the recovery runs. Return: diff --git a/laborious/workflows/simple_metrics.py b/laborious/workflows/simple_metrics.py index 279d77e..142f59d 100644 --- a/laborious/workflows/simple_metrics.py +++ b/laborious/workflows/simple_metrics.py @@ -77,7 +77,7 @@ class SimpleMetrics: if target_data and not enough_samples: workflow.logger.info( f'{len(target_data)} rows < simple_metrics_min_samples={min_samples}: ' - 'threshold alerting skipped' + 'threshold alerting and escalation skipped' ) simple_metrics = None @@ -114,17 +114,15 @@ class SimpleMetrics: start_to_close_timeout=timedelta(seconds=300), ) - if not thresholds: - # No thresholds = no breach possible, so skip the Mongo round trip. - return - 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). + # 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, diff --git a/tests/laborious/workflows/test_simple_metrics.py b/tests/laborious/workflows/test_simple_metrics.py index dc9b6bf..dfb6833 100644 --- a/tests/laborious/workflows/test_simple_metrics.py +++ b/tests/laborious/workflows/test_simple_metrics.py @@ -42,12 +42,17 @@ async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics): target_data = {'data': 'test_target_data'} simple_metrics_data = {'metrics': 'test_simple_metrics_data'} - workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None]) + workflow_mock.logger = MagicMock() + workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None, None]) workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data) # Act await simple_metrics.run(input_data) + # Assert - load, export and adaptive state (no thresholds still updates it) + assert workflow_mock.execute_activity_method.call_count == 3 + workflow_mock.logger.warning.assert_not_called() + # Assert - Check load_custom_query call # Query format matches psycopg2.sql output (identifiers with double quotes, literals with single quotes) expected_query = f""" @@ -130,13 +135,22 @@ async def test_run_empty_target_data(workflow_mock: AsyncMock, simple_metrics: S target_data = None + workflow_mock.logger = MagicMock() workflow_mock.execute_activity_method = AsyncMock(return_value=target_data) + workflow_mock.execute_local_activity_method = AsyncMock() # Act await simple_metrics.run(input_data) - # Assert - Should not call calculate_simple_metrics or export - assert workflow_mock.execute_activity_method.call_count == 1 + # Assert - No calculate or export; load + adaptive state (empty window = healthy run) + workflow_mock.execute_local_activity_method.assert_not_called() + calls = workflow_mock.execute_activity_method.call_args_list + assert [c.args[0] for c in calls] == [ + Activities.load_custom_query, + Activities.update_simple_metrics_adaptive_state, + ] + assert calls[1].args[1]['simple_metrics'] == [] + workflow_mock.logger.warning.assert_not_called() @mark.asyncio @@ -159,15 +173,22 @@ async def test_run_empty_simple_metrics(workflow_mock: AsyncMock, simple_metrics target_data = {'data': 'test_target_data'} simple_metrics_data = None + workflow_mock.logger = MagicMock() workflow_mock.execute_activity_method = AsyncMock(return_value=target_data) workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data) # Act await simple_metrics.run(input_data) - # Assert - Should call calculate_simple_metrics but not export - workflow_mock.execute_activity_method.assert_called_once() + # Assert - calculate returns None: no export, adaptive state gets [] (the `or []` is live) workflow_mock.execute_local_activity_method.assert_called_once() + calls = workflow_mock.execute_activity_method.call_args_list + assert [c.args[0] for c in calls] == [ + Activities.load_custom_query, + Activities.update_simple_metrics_adaptive_state, + ] + assert calls[1].args[1]['simple_metrics'] == [] + workflow_mock.logger.warning.assert_not_called() @mark.asyncio @@ -190,12 +211,17 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim target_data = {'data': 'test_target_data'} simple_metrics_data = {'metrics': 'test_simple_metrics_data'} - workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None]) + workflow_mock.logger = MagicMock() + workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None, None]) workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data) # Act await simple_metrics.run(input_data) + # Assert - all three activities ran (a swallowed exhausted side_effect would warn) + assert workflow_mock.execute_activity_method.call_count == 3 + workflow_mock.logger.warning.assert_not_called() + # Assert - Check calculate_simple_metrics call with default metrics workflow_mock.execute_activity_method.assert_any_call( Activities.load_custom_query, @@ -305,13 +331,21 @@ async def test_run_updates_adaptive_state_after_export(workflow_mock, simple_met @mark.asyncio @patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) -async def test_run_without_thresholds_skips_adaptive_state(workflow_mock, simple_metrics): - workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 3, None]) +async def test_run_without_thresholds_still_updates_adaptive_state(workflow_mock, simple_metrics): + # No thresholds = never a breach; still calling lets an escalated doc recover after the + # thresholds are removed. + workflow_mock.logger = MagicMock() + workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 3, None, None]) workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS) await simple_metrics.run(_input()) - assert workflow_mock.execute_activity_method.call_count == 2 + calls = workflow_mock.execute_activity_method.call_args_list + assert len(calls) == 3 + assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state + assert calls[2].args[1]['thresholds'] is None + assert calls[2].args[1]['simple_metrics'] == _BREACH_ROWS + workflow_mock.logger.warning.assert_not_called() @mark.asyncio