580 lines
25 KiB
Python
580 lines
25 KiB
Python
from temporalio import activity, workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
import time
|
|
import traceback
|
|
import warnings
|
|
from math import ceil
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
from pandas import DataFrame, Index, Series, to_datetime
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
|
from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
|
|
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now
|
|
from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError
|
|
from sientia_model.metrics.regression import RegressionMetrics
|
|
|
|
from laborious import metrics
|
|
from laborious.utils.dataframe_debug import build_dataframe_debug_message
|
|
|
|
warnings.filterwarnings('ignore', category=RuntimeWarning, message='Degrees of freedom <= 0')
|
|
warnings.filterwarnings(
|
|
'ignore', category=RuntimeWarning, message='invalid value encountered in scalar divide'
|
|
)
|
|
|
|
|
|
def threshold_breaches(
|
|
rows: list[dict[str, Any]], thresholds: dict[str, float] | None
|
|
) -> list[str]:
|
|
"""
|
|
Breach messages for metric rows outside their configured thresholds.
|
|
|
|
Convention: `<metric>_max` breaches when value > threshold, `<metric>_min` when value <
|
|
threshold. Shared by the breach alert and the adaptive-state rule so both agree on what a
|
|
breach is.
|
|
"""
|
|
thresholds = thresholds or {}
|
|
breaches = []
|
|
for row in rows:
|
|
metric_name, value = row['metric'], row['value']
|
|
max_key, min_key = f'{metric_name}_max', f'{metric_name}_min'
|
|
if max_key in thresholds and value > thresholds[max_key]:
|
|
breaches.append(f'{metric_name}={value} exceeds {max_key}={thresholds[max_key]}')
|
|
elif min_key in thresholds and value < thresholds[min_key]:
|
|
breaches.append(f'{metric_name}={value} below {min_key}={thresholds[min_key]}')
|
|
return breaches
|
|
|
|
|
|
_MAX_RECOVERY_RUNS = 12
|
|
_FREQUENCY_UNIT_SECONDS = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
|
|
|
|
|
|
def frequency_to_seconds(frequency: str) -> int:
|
|
"""`'5m'` -> 300. Same `'<int><s|m|h|d>'` format the orchestrator schedules with."""
|
|
# ponytail: copy of orchestrator/utils/converters.py:parse_frequency (no shared helper in
|
|
# sientia_do), same int() parsing so both repos accept the same strings, plus a > 0 guard;
|
|
# move it to sientia_do if a third repo needs it.
|
|
unit = _FREQUENCY_UNIT_SECONDS.get(frequency[-1:])
|
|
if unit is None:
|
|
raise ValueError(f'Invalid frequency: {frequency!r}')
|
|
value = int(frequency[:-1])
|
|
if value <= 0:
|
|
raise ValueError(f'Invalid frequency: {frequency!r}')
|
|
return value * unit
|
|
|
|
|
|
def recovery_runs(interval_minutes: int, frequency_alert: str) -> int:
|
|
"""
|
|
N of the hysteresis rule: healthy runs needed to renew the whole window without a breach,
|
|
capped at 12 (SIENTIAPDE-1995 decision 3).
|
|
"""
|
|
runs = ceil(interval_minutes * 60 / frequency_to_seconds(frequency_alert))
|
|
return min(_MAX_RECOVERY_RUNS, max(1, runs))
|
|
|
|
|
|
def next_adaptive_state(state: dict[str, Any], breached: bool, runs: int) -> dict[str, Any]:
|
|
"""
|
|
Hysteresis rule of the adaptive frequency (SIENTIAPDE-1995): escalate on the first breach,
|
|
recover after `runs` consecutive non-breach runs.
|
|
"""
|
|
if breached:
|
|
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
|
|
if streak >= runs:
|
|
return {'escalated': False, 'healthy_streak': 0}
|
|
return {'escalated': True, 'healthy_streak': streak}
|
|
|
|
|
|
class ModelMetrics(SientiaMonitoring):
|
|
"""
|
|
Metrics and statistical analysis activities for the Laborious pipeline.
|
|
|
|
This class centralizes drift/statistical computations and model-quality
|
|
aggregates used by scheduled workflows. Besides producing tabular outputs
|
|
for persistence, it also emits operational metrics (count, lag, error)
|
|
through ``SientiaMonitoring`` so execution health is observable in runtime.
|
|
"""
|
|
|
|
_MAX_DEBUG_DATAFRAME_ROWS = 100
|
|
|
|
# ponytail: same declaration as PredictionMetadata; the Activities composite sets the
|
|
# instance value in PredictionMetadata.__init__. A standalone ModelMetrics (unit tests) keeps
|
|
# None and the adaptive-state activity becomes a no-op.
|
|
mongo_repository: MongoDBRepository | None = None
|
|
_PIPELINES_COLLECTION = 'pipelines'
|
|
|
|
def __init__(
|
|
self,
|
|
logger: Logger,
|
|
notification_handler: NotificationHandler,
|
|
metrics_controller: MetricsController,
|
|
):
|
|
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
|
|
|
|
def close(self) -> None:
|
|
"""
|
|
Shutdown monitoring resources associated with model metrics activities.
|
|
|
|
This is invoked during worker teardown to flush/close metric controller
|
|
internals and prevent dangling telemetry tasks.
|
|
"""
|
|
SientiaMonitoring.shutdown(self)
|
|
|
|
def __del__(self):
|
|
self.close()
|
|
|
|
def _debug_dataframe(self, message: str, data: Any, metadata: dict[str, Any]) -> None:
|
|
"""
|
|
Log dataframe content only when row count is below the configured threshold
|
|
|
|
Args:
|
|
- message (str): Base log message to identify the dataframe in logs
|
|
- data (Any): Dataframe-like payload to be logged
|
|
- metadata (dict[str, Any]): Workflow metadata for contextual logging
|
|
"""
|
|
self.debug(
|
|
build_dataframe_debug_message(
|
|
message=message,
|
|
data=data,
|
|
max_rows=self._MAX_DEBUG_DATAFRAME_ROWS,
|
|
),
|
|
metadata,
|
|
)
|
|
|
|
def _drift_analyze_stage_error(
|
|
self,
|
|
exc: Exception,
|
|
context: str,
|
|
metadata: dict[str, Any],
|
|
core_labels: dict[str, Any],
|
|
) -> None:
|
|
"""
|
|
Log analyzer failure for a drift stage and increment the analyze error metric.
|
|
|
|
Args:
|
|
- exc (Exception): Failure raised by ``sientia_model``.
|
|
- context (str): Short label for the log line (e.g. univariate detection).
|
|
- metadata (dict[str, Any]): Workflow metadata for logging.
|
|
- core_labels (dict[str, Any]): Tags from ``get_core_labels`` for metrics.
|
|
"""
|
|
self.error(f'{context}: {exc}', metadata)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_ERROR_COUNT, tags=core_labels)
|
|
|
|
def get_drift_metrics(
|
|
self,
|
|
reference_data: DataFrame,
|
|
target_data: DataFrame,
|
|
target_name: str,
|
|
reference_columns: Index,
|
|
drift_metrics: list[str],
|
|
chunk_period: str,
|
|
metadata: dict[str, Any],
|
|
) -> DataFrame:
|
|
"""
|
|
Compute univariate and multivariate drift outputs and merge them into one dataframe.
|
|
|
|
The method orchestrates three analysis stages (univariate drift,
|
|
multivariate drift, and dataframe projection), emitting lag/count/error
|
|
metrics for each stage independently so failures are attributable.
|
|
|
|
Args:
|
|
- reference_data (DataFrame): Baseline dataset representing expected behavior.
|
|
- target_data (DataFrame): Current analysis dataset to compare against reference.
|
|
- target_name (str): Target column name used by ``DriftAnalysis`` config.
|
|
- reference_columns (Index): Feature columns evaluated for drift.
|
|
- drift_metrics (list[str]): Enabled univariate methods.
|
|
- chunk_period (str): Time bucket granularity used by analysis methods.
|
|
- metadata (dict[str, Any]): Workflow metadata for logs and notifications.
|
|
|
|
Return:
|
|
DataFrame: Consolidated drift dataframe from ``get_drift_metrics_dataframe`` using
|
|
``method`` / ``value`` (and optional ``threshold``, ``drift_type``), ready for
|
|
activity-level formatting before Postgres export.
|
|
"""
|
|
# ``DriftAnalysis`` uses truthiness checks on ``features`` (e.g. ``if not features``);
|
|
# a pandas ``Index`` is ambiguous in boolean context — normalize to a list.
|
|
feature_names: list[str] = list(reference_columns)
|
|
|
|
config = {
|
|
'target': target_name,
|
|
'prediction': 'prediction',
|
|
'timestamp': 'timestamp',
|
|
'features': feature_names,
|
|
}
|
|
|
|
drift_analysis = DriftAnalysis(config=config)
|
|
|
|
self._debug_dataframe(
|
|
f'Reference data: Size {reference_data.shape}', reference_data, metadata
|
|
)
|
|
|
|
self._debug_dataframe(f'Target data: Size {target_data.shape}', target_data, metadata)
|
|
|
|
core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift')
|
|
start_time = time.time()
|
|
try:
|
|
univariate_drift = drift_analysis.detect_univariate_drift(
|
|
reference_df=reference_data,
|
|
analysis_df=target_data,
|
|
features=feature_names,
|
|
timestamp_col=config['timestamp'],
|
|
methods=drift_metrics,
|
|
chunk_period=chunk_period,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
raise
|
|
self._drift_analyze_stage_error(
|
|
e, 'Error detecting univariate drift', metadata, core_labels
|
|
)
|
|
raise
|
|
self.observe_lag_sync(start_time, metrics.MODEL_ANALYZE_LAG, core_labels)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels)
|
|
|
|
core_labels = self.get_core_labels(metadata, operation_type='detect_multivariate_drift')
|
|
start_time = time.time()
|
|
try:
|
|
multivariate_drift = drift_analysis.detect_multivariate_drift(
|
|
reference_df=reference_data,
|
|
analysis_df=target_data,
|
|
features=feature_names,
|
|
timestamp_col=config['timestamp'],
|
|
chunk_period=chunk_period,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
raise
|
|
self._drift_analyze_stage_error(
|
|
e, 'Error detecting multivariate drift', metadata, core_labels
|
|
)
|
|
raise
|
|
self.observe_lag_sync(start_time, metrics.MODEL_ANALYZE_LAG, core_labels)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels)
|
|
|
|
start_time = time.time()
|
|
core_labels = self.get_core_labels(metadata, operation_type='get_drift_metrics_dataframe')
|
|
try:
|
|
drift_df = drift_analysis.get_drift_metrics_dataframe(
|
|
univariate_drift=univariate_drift,
|
|
multivariate_drift=multivariate_drift,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
raise
|
|
self._drift_analyze_stage_error(
|
|
e, 'Error building drift metrics dataframe', metadata, core_labels
|
|
)
|
|
raise
|
|
self.observe_lag_sync(start_time, metrics.MODEL_ANALYZE_LAG, core_labels)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels)
|
|
|
|
return drift_df
|
|
|
|
@staticmethod
|
|
def _to_naive_utc(series: Series) -> Series:
|
|
"""
|
|
Parse ``series`` as datetime and return a TZ-naive UTC copy.
|
|
|
|
``sientia_model.analytics.drift_analysis.DriftAnalysis`` preserves the
|
|
timezone of the input dataframe in its outputs, while target rows
|
|
loaded from PostgreSQL come in with ``+00:00``. Forcing both sides of
|
|
a comparison to TZ-naive UTC keeps ``isin`` / ``floor`` operations
|
|
deterministic regardless of how the analyzer constructs its
|
|
timestamps.
|
|
|
|
Args:
|
|
- series (Series): Input series containing datetime-parseable values.
|
|
|
|
Return:
|
|
Series: Datetime64 series with ``tz=None`` representing UTC instants.
|
|
"""
|
|
parsed = to_datetime(series)
|
|
if getattr(parsed.dt, 'tz', None) is not None:
|
|
parsed = parsed.dt.tz_convert('UTC').dt.tz_localize(None)
|
|
return parsed
|
|
|
|
@activity.defn(name='calculate_drift')
|
|
def calculate_drift(self, input_data: dict[str, Any]) -> list[dict]:
|
|
"""
|
|
Calculate drift metrics for a model.
|
|
|
|
Args:
|
|
input_data (dict[str, Any]): Input data containing:
|
|
- metadata (dict): Workflow execution metadata
|
|
- model_name (str): Name of the MLFlow model to calculate drift for
|
|
- reference_data (pd.DataFrame): Reference data for the model
|
|
- target_data (pd.DataFrame): Target data for calculating drift
|
|
- target_name (str): Name of the target column
|
|
- drift_metrics (list[str]): List of drift metrics to calculate
|
|
"""
|
|
metadata = input_data['metadata']
|
|
model_name = input_data['model_name']
|
|
model_id = input_data['model_id']
|
|
reference_raw_data = input_data['reference_data']
|
|
target_data = DataFrame(input_data['target_data'])
|
|
target_name = input_data['target_name']
|
|
drift_metrics = input_data['drift_metrics']
|
|
chunk_period = input_data['chunk_period']
|
|
|
|
if chunk_period not in ['min', 's']:
|
|
self.error(f'Invalid chunk period: {chunk_period}', metadata)
|
|
raise ValueError(f'Invalid chunk period: {chunk_period}, must be "min" or "s"')
|
|
|
|
self.info(f'Calculating drift for model {model_name}', metadata)
|
|
|
|
target_data = target_data.pivot(index='timestamp', columns='variable', values='value')
|
|
target_data['timestamp'] = target_data.index
|
|
# Keep timestamps as datetime: DriftAnalysis._chunk_dataframe relies on
|
|
# ``pd.Grouper(freq=...)`` which rejects string timestamp columns.
|
|
target_data['timestamp'] = to_datetime(target_data['timestamp'])
|
|
target_data = target_data.reset_index(drop=True)
|
|
target_data.dropna(inplace=True)
|
|
|
|
if reference_raw_data is not None:
|
|
self.info('Using reference data', metadata)
|
|
reference_data = DataFrame(reference_raw_data)
|
|
if 'timestamp' in reference_data.columns:
|
|
reference_data['timestamp'] = to_datetime(reference_data['timestamp'])
|
|
accurate = True
|
|
else:
|
|
# Get 30% first rows of target_data
|
|
self.warning('Using 30% first rows of target data as reference data', metadata)
|
|
target_data.sort_values(by='timestamp', ascending=True, inplace=True)
|
|
reference_data = target_data.head(int(len(target_data) * 0.3))
|
|
accurate = False
|
|
|
|
self.send_notification(
|
|
metadata=metadata,
|
|
notification_id='MODEL_METRICS_REFERENCE_DATA_WARNING',
|
|
message='Using 30% first rows of target data as reference data',
|
|
block='model_metrics',
|
|
level=NotificationLevel.WARNING,
|
|
attachment_content=reference_data.to_csv(),
|
|
)
|
|
|
|
reference_columns = reference_data.drop(
|
|
columns=[target_name, 'timestamp', 'target', 'prediction'], errors='ignore'
|
|
).columns
|
|
|
|
try:
|
|
drift_df = self.get_drift_metrics(
|
|
reference_data=reference_data,
|
|
target_data=target_data,
|
|
target_name=target_name,
|
|
reference_columns=reference_columns,
|
|
drift_metrics=drift_metrics,
|
|
chunk_period=chunk_period,
|
|
metadata=metadata,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
self.error(str(e), metadata)
|
|
notification_id = e.notification_id
|
|
notification_message = str(e)
|
|
else:
|
|
self.error(f'Error getting drift metrics: {e}', metadata)
|
|
notification_id = 'MODEL_METRICS_GET_DRIFT_METRICS_ERROR'
|
|
notification_message = f'Error getting drift metrics: {e}'
|
|
self.send_notification(
|
|
metadata=metadata,
|
|
notification_id=notification_id,
|
|
message=notification_message,
|
|
block='model_metrics',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=traceback.format_exc(),
|
|
)
|
|
raise
|
|
|
|
# Drop chunks whose floored timestamp does not appear in the analysis window.
|
|
# ``DriftAnalysis`` chunks over ``analysis_df``; this only excludes rows that
|
|
# do not belong to the current target window (e.g. stray merged reference rows).
|
|
target_floor = self._to_naive_utc(target_data['timestamp']).dt.floor(chunk_period)
|
|
drift_floor = self._to_naive_utc(drift_df['timestamp']).dt.floor(chunk_period)
|
|
drift_df = drift_df[drift_floor.isin(target_floor)]
|
|
|
|
if drift_df.empty:
|
|
self.warning(
|
|
'No drift metrics found after dropping rows where timestamp is not in target data',
|
|
metadata,
|
|
)
|
|
return []
|
|
|
|
# Analyzer emits diagnostic columns that are not stored in ``sientia_data.drift_metrics``.
|
|
drift_df = drift_df.drop(columns=['threshold', 'drift_type'], errors='ignore')
|
|
|
|
drift_df['model_id'] = str(model_id)
|
|
drift_df['accurate'] = accurate
|
|
|
|
# ``timestamp`` is overridden with the most recent target instant so
|
|
# every persisted row shares a single business timestamp (the run's
|
|
# logical "now"), matching what downstream consumers expect.
|
|
latest_target_timestamp = self._to_naive_utc(target_data['timestamp']).max()
|
|
drift_df['timestamp'] = (
|
|
pd.Timestamp(latest_target_timestamp)
|
|
.tz_localize('UTC')
|
|
.strftime(DATETIME_FORMAT_WITH_TZ)
|
|
)
|
|
|
|
# ``chunk_start_date`` / ``chunk_end_date`` may carry nanosecond
|
|
# precision (beyond ``timestamptz`` microseconds), so serialize as ISO
|
|
# text for the ``text`` Postgres columns.
|
|
for column in ('chunk_start_date', 'chunk_end_date'):
|
|
drift_df[column] = drift_df[column].apply(
|
|
lambda value: pd.Timestamp(value).isoformat() if pd.notna(value) else None
|
|
)
|
|
|
|
self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata)
|
|
|
|
return drift_df.to_dict(orient='records')
|
|
|
|
@activity.defn(name='calculate_simple_metrics')
|
|
def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
|
|
"""
|
|
Calculate simple regression metrics for a model using RegressionMetrics.
|
|
|
|
Args:
|
|
input_data: Input data containing:
|
|
- metadata (dict): Workflow execution metadata
|
|
- model_id (str): ID of the MLFlow model
|
|
- target_data (list[dict]): Target data with target, prediction, timestamp columns
|
|
- metrics (list[str]): Metric names to calculate
|
|
- interval_minutes (int): Window interval in minutes
|
|
- model_type (str | None): Model algorithm type (for r2 lock)
|
|
Returns:
|
|
list[dict]: Records with metric, value, model_id, timestamp, data_size, interval_minutes
|
|
"""
|
|
metadata = input_data['metadata']
|
|
model_id = input_data['model_id']
|
|
target_data = DataFrame(input_data['target_data'])
|
|
metric_names = list(input_data['metrics'])
|
|
interval_minutes = input_data['interval_minutes']
|
|
model_type = input_data.get('model_type')
|
|
|
|
data_size = target_data.shape[0]
|
|
|
|
# Filter r2 when model_type is known and unsupported
|
|
if (
|
|
model_type
|
|
and 'r2' in metric_names
|
|
and not RegressionMetrics.is_r2_supported(model_type)
|
|
):
|
|
metric_names = [m for m in metric_names if m != 'r2']
|
|
self.warning(
|
|
f'r2 excluded for model {model_id}: not supported for model_type={model_type}',
|
|
metadata,
|
|
)
|
|
self.send_notification(
|
|
metadata=metadata,
|
|
notification_id='SIMPLE_METRICS_R2_UNSUPPORTED',
|
|
message=f'r2 excluded: not a valid metric for model_type={model_type}',
|
|
block='model_metrics',
|
|
level=NotificationLevel.WARNING,
|
|
)
|
|
|
|
if not metric_names:
|
|
self.info(f'No metrics to calculate for model {model_id} after filtering', metadata)
|
|
return []
|
|
|
|
self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
|
|
|
|
# Build Series with DatetimeIndex for RegressionMetrics
|
|
timestamps = pd.to_datetime(target_data['timestamp'])
|
|
real_data = Series(target_data['target'].values, index=timestamps, dtype=float)
|
|
predictions = Series(target_data['prediction'].values, index=timestamps, dtype=float)
|
|
|
|
regression = RegressionMetrics(real_data, predictions)
|
|
output_data = regression.calculate(metric_names)
|
|
|
|
# Wrap with metadata columns matching the existing output schema
|
|
data = DataFrame(output_data)
|
|
data['model_id'] = model_id
|
|
data['timestamp'] = target_data['timestamp'].max()
|
|
data['data_size'] = data_size
|
|
data['interval_minutes'] = interval_minutes
|
|
|
|
self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata)
|
|
|
|
# Threshold alerting (optional — no crash when absent)
|
|
for breach_msg in threshold_breaches(output_data, input_data.get('thresholds')):
|
|
message = (
|
|
f'Threshold breach for model {model_id} '
|
|
f'(window {interval_minutes} min): {breach_msg}'
|
|
)
|
|
self.warning(message, metadata)
|
|
self.send_notification(
|
|
# Model-level trigger: the orchestrator dedups alerts on
|
|
# `<trigger>:<notification_id>`, so one breach seen by several windows
|
|
# (one schedule each) becomes one alert (SIENTIAPDE-1995).
|
|
metadata={**metadata, 'schedule_name': f'{model_id}-simple_metrics'},
|
|
notification_id='SIMPLE_METRICS_THRESHOLD_BREACH',
|
|
message=message,
|
|
block='model_metrics',
|
|
level=NotificationLevel.WARNING,
|
|
)
|
|
|
|
return data.to_dict(orient='records')
|
|
|
|
@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.
|
|
|
|
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
|
|
input only refreshes on the orchestrator tick and would be stale while escalated.
|
|
`updated_at` is bumped only when `escalated` flips: that is what makes the orchestrator
|
|
rewrite the schedule, and a streak change alone does not change the frequency.
|
|
|
|
Args:
|
|
input_data: Dict with:
|
|
- 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` (may be None: no breach).
|
|
- 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.
|
|
"""
|
|
metadata = input_data['metadata']
|
|
if self.mongo_repository is None:
|
|
return None
|
|
|
|
query = {'schedule_name': metadata['schedule_name']}
|
|
try:
|
|
pipeline = self.mongo_repository.find_one(self._PIPELINES_COLLECTION, query, metadata)
|
|
if not pipeline or 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'))
|
|
)
|
|
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
|
|
|
|
self.info(f'Adaptive state of {query["schedule_name"]}: {state} -> {new_state}', metadata)
|
|
return new_state
|