Short answer: there isn’t an official “Prefect ↔️ Datadog” integration package in Prefect 3.x, but yes—you can monitor work pools, queues, and flows with Datadog via: - Prometheus/OpenMetrics scraping of Prefect’s metrics endpoints - Prefect Events → Datadog (via webhooks) - Logs to Datadog (via standard Python logging/HTTP intake) - Optional: custom DogStatsD metrics using Prefect’s API What works today 1) Metrics (recommended starting point) - Prefect exposes Prometheus metrics you can scrape with the Datadog Agent’s OpenMetrics integration. - Enable server metrics (exposes /metrics on the API) and optional client metrics: -

PREFECT_API_ENABLE_METRICS=true

-

PREFECT_CLIENT_METRICS_ENABLED=true

(client endpoint defaults to http://localhost:4201/metrics) - Configure the Datadog Agent (OpenMetrics) to scrape these endpoints:

init_config:

  instances:
    - openmetrics_endpoint: <http://your-prefect-api:4200/metrics>
      namespace: prefect_api
      metrics:
        - prefect_*

    - openmetrics_endpoint: <http://localhost:4201/metrics>
      namespace: prefect_client
      metrics:
        - prefect_*

- Datadog docs: OpenMetrics (Prometheus) integration Notes: - The API server /metrics endpoint includes Prefect server-side metrics; the client endpoint exposes per-process client metrics (useful on workers). - From these, you can build dashboards/alerts (e.g., flow run events observed, worker websocket connections, etc.). If you need specific KPIs like “queue backlog,” consider #3 below. 2) Events → Datadog (for flow/pool/queue state changes) - Prefect emits structured events such as

prefect.flow-run.started/completed/failed

, worker claimed work, etc. - Use a Prefect Automation with a Webhook action to send selected events to Datadog Logs or Events intake (great for alerts on failures or stuck queues). - Datadog logs HTTP intake: Datadog Logs HTTP intake API 3) Custom metrics for work pools/queues - If you want explicit metrics like “queue length,” “late runs,” or “runs failing by deployment,” run a small periodic job that queries the Prefect API and emits DogStatsD gauges. Example outline:

from prefect import get_client
  from datadog import initialize, statsd

  # initialize(api_key=..., app_key=...) if using the Datadog library directly
  # or rely on the Datadog Agent’s DogStatsD on UDP :8125

  async def emit_metrics():
      async with get_client() as client:
          # work pools / queues
          work_pools = await client.read_work_pools()
          work_queues = await client.read_work_queues()

          statsd.gauge('prefect.work_pools.count', len(work_pools))
          statsd.gauge('prefect.work_queues.count', len(work_queues))

          # example: failed flow runs in the last window (filter as needed)
          # ...build filters and count, then emit:
          # statsd.gauge('prefect.flow_runs.failed.count', n_failed)

- This gives you precise KPIs for dashboards/alerts. 4) Logs to Datadog - Prefect uses standard Python logging; you can add a handler that forwards to Datadog (HTTP intake or Agent). This is useful if you want unified log search and alerts in Datadog. Next steps / questions to tailor the setup - Are you on Prefect Cloud or OSS Server? - Where do your workers run (Kubernetes, ECS, VMs)? This affects where to run the Datadog Agent and how to point it to the metrics endpoints. - Do you want just metrics, just logs, or both? And any key SLOs you want to alert on (e.g., queue backlog, late runs, failure rate)? If you share your environment, I can provide a drop-in Datadog Agent config and a small “queue backlog” exporter example that fits your deployment.

@Marvin We use Prefect Cl and run a mixture of Kubernetes and ECS agents. We want to alert on queue backlog and late runs

Got it — sounds like you’re on Prefect Cloud (and running workers on Kubernetes and ECS). Here’s a clean path to alerts in Datadog for both queue backlog and late runs: Plan overview - Late runs: Use a Prefect Cloud Automation with a Metric trigger on “lateness,” and send the alert to Datadog via a webhook. - Queue backlog: Run a small scheduled monitor (Kubernetes CronJob / ECS Scheduled Task) that counts scheduled runs per work queue via the Prefect Cloud API and emits DogStatsD metrics to Datadog. Create Datadog metric monitors on those series. 1) Late runs (no code required, done in Prefect Cloud) - In the Prefect Cloud UI → Automations → New automation: - Trigger: Metric trigger - Metric: lateness - Match: your work queue(s) (resource = Work queue; choose specific queue names) - Threshold: e.g., > 600 seconds for at least 5 minutes (tune as needed) - Action: Call Webhook to Datadog (Events or Logs intake) - Datadog Events API: POST https://api.datadoghq.com/api/v1/events?api_key=YOUR_DD_API_KEY with a JSON body like:

{
  "title": "Prefect Late Runs",
  "text": "High lateness on {{ event.resource.name }}",
  "tags": [
    "source:prefect",
    "type:lateness",
    "queue:{{ event.resource.name }}"
  ]
}

