1fanwang opened a new pull request, #66405:
URL: https://github.com/apache/airflow/pull/66405
## Description
Foundation discussion vehicle for AIP-97 (Infrastructure-Aware Task
Execution). Adds the executor-agnostic ``FailureDetails`` type and
extends the ``on_task_instance_failed`` hookspec to accept it as an
optional keyword argument, so listener authors can begin coding against
the API shape while each executor's surfacing PR iterates against a
fixed contract.
> **Note**: AIP-97 is currently in DRAFT on the cwiki. This PR is **not**
> implementing an accepted AIP — it is the smallest concrete artifact that
> demonstrates the listener-author API surface so the AIP discussion has
> something concrete to push back on. Marking the PR draft for that reason.
## The gap this addresses
Today ``on_task_instance_failed`` only sees the worker-side ``error``
exception. Failure causes that originate **outside** the worker process
are visible only to the executor and never reach the listener:
| Executor | Cause not in ``error`` |
|----------|------------------------|
| Kubernetes | ``OOMKilled``, ``PodEvicted``, ``ImagePullBackOff``,
``NodeOutOfMemory`` |
| Celery / CeleryKubernetes | ``WorkerLost``, ``SoftTimeLimit``, broker
disconnect |
| LocalExecutor | oom-killer ``SIGKILL``, subprocess termination by external
signal |
| Resource-managed clusters (YARN, GG, etc.) | preemption, eviction, queue
timeout |
The shape is the same across executors: a kind tag plus a categorical
reason plus structured metadata. ``FailureDetails`` formalises that shape.
## What this PR ships
- ``airflow.listeners.types.FailureDetails`` — frozen attrs class with
``executor_kind: str``, ``infra_reason: str | None = None``,
``infra_metadata: dict[str, Any] = factory(dict)``.
- ``on_task_instance_failed`` hookspec gains an optional
``failure_details: FailureDetails | None = None`` keyword argument.
pluggy dispatches by parameter name, so existing ``hookimpl``
implementations that don't declare ``failure_details`` keep working
unchanged.
- Newsfragment + parametrized tests covering the dataclass surface and
both listener variants (declares-it / doesn't-declare-it).
## Listener-author surface
\`\`\`python
from airflow.listeners import hookimpl
from airflow.listeners.types import FailureDetails
class InfraTrackingListener:
@hookimpl
def on_task_instance_failed(
self,
previous_state,
task_instance,
error,
failure_details: FailureDetails | None = None,
):
if failure_details and failure_details.infra_reason == \"OOMKilled\":
... # route to capacity-planning alert
\`\`\`
## Intentionally deferred (each is a follow-up PR candidate)
1. **Kubernetes executor wiring** — populate ``FailureDetails`` from
pod ``status.containerStatuses[0].state.terminated`` and pod
``status.reason`` when the executor marks a task FAILED.
2. **Celery executor wiring** — populate from ``WorkerLostError`` and
``SoftTimeLimitExceeded``.
3. **LocalExecutor wiring** — populate from ``Popen`` exit code +
signal (oom-killer detection on Linux).
4. **Resource-manager wiring** — preemption / eviction reasons from
YARN / Grid / etc.
5. **Schema additions** — should ``FailureDetails`` be persisted on
the TaskInstance (DB column / JSON field) so the UI can render it?
Out of scope for the listener-side foundation.
## Open questions for the discussion
- Should ``infra_reason`` be a free-form string or a constrained enum?
Free-form is more permissive but listeners can't pattern-match
reliably; an enum is brittle but searchable. Free-form for now;
open to switching.
- Should there be a corresponding ``FailureDetails`` on
``on_task_instance_success`` (success-with-warnings)? Out of scope
for this PR — the failure path is the demonstrably-broken one.
- Naming: ``FailureDetails`` vs ``InfraFailureDetails`` vs
``ExecutorFailureContext``. Open to whatever the AIP discussion
settles on.
## Testing
- Parametrized ``TestFailureDetails.test_construct`` covers five
realistic executor / reason / metadata shapes (k8s OOM, k8s
evicted, celery worker-lost, local SIGKILL, unknown reason).
- ``test_default_metadata_is_empty_dict`` and ``test_frozen`` lock
the dataclass invariants.
- ``TestOnTaskInstanceFailedAcceptsFailureDetails`` exercises the
pluggy-dispatch contract: the new listener receives the payload,
the legacy listener (no ``failure_details`` parameter) still gets
called.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
