Task memory diagnostics in Airflow 3: measurement scopes and trade-offs

[zhanqian-zhang]

Context and goal

Airflow 2.10 introduced periodic task CPU and memory observations in #39650.

Those metrics are no longer emitted in Airflow 3; their absence was reported in #49983. #51602 later removed stale documentation, and #56690 explored restoring CPU and memory metrics but was not merged.

Before attempting another implementation, I would like to clarify what a task memory metric should promise to measure.

The practical motivation is:

Help users notice tasks whose memory behavior may be worth investigating or optimizing.

There are several useful ways to answer that question, with different measurement boundaries and implementation costs.

The scopes below are design alternatives / possible extensions, not an implementation commitment or a fixed roadmap.

In particular, the smallest scope could provide useful evidence when high memory is actually observed, but it would not reliably rank executions by historical peak.

Scope 1 — Python Task Runner RSS diagnostic

The smallest useful signal would retain the lightweight periodic process-sampling idea from Airflow 2.10 while defining its meaning more narrowly.

An illustrative metric name is:

task.runner_rss_bytes

with the contract:

Periodically sampled Resident Set Size of one Python Task Runner process, reported in bytes.

Measurement boundary

Conceptually:

Supervisor
    |
    +-- Python Task Runner   <-- measured
            |
            +-- task/operator code
            +-- possible subprocesses

Only the Python Task Runner PID is measured.

It would not mean:

• total memory consumed by the execution workload;
• memory consumed by subprocesses;
• process-tree memory;
• container or Pod memory;
• host memory pressure;
• the total or maximum across concurrent executions of the same logical task;
• a guaranteed execution peak.

This distinction matters because some tasks perform most of their work in Bash, another Python interpreter, Java, ffmpeg, or another local/remote process.

Sampling placement and lifecycle

I do not think the exact sampling location should be decided in the metric contract.

Airflow 2.10 observed the task subprocess externally from its task runner. In Airflow 3, the Supervisor starts and tracks the Python Task Runner process, so Supervisor-side observation is one possible design.

Sampling from inside the Python runner is another possibility and has different access to task identity and initialized metrics.

Before choosing either, the implementation should validate:

• when sufficient task identity/tags become available;
• startup and finalization coverage;
• run_as_user / re-exec behavior;
• process permissions;
• shutdown/export behavior;
• thread/fork safety.

A realistic lifecycle contract is therefore:

Start observation once sufficient task identity and measurement context are available, and continue through finalization where feasible. Define the exact boundaries in the implementation.

Observation failures must remain strictly best-effort and must never affect task result, retry, deferral, or cleanup.

What Scope 1 can and cannot answer

For Python-heavy tasks, runner RSS may still be a useful low-cost signal:

"This Python runner was observed using substantial memory."

It cannot reliably answer:

"These are the tasks with the highest historical execution peaks."

For example:

0.0s    500 MiB
1.0s      8 GiB
2.0s    600 MiB
3.0s    exit

A periodic sampler may never observe the 8 GiB spike, and a sufficiently short execution may finish before a useful observation reaches the backend.

That is an inherent limitation of sampled current-state diagnostics.

The main unresolved Scope-1 problem: producer identity and concurrency

Consider two concurrent executions of the same logical task:

dag_id=foo
task_id=transform

execution A runner RSS = 900 MiB
execution B runner RSS = 200 MiB

Useful logical-task values could be:

current total = 1100 MiB
current max   = 900 MiB
active count  = 2

But two independent producers emitting individual Gauge values do not automatically produce any of those meanings.

(dag_id, task_id) is a useful business grouping key, but it is not necessarily sufficient producer/stream identity.

If independent runners appear as the same telemetry stream, their observations may be conflated.

The obvious alternative—adding run_id, PID, execution UUID, or another execution identity—distinguishes producers but introduces execution-level time-series cardinality and churn.

