gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(simple_metrics): fail the run explicitly when the adaptive state update fails (SIENTIAPDE-2191)

Browse Source 
Drop the try/except around update_simple_metrics_adaptive_state in the workflow and the broad except in the activity, so Mongo failures fail the run and show up in the Temporal metrics (PR #59 review). Invalid frequency_alert stays a warning + skip (the orchestrator ignores it too) and a malformed adaptive_state is read as not escalated.
This commit is contained in:
PedroHMCosme
2026-10-08 17:45:44 -03:00
parent 877a0c7543
commit d265e3328c
4 changed files with 98 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
laborious/activities/model_metrics.py
View File

@@ -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

37
laborious/workflows/simple_metrics.py
View File

@@ -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),
)

31
tests/laborious/activities/test_model_metrics.py
View File

@@ -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)

13
tests/laborious/workflows/test_simple_metrics.py
View File

@@ -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