https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> when using an ecs:push workpool I wan...
# ask-marvin
a
@Marvin when using an ecs:push workpool I want to track the ECS task id that is created for each flow run so that I can match them together in the future. I see that infrastructure_pid is filled in the flow run's metadata pointing to the ARN of the ECS task, however this is only populated while the flow run is actually executing and seems to be cleared as soon as it finishes. Is there a way that I can keep track of these ids long-term?
m
✅ thought for 151.6 seconds
Short answer: yes, but probably not via 
infrastructure_pid
alone. A few important points: * For ECS push work pools, 
infrastructure_pid
is the worker’s handle to the infrastructure, and for ECS it is typically formatted like 
cluster::task_arn
. * In practice, users often observe that it’s mainly useful while the run is active. * If you need a durable mapping between 
flow_run_id
and ECS 
taskArn
, the best long-term source is Prefect Events, not the transient execution metadata view. Recommended approach: use ECS infrastructure events Prefect emits ECS task lifecycle events that include both: * the ECS task identifier / ARN * the related Prefect flow run resource That gives you a historical record you can query later. Relevant docs: * AWS ECS infrastructure events * Prefect Events concepts Those ECS events include resource metadata like: * 
ecs.taskArn
* 
ecs.clusterArn
* a related resource for the flow run, like 
prefect.flow-run.<id>
So the durable correlation pattern is: 1. Flow run starts on ECS 2. Prefect emits 
prefect.ecs.task.*
events 3. Later, query events for the flow run and extract 
ecs.taskArn
If you want, you can also persist that mapping yourself into your own DB/S3/log sink as soon as the run starts. Example pattern:
Copy code
from prefect import get_client
from prefect.events.filters import EventFilter, EventNameFilter, EventRelatedFilter

async def get_ecs_tasks_for_flow_run(flow_run_id: str):
    async with get_client() as client:
        event_filter = EventFilter(
            event=EventNameFilter(prefix=["prefect.ecs.task"]),
            related=EventRelatedFilter(
                resources=[f"prefect.flow-run.{flow_run_id}"]
            ),
        )
        events = await client.read_events(filter=event_filter, limit=100)

    tasks = []
    for event in events.events:
        task_arn = event.resource.get("ecs.taskArn")
        if task_arn:
            tasks.append(
                {
                    "event": event.event,
                    "occurred": event.occurred,
                    "task_arn": task_arn,
                    "resource_id": event.resource.get("prefect.resource.id"),
                }
            )
    return tasks
A couple of caveats: * If you need exactly one ECS task ID per flow run, decide how to handle retries / re-submissions, since a flow run may have more than one infrastructure attempt over time. * If you need guaranteed long-term retention beyond your Prefect event retention window, copy the mapping into your own persistence layer. If you want the most robust setup, I’d recommend one of these: * Best for audit/history: query Prefect Events * Best for permanent internal tracking: write 
flow_run_id -> ecs_task_arn
to your own store when the run starts * Best for AWS-native observability: also tag or log the 
flow_run_id
into ECS task logs / CloudWatch context If you'd like, I can help you with either: * a concrete script to query historical ECS task ARNs for flow runs, or * a pattern to persist the mapping automatically at flow start.
a
@Marvin how can I emit an event with both the flow run id and ecs task id? show me some sample code
m
✅ thought for 96.0 seconds
Yes — the clean pattern is: * make the ECS task the primary 
resource
* attach the flow run as a 
related
resource * optionally also put the IDs in 
payload
for convenience Here’s a simple example from inside a flow.
Copy code
from prefect import flow
from prefect.events import emit_event
from prefect.runtime import flow_run


@flow
def my_flow():
    current_flow_run_id = flow_run.id

    # however you obtain this in your environment
    ecs_task_arn = "arn:aws:ecs:us-east-1:123456789012:task/my-cluster/abc123def456"
    ecs_task_id = ecs_task_arn.split("/")[-1]

    event = emit_event(
        event="acme.ecs.task.tracked",
        resource={
            "prefect.resource.id": f"ecs.task.{ecs_task_id}",
            "prefect.resource.name": ecs_task_id,
            "ecs.task_id": ecs_task_id,
            "ecs.task_arn": ecs_task_arn,
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{current_flow_run_id}",
                "prefect.resource.role": "flow-run",
            }
        ],
        payload={
            "flow_run_id": str(current_flow_run_id),
            "ecs_task_id": ecs_task_id,
            "ecs_task_arn": ecs_task_arn,
        },
    )

    return event
