digit50 opened a new issue, #75017:
URL: https://github.com/apache/airflow/issues/75017
### Under which category would you file this issue?
Task SDK
### Apache Airflow version
main (development), reproduced at d2c41b8c2027fcba5e632f00f5f80d86ec767981.
Also 3.3.2 with apache-airflow-task-sdk 1.3.2 and 3.4.0b1 with task-sdk
1.4.0b1, which carry the same helper. 3.3.1 and task-sdk 1.3.1 are not affected.
### What happened and how to reproduce it?
`process_log_messages_from_subprocess` in
`task-sdk/src/airflow/sdk/execution_time/supervisor.py` (lines 2909-2921) pops
`level`, `event` and `exception` from each JSON log event coming from the task
process and spreads every other key into the helper `_log_to_target(target,
level, msg, **event)` (line 2913, defined at line 2916). A log event that
carries a key named `target` or `msg` collides with the helper's own parameter
names and raises:
```
TypeError: _log_to_target() got multiple values for argument 'target'
```
Nothing on the way out catches it: `make_buffered_socket_reader.cb` catches
only `StopIteration`, `_service_subprocess` only `BrokenPipeError` and
`ConnectionResetError`, `ActivitySubprocess.wait` has only a `finally` that
closes the selector, `_PythonCoordinator.execute_task` has no handler and
`supervise_task` catches only `TaskLaunchError` and httpx request errors. So
`wait()` raises before `update_task_state_if_needed()` and `_upload_logs()`,
the heartbeat loop stops, and the child, which runs in its own process group,
is not killed.
An ordinary log call in a task is enough:
```python
structlog.get_logger().info("deploying", target="prod")
```
Airflow itself supplies a second trigger. `task_runner.py` line 1239 logs
`log.debug("Using serialized startup message from environment", msg=msg)` in
the `run_as_user` re-exec path, and debug events are forwarded to the
supervisor when `[logging] logging_level = DEBUG`. So every impersonated task
in a DEBUG-logging deployment should hit this with no Dag author involvement
(the debug forwarding of a `msg=` key was reproduced; the `sudo` re-exec itself
was not run).
Observed on the checkout, driving `supervise_task` with a real forked child
and a mock API transport that records every request:
```
info("deploying", target="prod") -> supervise_task raised TypeError after
0.35 s
requests sent: PATCH .../run, one PUT .../heartbeat; no state PATCH, no
log upload
child not killed: ran user code to the end, then its terminal message
failed with
ConnectionResetError; pid still present 9 s after the supervisor raised
info("deploying", msg="prod") -> same, "multiple values for argument
'msg'"
debug("deploying", msg="prod") -> rc 0 at logging_level=INFO, TypeError
at DEBUG
info("deploying", environment="prod") -> rc 0, PATCH .../state {"state":
"success"}
```
Under the standard executors the task instance does not stay running until
the heartbeat timeout: the LocalExecutor worker reports the workload's failure
state and the Celery wrapper re-raises, so the scheduler's
`process_executor_events` sees a failed event for a running task instance, logs
a state mismatch, counts `tasks.killed_externally` and calls `handle_failure`
(read from the code, the executor and scheduler were not run). The result is
that a task whose user code succeeds is recorded as failed or up for retry, a
retry can start while the orphaned process is still running user code, and the
task's final logs are never uploaded.
The same generator serves the Dag processor: `DagFileProcessorProcess`
registers it through `WatchedSubprocess._register_pipe_readers`, and
`DagFileProcessorManager._service_processor_sockets`
(`airflow-core/src/airflow/dag_processing/manager.py` lines 636-639) catches
only the same two socket errors while `DagProcessorJobRunner._execute` logs and
re-raises. A Dag file with `structlog.get_logger().info("parse",
target="prod")` at module level, run under `DagFileProcessorProcess.start`,
raised the same `TypeError` from the processor's selector callback; from the
code, that stops the Dag processor job and does so again on every restart.
The helper was added by #69523 (6d9cc2ed37) and backported by #72888
(3b6f63eef9). Before it, the code called `target.log(level, msg, **event)`
directly, where only `level` and `event` could collide and both are popped.
**How to reproduce**
1. A Dag with a Python task whose callable runs
`structlog.get_logger().info("deploying", target="prod")` and then sleeps a few
seconds.
2. Run it under the task SDK supervisor (any executor).
3. The supervisor logs the `TypeError` above and exits without sending a
terminal state; the task process keeps running; the task instance is then
failed by the scheduler's state-mismatch handling.
For the Dag processor: put the same log call at module level in a Dag file
and watch the Dag processor stop.
### What you think should happen instead?
A log event's keys should never be able to collide with the forwarding
helper's parameters. Making them positional-only removes both collisions:
```python
def _log_to_target(target, level, msg, /, **event):
```
Alternatively pass the event as one dict and call `target.log(level, msg,
**event)` inside. The existing `test_process_log_messages_from_subprocess`
sends only `level` and `event`; a case with `{"level": "info", "event": "x",
"target": "y", "msg": "z"}` would cover this.
### Operating System
Linux
### Deployment
Other
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
Not relevant.
### Official Helm Chart version
_No response_
### Kubernetes Version
Not Applicable
### Helm Chart configuration
Not Applicable
### Docker Image customizations
Not Applicable
### Anything else?
With `[logging] logging_level = DEBUG` and `run_as_user` set, the startup
debug line in `task_runner.py` (line 1239) carries `msg=` and should trigger
the same crash for every impersonated task.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
