This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ebe6c58d42e Fix triggerer CrashLoopBackOff when json_logs is enabled
(#68584)
ebe6c58d42e is described below
commit ebe6c58d42e6b6799a61b98312f0e38671204de1
Author: safaehar <[email protected]>
AuthorDate: Wed Jul 29 10:44:17 2026 +0200
Fix triggerer CrashLoopBackOff when json_logs is enabled (#68584)
* Fix triggerer CrashLoopBackOff when json_logs is enabled
TriggerRunnerSupervisor._process_log_messages_from_subprocess primes itself
by calling airflow.sdk.log.configure_logging() with no arguments.
json_output
defaults to False, so this reconfigures structlog globally and installs the
text WriteLogger factory -- overwriting the bytes BytesLogger factory that
startup set up from json_logs=True.
The stdout/stderr forwarders (_create_log_forwarder -> forward_to_log) were
already wrapped with the JSON (bytes) processor chain but bind their
underlying
logger lazily. As soon as a trigger subprocess writes to stdout/stderr --
for
example an import-time warning from a provider trigger that pulls in a heavy
client (kubernetes, boto3) -- the lazy bind resolves against the now-text
factory and WriteLogger.msg does `message + "\n"` on bytes from the JSON
renderer, raising `TypeError: can't concat str to bytes` and crash-looping
the
triggerer.
Pass json_output from the logging.json_logs config so the global structlog
factory stays consistent with the rest of the process.
* Add newsfragment
* Refactor docstring for test_process_log_messages_configures_logging
Updated the docstring to clarify the behavior of
`_process_log_messages_from_subprocess()` regarding JSON logging configuration.
* Delete airflow-core/newsfragments/68584.bugfix.rst
* Remove comments about logging configuration
Removed comments explaining the configure_logging function's behavior with
json_logs setting.
---
.../src/airflow/jobs/triggerer_job_runner.py | 2 +-
airflow-core/tests/unit/jobs/test_triggerer_job.py | 21 +++++++++++++++++++++
2 files changed, 22 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index f851e491a69..8f8e930b07f 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -1018,7 +1018,7 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
from airflow.sdk.log import configure_logging
- configure_logging()
+ configure_logging(json_output=conf.getboolean("logging", "json_logs",
fallback=False))
fallback_log = structlog.get_logger(logger_name=__name__)
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 31a77b50b65..dcaaf7b3951 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -362,6 +362,27 @@ def test_run_invokes_seams_in_order(supervisor_builder,
mocker):
assert events == ["enter", "tick-1", "tick-2", "tick-3", "exit"]
[email protected]("json_logs", [True, False])
+def
test_process_log_messages_configures_logging_matching_json_logs(supervisor_builder,
mocker, json_logs):
+ """_process_log_messages_from_subprocess() must reconfigure logging using
the
+ configured ``logging.json_logs`` value rather than the default
+ ``json_output=False``.
+ This generator reconfigures structlog globally when first primed. Failing
to
+ propagate the configured JSON logging mode can leave the triggerer with an
+ inconsistent logging configuration and break subprocess log forwarding.
+ """
+ supervisor = supervisor_builder()
+
+ configure_logging = mocker.patch("airflow.sdk.log.configure_logging")
+ mocker.patch("airflow.sdk.log.logging_processors")
+
+ with conf_vars({("logging", "json_logs"): str(json_logs)}):
+ gen = supervisor._process_log_messages_from_subprocess()
+ next(gen) # prime the generator -- this is what calls
configure_logging()
+
+ configure_logging.assert_called_once_with(json_output=json_logs)
+
+
def test_client_delegates_to_make_client_and_caches_result(supervisor_builder,
mocker):
"""``supervisor.client`` delegates to ``make_client`` (the
subclass-override hook)
and caches the result across accesses."""