If you want the full flow-run context attached with a name too:
Copy code
from prefect import flow
from prefect.events import emit_event
from prefect.context import FlowRunContext


@flow
def my_flow():
    ctx = FlowRunContext.get()
    flow_run_id = str(ctx.flow_run.id)
    flow_name = ctx.flow.name

    ecs_task_arn = "arn:aws:ecs:us-east-1:123456789012:task/my-cluster/abc123def456"
    ecs_task_id = ecs_task_arn.split("/")[-1]

    emit_event(
        event="acme.ecs.task.started",
        resource={
            "prefect.resource.id": f"ecs.task.{ecs_task_id}",
            "prefect.resource.name": ecs_task_id,
            "ecs.task_id": ecs_task_id,
            "ecs.task_arn": ecs_task_arn,
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
                "prefect.resource.role": "flow-run",
                "prefect.resource.name": flow_name,
            }
        ],
        payload={
            "flow_run_id": flow_run_id,
            "ecs_task_id": ecs_task_id,
            "ecs_task_arn": ecs_task_arn,
        },
    )
If you’re doing this in a task instead of directly in the flow:
Copy code
from prefect import flow, task
from prefect.events import emit_event
from prefect.runtime import flow_run


@task
def emit_ecs_mapping(ecs_task_arn: str):
    current_flow_run_id = flow_run.id
    ecs_task_id = ecs_task_arn.split("/")[-1]

    emit_event(
        event="acme.ecs.task.mapping.created",
        resource={
            "prefect.resource.id": f"ecs.task.{ecs_task_id}",
            "prefect.resource.name": ecs_task_id,
            "ecs.task_id": ecs_task_id,
            "ecs.task_arn": ecs_task_arn,
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{current_flow_run_id}",
                "prefect.resource.role": "flow-run",
            }
        ],
        payload={
            "flow_run_id": str(current_flow_run_id),
            "ecs_task_id": ecs_task_id,
            "ecs_task_arn": ecs_task_arn,
        },
    )


@flow
def my_flow():
    ecs_task_arn = "arn:aws:ecs:us-east-1:123456789012:task/my-cluster/abc123def456"
    emit_ecs_mapping(ecs_task_arn)
A few tips: * 
prefect.resource.id
is required on the primary 
resource
* 
prefect.resource.role
is required on each 
related
resource * Put fields you want easy filtering on in 
resource
/ 
related
* Put extra details in 
payload
If your main goal is correlation later, I’d use a shape like this: * Event name: 
acme.ecs.task.mapping.created
* Primary resource: 
ecs.task.<task_id>
* Related resource: 
prefect.flow-run.<flow_run_id>
That makes it easy to query “what ECS task belonged to this flow run?” or “what flow run owned this ECS task?” Docs: * Prefect Events * ECS infrastructure events One important caveat: from inside the running flow code, you may not always have a reliable way to discover the ECS task ARN automatically unless you read it from the environment or metadata service. If you want, I can show you a version that fetches the ECS task ID/ARN automatically from the ECS container metadata endpoint.
a
@Marvin yes please, I'd like to see how to read it from the ECS container metadata endpoint.
m
✅ thought for 16.8 seconds
Yep — on ECS the usual way is to read the container metadata URI from the environment and then call the task metadata endpoint. Inside an ECS task, AWS typically provides one of: * 
ECS_CONTAINER_METADATA_URI_V4
* 
ECS_CONTAINER_METADATA_URI
Then you can call 
<metadata_uri>/task
to get task-level metadata, including the task ARN. Here’s a Prefect example that emits a custom event with both the flow run id and ECS task metadata.
Copy code
import os
import requests

from prefect import flow, task
from prefect.events import emit_event
from prefect.runtime import flow_run