Docs: Datadog Events API - Or Datadog Logs intake: POST https://http-intake.logs.datadoghq.com/v1/input/YOUR_DD_API_KEY with headers

Content-Type: application/json

and a JSON body similar to:

{
  "message": "Prefect late runs over threshold",
  "ddsource": "prefect",
  "ddtags": "queue:{{ event.resource.name }}"
}

Docs: Datadog Logs HTTP intake 2) Queue backlog (small monitor + DogStatsD metrics) - Backlog = flow runs in Scheduled state that haven’t been picked up yet. - Overdue backlog = scheduled in the past (scheduled start time < now). - Late runs = flow runs in state Late (Cloud marks these automatically after a grace period). Use this tiny async job (run every 1–5 minutes) to emit metrics to Datadog: ``` import asyncio import os from datetime import datetime, timezone from datadog import initialize, statsd from prefect import get_client from prefect.client.schemas.filters import ( FlowRunFilter, FlowRunFilterState, FlowRunFilterStateName, FlowRunFilterNextScheduledStartTime, WorkQueueFilter, WorkQueueFilterName, ) # If you prefer DogStatsD via Agent (recommended), you do NOT need initialize() # Just ensure the Agent is reachable on UDP 8125 and DD_AGENT_HOST/DD_DOGSTATSD_PORT are set. # If you'd rather use the Datadog Metrics API directly, uncomment and set your keys: # initialize(api_key=os.environ["DD_API_KEY"], app_key=os.environ["DD_APP_KEY"]) WORK_QUEUE_NAMES = os.getenv("WORK_QUEUE_NAMES", "default").split(",") async def emit_queue_metrics(): now = datetime.now(timezone.utc) async with get_client() as client: for queue in [q.strip() for q in WORK_QUEUE_NAMES if q.strip()]: # Total backlog: all Scheduled backlog = await client.flow_runs.count_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Scheduled"]) ) ), work_queue_filter=WorkQueueFilter( name=WorkQueueFilterName(any_=[queue]) ), ) # Overdue backlog: Scheduled with next_scheduled_start_time < now overdue = await client.flow_runs.count_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Scheduled"]) ), next_scheduled_start_time=FlowRunFilterNextScheduledStartTime( before_=now

), ), work_queue_filter=WorkQueueFilter( name=WorkQueueFilterName(any_=[queue]) ), ) # Late runs: state name = Late late = await client.flow_runs.count_flow_runs( flow_run_filter=FlowRunFilter( state=FlowRunFilterState( name=FlowRunFilterStateName(any_=["Late"]) ) ), work_queue_filter=WorkQueueFilter( name=WorkQueueFilterName(any_=[queue]) ), ) # Emit DogStatsD metrics (tags include the queue name) tags = [f"queue:{queue}"] statsd.gauge("prefect.queue.backlog", backlog, tags=tags) statsd.gauge("prefect.queue.backlog_overdue", overdue, tags=tags) statsd.gauge("prefect.queue.late_runs", late, tags=tags) if name == "__main__": # Requires PREFECT_API_URL and PREFECT_API_KEY set for Prefect Cloud asyncio.run(emit_queue_metrics()) ``` Deployment notes - Kubernetes: run the script as a CronJob; ensure the Datadog Agent DaemonSet is deployed and DogStatsD is reachable (8125/UDP). OpenMetrics scraping isn’t useful for backlog from Cloud because Prefect Cloud’s server metrics aren’t exposed to scrape; the API approach above is the right way for backlog counts. - ECS (Fargate): run as a Scheduled Task. If you use a Datadog Agent sidecar, DogStatsD is at localhost:8125; otherwise, point DD_AGENT_HOST to the node/Agent. Suggested Datadog monitors - Queue backlog: - Metric: sum(avgprefect.queue.backlog{queue<your-queue>}) by queue - Example threshold: > 10 for 5 minutes - Overdue backlog (often a better signal of trouble): - Metric: sum(avgprefect.queue.backlog overdue{queue<your-queue>}) - Example threshold: > 0 for 10 minutes - Late runs (if you prefer Datadog-only alerts instead of Automations): - Metric: sum(avgprefect.queue.late runs{queue<your-queue>}) > 0 for 5 minutes Optional: stick with Prefect Automations for “lateness” - It’s very effective to let Prefect calculate lateness and just forward the alert to Datadog via webhook (as in step 1). Keep Datadog monitors focused on backlog metrics from the script. Quick checks so we can tailor this for you - Which work queue names should we monitor? - What thresholds feel right for backlog and lateness? - Do you already run the Datadog Agent on your k8s nodes and in ECS (sidecar)? If yes, I can give you a CronJob and ECS task definition you can drop in. If not, I can outline that too. Datadog refs - OpenMetrics (Prometheus) integration: Datadog OpenMetrics - Logs intake: Datadog Logs HTTP intake - Events API: Datadog Events API