From d1786085d0f8c91172efcb0ee6ba430b753ac794 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 27 Oct 2026 16:19:51 -0300 Subject: [PATCH] SIENTIAPDE-2212 SIENTIAPDE-2212-drift-janela-e-chunk-period: Add configurable chunk period and dynamic window calculation for drift, with warnings for invalid periods. --- .gitignore | 3 +- orchestrator/activities/formatters.py | 19 ++++- orchestrator/activities/temporal_manager.py | 3 + orchestrator/utils/orchestrator_functions.py | 60 +++++++++++++-- .../activities/test_formatters.py | 43 +++++++++++ .../activities/test_temporal_manager.py | 39 ++++++++++ .../utils/test_orchestrator_functions.py | 75 +++++++++++++++++++ 7 files changed, 235 insertions(+), 7 deletions(-) diff --git a/.gitignore b/.gitignore index 372699a..f8e8af5 100644 --- a/.gitignore +++ b/.gitignore @@ -49,4 +49,5 @@ git_log .env openspec/ -.cursor/ \ No newline at end of file +.cursor/ +.claude/ \ No newline at end of file diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 17a8187..8a06fea 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -19,6 +19,7 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.utils.orchestrator_functions import ( build_tag_config, + chunk_period_warning, drift, frequency_alert_warning, gather_read_tags, @@ -371,7 +372,9 @@ class Formatters(SientiaMonitoring): configuration to determine which schedules need to be created, updated, or deleted. Uses timestamp comparison to identify schedules that have changed. Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules - whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored). + whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored), + and a DRIFT_CHUNK_PERIOD_INVALID WARNING for drift schedules whose configured + chunk_period is not supported by laborious (replaced by 'min'). Args: input_data (dict[str, Any]): The input data containing: @@ -430,6 +433,20 @@ class Formatters(SientiaMonitoring): block='create_schedule_config', level=NotificationLevel.WARNING, ) + warning = chunk_period_warning(schedule.get('chunk_period_configured')) + if warning: + self.send_notification( + metadata={ + **metadata, + 'schedule_name': schedule_name, + 'model_id': schedule.get('model_id'), + 'model_name': schedule.get('model_name'), + }, + notification_id='DRIFT_CHUNK_PERIOD_INVALID', + message=warning, + block='create_schedule_config', + level=NotificationLevel.WARNING, + ) for namespace, schedules in current_schedule_config.items(): for schedule_name in schedules: diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 8fcc95c..ec622e4 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -220,6 +220,9 @@ class TemporalManager(SientiaMonitoring): task_queue_name = self._build_task_queue_name(workflow_type, schedule) schedule['task_queue'] = task_queue_name + # No `policy`: schedules rely on Temporal's default SKIP overlap. The + # laborious `DriftAnalysis` cache assumes at most one running drift + # workflow per model; revisit its locking if a policy is ever added here. await client.create_schedule( schedule_name, Schedule( diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index c5df001..bdbb592 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -1,3 +1,4 @@ +from math import ceil from typing import Any from orchestrator.utils.converters import parse_frequency @@ -43,6 +44,28 @@ def common_config(config: dict[str, Any]): } +# Kept in sync by hand with the `chunk_period` validation in +# sientia-dataops-laborious/laborious/activities/model_metrics.py (no shared contract). +DRIFT_CHUNK_PERIODS = ('min', 's') +# Re-reads the tail of the previous window so rows ingested up to 1 minute late still count. +DRIFT_WINDOW_MARGIN_MINUTES = 1 + + +def _drift_interval_minutes(config: dict[str, Any], frequency: Any) -> int: + """ + Drift window in minutes: `interval_minutes` as configured (no margin), otherwise the + schedule frequency rounded up to whole minutes (min 1) plus DRIFT_WINDOW_MARGIN_MINUTES. + """ + if 'interval_minutes' in config: + return config['interval_minutes'] + try: + seconds = parse_frequency(str(frequency)) + except ValueError: + # create_schedules fails this schedule on its own; don't kill the tick for every model. + return 60 + return max(1, ceil(seconds / 60)) + DRIFT_WINDOW_MARGIN_MINUTES + + def drift(config: dict[str, Any]): """ Build drift configuration from pipeline config. @@ -52,25 +75,52 @@ def drift(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - - interval_minutes (int, optional): Time window for data comparison in minutes (default: 60) + - interval_minutes (int, optional): Time window for data comparison in minutes, + used as-is (no margin). Default: `frequency` rounded up to whole minutes + (min 1) plus a 1-minute margin for late data (2 with the default '1m'). + Pipelines whose ingestion lag exceeds 1 minute should set it explicitly + (frequency + lag); `interval_minutes: 60` keeps the legacy window. + - chunk_period (str, optional): Drift bucket size, 'min' or 's' (default: 'min'). + Any other value falls back to 'min' and create_schedule_config warns. - drift_metrics (list[str], optional): Statistical metrics to compute (default: ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']) - Additional fields from common_config Returns: dict[str, Any]: Drift detection configuration with source/target tables, - time interval, and metrics specifications. Results stored in 'drift_metrics' table + time interval, chunk period and metrics specifications. Results stored in + 'drift_metrics' table """ - return { - **common_config(config), + base = common_config(config) + configured_chunk_period = config.get('chunk_period') + chunk_period_valid = configured_chunk_period in DRIFT_CHUNK_PERIODS + output = { + **base, 'schema': 'sientia_data', 'source_table_name': 'laborious_data', 'target_table_name': 'drift_metrics', - 'interval': config.get('interval_minutes', 60), + 'interval': _drift_interval_minutes(config, base['frequency']), + 'chunk_period': configured_chunk_period if chunk_period_valid else 'min', 'drift_metrics': config.get( 'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'] ), } + if configured_chunk_period is not None and not chunk_period_valid: + # Forwarded so create_schedule_config can warn without re-reading the pipeline. + output['chunk_period_configured'] = configured_chunk_period + return output + + +def chunk_period_warning(chunk_period: Any) -> str | None: + """ + Warning message when a configured drift `chunk_period` is not supported by laborious, + otherwise None. Warn only: `drift()` already replaced it with 'min'. + """ + if chunk_period is None or chunk_period in DRIFT_CHUNK_PERIODS: + return None + return ( + f"chunk_period={chunk_period!r} is not one of {DRIFT_CHUNK_PERIODS}; falling back to 'min'" + ) def simple_metrics(config: dict[str, Any]): diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 4891eb2..320852c 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -1073,3 +1073,46 @@ def test_create_schedule_config_no_warning_at_or_above_floor(formatters): formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m')) formatters.send_notification.assert_not_called() + + +def _drift_schedule_config(chunk_period_configured): + return { + 'metadata': {}, + 'current_schedule_config': {'laborious': {}}, + 'schedule_config': { + 'laborious': { + 'm1-drift-1m': { + 'workflow_type': 'drift', + 'model_id': 'm1', + 'model_name': 'm1', + 'frequency': '1m', + 'chunk_period': 'min', + 'chunk_period_configured': chunk_period_configured, + 'updated_at': '2026-10-27', + } + }, + 'scouter': {}, + }, + } + + +def test_create_schedule_config_warns_on_invalid_chunk_period(formatters): + result = formatters.create_schedule_config(_drift_schedule_config('hour')) + + assert 'm1-drift-1m' in result['to_create']['laborious'] + formatters.send_notification.assert_called_once() + kwargs = formatters.send_notification.call_args.kwargs + assert kwargs['notification_id'] == 'DRIFT_CHUNK_PERIOD_INVALID' + assert kwargs['level'] == NotificationLevel.WARNING + assert kwargs['metadata']['schedule_name'] == 'm1-drift-1m' + assert kwargs['metadata']['model_id'] == 'm1' + assert kwargs['metadata']['model_name'] == 'm1' + + +def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatters): + input_data = _drift_schedule_config('hour') + input_data['current_schedule_config'] = {'laborious': {'m1-drift-1m': '2026-10-27'}} + + formatters.create_schedule_config(input_data) + + formatters.send_notification.assert_not_called() diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index 337ff1a..f3c44ed 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -3,6 +3,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel +from temporalio.client import ScheduleOverlapPolicy from orchestrator.activities.temporal_manager import TemporalManager from orchestrator.utils.converters import parse_frequency @@ -684,3 +685,41 @@ async def test_create_schedules_tenant_runtime_queue( task_timeout=timedelta(seconds=300), typed_search_attributes=_mock_typed_search_attributes.return_value, ) + + +@mark.asyncio +@patch('orchestrator.activities.temporal_manager.Schedule') +@patch('orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow') +@patch('orchestrator.activities.temporal_manager.ScheduleIntervalSpec') +@patch('orchestrator.activities.temporal_manager.ScheduleSpec') +@patch('orchestrator.activities.temporal_manager.TypedSearchAttributes') +@patch('orchestrator.activities.temporal_manager.SearchAttributePair') +async def test_create_schedules_drift_keeps_default_skip_overlap( + _mock_search_attribute_pair, + _mock_typed_search_attributes, + _mock_schedule_spec, + _mock_schedule_interval_spec, + _mock_schedule_action_start_workflow, + mock_schedule, + temporal_manager, +): + # The laborious DriftAnalysis cache assumes no two drift runs of a model overlap. + input_data = { + 'schedules': { + 'laborious': { + 'm1-drift-1m': { + 'model_id': 'm1', + 'model_name': 'm1', + 'workflow_type': 'drift', + 'frequency': '1m', + } + } + } + } + temporal_manager.temporal_clients['laborious'].create_schedule = AsyncMock() + + await temporal_manager.create_schedules(input_data) + + mock_schedule.assert_called_once() + policy = mock_schedule.call_args.kwargs.get('policy') + assert policy is None or policy.overlap == ScheduleOverlapPolicy.SKIP diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 45209ee..769c679 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -5,6 +5,7 @@ from pytest import mark from orchestrator.utils.orchestrator_functions import ( base_scouter, build_tag_config, + chunk_period_warning, common_config, drift, frequency_alert_warning, @@ -79,6 +80,7 @@ def test_drift(): 'source_table_name': 'laborious_data', 'target_table_name': 'drift_metrics', 'interval': 120, + 'chunk_period': 'min', 'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'], 'execution_timeout_seconds': 300, 'task_timeout_seconds': 300, @@ -87,6 +89,79 @@ def test_drift(): assert result == expected +def _drift_config(**fields): + return { + 'workflow_type': 'drift', + 'schedule_name': 'm1-drift', + 'model_id': 'm1', + 'model': {'name': 'm1'}, + **fields, + } + + +def test_drift_default_window_is_frequency_plus_margin(): + result = drift(_drift_config()) + + assert result['frequency'] == '1m' + assert result['interval'] == 2 + + +@mark.parametrize( + ('frequency', 'interval'), + [('1h', 61), ('30s', 2), ('90s', 3)], +) +def test_drift_window_derived_from_frequency(frequency, interval): + assert drift(_drift_config(frequency=frequency))['interval'] == interval + + +def test_drift_explicit_interval_minutes_wins_without_margin(): + result = drift(_drift_config(frequency='1m', interval_minutes=60)) + + assert result['interval'] == 60 + + +def test_drift_invalid_frequency_falls_back_to_legacy_window(): + result = drift(_drift_config(frequency='abc')) + + assert result['frequency'] == 'abc' + assert result['interval'] == 60 + + +def test_drift_chunk_period_defaults_to_min(): + result = drift(_drift_config()) + + assert result['chunk_period'] == 'min' + assert 'chunk_period_configured' not in result + + +def test_drift_chunk_period_seconds_kept(): + result = drift(_drift_config(chunk_period='s')) + + assert result['chunk_period'] == 's' + assert 'chunk_period_configured' not in result + + +@mark.parametrize('chunk_period', ['hour', 5]) +def test_drift_invalid_chunk_period_falls_back_to_min(chunk_period): + result = drift(_drift_config(chunk_period=chunk_period)) + + assert result['chunk_period'] == 'min' + assert result['chunk_period_configured'] == chunk_period + + +@mark.parametrize('chunk_period', [None, 'min', 's']) +def test_chunk_period_warning_none_when_absent_or_valid(chunk_period): + assert chunk_period_warning(chunk_period) is None + + +@mark.parametrize('chunk_period', ['hour', 5]) +def test_chunk_period_warning_message_when_invalid(chunk_period): + warning = chunk_period_warning(chunk_period) + + assert warning is not None + assert repr(chunk_period) in warning + + def test_simple_metrics(): config = { 'workflow_type': 'simple_metrics',