diff --git a/airflow_monitor/actions.py b/airflow_monitor/actions.py index 9b6bd24..d383e7f 100644 --- a/airflow_monitor/actions.py +++ b/airflow_monitor/actions.py @@ -211,54 +211,54 @@ class ActionHandler: return "needs_alert" def collect_and_alert(self, issues: list[DagIssue]): - """Export alerts for issues that need alerting. + """Export all current issues and mark new ones as alerted. - Filters to only unalerted issues, builds JSON payload, - writes to file for Zabbix agent, and marks as alerted in state. + Always exports the full list of current problems so Zabbix sees them + until they are resolved. Only logs warnings for newly discovered issues. """ - to_alert = [] + # Build export payload for ALL current issues + all_problems = [] + new_issues = [] for issue in issues: + entry = self._state.ensure_entry(issue.dag_id, issue.dag_run_id) + alert_data = { + "dag_id": issue.dag_id, + "dag_run_id": issue.dag_run_id, + "issue_type": issue.issue_type, + "status": issue.state, + "duration_seconds": issue.duration_seconds, + "error_info": issue.error_info, + "retry_count": entry["retry_count"], + } + all_problems.append(alert_data) if not self._state.is_alerted(issue.dag_id, issue.dag_run_id): - entry = self._state.ensure_entry(issue.dag_id, issue.dag_run_id) - alert_data = { - "dag_id": issue.dag_id, - "dag_run_id": issue.dag_run_id, - "issue_type": issue.issue_type, - "status": issue.state, - "duration_seconds": issue.duration_seconds, - "error_info": issue.error_info, - "retry_count": entry["retry_count"], - } - to_alert.append(alert_data) - logger.debug( - "Issue queued for alert: %s/%s type=%s retries=%d", - issue.dag_id, issue.dag_run_id, - issue.issue_type, entry["retry_count"], - ) + new_issues.append(issue) - if not to_alert: - # Write empty list to clear Zabbix trigger + if not all_problems: self._exporter.export_problems([]) - logger.info("No issues to alert, exported empty problem list") + logger.info("No issues, exported empty problem list") return - logger.warning( - "Exporting %d problematic DAG runs for Zabbix: %s", - len(to_alert), - ", ".join( - f"{p['dag_id']}/{p['dag_run_id']}({p['issue_type']})" - for p in to_alert - ), + # Log new issues at warning level + if new_issues: + logger.warning( + "New problematic DAG runs: %s", + ", ".join( + f"{i.dag_id}/{i.dag_run_id}({i.issue_type})" + for i in new_issues + ), + ) + + logger.debug("Alert payload:\n%s", json.dumps(all_problems, indent=2, ensure_ascii=False)) + + self._exporter.export_problems(all_problems) + + # Mark new issues as alerted + for issue in new_issues: + self._state.mark_alerted(issue.dag_id, issue.dag_run_id) + logger.debug("Marked as alerted: %s/%s", issue.dag_id, issue.dag_run_id) + + logger.info( + "Exported %d issues to problems file (%d new)", + len(all_problems), len(new_issues), ) - - # Log full alert payload at debug level - logger.debug("Alert payload:\n%s", json.dumps(to_alert, indent=2, ensure_ascii=False)) - - self._exporter.export_problems(to_alert) - - for issue in issues: - if not self._state.is_alerted(issue.dag_id, issue.dag_run_id): - self._state.mark_alerted(issue.dag_id, issue.dag_run_id) - logger.debug("Marked as alerted: %s/%s", issue.dag_id, issue.dag_run_id) - - logger.info("Exported %d issues to problems file for Zabbix agent", len(to_alert)) diff --git a/zabbix/zbx_template_airflow_monitor_5x.xml b/zabbix/zbx_template_airflow_monitor_5x.xml index c9a0cca..743b57e 100644 --- a/zabbix/zbx_template_airflow_monitor_5x.xml +++ b/zabbix/zbx_template_airflow_monitor_5x.xml @@ -1,6 +1,6 @@ - 5.4 + 5.0 Templates/Applications @@ -17,7 +17,7 @@ - Установлен airflow-monitor (/opt/airflow-monitor) - Установлен конфиг /etc/zabbix/zabbix_agentd.conf.d/airflow-monitor.conf -Версия шаблона: 1.0.0 (Zabbix 5.x) +Версия шаблона: 1.0.1 (Zabbix 5.0) Templates/Applications @@ -32,12 +32,6 @@ 7d TEXT JSON-массив проблемных DAG-запусков. [] = нет проблем. - - - component - airflow - - @@ -45,16 +39,10 @@ Airflow: Problematic DAGs count airflow.dag.problems.count 1m - 90d + 7d 365d UNSIGNED Количество DAG-запусков с проблемами. - - - component - airflow - - @@ -63,14 +51,9 @@ airflow.monitor.heartbeat 1m 7d + 365d UNSIGNED Epoch timestamp последнего успешного цикла мониторинга. - - - component - dag-monitor - - @@ -81,12 +64,6 @@ 7d TEXT JSON со статусом монитора. - - - component - dag-monitor - - @@ -94,19 +71,13 @@ Airflow Monitor: Service alive airflow.monitor.alive 1m - 90d + 7d 365d UNSIGNED 1 = сервис запущен, 0 = остановлен. Service state - - - component - dag-monitor - - @@ -114,7 +85,7 @@ Airflow Monitor: DAG count DEPENDENT airflow.monitor.dag_count - 90d + 7d 365d UNSIGNED Количество отслеживаемых DAG-ов. @@ -124,23 +95,13 @@ JSONPATH - - $.dag_count - + $.dag_count DISCARD_UNCHANGED_HEARTBEAT - - 1h - + 1h - - - component - airflow - - @@ -148,7 +109,7 @@ Airflow Monitor: Cycle time (seconds) DEPENDENT airflow.monitor.cycle_time - 90d + 7d 365d FLOAT s @@ -159,17 +120,9 @@ JSONPATH - - $.cycle_time_seconds - + $.cycle_time_seconds - - - component - dag-monitor - - @@ -177,7 +130,7 @@ Airflow Monitor: Cycle count DEPENDENT airflow.monitor.cycle_count - 90d + 7d 365d UNSIGNED Номер текущего цикла мониторинга. @@ -187,133 +140,16 @@ JSONPATH - - $.cycle_count - + $.cycle_count DISCARD_UNCHANGED_HEARTBEAT - - 1h - + 1h - - - component - dag-monitor - - - - - - {Airflow DAG Monitor:airflow.dag.problems.count.last()}>0 - {Airflow DAG Monitor:airflow.dag.problems.count.last()}=0 - Airflow: {ITEM.LASTVALUE1} problematic DAG run(s) detected - HIGH - Обнаружены проблемные DAG-запуски (failed или long_running). -Подробности в item "Airflow: Problematic DAGs (JSON)". -Триггер снимается автоматически при разрешении всех проблем. - - - scope - availability - - - - - - - {Airflow DAG Monitor:airflow.monitor.heartbeat.nodata(600)}=1 or (({Airflow DAG Monitor:airflow.monitor.heartbeat.last()}>0) and ({Airflow DAG Monitor:airflow.monitor.heartbeat.fuzzytime(600)}=0)) - Airflow Monitor: Data is stale (no update for 10+ minutes) - DISASTER - Данные мониторинга не обновлялись более 10 минут. -Возможные причины: сервис остановлен, Airflow API недоступен. - - - Airflow Monitor: Service is not running - {Airflow DAG Monitor:airflow.monitor.alive.last()}=0 - - - - - scope - availability - - - - - - - {Airflow DAG Monitor:airflow.monitor.alive.last()}=0 - Airflow Monitor: Service is not running - HIGH - Сервис airflow-monitor не запущен. -Запустить: sudo systemctl start airflow-monitor - - - scope - availability - - - - - - - {Airflow DAG Monitor:airflow.monitor.cycle_time.last()}>60 - Airflow Monitor: Cycle time is high ({ITEM.LASTVALUE1}s) - WARNING - Цикл мониторинга выполняется дольше 60 секунд. - - - scope - performance - - - - - - - - - Airflow: Problem DAG count - - - FF0000 - - Airflow DAG Monitor - airflow.dag.problems.count - - - - 00AA00 - RIGHT - - Airflow DAG Monitor - airflow.monitor.dag_count - - - - - - - - Airflow Monitor: Cycle time - - - 0000FF - - Airflow DAG Monitor - airflow.monitor.cycle_time - - - - - - {$AIRFLOW.STALE_THRESHOLD} @@ -329,6 +165,89 @@ + + + + {Airflow DAG Monitor:airflow.dag.problems.count.last()}>0 + + Airflow: {ITEM.LASTVALUE1} problematic DAG run(s) detected + HIGH + Обнаружены проблемные DAG-запуски (failed или long_running). +Подробности в item "Airflow: Problematic DAGs (JSON)". +Триггер снимается автоматически при разрешении всех проблем. + + + + + {Airflow DAG Monitor:airflow.monitor.heartbeat.nodata(600)}=1 or (({Airflow DAG Monitor:airflow.monitor.heartbeat.last()}>0) and ({Airflow DAG Monitor:airflow.monitor.heartbeat.fuzzytime(600)}=0)) + Airflow Monitor: Data is stale (no update for 10+ minutes) + DISASTER + Данные мониторинга не обновлялись более 10 минут. +Возможные причины: сервис остановлен, Airflow API недоступен. + + + Airflow Monitor: Service is not running + {Airflow DAG Monitor:airflow.monitor.alive.last()}=0 + + + + + + + {Airflow DAG Monitor:airflow.monitor.alive.last()}=0 + Airflow Monitor: Service is not running + HIGH + Сервис airflow-monitor не запущен. +Запустить: sudo systemctl start airflow-monitor + + + + + {Airflow DAG Monitor:airflow.monitor.cycle_time.last()}>60 + Airflow Monitor: Cycle time is high ({ITEM.LASTVALUE1}s) + WARNING + Цикл мониторинга выполняется дольше 60 секунд. + + + + + + + Airflow: Problem DAG count + + + FF0000 + + Airflow DAG Monitor + airflow.dag.problems.count + + + + 00AA00 + RIGHT + + Airflow DAG Monitor + airflow.monitor.dag_count + + + + + + + + Airflow Monitor: Cycle time + + + 0000FF + + Airflow DAG Monitor + airflow.monitor.cycle_time + + + + + + Service state