def get_ecs_task_metadata():
    metadata_base = (
        os.environ.get("ECS_CONTAINER_METADATA_URI_V4")
        or os.environ.get("ECS_CONTAINER_METADATA_URI")
    )

    if not metadata_base:
        raise RuntimeError(
            "ECS container metadata endpoint not available. "
            "This code must run inside an ECS task with metadata enabled."
        )

    response = requests.get(f"{metadata_base}/task", timeout=5)
    response.raise_for_status()
    return response.json()


@task
def emit_ecs_task_mapping_event():
    task_metadata = get_ecs_task_metadata()

    flow_run_id = str(flow_run.id)
    task_arn = task_metadata["TaskARN"]
    cluster = task_metadata.get("Cluster")
    family = task_metadata.get("Family")
    revision = task_metadata.get("Revision")

    ecs_task_id = task_arn.split("/")[-1]

    event = emit_event(
        event="acme.ecs.task.mapping.created",
        resource={
            "prefect.resource.id": f"ecs.task.{ecs_task_id}",
            "prefect.resource.name": ecs_task_id,
            "ecs.task_id": ecs_task_id,
            "ecs.task_arn": task_arn,
            "ecs.cluster": cluster or "",
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
                "prefect.resource.role": "flow-run",
            }
        ],
        payload={
            "flow_run_id": flow_run_id,
            "ecs_task_id": ecs_task_id,
            "ecs_task_arn": task_arn,
            "cluster": cluster,
            "family": family,
            "revision": revision,
            "raw_metadata": task_metadata,
        },
    )

    return {
        "flow_run_id": flow_run_id,
        "ecs_task_id": ecs_task_id,
        "ecs_task_arn": task_arn,
        "event_id": str(event.id) if event else None,
    }


@flow
def my_flow():
    return emit_ecs_task_mapping_event()


if __name__ == "__main__":
    print(my_flow())
A lighter-weight version if you just want the ARN:
Copy code
import os
import requests

from prefect import flow
from prefect.events import emit_event
from prefect.runtime import flow_run


def get_ecs_task_arn():
    metadata_base = (
        os.environ.get("ECS_CONTAINER_METADATA_URI_V4")
        or os.environ.get("ECS_CONTAINER_METADATA_URI")
    )
    if not metadata_base:
        return None

    response = requests.get(f"{metadata_base}/task", timeout=5)
    response.raise_for_status()
    return response.json()["TaskARN"]


@flow
def my_flow():
    task_arn = get_ecs_task_arn()
    if not task_arn:
        return

    task_id = task_arn.split("/")[-1]

    emit_event(
        event="acme.ecs.task.mapping.created",
        resource={
            "prefect.resource.id": f"ecs.task.{task_id}",
            "prefect.resource.name": task_id,
            "ecs.task_arn": task_arn,
        },
        related=[
            {
                "prefect.resource.id": f"prefect.flow-run.{flow_run.id}",
                "prefect.resource.role": "flow-run",
            }
        ],
        payload={
            "flow_run_id": str(flow_run.id),
            "ecs_task_arn": task_arn,
            "ecs_task_id": task_id,
        },
    )
A few practical notes: * This only works inside the ECS task/container. * If you run locally, those env vars won’t exist, so you should handle that case gracefully.
* On ECS/Fargate, 
/task
returns task-level metadata; 
TaskARN
is the field you want. * You may also want to include: * 
AvailabilityZone
* 
LaunchType
* container image/name * task definition family/revision If you want a safer production version, use graceful fallback instead of raising:
Copy code
def try_get_ecs_task_metadata():
    metadata_base = (
        os.environ.get("ECS_CONTAINER_METADATA_URI_V4")
        or os.environ.get("ECS_CONTAINER_METADATA_URI")
    )
    if not metadata_base:
        return None

    try:
        response = requests.get(f"{metadata_base}/task", timeout=5)
        response.raise_for_status()
        return response.json()
    except Exception:
        return None
