如何通过Datadog实现Airflow DAG被暂停时的自动告警

可替代的DAG状态获取方式 #

除了直接查询Airflow元数据库Postgres外，还有3种更安全、维护成本更低的方案：

• Airflow REST API调用：Airflow 2.0+版本自带官方稳定REST接口，调用GET /dags接口即可获取全量DAG的基础信息，返回字段中is_paused直接对应DAG的暂停状态，无需进入Postgres Pod操作，权限管控更灵活。
• Airflow CLI命令读取：在Airflow调度器或Webserver节点执行airflow dags list -o json，输出的结构化结果中包含每个DAG的paused状态字段，适合无法直接访问元数据库、也无法调用API的场景。
• Airflow事件回调钩子：自定义全局on_dag_paused事件回调函数，绑定到Airflow配置后，只要有DAG被暂停就会触发回调逻辑，可直接推送告警事件到Datadog，无需轮询，告警延迟最低。

新增DAG适配逻辑 #

上述所有方案都不需要为新增DAG做单独适配：

• 如果用API/CLI/数据库查询的轮询方案：每次轮询都会拉取全量DAG的最新状态，新增DAG会自动纳入检测范围。你只需要在自定义脚本中维护一份上一次轮询的状态快照，每次对比发现原本is_paused=false的DAG本次变为true，就直接触发告警即可。
• 如果用回调钩子方案：全局配置生效后，所有新部署的DAG触发状态变更事件时都会自动走回调逻辑，不需要修改单DAG的代码。

现有元数据库查询方案优化 #

你当前用到的查询语句可以做增量优化，减少不必要的状态对比开销，可调整为：

SELECT dag_id, is_paused, last_paused_at 
FROM dag 
WHERE is_paused = true 
AND last_paused_at > NOW() - INTERVAL '5分钟'; -- 时间间隔和你的脚本轮询频率保持一致

该语句可直接筛选出最近轮询周期内被暂停的DAG，无需额外维护历史状态快照即可实现告警触发。

内容的提问来源于stack exchange，提问作者Avenger