tasks log is not correctly captured

pingzh commented 2 years ago

Apache Airflow version

2.2.4

What happened

When the task is executed via exec https://github.com/apache/airflow/blob/2.2.4/airflow/task/task_runner/standard_task_runner.py#L43 , the task std log is not captured.

Please see the image:

In the log, i don't see run_id=

What you think should happen instead

No response

How to reproduce

You can set the run_as_user on your task or edit the installed pkg to only use _start_by_exec in the StandardTaskRunner.

Operating System

Apple M1 Max, version: 12.2

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

on my local Mac.

Anything else

Python version:

Python 3.9.7

I used the default airflow_local_setttings and default logging class.

Are you willing to submit PR?

[x] Yes I am willing to submit a PR!

Code of Conduct

[X] I agree to follow this project's Code of Conduct

pingzh commented 2 years ago

while investigating this issue, i found that if i set DONOT_MODIFY_HANDLERS to True, the airflow tasks run will hang:

for example, if my local:

i changed the code of DONOT_MODIFY_HANDLERS to True and forced the StandardTaskRunner to use _start_by_exec.

airflow tasks run example_bash_operator also_run_this scheduled__2022-05-05T00:00:00+00:00 --job-id 272 --local --subdir /Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/example_dags/example_bash_operator.py -f

it stuck without making any progress:

i had to force killing it, it looks like there was an infinite loop:

  File "/Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/utils/log/logging_mixin.py", line 109, in _propagate_log
    self.logger.log(self.level, remove_escape_codes(message))
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1512, in log
    self._log(level, msg, args, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1589, in _log
    self.handle(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1599, in handle
    self.callHandlers(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1661, in callHandlers
    hdlr.handle(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 952, in handle
    self.emit(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1086, in emit
    stream.write(msg + self.terminator)
  File "/Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/utils/log/logging_mixin.py", line 121, in write
    self._propagate_log(self._buffer.rstrip())
  File "/Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/utils/log/logging_mixin.py", line 109, in _propagate_log
    self.logger.log(self.level, remove_escape_codes(message))
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1512, in log
    self._log(level, msg, args, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1589, in _log
    self.handle(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1599, in handle
    self.callHandlers(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1661, in callHandlers
    hdlr.handle(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 952, in handle
    self.emit(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1086, in emit
    stream.write(msg + self.terminator)
  File "/Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/utils/log/logging_mixin.py", line 121, in write
    self._propagate_log(self._buffer.rstrip())
  File "/Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/utils/log/logging_mixin.py", line 109, in _propagate_log
    self.logger.log(self.level, remove_escape_codes(message))
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1512, in log
    self._log(level, msg, args, **kwargs)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1589, in _log
    self.handle(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1599, in handle
    self.callHandlers(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1661, in callHandlers
    hdlr.handle(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 952, in handle
    self.emit(record)
  File "/Users/ping_zhang/miniforge3/envs/airflow-rb2.2.4/lib/python3.9/logging/__init__.py", line 1086, in emit
    stream.write(msg + self.terminator)
  File "/Users/ping_zhang/airlab/repos/airflow-internal-1.10.4/airflow/utils/log/logging_mixin.py", line 120, in write
    self._buffer += message
KeyboardInterrupt

pingzh commented 2 years ago

found the root cause. working on a fix.

fixed in this PR: https://github.com/apache/airflow/pull/23684/

pingzh commented 2 years ago

fixed in https://github.com/apache/airflow/pull/23684

apache / airflow