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