1fanwang opened a new pull request, #71566:
URL: https://github.com/apache/airflow/pull/71566
A worker pod that dies during task-runner startup — an import error, an OOM
kill landing on PID 1, a malformed entrypoint — never writes a local or
remote
log. The log reader only asks the executor for its container log while the
task
is `RUNNING`, so once the attempt lands in `FAILED` the pod log is never
consulted and the task renders a blank log page. The traceback is only
reachable
out of band, via `kubectl logs` or a log aggregator, and only until the pod
is
reaped.
Kubernetes reports nothing useful for this shape of failure either — the
container status is `exit_code=1, reason="Error", message=None` — so the
structured failure details added in #54115 have nothing to show. The pod's
stdout is the only record of what went wrong.
This wires the executor log source into the fallback core already has. When
an
attempt failed and neither a local nor a remote log turned up, the reader
asks
the executor for its container log, the same last-resort condition already
used
a few lines below for served logs.
`KubernetesExecutor.get_streaming_task_log`
and `[kubernetes_executor] running_pod_log_lines` do the actual read, so
there
is no new configuration and no extra call on the scheduler's path. Any
executor
implementing `get_task_log` benefits; the base implementation returns
nothing,
so executors that don't are unaffected.
Successful and upstream-failed tasks keep their current behaviour, and a
failed
task that does have a worker log still skips the executor entirely, so the
API
call only happens when there is genuinely nothing else to show. A finished
task
naming an executor that has since been dropped from the config no longer
raises
out of the log reader — that path was previously unreachable for finished
tasks.
closes: #66795
# Testing Done
Live end-to-end against a real kind cluster (`breeze k8s create-cluster`,
v1.30.13), driving `FileTaskHandler._read()` through the real
`KubernetesExecutor` against a real crashed pod. Identical environment for
both
runs; the only variable is the patch.
| # | Scenario | Result |
|---|---|---|
| 1 | Failed pod, unpatched | Blank log |
| 2 | Failed pod, patched | Pod traceback surfaced |
| 3 | Unit suite, `test_file_task_handler.py` | 18 passed |
| 4 | Unit suite, k8s `log_handlers/` | 8 passed |
<details>
<summary>Raw logs</summary>
```console
$ kubectl get nodes
NAME STATUS ROLES AGE
VERSION
airflow-python-3.10-v1.30.13-control-plane Ready control-plane 36s
v1.30.13
airflow-python-3.10-v1.30.13-worker Ready <none> 15s
v1.30.13
$ git checkout origin/main --
airflow-core/src/airflow/utils/log/file_task_handler.py
$ python dev/repro_66795.py
== creating a worker pod that crashes at startup ==
pod phase=Failed
container exit_code=1 reason='Error' message=None
== FileTaskHandler._read() for the failed task ==
::group::Log message source details
::endgroup::
RESULT: pod traceback in task log = False
$ git checkout HEAD --
airflow-core/src/airflow/utils/log/file_task_handler.py
$ python dev/repro_66795.py
== creating a worker pod that crashes at startup ==
pod phase=Failed
container exit_code=1 reason='Error' message=None
== FileTaskHandler._read() for the failed task ==
::group::Log message source details
Attempting to fetch logs from pod repro-66795-worker through kube API
Found logs through kube API
::endgroup::
Traceback (most recent call last):
File "<string>", line 1, in <module>
ModuleNotFoundError: No module named 'pandas_does_not_exist'
RESULT: pod traceback in task log = True
```
Unit suites, and the same two new cases failing on unpatched source:
```console
$ uv run --project airflow-core pytest
airflow-core/tests/unit/utils/log/test_file_task_handler.py -q
18 passed, 1 warning in 38.54s
$ uv run --project providers/cncf/kubernetes pytest \
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/log_handlers/ -q
8 passed, 1 warning in 23.58s
# with `git checkout origin/main --
airflow-core/src/airflow/utils/log/file_task_handler.py`:
FAILED ...::test_reads_executor_logs_when_attempt_left_no_other_logs[failed]
FAILED
...::test_reads_executor_logs_when_attempt_left_no_other_logs[up_for_retry]
2 failed, 6 passed
```
</details>
<details>
<summary>Reproducer used above (<code>dev/repro_66795.py</code>, not
committed)</summary>
Run against any kind cluster with:
```bash
export KUBECONFIG=<your kubeconfig>
export AIRFLOW__CORE__EXECUTOR=KubernetesExecutor
export AIRFLOW__KUBERNETES_EXECUTOR__NAMESPACE=default
export AIRFLOW__KUBERNETES_EXECUTOR__IN_CLUSTER=False
export AIRFLOW__KUBERNETES_EXECUTOR__CONFIG_FILE=$KUBECONFIG
python dev/repro_66795.py
```
```python
"""Live repro for issue #66795 against a real kind cluster.
Creates a worker pod that dies the way a task runner does on an import
error, then asks
FileTaskHandler._read() for that task's log. Before the fix the log is
empty; after it the
pod's traceback comes through.
python dev/repro_66795.py
"""
from __future__ import annotations
import sys
import tempfile
import time
from unittest.mock import MagicMock
from kubernetes import client, config
from airflow.utils.state import TaskInstanceState
DAG_ID = "repro_66795"
TASK_ID = "crashing_task"
RUN_ID = "manual__2026-01-01T00-00-00-00-00"
TRY_NUMBER = 1
WORKER_ID = "1"
NAMESPACE = "default"
POD_NAME = "repro-66795-worker"
CRASH = "import pandas_does_not_exist # noqa: F401"
def make_pod(core: client.CoreV1Api) -> None:
try:
core.delete_namespaced_pod(name=POD_NAME, namespace=NAMESPACE)
time.sleep(5)
except client.rest.ApiException:
pass
core.create_namespaced_pod(
namespace=NAMESPACE,
body=client.V1Pod(
metadata=client.V1ObjectMeta(
name=POD_NAME,
labels={
"dag_id": DAG_ID,
"task_id": TASK_ID,
"run_id": RUN_ID,
"try_number": str(TRY_NUMBER),
"kubernetes_executor": "True",
"airflow-worker": WORKER_ID,
},
),
spec=client.V1PodSpec(
restart_policy="Never",
containers=[
client.V1Container(
name="base",
image="python:3.12-slim",
command=["python", "-c", CRASH],
)
],
),
),
)
for _ in range(60):
pod = core.read_namespaced_pod(name=POD_NAME, namespace=NAMESPACE)
if pod.status.phase in ("Failed", "Succeeded"):
print(f"pod phase={pod.status.phase}")
state = pod.status.container_statuses[0].state.terminated
print(f"container exit_code={state.exit_code}
reason={state.reason!r} message={state.message!r}")
return
time.sleep(2)
raise SystemExit("pod never reached a terminal phase")
def read_task_log() -> list[str]:
from airflow.utils.log.file_task_handler import FileTaskHandler
ti = MagicMock()
ti.dag_id = DAG_ID
ti.task_id = TASK_ID
ti.run_id = RUN_ID
ti.map_index = -1
ti.try_number = TRY_NUMBER
ti.queued_by_job_id = WORKER_ID
ti.executor = "KubernetesExecutor"
ti.executor_config = {}
ti.hostname = POD_NAME
ti.state = TaskInstanceState.FAILED
# An empty local base and no remote logging is exactly the state a
worker that died at
# startup leaves behind: nothing was ever written anywhere.
handler = FileTaskHandler(base_log_folder=tempfile.mkdtemp())
handler._render_filename = lambda ti, try_number:
f"{DAG_ID}/{RUN_ID}/{TASK_ID}/{try_number}.log"
handler._read_from_logs_server = lambda ti, path: ([], [])
stream, _ = handler._read(ti=ti, try_number=TRY_NUMBER)
return [msg.event for msg in stream]
def main() -> int:
config.load_kube_config()
core = client.CoreV1Api()
print("== creating a worker pod that crashes at startup ==")
make_pod(core)
print("\n== FileTaskHandler._read() for the failed task ==")
events = read_task_log()
for event in events:
print(f" {event}")
hit = any("ModuleNotFoundError" in e for e in events)
print(f"\nRESULT: pod traceback in task log = {hit}")
return 0 if hit else 1
if __name__ == "__main__":
sys.exit(main())
```
</details>
--
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]