Different metrics backends expose this problem differently. That is an implementation concern, but the underlying semantic problem is backend-independent.

So Scope 1 has an important prerequisite:

Can Airflow export independent runner observations with unambiguous and reasonably bounded identity?

If not, the apparently simple runner metric may not be a sound abstraction, and aggregation before export may be preferable.

I would prefer to decide this semantic question first and then determine the appropriate StatsD/OpenTelemetry implementation, rather than letting a telemetry primitive define the metric contract.

Scope 2 — aggregate current runner observations

A stronger design answers:

"How much runner memory are the currently active executions of this logical task using?"

This requires aggregation before the low-cardinality logical-task metric is exported.

Conceptually:

execution A --+
              |
execution B --+--> stable local aggregation owner
              |          |
execution C --+          +-- group by (dag_id, task_id)
                         |
                         +-- SUM current runner RSS
                         +-- MAX current runner RSS
                         +-- COUNT active executions

The important property is that the aggregation owner has a lifetime longer than one task execution and can maintain bounded state for currently active executions.

Execution-level information can exist internally:

execution identity
PID / process identity
dag_id
task_id

without becoming permanent metric dimensions.

This separates:

high-cardinality execution identity
        -> bounded local state

logical task identity
        -> exported metrics

Possible metric semantics

For each logical task within one aggregation owner:

Illustrative signal	Meaning
task.active_runner_rss_bytes	Sum of successfully observed active runner RSS values
task.max_active_runner_rss_bytes	Maximum of successfully observed active runner RSS values
task.active_executions	Executions known by the owner to be active
task.observed_executions	Active executions whose RSS was successfully read

For example, if three executions are alive but only two can be sampled:

active_executions   = 3
observed_executions = 2

The RSS total/max then describe the observed subset.

Reporting active_executions = 2 would incorrectly redefine "active" as "successfully observed".

Also, even here, summing process RSS should not be interpreted as deduplicated physical memory.

Multiple aggregation owners

"Local" does not imply deployment-wide aggregation.

Suppose:

owner A:
    total = 1100 MiB
    max   = 900 MiB
    count = 2

owner B:
    total = 500 MiB
    max   = 500 MiB
    count = 1

A global logical-task view would conceptually require:

total -> SUM = 1600 MiB
max   -> MAX = 900 MiB
count -> SUM = 3

The sources must therefore remain distinguishable until that second aggregation occurs.

This cross-owner identity/query contract is part of Scope 2; it is not implied merely by choosing metric names.

Lifecycle complexity

The owner would need explicit handling for:

• execution registration/unregistration;
• retry;
• defer/resume;
• unexpected process termination;
• Supervisor/worker shutdown;
• stale registrations;
• PID reuse;
• collection racing with execution termination.

PID alone may be insufficient as durable process identity because PIDs can be reused.

Executor implications

The semantic contract need not be executor-specific, but the concrete aggregation owner may be.

Possible candidates include:

• CeleryExecutor: a long-lived Celery worker domain is a plausible first aggregation boundary;
• LocalExecutor: a long-lived parent/management layer may provide an equivalent boundary, but this needs code-level validation;
• KubernetesExecutor: task-per-Pod execution has no equivalent cross-execution owner inside each task Pod, so aggregation would require a different architecture.

These are design candidates, not claims that Airflow already provides such aggregation facilities.

A practical implementation could initially support only execution paths for which a correct stable owner has been identified.

An unsupported execution path should not silently emit runner-local readings under the same metric name as a metric promising logical-task aggregation.

Scope 3 — execution-workload memory accounting

Scopes 1 and 2 still measure Python Task Runner processes.

That may substantially underestimate executions such as:

Task Runner       300 MiB
    |
    +-- Python    700 MiB
    +-- Java      1.5 GiB
    +-- ffmpeg    2.0 GiB

If the intended question is:

"How much local memory did this execution workload consume?"

then the measurement boundary itself needs to change.

Process discovery as an approximation

