[raghavan-arvind]

Why

The existing DatadogSensor queries the event stream for transition events in a time window. That shape cannot answer "is this monitor healthy right now": a monitor that entered Alert before the window (or recovered before it) emits no event inside it, so gating a pipeline on monitor health requires reconstructing state from transitions.

This adds DatadogMonitorSensorAsync, which checks a monitor's current overall_state — an already-alerting or already-recovered monitor is observed regardless of when it last changed state.

What

• DatadogMonitorSensorAsync: waits for a monitor to reach one of target_states (default ("OK",)). Deferrable-only, and named *Async to make that explicit: it waits in the triggerer via a new DatadogMonitorTrigger and never occupies a worker slot while waiting; there is deliberately no poke-mode fallback.
• DatadogMonitorTrigger: polls the monitor state off the event loop (asyncio.to_thread, since the datadog client is sync) and fires when a target state is reached.
• soft_fail on the deferral path applies only to timeouts, matching poke/reschedule semantics elsewhere: a trigger crash (bad monitor id, auth failure) fails the task loudly instead of being converted into a skip that silently skips the downstream branch. Works on both Airflow 3.x (TaskDeferralTimeout) and 2.11 (timeout arrives as TaskDeferralError("Trigger/execution timeout")).
• The datadog library returns API errors as a payload instead of raising; they surface as a clear AirflowException (e.g. Datadog monitor 999999999: ['Monitor not found']).
• provider.yaml gains a triggers section; unit tests for the sensor (defer, completion, soft_fail semantics) and trigger (serialize round-trip, poll-until-target, API error).

Testing

uv run --project providers/datadog pytest providers/datadog/tests/unit/datadog: 18 passed. Also validated end to end in a production-like Airflow 3.3 deployment with canary tasks covering pass / timeout / timeout-with-soft_fail / missing-monitor / missing-with-soft_fail.

🤖 Generated with Claude Code

[raghavan-arvind]

Verified all these sensors return the correct result:

# Datadog sensor canaries: every (case x mode) combination, alerting when
# the outcome differs from the expected one encoded in the task name.
DATADOG_CANARY_CASES = {
    # name -> (monitor_id, target_states, soft_fail, expected outcome)
    'ok_should_pass': (DATADOG_OK_MONITOR_ID, ('OK',), False, 'success'),
    'alert_timeout_should_fail': (DATADOG_OK_MONITOR_ID, ('Alert',), False, 'failed'),
    'alert_timeout_soft_fail_should_skip': (DATADOG_OK_MONITOR_ID, ('Alert',), True, 'skipped'),
    'missing_should_fail': (DATADOG_INVALID_MONITOR_ID, ('OK',), False, 'failed'),
    'missing_soft_fail_should_fail': (DATADOG_INVALID_MONITOR_ID, ('OK',), True, 'failed'),
}

for case_name, (monitor_id, target_states, soft_fail, expected) in DATADOG_CANARY_CASES.items():
  for sensor_mode in ('poke', 'reschedule', 'deferrable'):
    callbacks = {
        'success': {},
        'failed': {
            'on_failure_callback': None,
            'on_success_callback': alert_on_failure,
            'on_skipped_callback': alert_on_failure,
        },
        'skipped': {'on_success_callback': alert_on_failure},
    }[expected]
    DatadogMonitorSensor(
        dag=dag,
        task_id=f'datadog_monitor_{case_name}_{sensor_mode}',
        monitor_id=monitor_id,
        target_states=target_states,
        soft_fail=soft_fail,
        timeout=timedelta(seconds=90),
        poke_interval=30,
        deferrable=sensor_mode == 'deferrable',
        **({'mode': sensor_mode} if sensor_mode != 'deferrable' else {}),
        **callbacks,
    )

[raghavan-arvind]

The soft_fail tests are reliant on some fix like:
#71256

Would like to discuss that PR first before this one

[raghavan-arvind]

Ideally we can remove this and merge some variant of #71256 first.

[raghavan-arvind]

Would like to only support deferrable for this sensor -- is that okay? Any existing patterns for this? For retrying transient errors, 500, 429, the semantics of changing the poll interval temporarily into an exponential backoff are more straightforward that way

I know TimeSensorAsync is deprecated in favor of TimeSensor with deferrable=True, but would prefer to write a DatadogMonitorSensorAsync to standardize on deferrable going forward.

edit: updated

[kaxil]

Is there a reason this needs a separate DatadogMonitorSensorAsync class rather than deferrable=True on the existing DatadogSensor? That's the pattern the project has moved to (TimeSensorAsync is deprecated in favor of TimeSensor(deferrable=True)), and a separate *Async sensor plus a new triggers module are permanent provider surface. The resume_execution override that skips the base sensor's soft-fail handling also depends on #71256, which is closed. Is there an issue or a user asking for deferrable Datadog monitors? Converting to draft until there is.