如何通过Airflow实现DAG失败时自动创建ServiceNow工单

1. Airflow Connection配置 #

先在Airflow中配置ServiceNow的连接信息，避免硬编码凭证：

• 打开Airflow UI，进入Admin > Connections页面
• 点击Add a new record，填写以下关键信息：
  • Conn Id：自定义一个标识，比如servicenow_rest_conn
  • Conn Type：选择HTTP类型
  • Host：你的ServiceNow实例域名（例如dev12345.service-now.com）
  • Login：拥有创建工单权限的ServiceNow账号用户名
  • Password：对应账号的密码或API专用令牌
  • Extra：可选，添加JSON格式的额外请求头，如{"headers": {"Content-Type": "application/json"}}

2. ServiceNow API调用实现 #

使用Airflow的HttpHook复用上述连接配置，编写创建工单的函数：

from airflow.providers.http.hooks.http import HttpHook
import json

def create_servicenow_ticket(context):
    # 从上下文提取DAG失败相关信息
    dag_id = context.get('dag').dag_id
    task_id = context.get('task_instance').task_id
    execution_date = context.get('execution_date').isoformat()
    error_msg = str(context.get('exception'))

    # 构造ServiceNow工单请求体（根据你的实例需求调整字段）
    payload = {
        "short_description": f"Airflow DAG失败告警: {dag_id} - {task_id}",
        "description": f"DAG ID: {dag_id}\n任务ID: {task_id}\n执行时间: {execution_date}\n错误信息: {error_msg}",
        "category": "Application",
        "subcategory": "Automation",
        "assignment_group": "你的运维组名称",
        "urgency": "2"  # 1=高, 2=中, 3=低
    }

    # 初始化HttpHook，使用之前配置的Conn Id
    hook = HttpHook(http_conn_id='servicenow_rest_conn', method='POST')
    # ServiceNow创建工单的默认API端点
    endpoint = '/api/now/table/incident'

    try:
        # 发送请求，启用基本认证
        response = hook.run(
            endpoint=endpoint,
            data=json.dumps(payload),
            headers={"Content-Type": "application/json"}
        )
        print(f"ServiceNow工单创建成功，响应: {response.text}")
    except Exception as e:
        print(f"创建ServiceNow工单失败: {str(e)}")
        # 可根据需求添加重试或额外告警逻辑

3. 在DAG中集成失败回调 #

将上述函数作为on_failure_callback参数，绑定到整个DAG或单个任务：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def sample_task():
    # 模拟任务失败场景
    raise ValueError("测试任务失败触发ServiceNow告警")

with DAG(
    dag_id='servicenow_failure_alert_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    on_failure_callback=create_servicenow_ticket,  # 整个DAG失败时触发
    catchup=False
) as dag:
    task = PythonOperator(
        task_id='sample_failing_task',
        python_callable=sample_task,
        # 也可以单独给任务设置失败回调：on_failure_callback=create_servicenow_ticket
    )

参考资料 #

• ServiceNow REST API官方文档：重点查看incident表的创建接口说明，明确必填字段、可选字段的定义规则
• Airflow HttpHook文档：学习如何通过Airflow连接管理HTTP请求，实现凭证复用与请求标准化
• Airflow Connection官方指南：了解不同类型连接的配置参数与安全最佳实践

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