What I’d suggest is: don’t start with the Datadog sidecar inside each flow-run ECS task. Start from Prefect’s event system and push flow-run failure events into Datadog. That gives you the simplest path to “alert when flows fail” without adding metrics code to every flow. Recommended approach Use: Prefect events → Automation → Webhook to Datadog Why this is a good fit for your setup: - your flow infrastructure is ephemeral ECS tasks - logs are going to

awslogs

, so the Datadog agent sidecar won’t automatically see much unless you rework log collection - you want simple failure metrics / monitors with minimal per-flow code - Prefect already emits flow lifecycle events like

prefect.flow-run.Failed

So instead of trying to scrape logs or inject DogStatsD into every flow container, let Prefect observe failures centrally and then forward them. What Prefect gives you natively In Prefect 3.x, there is not a built-in Datadog integration/package, but there is native support for: - Events for flow/task state changes - Automations that react to those events - Webhook actions that can POST to external systems That means you can create an automation that listens for: -

prefect.flow-run.Failed

- optionally

prefect.flow-run.Crashed

and POSTs a payload to Datadog. Relevant docs: - Events - Automations - Custom webhook / notification actions What I would do first Option 1: Send Prefect failures to Datadog Events - easiest to reason about - good if your main goal is monitors/alerts and visibility - each Prefect flow failure becomes a Datadog event Option 2: Send Prefect failures as a custom metric - better if you want dashboards like “failure count by flow/deployment over time” - Datadog monitor can alert on count/rate If you’re “starting small”, I’d actually begin with Datadog Events or a single custom metric named something like

prefect.flow_failures

. Architecture

text
Flow run fails in ECS task
  -> Prefect emits `prefect.flow-run.Failed`
  -> Prefect Automation matches event
  -> Automation calls Datadog webhook/API
  -> Datadog receives event or metric
  -> Datadog monitor alerts

No flow code changes required. Why the agent sidecar isn’t helping much yet That’s expected. A Datadog agent sidecar is most useful when: - it can collect container logs directly - it can scrape metrics endpoints - your app emits DogStatsD / APM / OTLP to it In your current setup: - logs go to CloudWatch via awslogs - ephemeral ECS tasks may not expose scrape targets long enough to be useful - Prefect itself doesn’t automatically emit “flow failure count” as a Datadog metric from the worker container So the sidecar is not the best first lever for this specific goal. Server-side metrics: useful, but not enough for your goal Prefect server can expose Prometheus-style metrics at

/api/metrics

if enabled. That can be scraped by Datadog, but those metrics are more about Prefect server internals / infrastructure, not “count failed flow runs by deployment”. So I’d view that as nice later, not your primary solution for failure monitors. Good first implementation plan 1. Make sure events/automations are enabled on your self-hosted Prefect server 2. Create an automation that watches for: -

prefect.flow-run.Failed

- maybe also

prefect.flow-run.Crashed

3. Have the automation call: - Datadog Events API, or - Datadog Metrics API 4. Add tags like: - flow name - deployment name - work pool - environment 5. Build a Datadog monitor on top of that My recommendation If you want the least moving parts: Start with Datadog Events API because: - payload is simpler - easier to debug - lets you validate the end-to-end signal quickly Then once that works, move to:

- Metrics API for dashboards/threshold monitors What to send For a failure event/metric, I’d include tags like: -

flow:<flow-name>

-

deployment:<deployment-name>

-

work_pool:<work-pool>

-

env:<env>

- maybe

state:failed

That makes monitors and faceting much more useful later. What not to do first - Don’t add StatsD calls into every flow unless you really need app-specific metrics - Don’t rely on scraping ephemeral ECS task metrics for flow failures - Don’t rework your logging pipeline just to get failure alerts If you want, I can help with the next step I can draft either of these for you: 1. a Prefect automation + Datadog Events webhook example 2. a Prefect automation + Datadog Metrics API example 3. a recommendation for server config needed in a self-hosted Prefect setup for automations/events If you want, I can give you a concrete example for “alert on any failed flow run in Datadog” next.

Absolutely — below are two concrete patterns: 1. Prefect -> Datadog Events API 2. Prefect -> Datadog Metrics API And then I’ll answer your ECS tagging question. Short answer on tags at task-definition level Yes, for data emitted via the Datadog agent sidecar, you can usually apply common Datadog tags at the ECS task/container level using Datadog env vars like

DD_TAGS

