This is an automated email from the ASF dual-hosted git repository.
ashb 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 9068fba9cb7 Fix remote processor injection happening before dictConfig
runs in configure_logging (#66633)
9068fba9cb7 is described below
commit 9068fba9cb79496d726fc78dd57f6400b7544482
Author: Idris Ibrahim <[email protected]>
AuthorDate: Thu Oct 8 08:50:29 2026 +0100
Fix remote processor injection happening before dictConfig runs in
configure_logging (#66633)
In configure_logging(), getattr(remote, 'processors') was called before
dictConfig() ran. dictConfig() calls _clearExistingHandlers() which closes
every handler in logging._handlerList — including the remote handler built
moments earlier. This caused CloudWatch/Watchtower logs to be silently
dropped when using the Task SDK with remote logging on ECS/Kubernetes
workers.
Fix: move the remote processor injection to after dictConfig() has run via
a second structlog.configure() call. Also gate on not sending_to_supervisor
to avoid unmasked events from the task subprocess, and add a None default
to getattr() to handle third-party RemoteLogIO objects.
Provider-side fix: #68779
Closes #66475
* Keep remote task log handlers alive during setup
dictConfig can close a remote handler before it receives its first
task-log record, silently dropping remote logs. The logging setup must
avoid that lifecycle conflict while remaining compatible with remote
handlers that do not provide custom processors.
* Cover remote handlers without custom processors
Handlers supplied by third-party remote logging implementations may not
define custom processors. Keep the regression coverage explicit so that
compatibility does not silently change.
* Cover remote handlers without custom processors
Handlers supplied by third-party remote logging implementations may not
define custom processors. Keep the regression coverage explicit so that
compatibility does not silently change.
---
airflow-core/newsfragments/66633.bugfix.rst | 1 +
task-sdk/src/airflow/sdk/log.py | 15 +++--
task-sdk/tests/task_sdk/test_log.py | 89 +++++++++++++++++++++++++++++
3 files changed, 101 insertions(+), 4 deletions(-)
diff --git a/airflow-core/newsfragments/66633.bugfix.rst
b/airflow-core/newsfragments/66633.bugfix.rst
new file mode 100644
index 00000000000..5ea85be3152
--- /dev/null
+++ b/airflow-core/newsfragments/66633.bugfix.rst
@@ -0,0 +1 @@
+Fix remote task logs being silently dropped when ``dictConfig`` closes a
remote logging handler during configuration.
diff --git a/task-sdk/src/airflow/sdk/log.py b/task-sdk/src/airflow/sdk/log.py
index e105fd70d76..849a7eca209 100644
--- a/task-sdk/src/airflow/sdk/log.py
+++ b/task-sdk/src/airflow/sdk/log.py
@@ -74,7 +74,7 @@ def logging_processors(
if mask_secrets:
extra_processors += (mask_logs,)
- if (remote := load_remote_log_handler()) and (remote_processors :=
getattr(remote, "processors")):
+ if (remote := load_remote_log_handler()) and (remote_processors :=
getattr(remote, "processors", None)):
extra_processors += remote_processors
procs, _, final_writer = structlog_processors(
@@ -119,9 +119,6 @@ def configure_logging(
if mask_secrets:
extra_processors += (mask_logs,)
- if (remote := load_remote_log_handler()) and (remote_processors :=
getattr(remote, "processors")):
- extra_processors += remote_processors
-
configure_logging(
json_output=json_output,
log_level=log_level,
@@ -134,6 +131,16 @@ def configure_logging(
callsite_parameters=callsite_params,
)
+ # Build the remote handler after dictConfig(), which closes every
previously registered handler.
+ if (
+ not sending_to_supervisor
+ and (remote := load_remote_log_handler())
+ and (remote_processors := getattr(remote, "processors", None))
+ ):
+ current_processors = list(structlog.get_config()["processors"])
+ updated_processors = current_processors[:-1] + list(remote_processors)
+ [current_processors[-1]]
+ structlog.configure(processors=updated_processors)
+
def logger_at_level(name: str, level: int) -> Logger:
"""Create a new logger at the given level."""
diff --git a/task-sdk/tests/task_sdk/test_log.py
b/task-sdk/tests/task_sdk/test_log.py
new file mode 100644
index 00000000000..aa955a87a60
--- /dev/null
+++ b/task-sdk/tests/task_sdk/test_log.py
@@ -0,0 +1,89 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+import structlog
+
+from airflow.sdk import log as sdk_log
+
+
+class TestConfigureLogging:
+ @pytest.mark.parametrize("sending_to_supervisor", [True, False])
+ @mock.patch("airflow.sdk.log.load_remote_log_handler")
+ @mock.patch("airflow.sdk._shared.logging.configure_logging")
+ def test_injects_remote_processors_after_dictconfig(
+ self, mock_configure_logging, mock_load_remote_log_handler,
sending_to_supervisor
+ ):
+ initial_processor = mock.Mock()
+ final_renderer = mock.Mock()
+ remote_processor = mock.Mock()
+ remote_handler = mock.Mock(spec=["processors"])
+ remote_handler.processors = (remote_processor,)
+ calls = []
+
+ def configure_structlog(**_):
+ calls.append("dictConfig")
+ structlog.configure(processors=[initial_processor, final_renderer])
+
+ def load_remote_handler():
+ calls.append("load_remote_log_handler")
+ return remote_handler
+
+ mock_configure_logging.side_effect = configure_structlog
+ mock_load_remote_log_handler.side_effect = load_remote_handler
+ original_processors = structlog.get_config()["processors"]
+ sdk_log.configure_logging.cache_clear()
+
+ try:
+
sdk_log.configure_logging(sending_to_supervisor=sending_to_supervisor)
+
+ assert mock_configure_logging.called
+ if sending_to_supervisor:
+ assert calls == ["dictConfig"]
+ assert structlog.get_config()["processors"] ==
[initial_processor, final_renderer]
+ else:
+ assert calls == ["dictConfig", "load_remote_log_handler"]
+ assert structlog.get_config()["processors"] == [
+ initial_processor,
+ remote_processor,
+ final_renderer,
+ ]
+ finally:
+ structlog.configure(processors=original_processors)
+ sdk_log.configure_logging.cache_clear()
+
+ @mock.patch("airflow.sdk._shared.logging.structlog.structlog_processors")
+ @mock.patch("airflow.sdk.log.load_remote_log_handler",
return_value=object())
+ def test_allows_remote_handler_without_processors(
+ self, mock_load_remote_log_handler, mock_structlog_processors
+ ):
+ initial_processor = mock.Mock()
+ final_renderer = mock.Mock()
+ mock_structlog_processors.return_value = ([initial_processor], None,
final_renderer)
+ sdk_log.logging_processors.cache_clear()
+
+ try:
+ processors = sdk_log.logging_processors(json_output=False)
+ finally:
+ sdk_log.logging_processors.cache_clear()
+
+ mock_load_remote_log_handler.assert_called_once_with()
+ assert processors == (initial_processor, sdk_log.mask_logs,
final_renderer)
