airflow_datadog_sensor_and_hook_event_monitoring_dag.py

1import datetime
2
3from airflow import DAG
4from airflow.providers.datadog.hooks.datadog import DatadogHook
5from airflow.providers.datadog.sensors.datadog import DatadogSensor
6from airflow.operators.python import PythonOperator
7
8def send_metric_to_datadog():
9    """
10    Example of using DatadogHook to send a custom metric.
11    """
12    hook = DatadogHook(datadog_conn_id="datadog_default")
13    hook.post_event(
14        title="Airflow Task Alert",
15        text="A custom metric/event has been sent from Airflow.",
16        tags=["env:production", "tool:airflow"]
17    )
18
19with DAG(
20    dag_id="example_datadog_operator",
21    start_date=datetime.datetime(2023, 1, 1),
22    schedule=None,
23    catchup=False,
24) as dag:
25
26    # Sensor to wait for a specific event in Datadog
27    # It will keep poking until an event with the specified tags/sources is found
28    check_for_event = DatadogSensor(
29        task_id="check_for_event",
30        datadog_conn_id="datadog_default",
31        tags=["status:ready"],
32        source_type_name="airflow",
33        timeout=600,
34        poke_interval=30,
35    )
36
37    # Task using the Hook to send data back to Datadog
38    send_metric = PythonOperator(
39        task_id="send_metric_to_datadog",
40        python_callable=send_metric_to_datadog,
41    )
42
43    check_for_event >> send_metric