. But: if you send data directly from a Prefect automation webhook to Datadog’s API, those task-definition tags are not automatically applied. In that case, you need to include tags in the webhook payload yourself. So there are really two tagging planes: - Agent-collected telemetry from the ECS task -> can inherit task/container Datadog tags - Webhook/API-submitted events/metrics from Prefect automations -> must include tags explicitly in the JSON payload That distinction is the main gotcha. --- Recommendation for your setup I’d do this: - Use Prefect automations for failure alerts/metrics - Define a small shared tag set in the automation payload, e.g. -

env:prod

-

service:prefect

-

team:data-eng

- If you also run the Datadog agent sidecar, set the same tags there with

DD_TAGS

so dashboards line up That gives you consistency without needing perfect tag propagation from AWS -> Datadog -> Prefect. --- Example 1: Send failed flow runs to Datadog Events API This is the easiest thing to stand up first. You’ll need: - a Datadog API key - network egress from your Prefect server/automation service to Datadog - automations enabled in your self-hosted Prefect environment Docs: - Prefect automations - Webhook actions - Datadog Events API

import os
from datetime import timedelta

from prefect.automations import Automation, EventTrigger, Posture
from prefect.events.actions import CallWebhook
from prefect.blocks.webhook import Webhook

# Create a webhook block pointing at Datadog Events API
webhook = Webhook(
    method="POST",
    url="<https://api.datadoghq.com/api/v1/events>",
    headers={
        "DD-API-KEY": os.environ["DATADOG_API_KEY"],
        "Content-Type": "application/json",
    },
    allow_private_urls=False,
    verify=True,
)

webhook_block_id = webhook.save("datadog-events-webhook", overwrite=True)

# Create an automation that fires on failed flow runs
automation = Automation(
    name="Datadog - Flow Run Failures as Events",
    description="Send Prefect flow run failures to Datadog Events",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Failed", "prefect.flow-run.Crashed"},
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(seconds=0),
        match={
            "prefect.resource.id": ["prefect.flow-run.*"]
        },
        for_each={"prefect.resource.id"},
    ),
    actions=[
        CallWebhook(
            block_document_id=webhook_block_id,
            payload="""
{
  "title": "Prefect flow run failed",
  "text": "Flow '{{ flow.name }}' failed in deployment '{{ deployment.name }}'.\\n\\nFlow run: {{ flow_run.name }}\\nState: {{ flow_run.state.name }}\\nMessage: {{ flow_run.state.message }}",
  "priority": "normal",
  "alert_type": "error",
  "source_type_name": "prefect",
  "aggregation_key": "prefect-flow-failure-{{ flow.id }}",
  "tags": [
    "service:prefect",
    "source:prefect",
    "status:failed",
    "flow:{{ flow.name }}",
    "deployment:{{ deployment.name }}"{% for tag in flow_run.tags %},
    "{{ tag }}"
    {% endfor %}
  ]
}
"""
        )
    ],
)

created = automation.create()
print(created)

Notes on that example - It reacts to both

Failed

and

Crashed

- It adds

flow_run.tags

into the Datadog tag list - If you already use Prefect flow/deployment tags, those are a nice bridge into Datadog

- This is generally the best “quick win” --- Example 2: Send failed flow runs as a Datadog custom metric This is better if you want Datadog monitors like: - “alert if failures > 0 in the last 15m” - “group by flow” - “group by deployment” Docs: - Datadog Metrics API Datadog’s series endpoint is typically: -

<https://api.datadoghq.com/api/v2/series>

Example:

import os
from datetime import timedelta

from prefect.automations import Automation, EventTrigger, Posture
from prefect.events.actions import CallWebhook
from prefect.blocks.webhook import Webhook

webhook = Webhook(
    method="POST",
    url="<https://api.datadoghq.com/api/v2/series>",
    headers={
        "DD-API-KEY": os.environ["DATADOG_API_KEY"],
        "Content-Type": "application/json",
    },
    allow_private_urls=False,
    verify=True,
)

webhook_block_id = webhook.save("datadog-metrics-webhook", overwrite=True)