One possible approach is to discover processes associated with an execution through parent/child relationships or a process-group boundary where one exists.

This could improve coverage of common local subprocesses, but it should be treated as:

a process-discovery approach with incomplete attribution,

not as a ready-made resource-accounting boundary.

Limitations include:

• processes may detach or escape the discovered hierarchy/group;
• other coordinators/runtimes may use different process models;
• summing RSS can double-count shared resident pages;
• PSS has different attribution semantics and higher collection cost/availability constraints.

Any metric based on this approach should state exactly which processes and memory measure it represents.

Dedicated per-execution cgroup

A stronger Linux-specific design would establish a dedicated cgroup for an execution before the workload starts:

worker
 |
 +-- execution A cgroup
 |      +-- Task Runner
 |      +-- bash
 |      +-- ffmpeg
 |
 +-- execution B cgroup
        +-- Task Runner
        +-- java

The kernel could then account memory to that execution resource group and its descendants.

Relevant cgroup-v2 information to investigate includes:

memory.current
memory.stat
memory.peak
memory.events

A working-set-style value could also be considered if that is the desired semantic.

The important improvement is not that a cgroup creates one universally "correct" memory value—shared memory, cache and charge ownership still have semantics that need documentation.

The improvement is the measurement boundary:

memory is accounted to the execution workload rather than inferred from one selected process.

Historical execution peaks

A kernel-maintained value such as memory.peak may also capture peaks that periodic polling misses.

That could support a separate concept such as:

task.execution_peak_memory_bytes

with one peak observation for each execution whose final resource state was successfully captured.

This is different from periodically recording RSS samples into a Histogram:

periodic RSS Histogram
    -> distribution of sampling points

per-execution peak distribution
    -> distribution of execution peaks

The second is more useful for historical memory-sizing analysis.

However, this is not automatic.

A surviving lifecycle owner must read and export the peak before the cgroup is removed.

In particular, a design that only records the peak after successful task completion would miss some of the most interesting executions:

• failures;
• kills;
• OOM termination.

Reliable peak collection therefore requires an observer/lifecycle design that survives long enough to handle abnormal termination.

It would also require an appropriate bytes-valued distribution API and backend representation; a duration-oriented timing API should not be reused merely because it already exposes a Histogram-like primitive.

Platform cost

Per-execution cgroups introduce substantial platform questions:

• cgroup-v2 availability;
• delegation and permissions;
• workers running inside containers;
• ownership and cleanup;
• abnormal termination;
• unsupported operating systems/deployments.

If this scope is implemented, lack of cgroup support should not silently change the same metric into runner RSS.

The scopes should remain semantically distinct

These are different measurements:

one runner's RSS

aggregate current RSS of active runners

process-tree/process-group approximation

execution-cgroup memory

They answer different questions.

They should not all silently become implementations of:

task.memory_usage

depending on executor, platform, or available privileges.

Likewise, changing a published metric from:

one runner observation

to:

sum of all active runners

would materially change dashboard and alert semantics.

If more than one scope is useful, separate metric names or an explicit migration strategy would be safer.

Questions for the community

I would particularly appreciate feedback on these points:

1. Is a precisely named Python Task Runner RSS diagnostic useful in Airflow core at all?
2. If so, can independent concurrent runner observations be exported with sufficiently clear and bounded producer identity?
   If not, should a stable local owner and Scope-2 aggregation be the minimum design?
3. Should a **(dag_id, task_id)** memory metric be expected to represent correctly aggregated concurrent state from its first introduction?
4. Longer term, is there value in exposing both runner diagnostics and a separate execution-workload memory signal?
5. If workload-level accounting is desirable, is an Airflow-managed per-execution cgroup worth investigating, or should that responsibility remain with the deployment/container layer?

My current inclination is to keep any first contribution as small as possible, but not at the cost of publishing a metric whose measurement boundary, producer identity, or aggregation semantics are ambiguous.