GitHub user zhanqian-zhang created a discussion: Request for feedback: task
memory diagnostics design for Airflow 3
## Context and goal
Airflow 2.10 introduced task CPU and memory metrics in #39650.
These metrics are no longer emitted in Airflow 3. Their absence was reported in
#49983, and #56690 later proposed restoring CPU and memory metrics but was not
merged.
Before attempting another implementation, I would like to clarify what a task
memory metric should actually mean.
The practical motivation is:
> Help users notice tasks whose memory behavior may be worth investigating or
> optimizing.
There are multiple useful answers to that question, with substantially
different semantic guarantees and implementation costs.
The three scopes below are **design alternatives / possible extensions, not an
implementation commitment or a fixed roadmap**.
In particular, the smallest scope can provide positive diagnostic evidence when
high memory is observed, but it cannot reliably rank task executions by
historical peak.
---
## Scope 1: Python Task Runner RSS diagnostic
The smallest useful signal would follow the lightweight process-sampling idea
from Airflow 2.10, but define the measurement much more narrowly.
For example:
```text
task.runner_rss_bytes
```
Meaning:
> Periodically sampled Resident Set Size of the Python Task Runner process.
The metric would use bytes rather than a percentage.
### Measurement boundary
Conceptually:
```text
Supervisor
|
+-- Python Task Runner <-- measured
|
+-- task code
+-- possible subprocesses
```
Only the Python Task Runner process itself is measured.
It would **not** mean:
- total memory consumed by the task workload;
- memory consumed by subprocesses;
- process-tree memory;
- container or Pod memory;
- host memory pressure;
- total memory of all concurrent executions of the same logical task;
- true execution peak memory.
This distinction is important because a task may delegate most of its work to
Bash, another Python interpreter, Java, ffmpeg, or another local/remote process.
### Sampling lifecycle
I would not define this as "the entire runner lifetime", because some task
identity and metric context becomes available only after part of startup has
already occurred.
A more realistic contract is:
> Start resource observation once sufficient task identity is available in the
> Python runner, and continue through finalization where feasible. The exact
> start/stop boundaries should be defined by the implementation.
Observation failures must remain strictly best-effort and must never change
task outcome, retries, deferral, or cleanup behavior.
### What this scope is good for
For Python-heavy workloads, runner RSS may still be a useful diagnostic for
finding obviously large Python processes.
Its attraction is that it can remain relatively small:
- process RSS is cheap to read;
- it does not require cgroup support;
- it does not attempt workload accounting;
- it does not require introducing a general resource-aggregation subsystem.
However, this remains a small implementation **only if the export-identity
problem below has a satisfactory solution**.
### Sampling limitation
Periodic sampling cannot guarantee peaks.
For example:
```text
0.0s 500 MiB
1.0s 8 GiB
2.0s 600 MiB
3.0s exit
```
A five-second sampler can miss the entire execution, including the 8 GiB spike.
Therefore this signal could answer:
> "We observed this runner using a lot of memory."
It cannot reliably answer:
> "These are the top tasks by historical peak memory."
---
## The main unresolved issue for Scope 1: producer identity and concurrency
Suppose two executions of the same logical task run concurrently:
```text
dag_id=foo
task_id=transform
execution A: 900 MiB
execution B: 200 MiB
```
Useful aggregate values might be:
```text
current total = 1100 MiB
current max = 900 MiB
active count = 2
```
But two independent processes emitting the same logical metric do not
automatically produce any of those meanings.
The problem is deeper than deciding whether users are willing to accept a
non-aggregated metric.
If the two producers have the same metric-stream identity, the telemetry system
may not have enough information to distinguish them correctly.
Conversely, adding an execution identifier such as `run_id`, PID, or execution
UUID makes the producers unambiguous but introduces execution-level time-series
cardinality and churn.
Different backends expose this problem differently. For example, OTel has
explicit stream/resource identity considerations, while plain StatsD has much
weaker identity/tag semantics.
I therefore see an unresolved Scope-1 design question:
> Can Airflow represent independent runner observations with a stable and
> acceptable metric identity without creating execution-level cardinality?
If not, the apparently simple Scope 1 may not actually be the right
abstraction, and aggregation before export may be preferable.
I think the semantic decision should come first; the exact StatsD/OTel
implementation should follow from it.
---
## Scope 2: current logical-task aggregation
A stronger metric could answer:
> "How much runner memory are all currently active executions of this logical
> task using?"
That requires aggregation before the low-cardinality metric is exported.
Conceptually:
```text
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
```
Possible semantics would be:
```text
task.active_runner_rss_bytes
= sum of current RSS of observed active runners
task.max_active_runner_rss_bytes
= maximum current RSS among observed active runners
task.active_executions
= number of known active executions
```
It may also be useful to distinguish:
```text
active_executions
observed_executions
```
For example, if three executions are known to be alive but RSS can only be read
for two, reporting `active_executions = 2` would incorrectly redefine "active"
as "successfully observed".
### Why a stable aggregation owner helps
The owner can maintain bounded execution-level state internally:
```text
execution identity
process identity
dag_id
task_id
```
without exporting those high-cardinality identities as metric dimensions.
This separates two concerns:
```text
execution identity
-> local bounded state
logical task identity
-> exported metric
```
That avoids solving writer identity by adding a new time series for every
execution.
### Lifecycle cost
This stronger semantic requires significantly more lifecycle work:
- register/unregister;
- retries;
- defer/resume;
- unexpected process death;
- stale registrations;
- PID reuse;
- worker restart;
- collection racing with execution termination.
A PID alone may not be a sufficient long-lived process identity because PIDs
can be reused.
### Executor implications
The metric contract could remain executor-independent, while the concrete
aggregation owner may differ by execution architecture.
For example:
- a long-lived Celery worker domain is a plausible aggregation boundary for
concurrent Celery executions;
- LocalExecutor may have a suitable long-lived management/parent boundary,
which would need code-level validation;
- KubernetesExecutor is structurally different because an execution is
typically associated with its own worker Pod, so there is no equivalent
cross-execution owner inside that Pod.
A practical first implementation of this scope might therefore support only
execution paths for which a correct aggregation owner has been identified.
Unsupported paths should not emit the same metric name with weaker semantics.
---
## Scope 3: execution-workload accounting
Scopes 1 and 2 still measure **runner processes**.
That can significantly underestimate workloads such as:
```text
Task Runner 300 MiB
|
+-- Python 700 MiB
+-- Java 1.5 GiB
+-- ffmpeg 2.0 GiB
```
If the intended question is instead:
> "How much local memory pressure did this task execution workload cause?"
then the measurement boundary has to change.
### Possible process-discovery approach
One possible approach is to discover processes associated with the execution,
for example using parent/child relationships or an execution-related
process-group boundary where one exists.
This could improve coverage of common subprocess workloads, but it should
**not** be treated as complete workload accounting.
Limitations include:
- subprocesses may detach or otherwise escape the discovered hierarchy/group;
- another coordinator/runtime may use a different process model;
- simply summing process RSS double-counts shared resident pages;
- PSS offers different attribution semantics but is more expensive and
platform-dependent.
So I would view process-group/process-tree inspection as:
> a possible process-discovery strategy with incomplete attribution,
not as a ready-made workload accounting model.
### Dedicated per-execution cgroup
A stronger Linux-specific design would establish a dedicated cgroup for an
execution before its workload starts:
```text
worker
|
+-- execution A cgroup
| +-- runner
| +-- bash
| +-- ffmpeg
|
+-- execution B cgroup
+-- runner
+-- java
```
The kernel would then account memory against the execution resource group and
its descendants.
Relevant cgroup-v2 information may include:
```text
memory.current
memory.stat
memory.peak
memory.events
```
A working-set-style value could also be considered if that is the desired
semantic.
The advantage is not that cgroups provide one universally "correct" definition
of memory. Shared memory, cache and charge ownership still have semantics that
need documentation.
The major improvement is the boundary:
> memory is accounted to an execution workload rather than inferred from one
> selected process.
### Per-execution peaks
A dedicated cgroup may also provide a better way to capture short-lived memory
peaks than periodic polling.
For example, `memory.peak` can retain a peak that occurred between sampling
intervals.
However, this still requires correct lifecycle handling:
> a surviving observer must read and export the peak before the execution
> cgroup is removed.
A naive "emit one peak observation when the task completes successfully" design
would be insufficient because failed, killed, or OOM executions are precisely
some of the most important cases.
Peak collection therefore needs a lifecycle owner that can survive long enough
to observe abnormal termination as well.
If this can be made reliable, a separate historical metric could represent one
peak observation per observed execution, potentially enabling distributions
such as P50/P95/P99 peak memory.
That is different from periodically putting RSS samples into a Histogram:
```text
periodic RSS histogram
-> distribution of samples
per-execution peak histogram
-> distribution of execution peaks
```
The latter is much closer to historical memory-sizing analysis.
### Platform cost
A dedicated cgroup design is much more invasive and platform-dependent.
It would require answers around:
- cgroup v2 availability;
- delegation/permissions;
- running inside containers;
- lifecycle and cleanup;
- unsupported platforms;
- abnormal execution termination.
It should not silently fall back to runner RSS under the same metric name when
cgroup accounting is unavailable.
---
## These scopes should not silently share one semantic
The following values answer different questions:
```text
runner RSS
aggregate RSS of active runners
process-tree/process-group approximation
execution-cgroup memory
```
They should not all be hidden behind a generic metric such as:
```text
task.memory_usage
```
if the meaning changes depending on executor or platform.
Similarly, changing a published metric from:
```text
one runner observation
```
to:
```text
sum of all active runners
```
would change dashboard and alert semantics.
If more than one scope is eventually useful, separate metrics or an explicit
migration strategy would be safer.
---
## Questions for the community
The main questions I would like feedback on are:
1. **Is a narrowly defined Python Task Runner RSS diagnostic useful in Airflow
core at all?**
2. **If so, what producer/export identity would make independent concurrent
runner observations well-defined without introducing unacceptable
execution-level cardinality?**
3. **Should a `(dag_id, task_id)` memory metric be expected to represent
correctly aggregated concurrent state from its first introduction?**
If yes, the aggregation-owner design may be the appropriate minimum scope.
4. **Longer term, is there value in a distinct execution-workload memory signal
in addition to runner diagnostics?**
5. **For workload-level accounting, is an Airflow-managed per-execution cgroup
something worth investigating, or is that responsibility better left to the
deployment/container layer?**
My current preference is to keep any first contribution as small as possible,
but not at the cost of publishing a metric whose identity or semantics are
ambiguous.
GitHub link: https://github.com/apache/airflow/discussions/73623
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