Then emit only when metadata is present. AWS docs for the endpoint: * ECS task metadata endpoint v4 If you want, I can also show a version that: * stores this as a Prefect artifact/log, * emits it once at flow start automatically, or * queries these custom events later by 
flow_run_id
.
a
@Marvin and do you receommend to use the ECS task API, or what about reading the infrastructure_pid metadata from the flow run, using the flow run id? it requires a look up to the prefect API but I assume at this time it will already have been populated. What are the benefits/tradeoffs to each approach?
m
✅ thought for 199.9 seconds
Yes — between the two, I’d generally recommend: use the ECS metadata endpoint if you are inside the ECS task and want the task identity for the current run. Then treat 
infrastructure_pid
as: a convenient Prefect-side correlation field, but not the most direct source from inside the container. Why I’d lean that way: ECS metadata endpoint * Pros * direct from AWS / the running container * no Prefect API round-trip * no need to parse Prefect-specific formatting * tells you the actual container’s current task ARN * less coupling to Prefect internals * Cons * ECS-specific * only available from inside the running ECS environment * if you want to look it up later from outside the task, you need to persist it yourself 
infrastructure_pid
from Prefect API * Pros * available centrally from Prefect * useful from outside the container too * infrastructure-agnostic concept * already tied to the flow run object * Cons * requires a Prefect API lookup * format is worker-specific; for ECS it’s 
cluster::task_arn
* more coupled to Prefect implementation details * if you’re asking from inside the task, it’s a little indirect On your specific assumption:
I assume at this time it will already have been populated
Usually yes by the time user flow code is running, but I would still call it less robust than the ECS metadata endpoint. The worker launches the ECS task, then updates the flow run’s 
infrastructure_pid
via the Prefect API. In normal conditions, by the time your flow code starts, that field is typically populated. But conceptually that still depends on: * the worker successfully writing it back * the Prefect API being reachable * timing/order of operations So the tradeoff is: * If you want the current task’s identity from inside the task, metadata endpoint is the strongest source. * If you want the Prefect record of which infrastructure was attached to a flow run, 
infrastructure_pid
is the better source. * If you want durable correlation later, best practice is to read the metadata endpoint, then emit/persist that mapping yourself. My practical recommendation: Inside the flow running on ECS 1. Read task ARN from ECS metadata endpoint 2. Get 
flow_run.id
from Prefect runtime 3. Emit a custom event or persist the mapping externally Outside the flow / for retrospective lookup 1. Check Prefect events or your persisted mapping 2. Optionally fall back to 
infrastructure_pid
if still present So I’d summarize it as: Best source of truth for the running task: ECS metadata endpoint Best Prefect-side correlation handle: 
infrastructure_pid
Best long-term strategy: emit/persist your own mapping event A robust hybrid helper would look like this:
Copy code
import os
import requests

from prefect.context import FlowRunContext


def get_ecs_task_arn_from_metadata():
    metadata_base = (
        os.environ.get("ECS_CONTAINER_METADATA_URI_V4")
        or os.environ.get("ECS_CONTAINER_METADATA_URI")
    )
    if not metadata_base:
        return None

    response = requests.get(f"{metadata_base}/task", timeout=5)
    response.raise_for_status()
    return response.json().get("TaskARN")


def get_ecs_task_arn_from_infrastructure_pid():
    ctx = FlowRunContext.get()
    if not ctx or not ctx.flow_run or not ctx.flow_run.infrastructure_pid:
        return None

    infrastructure_pid = ctx.flow_run.infrastructure_pid
    if "::" not in infrastructure_pid:
        return None

    _, task_arn = infrastructure_pid.split("::", 1)
    return task_arn


def get_ecs_task_arn():
    return (
        get_ecs_task_arn_from_metadata()
        or get_ecs_task_arn_from_infrastructure_pid()
    )
That gives you: * first choice: direct AWS metadata * fallback: Prefect’s recorded infrastructure id Relevant docs: * ECS task metadata endpoint v4
* Prefect Events * ECS infrastructure events If you want, I can also give you a complete utility module that: * detects ECS, * gets the task ARN, * emits the custom Prefect event, * and safely no-ops when running locally.
Open in Slack
PreviousNext
Powered by