From e99fe6afc38b18e59f5ff9df9273bdaa007f3cf5 Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Wed, 8 Apr 2026 18:54:33 +0700 Subject: [PATCH] =?UTF-8?q?fix:=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8?= =?UTF-8?q?=D0=BB=20max=5Ffailed=5Fage=20=E2=80=94=20=D0=B8=D0=B3=D0=BD?= =?UTF-8?q?=D0=BE=D1=80=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D1=82=D1=8C=20=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D1=80=D1=8B=D0=B5=20failed=20runs=20(>24=D1=87)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - failed runs старше 24 часов пропускаются как историческая ошибка - предотвращает массовый перезапуск DAG-ов от 2024 года - параметр max_failed_age настраивается в config.yaml --- airflow_monitor/analyzer.py | 14 ++++++++++++-- airflow_monitor/config.py | 1 + airflow_monitor/monitor.py | 5 ++++- config.yaml | 4 ++++ 4 files changed, 21 insertions(+), 3 deletions(-) diff --git a/airflow_monitor/analyzer.py b/airflow_monitor/analyzer.py index 276b9d5..8592601 100644 --- a/airflow_monitor/analyzer.py +++ b/airflow_monitor/analyzer.py @@ -27,11 +27,13 @@ class DagIssue: class DagAnalyzer: """Analyzes DAG runs and produces a list of issues.""" - def __init__(self, long_running_threshold: int): + def __init__(self, long_running_threshold: int, max_failed_age: int = 86400): self._threshold = long_running_threshold + self._max_failed_age = max_failed_age # ignore failed runs older than this (seconds) logger.debug( - "DagAnalyzer initialized: long_running_threshold=%ds (%.1f min)", + "DagAnalyzer initialized: long_running_threshold=%ds (%.1f min), max_failed_age=%ds (%.1f hours)", long_running_threshold, long_running_threshold / 60, + max_failed_age, max_failed_age / 3600, ) def analyze_dag_runs( @@ -81,6 +83,14 @@ class DagAnalyzer: ) if state == "failed": + # Skip old failed runs — only alert on recent failures + if duration > self._max_failed_age: + logger.debug( + " → SKIP: failed run too old (%.0fs > %ds max_failed_age)", + duration, self._max_failed_age, + ) + continue + error_info = self._extract_error(dag_id, run_id, task_instances_fn) issue = DagIssue( dag_id=dag_id, diff --git a/airflow_monitor/config.py b/airflow_monitor/config.py index 7ea8781..105dff4 100644 --- a/airflow_monitor/config.py +++ b/airflow_monitor/config.py @@ -25,6 +25,7 @@ class MonitorConfig: cycle_interval: int = 300 # seconds between full cycles long_running_threshold: int = 1800 # 30 minutes + max_failed_age: int = 86400 # ignore failed runs older than 24 hours retry_wait: int = 120 # wait after retry before re-check max_retries: int = 1 state_file: str = "/var/lib/airflow-monitor/state.json" diff --git a/airflow_monitor/monitor.py b/airflow_monitor/monitor.py index 659cd82..53dc759 100644 --- a/airflow_monitor/monitor.py +++ b/airflow_monitor/monitor.py @@ -32,7 +32,10 @@ class Monitor: self._shutdown = shutdown_event self._client = AirflowClient(config.airflow) self._state = StateManager(config.monitor.state_file, config.monitor.state_max_age) - self._analyzer = DagAnalyzer(config.monitor.long_running_threshold) + self._analyzer = DagAnalyzer( + config.monitor.long_running_threshold, + config.monitor.max_failed_age, + ) self._exporter = DataExporter(config.zabbix) self._actions = ActionHandler( self._client, self._state, self._exporter, config.monitor, diff --git a/config.yaml b/config.yaml index b04843e..09b672a 100644 --- a/config.yaml +++ b/config.yaml @@ -36,6 +36,10 @@ monitor: # DAG run duration threshold to consider as "long running" (seconds) long_running_threshold: 1800 # 30 minutes + # Игнорировать failed runs старше этого порога (seconds) + # Старые ошибки — это история, не текущая проблема + max_failed_age: 86400 # 24 hours + # Time to wait after retry before re-checking (seconds) retry_wait: 120 # 2 minutes