automation = Automation(
    name="Datadog - Flow Run Failures as Metrics",
    description="Emit a custom Datadog metric when a Prefect flow run fails",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Failed", "prefect.flow-run.Crashed"},
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(seconds=0),
        match={
            "prefect.resource.id": ["prefect.flow-run.*"]
        },
        for_each={"prefect.resource.id"},
    ),
    actions=[
        CallWebhook(
            block_document_id=webhook_block_id,
            payload="""
{
  "series": [
    {
      "metric": "prefect.flow_run.failed",
      "type": 1,
      "points": [
        {
          "timestamp": {{ event.occurred.timestamp() | int }},
          "value": 1
        }
      ],
      "tags": [
        "service:prefect",
        "source:prefect",
        "status:failed",
        "flow:{{ flow.name }}",
        "deployment:{{ deployment.name }}"{% for tag in flow_run.tags %},
        "{{ tag }}"
        {% endfor %}
      ]
    }
  ]
}
"""
        )
    ],
)

created = automation.create()
print(created)

Notes - This creates a custom metric

prefect.flow_run.failed

- In Datadog, you can build a monitor on the sum/count over a time window - I’d personally use tags that are stable and low-cardinality - Avoid putting unique IDs like flow run UUIDs in metric tags --- Suggested Datadog monitor ideas For the custom metric approach: - Alert if

sum(last_15m):sum:prefect.flow_run.failed{env:prod} > 0

- Group by: -

flow

-

deployment

-

team

For the events approach: - event monitor on Prefect failures - useful for immediate alerting and investigation If you want, I can also draft the actual Datadog monitor queries. --- How to use your existing AWS/ECS tags This is the part to be careful with. If telemetry goes through the Datadog agent sidecar You can usually set shared Datadog tags in the ECS task/container definition using env vars like:

{
  "name": "DD_TAGS",
  "value": "env:prod service:prefect team:data-eng"
}

Sometimes people also use other Datadog ECS/AWS tag collection features, but the most reliable simple thing is: - explicitly set

DD_TAGS

on the Datadog agent container - optionally also on the app/flow container if needed for traces/logs Example task-def snippet: ``` { "containerDefinitions": [ { "name": "flow", "image": "your-flow-image", "environment": [ { "name": "DD_TAGS", "value": "env:prod service:prefect team:data-eng" } ] }, { "name": "datadog-agent", "image": "datadog/agent:latest", "environment": [ { "name": "DD_API_KEY", "value": "..." }, { "name": "DD_SITE", "value": "datadoghq.com" }, { "name": "DD_TAGS", "value": "env:prod service:prefect team:data-eng" } ] } ], "tags": [

{ "key": "env", "value": "prod" }, { "key": "service", "value": "prefect" }, { "key": "team", "value": "data-eng" } ] }

*But for Prefect automation webhooks*

Those are executed by Prefect’s automation system, not by the ECS task’s Datadog agent. So:
- ECS task definition tags do *not automatically attach*
- AWS resource tags do *not automatically attach*
- you must include tags directly in the webhook JSON payload

So if you want consistent tags, I’d define a shared set and repeat them:

"tags": [ "env:prod", "service:prefect", "team:data-eng", "flow:{{ flow.name }}", "deployment:{{ deployment.name }}" ] ``` --- Can Prefect use your AWS tags automatically? Not directly in a built-in way, at least not for these automation webhooks. What can work: - if the relevant tags already exist as Prefect flow/deployment tags, include

flow_run.tags

- if you have a small known set of environment/team/service tags, hardcode or template them into the webhook payload - if you really want dynamic AWS tag lookup, you’d need a custom service in the middle: - Prefect automation -> your webhook endpoint - your service enriches with ECS/AWS tags - your service forwards to Datadog That’s more complex than I’d recommend for phase 1. --- Best practical tagging strategy I’d suggest this: 1. Pick a canonical shared tag set: -

env

-

service

-

team

-

region

-

flow

-

deployment

2. Put those in: -

DD_TAGS

on ECS/Datadog-agent side - Prefect automation payload tags 3. If you already use Prefect tags, append them too: -

flow_run.tags

That gives you enough consistency for dashboards and monitors. --- One more useful variant: filter to only prod or selected deployments If you only want certain Prefect-tagged runs to trigger Datadog, you can keep the automation broad and use payload tags, or you can narrow the trigger based on matching rules. If you want, I can help refine the automation to only trigger for: - specific flows - specific deployments - specific Prefect tags --- My recommendation for you right now Start with Events API first, because it’s easiest to validate. Then add the Metrics API automation once you know the signal is correct. --- If you want, I can next give you: 1. a single script that creates both automations together, or 2. a version tailored to ECS/Fargate + your existing env/team/service tags.