From 4b84e61e818886b226525f4245f2471389e22ebe Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sat, 16 May 2026 17:25:10 +0800 Subject: [PATCH 1/7] Decouple remote logging configuration from core airflow Fix type annotation --- .../src/airflow/provider.yaml.schema.json | 19 ++ .../src/airflow/provider_info.schema.json | 19 ++ airflow-core/src/airflow/providers_manager.py | 55 ++++ .../unit/always/test_providers_manager.py | 81 ++++++ .../src/airflow_shared/logging/factory.py | 186 ++++++++++++++ shared/logging/tests/logging/test_factory.py | 242 ++++++++++++++++++ task-sdk/src/airflow/sdk/log.py | 13 +- .../airflow/sdk/providers_manager_runtime.py | 57 ++++- .../test_providers_manager_runtime.py | 68 ++++- 9 files changed, 731 insertions(+), 9 deletions(-) create mode 100644 shared/logging/src/airflow_shared/logging/factory.py create mode 100644 shared/logging/tests/logging/test_factory.py diff --git a/airflow-core/src/airflow/provider.yaml.schema.json b/airflow-core/src/airflow/provider.yaml.schema.json index 5714b8db658c5..648aedc1a55c6 100644 --- a/airflow-core/src/airflow/provider.yaml.schema.json +++ b/airflow-core/src/airflow/provider.yaml.schema.json @@ -529,6 +529,25 @@ "type": "string" } }, + "remote-logging": { + "type": "array", + "description": "Remote logging IO handlers contributed by the provider. Each entry registers a RemoteLogIO implementation that ProvidersManager dispatches by URL scheme.", + "items": { + "type": "object", + "required": ["classpath", "scheme"], + "additionalProperties": false, + "properties": { + "classpath": { + "type": "string", + "description": "Fully-qualified class name of the RemoteLogIO implementation." + }, + "scheme": { + "type": "string", + "description": "URL scheme (without ://) of [logging] remote_base_log_folder that this handler claims." + } + } + } + }, "auth-backends": { "type": "array", "description": "API Auth Backend module names", diff --git a/airflow-core/src/airflow/provider_info.schema.json b/airflow-core/src/airflow/provider_info.schema.json index 86fc726a05168..b3e9bf74b6284 100644 --- a/airflow-core/src/airflow/provider_info.schema.json +++ b/airflow-core/src/airflow/provider_info.schema.json @@ -331,6 +331,25 @@ "type": "string" } }, + "remote-logging": { + "type": "array", + "description": "Remote logging IO handlers contributed by the provider. Each entry registers a RemoteLogIO implementation that ProvidersManager dispatches by URL scheme.", + "items": { + "type": "object", + "required": ["classpath", "scheme"], + "additionalProperties": false, + "properties": { + "classpath": { + "type": "string", + "description": "Fully-qualified class name of the RemoteLogIO implementation." + }, + "scheme": { + "type": "string", + "description": "URL scheme (without ://) of [logging] remote_base_log_folder that this handler claims." + } + } + } + }, "auth-backends": { "type": "array", "description": "API Auth Backend module names", diff --git a/airflow-core/src/airflow/providers_manager.py b/airflow-core/src/airflow/providers_manager.py index 6fefcbc39b06d..118ac9d9102d8 100644 --- a/airflow-core/src/airflow/providers_manager.py +++ b/airflow-core/src/airflow/providers_manager.py @@ -223,6 +223,14 @@ class NotificationInfo(NamedTuple): package_name: str +class RemoteLoggingInfo(NamedTuple): + """Remote logging IO handler registered by a provider.""" + + classpath: str + scheme: str + package_name: str + + class PluginInfo(NamedTuple): """Plugin class, name and provider it comes from.""" @@ -432,6 +440,8 @@ def __init__(self): self._cli_command_provider_name_set: set[str] = set() self._extra_link_class_name_set: set[str] = set() self._logging_class_name_set: set[str] = set() + self._remote_logging_info_list: list[RemoteLoggingInfo] = [] + self._remote_logging_by_scheme: dict[str, RemoteLoggingInfo] = {} self._auth_manager_class_name_set: set[str] = set() self._auth_manager_without_check_set: set[tuple[str, str]] = set() self._secrets_backend_class_name_set: set[str] = set() @@ -571,6 +581,12 @@ def initialize_providers_logging(self): self.initialize_providers_list() self._discover_logging() + @provider_info_cache("remote_logging") + def initialize_providers_remote_logging(self): + """Lazy initialization of providers remote logging IO handlers.""" + self.initialize_providers_list() + self._discover_remote_logging() + @provider_info_cache("secrets_backends") def initialize_providers_secrets_backends(self): """Lazy initialization of providers secrets_backends information.""" @@ -1240,6 +1256,32 @@ def _discover_logging(self) -> None: if _correctness_check(provider_package, logging_class_name, provider): self._logging_class_name_set.add(logging_class_name) + def _discover_remote_logging(self) -> None: + """Retrieve all remote logging IO handlers defined in the providers.""" + for provider_package, provider in self._provider_dict.items(): + entries = provider.data.get("remote-logging") or [] + for entry in entries: + classpath = entry["classpath"] + if not _correctness_check(provider_package, classpath, provider): + continue + info = RemoteLoggingInfo( + classpath=classpath, + scheme=entry["scheme"], + package_name=provider_package, + ) + self._remote_logging_info_list.append(info) + existing = self._remote_logging_by_scheme.get(info.scheme) + if existing is not None: + log.warning( + "Remote logging scheme '%s' is already registered by %s; ignoring " + "duplicate registration from %s.", + info.scheme, + existing.package_name, + info.package_name, + ) + continue + self._remote_logging_by_scheme[info.scheme] = info + def _discover_secrets_backends(self) -> None: """Retrieve all secrets backends defined in the providers.""" for provider_package, provider in self._provider_dict.items(): @@ -1450,6 +1492,17 @@ def logging_class_names(self) -> list[str]: self.initialize_providers_logging() return sorted(self._logging_class_name_set) + @property + def remote_logging_handlers(self) -> list[RemoteLoggingInfo]: + """Return all remote logging IO handlers contributed by providers.""" + self.initialize_providers_remote_logging() + return list(self._remote_logging_info_list) + + def remote_logging_handler_by_scheme(self, scheme: str) -> RemoteLoggingInfo | None: + """Return the remote logging IO handler registered for the given URL scheme, if any.""" + self.initialize_providers_remote_logging() + return self._remote_logging_by_scheme.get(scheme) + @property def secrets_backend_class_names(self) -> list[str]: """Returns set of secret backend class names.""" @@ -1532,6 +1585,8 @@ def _cleanup(self): self._field_behaviours.clear() self._extra_link_class_name_set.clear() self._logging_class_name_set.clear() + self._remote_logging_info_list.clear() + self._remote_logging_by_scheme.clear() self._auth_manager_class_name_set.clear() self._auth_manager_without_check_set.clear() self._secrets_backend_class_name_set.clear() diff --git a/airflow-core/tests/unit/always/test_providers_manager.py b/airflow-core/tests/unit/always/test_providers_manager.py index afa473e80a4f0..5b915162a09ce 100644 --- a/airflow-core/tests/unit/always/test_providers_manager.py +++ b/airflow-core/tests/unit/always/test_providers_manager.py @@ -37,6 +37,7 @@ PluginInfo, ProviderInfo, ProvidersManager, + RemoteLoggingInfo, ) from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker @@ -47,6 +48,16 @@ from airflow.cli.cli_config import CLICommand +class FakeRemoteLogIO: + """Importable stub used by remote-logging discovery tests.""" + + processors: tuple = () + + @classmethod + def from_config(cls): + return cls() + + def test_cleanup_providers_manager(cleanup_providers_manager): """Check the cleanup provider manager functionality.""" provider_manager = ProvidersManager() @@ -163,6 +174,76 @@ def test_providers_manager_register_dialects(self): ), ) + def test_providers_manager_register_remote_logging_by_scheme(self): + providers_manager = ProvidersManager() + providers_manager._provider_dict = LazyDictWithCache() + providers_manager._provider_dict["fake.remote.logging"] = ProviderInfo( + version="0.0.1", + data={ + "remote-logging": [ + { + "classpath": f"{__name__}.FakeRemoteLogIO", + "scheme": "fake", + } + ] + }, + ) + providers_manager._discover_remote_logging() + assert len(providers_manager._remote_logging_info_list) == 1 + assert providers_manager._remote_logging_by_scheme["fake"] == RemoteLoggingInfo( + classpath=f"{__name__}.FakeRemoteLogIO", + scheme="fake", + package_name="fake.remote.logging", + ) + assert "unknown" not in providers_manager._remote_logging_by_scheme + + def test_providers_manager_register_remote_logging_duplicate_scheme_first_wins(self): + providers_manager = ProvidersManager() + providers_manager._provider_dict = LazyDictWithCache() + providers_manager._provider_dict["fake.remote.logging.first"] = ProviderInfo( + version="0.0.1", + data={ + "remote-logging": [ + { + "classpath": f"{__name__}.FakeRemoteLogIO", + "scheme": "dup", + } + ] + }, + ) + providers_manager._provider_dict["fake.remote.logging.second"] = ProviderInfo( + version="0.0.1", + data={ + "remote-logging": [ + { + "classpath": f"{__name__}.FakeRemoteLogIO", + "scheme": "dup", + } + ] + }, + ) + providers_manager._discover_remote_logging() + winner = providers_manager._remote_logging_by_scheme["dup"] + assert winner.package_name == "fake.remote.logging.first" + + def test_providers_manager_remote_logging_bad_class_filtered(self): + providers_manager = ProvidersManager() + providers_manager._provider_dict = LazyDictWithCache() + providers_manager._provider_dict["fake.remote.logging"] = ProviderInfo( + version="0.0.1", + data={ + "remote-logging": [ + { + "classpath": "fake.module.does.not.exist.FakeRemoteLogIO", + "scheme": "bad", + } + ] + }, + ) + providers_manager._discover_remote_logging() + assert "bad" not in providers_manager._remote_logging_by_scheme + assert providers_manager._remote_logging_info_list == [] + def test_connection_form_widgets(self, yaml_ui_metadata_counts): yaml_widgets, _ = yaml_ui_metadata_counts provider_manager = ProvidersManager() diff --git a/shared/logging/src/airflow_shared/logging/factory.py b/shared/logging/src/airflow_shared/logging/factory.py new file mode 100644 index 0000000000000..babe286c0c825 --- /dev/null +++ b/shared/logging/src/airflow_shared/logging/factory.py @@ -0,0 +1,186 @@ +# 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. +""" +Build a remote logging IO handler from a provider-supplied registry. + +This module owns the runtime-independent logic for resolving and instantiating +``RemoteLogIO`` handlers. Both ``airflow.logging_config`` (core) and +``airflow.sdk.log._load_logging_config`` (Task SDK) call +:func:`resolve_remote_task_log` with their own ``ProvidersManager`` instance and +``import_string`` implementation; that function in turn delegates to +:func:`_build_remote_task_log_from_provider` for the provider-dispatch step and to +:func:`airflow_shared.logging.remote.discover_remote_log_handler` for the +user-defined ``logging_config_class`` and legacy fallbacks. +""" + +from __future__ import annotations + +import logging +import os +from collections.abc import Callable +from typing import TYPE_CHECKING, Any +from urllib.parse import urlsplit + +from airflow_shared.logging.remote import discover_remote_log_handler + +if TYPE_CHECKING: + from airflow.configuration import AirflowConfigParser + from airflow.logging.remote import RemoteLogIO + from airflow.providers_manager import ProvidersManager, RemoteLoggingInfo as _CoreRemoteLoggingInfo + from airflow.sdk.configuration import AirflowSDKConfigParser + from airflow.sdk.providers_manager_runtime import ( + ProvidersManagerTaskRuntime, + RemoteLoggingInfo as _SDKRemoteLoggingInfo, + ) + +log = logging.getLogger(__name__) + +DEFAULT_LOGGING_CONFIG_PATH = "airflow.config_templates.airflow_local_settings.DEFAULT_LOGGING_CONFIG" +# Default ``[logging] logging_config_class`` value when the user has not overridden it. + + +def _instantiate_handler( + info: _CoreRemoteLoggingInfo | _SDKRemoteLoggingInfo, + *, + import_string: Callable[[str], Any], +) -> Any | None: + try: + cls = import_string(info.classpath) + except Exception as err: + log.info( + "Remote task logs will not be available; failed to import %s: %s", + info.classpath, + err, + ) + if "PYTEST_CURRENT_TEST" in os.environ: + raise + return None + + factory = getattr(cls, "from_config", None) + if factory is None: + log.warning( + "Provider %s registered %s as a remote logging handler, but the class does not " + "implement classmethod 'from_config'. The handler will be skipped.", + info.package_name, + info.classpath, + ) + return None + + try: + return factory() + except Exception as err: + log.info( + "Remote task logs will not be available; %s.from_config raised: %s", + info.classpath, + err, + ) + if "PYTEST_CURRENT_TEST" in os.environ: + raise + return None + + +def _build_remote_task_log_from_provider( + *, + remote_base_log_folder: str | None, + providers_manager: ProvidersManager | ProvidersManagerTaskRuntime, + import_string: Callable[[str], Any], +) -> tuple[RemoteLogIO | None, str | None]: + """ + Resolve a remote logging IO handler from its URL scheme. + + The shared layer only routes to the right provider class; each provider's + ``from_config`` classmethod imports ``conf`` itself and reads its own + backend-specific keys. + + :param remote_base_log_folder: Value of ``[logging] remote_base_log_folder``. + Its URL scheme is the dispatch key. + :param providers_manager: The ``ProvidersManager`` (core) or + ``ProvidersManagerTaskRuntime`` (Task SDK) whose + ``remote_logging_handler_by_scheme`` method resolves the scheme. + :param import_string: ``import_string`` callable from the caller's runtime. + :returns: ``(handler, default_remote_conn_id)``. Both elements are ``None`` + when no provider claims the scheme. + """ + if not remote_base_log_folder: + return None, None + + scheme = urlsplit(remote_base_log_folder).scheme + if not scheme: + return None, None + + info = providers_manager.remote_logging_handler_by_scheme(scheme) + if info is None: + return None, None + + handler = _instantiate_handler(info, import_string=import_string) + if handler is None: + return None, None + + # Provider dispatch does not produce a default conn id. Users who need one + # set ``[logging] remote_log_conn_id`` explicitly; providers that want a + # backend-specific default can read from their own hook inside ``from_config``. + return handler, None + + +def resolve_remote_task_log( + *, + conf: AirflowConfigParser | AirflowSDKConfigParser, + providers_manager: ProvidersManager | ProvidersManagerTaskRuntime, + import_string: Callable[[str], Any], +) -> tuple[RemoteLogIO | None, str | None]: + """ + Resolve the active remote log handler via the three-tier precedence rule. + + Order: + + 1. A user-defined ``logging_config_class`` module that exports + ``REMOTE_TASK_LOG`` (or ``DEFAULT_REMOTE_CONN_ID``) wins — preserves + existing custom configs unchanged. + 2. ProvidersManager dispatch (provider yaml ``remote-logging:`` blocks). + 3. Legacy attr-path against the default logging module. Transitional while + ``airflow_local_settings.py`` still carries the per-scheme if/elif chain; + removed when every provider has migrated. + + :param conf: AirflowConfigParser-like object exposing ``get``. + :param providers_manager: Caller's ``ProvidersManager`` / + ``ProvidersManagerTaskRuntime``. + :param import_string: ``import_string`` callable from the caller's runtime. + :returns: ``(remote_task_log, default_remote_conn_id)``. Either element may + be ``None`` when no path produces a result. + """ + logging_class_path = ( + conf.get("logging", "logging_config_class", fallback=DEFAULT_LOGGING_CONFIG_PATH) + or DEFAULT_LOGGING_CONFIG_PATH + ) + user_defined = logging_class_path != DEFAULT_LOGGING_CONFIG_PATH + + if user_defined: + remote_task_log, default_remote_conn_id = discover_remote_log_handler( + logging_class_path, DEFAULT_LOGGING_CONFIG_PATH, import_string + ) + if remote_task_log is not None or default_remote_conn_id is not None: + return remote_task_log, default_remote_conn_id + + remote_task_log, default_remote_conn_id = _build_remote_task_log_from_provider( + remote_base_log_folder=conf.get("logging", "remote_base_log_folder", fallback=None), + providers_manager=providers_manager, + import_string=import_string, + ) + if remote_task_log is not None: + return remote_task_log, default_remote_conn_id + + return discover_remote_log_handler(logging_class_path, DEFAULT_LOGGING_CONFIG_PATH, import_string) diff --git a/shared/logging/tests/logging/test_factory.py b/shared/logging/tests/logging/test_factory.py new file mode 100644 index 0000000000000..711121334245a --- /dev/null +++ b/shared/logging/tests/logging/test_factory.py @@ -0,0 +1,242 @@ +# +# 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 types import SimpleNamespace +from typing import Any + +import pytest + +from airflow_shared.logging.factory import ( + DEFAULT_LOGGING_CONFIG_PATH, + _build_remote_task_log_from_provider, + resolve_remote_task_log, +) + + +class FakeProvidersManager: + """Stub manager satisfying the ProvidersManagerLike protocol.""" + + def __init__(self, by_scheme: dict[str, Any] | None = None) -> None: + self._by_scheme = by_scheme or {} + + def remote_logging_handler_by_scheme(self, scheme: str): + return self._by_scheme.get(scheme) + + +class ExplodingProvidersManager: + """Stub manager whose lookup must never be called by the test.""" + + def remote_logging_handler_by_scheme(self, scheme: str): # pragma: no cover - sentinel + pytest.fail("lookup must not run") + + +class FakeRemoteLogIO: + """Importable stub used as the registered handler class.""" + + last_called: bool = False + + def __init__(self): + pass + + @classmethod + def from_config(cls): + FakeRemoteLogIO.last_called = True + return cls() + + +class FakeRemoteLogIONoFactory: + """Stub missing the from_config classmethod.""" + + +_LOOKUP: dict[str, Any] = { + "fakeremotelogio": FakeRemoteLogIO, + "fakeremotelogionofactory": FakeRemoteLogIONoFactory, +} + + +def _fake_import_string(path: str) -> Any: + # Tests pass simple names like "fakeremotelogio"; route them through a dict + # to avoid needing real importable modules. + return _LOOKUP[path.lower()] + + +def _info( + *, + classpath: str = "fakeremotelogio", + scheme: str = "fake", + package_name: str = "fake-provider", +) -> SimpleNamespace: + return SimpleNamespace( + classpath=classpath, + scheme=scheme, + package_name=package_name, + ) + + +@pytest.fixture(autouse=True) +def _reset_fake(): + FakeRemoteLogIO.last_called = False + yield + FakeRemoteLogIO.last_called = False + + +def test_provider_dispatch_resolves_by_scheme(): + info = _info(scheme="fake") + handler, conn_id = _build_remote_task_log_from_provider( + remote_base_log_folder="fake://bucket/path", + providers_manager=FakeProvidersManager({"fake": info}), + import_string=_fake_import_string, + ) + assert isinstance(handler, FakeRemoteLogIO) + assert FakeRemoteLogIO.last_called is True + # Provider dispatch no longer derives a default conn id; users explicitly + # set [logging] remote_log_conn_id when they need one. + assert conn_id is None + + +def test_provider_dispatch_returns_none_when_scheme_unknown(): + handler, conn_id = _build_remote_task_log_from_provider( + remote_base_log_folder="unknown://x", + providers_manager=FakeProvidersManager(), + import_string=_fake_import_string, + ) + assert handler is None + assert conn_id is None + + +def test_provider_dispatch_returns_none_when_remote_base_unset(): + handler, conn_id = _build_remote_task_log_from_provider( + remote_base_log_folder=None, + providers_manager=ExplodingProvidersManager(), + import_string=_fake_import_string, + ) + assert handler is None + assert conn_id is None + + +def test_provider_dispatch_returns_none_when_no_scheme_in_url(): + handler, conn_id = _build_remote_task_log_from_provider( + remote_base_log_folder="/local/path/no/scheme", + providers_manager=ExplodingProvidersManager(), + import_string=_fake_import_string, + ) + assert handler is None + assert conn_id is None + + +def test_provider_dispatch_skips_handler_without_factory(): + info = _info(classpath="fakeremotelogionofactory") + handler, conn_id = _build_remote_task_log_from_provider( + remote_base_log_folder="fake://b", + providers_manager=FakeProvidersManager({"fake": info}), + import_string=_fake_import_string, + ) + assert handler is None + assert conn_id is None + + +# --------------------------------------------------------------------------- +# Tests for resolve_remote_task_log (three-tier precedence orchestrator) +# --------------------------------------------------------------------------- + +_FAKE_LOGGING_DICT: dict[str, Any] = {"version": 1} + + +class FakeConf: + """Minimal ConfLike stub keyed by ``(section, key)``.""" + + def __init__(self, values: dict[tuple[str, str], Any] | None = None) -> None: + self._values = values or {} + + def get(self, section, key, **kwargs): + return self._values.get((section, key), kwargs.get("fallback")) + + +def test_resolve_user_defined_module_wins(): + """If the user's logging_config_class module exposes REMOTE_TASK_LOG, it wins.""" + user_path = "fake.user.module.DEFAULT_LOGGING_CONFIG" + sentinel_remote = object() + + def fake_import_string(path: str): + if path == user_path: + return _FAKE_LOGGING_DICT + if path == "fake.user.module": + return SimpleNamespace(REMOTE_TASK_LOG=sentinel_remote, DEFAULT_REMOTE_CONN_ID="user_conn") + raise AssertionError(f"unexpected import: {path}") + + import sys + + sys.modules["fake.user.module"] = SimpleNamespace( + REMOTE_TASK_LOG=sentinel_remote, DEFAULT_REMOTE_CONN_ID="user_conn" + ) + try: + handler, conn_id = resolve_remote_task_log( + conf=FakeConf({("logging", "logging_config_class"): user_path}), + providers_manager=ExplodingProvidersManager(), # must not be consulted + import_string=fake_import_string, + ) + finally: + sys.modules.pop("fake.user.module", None) + + assert handler is sentinel_remote + assert conn_id == "user_conn" + + +def test_resolve_falls_through_to_provider_dispatch(): + """Default logging_class_path → user_defined is False → provider dispatch runs.""" + info = _info(scheme="fake") + handler, conn_id = resolve_remote_task_log( + conf=FakeConf({("logging", "remote_base_log_folder"): "fake://bucket"}), + providers_manager=FakeProvidersManager({"fake": info}), + import_string=_fake_import_string, + ) + assert isinstance(handler, FakeRemoteLogIO) + assert conn_id is None + + +def test_resolve_falls_back_to_legacy_attr_path_when_dispatch_returns_none(): + """When provider dispatch yields nothing, the legacy module-attr path is consulted.""" + sentinel_remote = object() + + def fake_import_string(path: str): + if path == DEFAULT_LOGGING_CONFIG_PATH: + return _FAKE_LOGGING_DICT + raise AssertionError(f"unexpected import: {path}") + + import sys + + parent_module = DEFAULT_LOGGING_CONFIG_PATH.rsplit(".", 1)[0] + saved = sys.modules.get(parent_module) + sys.modules[parent_module] = SimpleNamespace( + REMOTE_TASK_LOG=sentinel_remote, DEFAULT_REMOTE_CONN_ID="legacy_conn" + ) + try: + handler, conn_id = resolve_remote_task_log( + conf=FakeConf({("logging", "remote_base_log_folder"): "unknown://b"}), + providers_manager=FakeProvidersManager(), # empty registry + import_string=fake_import_string, + ) + finally: + if saved is not None: + sys.modules[parent_module] = saved + else: + sys.modules.pop(parent_module, None) + + assert handler is sentinel_remote + assert conn_id == "legacy_conn" diff --git a/task-sdk/src/airflow/sdk/log.py b/task-sdk/src/airflow/sdk/log.py index f93abe1a72240..f16d35fae5baa 100644 --- a/task-sdk/src/airflow/sdk/log.py +++ b/task-sdk/src/airflow/sdk/log.py @@ -173,16 +173,15 @@ def init_log_file(local_relative_path: str) -> Path: def _load_logging_config() -> None: """Load and cache the remote logging configuration from SDK config.""" - from airflow.sdk._shared.logging.remote import discover_remote_log_handler + from airflow.sdk._shared.logging.factory import resolve_remote_task_log from airflow.sdk._shared.module_loading import import_string from airflow.sdk.configuration import conf + from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime - fallback = "airflow.config_templates.airflow_local_settings.DEFAULT_LOGGING_CONFIG" - logging_class_path = conf.get("logging", "logging_config_class", fallback=fallback) - - # Load remote logging configuration using shared discovery logic - remote_task_log, default_remote_conn_id = discover_remote_log_handler( - logging_class_path, fallback, import_string + remote_task_log, default_remote_conn_id = resolve_remote_task_log( + conf=conf, + providers_manager=ProvidersManagerTaskRuntime(), + import_string=import_string, ) _ActiveLoggingConfig.set(remote_task_log, default_remote_conn_id) diff --git a/task-sdk/src/airflow/sdk/providers_manager_runtime.py b/task-sdk/src/airflow/sdk/providers_manager_runtime.py index e28ed3fe14a83..ba3382b221463 100644 --- a/task-sdk/src/airflow/sdk/providers_manager_runtime.py +++ b/task-sdk/src/airflow/sdk/providers_manager_runtime.py @@ -24,7 +24,7 @@ import traceback import warnings from collections.abc import Callable, MutableMapping -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, NamedTuple from urllib.parse import SplitResult import structlog @@ -55,6 +55,14 @@ log = structlog.getLogger(__name__) +class RemoteLoggingInfo(NamedTuple): + """Remote logging IO handler registered by a provider.""" + + classpath: str + scheme: str + package_name: str + + def _correctness_check(provider_package: str, class_name: str, provider_info: ProviderInfo) -> Any: """ Perform coherence check on provider classes. @@ -150,6 +158,8 @@ def __init__(self): # Keeps dict of hooks keyed by connection type. They are lazy evaluated at access time self._hooks_lazy_dict: LazyDictWithCache[str, HookInfo | Callable] = LazyDictWithCache() self._plugins_set: set[PluginInfo] = set() + self._remote_logging_info_list: list[RemoteLoggingInfo] = [] + self._remote_logging_by_scheme: dict[str, RemoteLoggingInfo] = {} self._provider_schema_validator = _create_provider_info_schema_validator() self._init_airflow_core_hooks() # Populated by initialize_provider_configs(); holds provider-contributed config sections. @@ -214,6 +224,12 @@ def initialize_providers_plugins(self): self.initialize_providers_list() self._discover_plugins() + @provider_info_cache("remote_logging") + def initialize_providers_remote_logging(self): + """Lazy initialization of providers remote logging IO handlers.""" + self.initialize_providers_list() + self._discover_remote_logging() + @provider_info_cache("taskflow_decorators") def initialize_providers_taskflow_decorator(self): """Lazy initialization of providers taskflow decorators.""" @@ -534,6 +550,32 @@ def _discover_plugins(self) -> None: ) ) + def _discover_remote_logging(self) -> None: + """Retrieve all remote logging IO handlers defined in the providers.""" + for provider_package, provider in self._provider_dict.items(): + entries = provider.data.get("remote-logging") or [] + for entry in entries: + classpath = entry["classpath"] + if not _correctness_check(provider_package, classpath, provider): + continue + info = RemoteLoggingInfo( + classpath=classpath, + scheme=entry["scheme"], + package_name=provider_package, + ) + self._remote_logging_info_list.append(info) + existing = self._remote_logging_by_scheme.get(info.scheme) + if existing is not None: + log.warning( + "Remote logging scheme '%s' is already registered by %s; ignoring " + "duplicate registration from %s.", + info.scheme, + existing.package_name, + info.package_name, + ) + continue + self._remote_logging_by_scheme[info.scheme] = info + def _discover_taskflow_decorators(self) -> None: for name, info in self._provider_dict.items(): for taskflow_decorator in info.data.get("task-decorators", []): @@ -611,6 +653,17 @@ def plugins(self) -> list[PluginInfo]: self.initialize_providers_plugins() return sorted(self._plugins_set, key=lambda x: x.plugin_class) + @property + def remote_logging_handlers(self) -> list[RemoteLoggingInfo]: + """Return all remote logging IO handlers contributed by providers.""" + self.initialize_providers_remote_logging() + return list(self._remote_logging_info_list) + + def remote_logging_handler_by_scheme(self, scheme: str) -> RemoteLoggingInfo | None: + """Return the remote logging IO handler registered for the given URL scheme, if any.""" + self.initialize_providers_remote_logging() + return self._remote_logging_by_scheme.get(scheme) + @property def provider_configs(self) -> list[tuple[str, dict[str, Any]]]: self.initialize_provider_configs() @@ -640,6 +693,8 @@ def _cleanup(self): self._hook_provider_dict.clear() self._hooks_lazy_dict.clear() self._plugins_set.clear() + self._remote_logging_info_list.clear() + self._remote_logging_by_scheme.clear() self._asset_uri_handlers.clear() self._asset_factories.clear() self._asset_to_openlineage_converters.clear() diff --git a/task-sdk/tests/task_sdk/test_providers_manager_runtime.py b/task-sdk/tests/task_sdk/test_providers_manager_runtime.py index 1cae21d53c764..dfa18c3d9d7e5 100644 --- a/task-sdk/tests/task_sdk/test_providers_manager_runtime.py +++ b/task-sdk/tests/task_sdk/test_providers_manager_runtime.py @@ -31,7 +31,7 @@ LazyDictWithCache, ProviderInfo, ) -from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime +from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime, RemoteLoggingInfo from tests_common.test_utils.markers import skip_if_force_lowest_dependencies_marker, skip_if_not_on_main from tests_common.test_utils.paths import AIRFLOW_ROOT_PATH @@ -39,6 +39,16 @@ PY313 = sys.version_info >= (3, 13) +class FakeRemoteLogIO: + """Importable stub used by remote-logging discovery tests.""" + + processors: tuple = () + + @classmethod + def from_config(cls): + return cls() + + def test_cleanup_providers_manager_runtime(cleanup_providers_manager): """Check the cleanup provider manager functionality.""" provider_manager = ProvidersManagerTaskRuntime() @@ -277,3 +287,59 @@ def initialize_provider_configs() -> None: assert conf.get("test_sdk_provider", "test_option") == "provider-default" finally: conf.invalidate_cache() + + def test_register_remote_logging_by_scheme(self): + providers_manager = ProvidersManagerTaskRuntime() + providers_manager._provider_dict = LazyDictWithCache() + providers_manager._provider_dict["fake.remote.logging"] = ProviderInfo( + version="0.0.1", + data={ + "remote-logging": [ + { + "classpath": f"{__name__}.FakeRemoteLogIO", + "scheme": "fake", + } + ] + }, + ) + providers_manager._discover_remote_logging() + assert len(providers_manager._remote_logging_info_list) == 1 + assert providers_manager._remote_logging_by_scheme["fake"] == RemoteLoggingInfo( + classpath=f"{__name__}.FakeRemoteLogIO", + scheme="fake", + package_name="fake.remote.logging", + ) + + def test_register_remote_logging_duplicate_scheme_first_wins(self): + providers_manager = ProvidersManagerTaskRuntime() + providers_manager._provider_dict = LazyDictWithCache() + providers_manager._provider_dict["fake.remote.logging.first"] = ProviderInfo( + version="0.0.1", + data={"remote-logging": [{"classpath": f"{__name__}.FakeRemoteLogIO", "scheme": "dup"}]}, + ) + providers_manager._provider_dict["fake.remote.logging.second"] = ProviderInfo( + version="0.0.1", + data={"remote-logging": [{"classpath": f"{__name__}.FakeRemoteLogIO", "scheme": "dup"}]}, + ) + providers_manager._discover_remote_logging() + assert providers_manager._remote_logging_by_scheme["dup"].package_name == ( + "fake.remote.logging.first" + ) + + def test_register_remote_logging_bad_class_filtered(self): + providers_manager = ProvidersManagerTaskRuntime() + providers_manager._provider_dict = LazyDictWithCache() + providers_manager._provider_dict["fake.remote.logging"] = ProviderInfo( + version="0.0.1", + data={ + "remote-logging": [ + { + "classpath": "fake.module.does.not.exist.FakeRemoteLogIO", + "scheme": "bad", + } + ] + }, + ) + providers_manager._discover_remote_logging() + assert "bad" not in providers_manager._remote_logging_by_scheme + assert providers_manager._remote_logging_info_list == [] From 7b8f8354b256bb9a274f785409d1604ef1757b46 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sat, 16 May 2026 17:34:50 +0800 Subject: [PATCH 2/7] Align get_default_remote_conn_id with load_remote_conn_id --- airflow-core/src/airflow/logging_config.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/airflow-core/src/airflow/logging_config.py b/airflow-core/src/airflow/logging_config.py index 0da017d9b0826..261249f4f48bb 100644 --- a/airflow-core/src/airflow/logging_config.py +++ b/airflow-core/src/airflow/logging_config.py @@ -54,6 +54,11 @@ def get_remote_task_log() -> RemoteLogIO | None: def get_default_remote_conn_id() -> str | None: + from airflow.configuration import conf + + if conn_id := conf.get("logging", "remote_log_conn_id", fallback=None): + return conn_id + if not _ActiveLoggingConfig.logging_config_loaded: load_logging_config() return _ActiveLoggingConfig.default_remote_conn_id From e8bb6b7a08eefe0b28b96731fe959736115ccfee Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sat, 16 May 2026 17:49:29 +0800 Subject: [PATCH 3/7] Make _build_remote_task_log_from_provider return RemoteLogIO without conn id --- .../src/airflow_shared/logging/factory.py | 35 ++++++++----------- shared/logging/tests/logging/test_factory.py | 17 +++------ 2 files changed, 20 insertions(+), 32 deletions(-) diff --git a/shared/logging/src/airflow_shared/logging/factory.py b/shared/logging/src/airflow_shared/logging/factory.py index babe286c0c825..9d4c00654d7e6 100644 --- a/shared/logging/src/airflow_shared/logging/factory.py +++ b/shared/logging/src/airflow_shared/logging/factory.py @@ -98,7 +98,7 @@ def _build_remote_task_log_from_provider( remote_base_log_folder: str | None, providers_manager: ProvidersManager | ProvidersManagerTaskRuntime, import_string: Callable[[str], Any], -) -> tuple[RemoteLogIO | None, str | None]: +) -> RemoteLogIO | None: """ Resolve a remote logging IO handler from its URL scheme. @@ -106,34 +106,29 @@ def _build_remote_task_log_from_provider( ``from_config`` classmethod imports ``conf`` itself and reads its own backend-specific keys. + Provider dispatch does not produce a default conn id. Users who need one + set ``[logging] remote_log_conn_id`` explicitly; providers that want a + backend-specific default can read from their own hook inside ``from_config``. + :param remote_base_log_folder: Value of ``[logging] remote_base_log_folder``. Its URL scheme is the dispatch key. :param providers_manager: The ``ProvidersManager`` (core) or ``ProvidersManagerTaskRuntime`` (Task SDK) whose ``remote_logging_handler_by_scheme`` method resolves the scheme. :param import_string: ``import_string`` callable from the caller's runtime. - :returns: ``(handler, default_remote_conn_id)``. Both elements are ``None`` - when no provider claims the scheme. + :returns: The instantiated handler, or ``None`` when no provider claims the + scheme. """ if not remote_base_log_folder: - return None, None - - scheme = urlsplit(remote_base_log_folder).scheme - if not scheme: - return None, None + return None - info = providers_manager.remote_logging_handler_by_scheme(scheme) - if info is None: - return None, None + if not (scheme := urlsplit(remote_base_log_folder).scheme): + return None - handler = _instantiate_handler(info, import_string=import_string) - if handler is None: - return None, None + if (info := providers_manager.remote_logging_handler_by_scheme(scheme)) is None: + return None - # Provider dispatch does not produce a default conn id. Users who need one - # set ``[logging] remote_log_conn_id`` explicitly; providers that want a - # backend-specific default can read from their own hook inside ``from_config``. - return handler, None + return _instantiate_handler(info, import_string=import_string) def resolve_remote_task_log( @@ -175,12 +170,12 @@ def resolve_remote_task_log( if remote_task_log is not None or default_remote_conn_id is not None: return remote_task_log, default_remote_conn_id - remote_task_log, default_remote_conn_id = _build_remote_task_log_from_provider( + remote_task_log = _build_remote_task_log_from_provider( remote_base_log_folder=conf.get("logging", "remote_base_log_folder", fallback=None), providers_manager=providers_manager, import_string=import_string, ) if remote_task_log is not None: - return remote_task_log, default_remote_conn_id + return remote_task_log, conf.get("logging", "remote_log_conn_id", fallback=None) return discover_remote_log_handler(logging_class_path, DEFAULT_LOGGING_CONFIG_PATH, import_string) diff --git a/shared/logging/tests/logging/test_factory.py b/shared/logging/tests/logging/test_factory.py index 711121334245a..03dae31ea5e08 100644 --- a/shared/logging/tests/logging/test_factory.py +++ b/shared/logging/tests/logging/test_factory.py @@ -98,57 +98,50 @@ def _reset_fake(): def test_provider_dispatch_resolves_by_scheme(): info = _info(scheme="fake") - handler, conn_id = _build_remote_task_log_from_provider( + handler = _build_remote_task_log_from_provider( remote_base_log_folder="fake://bucket/path", providers_manager=FakeProvidersManager({"fake": info}), import_string=_fake_import_string, ) assert isinstance(handler, FakeRemoteLogIO) assert FakeRemoteLogIO.last_called is True - # Provider dispatch no longer derives a default conn id; users explicitly - # set [logging] remote_log_conn_id when they need one. - assert conn_id is None def test_provider_dispatch_returns_none_when_scheme_unknown(): - handler, conn_id = _build_remote_task_log_from_provider( + handler = _build_remote_task_log_from_provider( remote_base_log_folder="unknown://x", providers_manager=FakeProvidersManager(), import_string=_fake_import_string, ) assert handler is None - assert conn_id is None def test_provider_dispatch_returns_none_when_remote_base_unset(): - handler, conn_id = _build_remote_task_log_from_provider( + handler = _build_remote_task_log_from_provider( remote_base_log_folder=None, providers_manager=ExplodingProvidersManager(), import_string=_fake_import_string, ) assert handler is None - assert conn_id is None def test_provider_dispatch_returns_none_when_no_scheme_in_url(): - handler, conn_id = _build_remote_task_log_from_provider( + handler = _build_remote_task_log_from_provider( remote_base_log_folder="/local/path/no/scheme", providers_manager=ExplodingProvidersManager(), import_string=_fake_import_string, ) assert handler is None - assert conn_id is None def test_provider_dispatch_skips_handler_without_factory(): info = _info(classpath="fakeremotelogionofactory") - handler, conn_id = _build_remote_task_log_from_provider( + handler = _build_remote_task_log_from_provider( remote_base_log_folder="fake://b", providers_manager=FakeProvidersManager({"fake": info}), import_string=_fake_import_string, ) assert handler is None - assert conn_id is None # --------------------------------------------------------------------------- From 5cb1e0ad20d0e8d8db976b02a11b6db353948e6d Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sun, 17 May 2026 00:15:13 +0800 Subject: [PATCH 4/7] Split load_logging_config into _get_logging_config and _load_logging_config Decouple importing the [logging] logging_config_class dict from remote-handler resolution so each can run independently. The deprecated load_logging_config wrapper delegates to both and keeps the old return shape. Wire the lazy remote-handler load through the shared resolve_remote_task_log factory. --- airflow-core/src/airflow/logging_config.py | 68 ++++-- .../tests/unit/logging/test_logging_config.py | 210 ++++++++++++++++++ 2 files changed, 254 insertions(+), 24 deletions(-) create mode 100644 airflow-core/tests/unit/logging/test_logging_config.py diff --git a/airflow-core/src/airflow/logging_config.py b/airflow-core/src/airflow/logging_config.py index 261249f4f48bb..7f4ae605cb21a 100644 --- a/airflow-core/src/airflow/logging_config.py +++ b/airflow-core/src/airflow/logging_config.py @@ -21,7 +21,7 @@ import warnings from typing import TYPE_CHECKING, Any -from airflow._shared.logging.remote import discover_remote_log_handler +from airflow._shared.logging.factory import DEFAULT_LOGGING_CONFIG_PATH, resolve_remote_task_log from airflow._shared.module_loading import import_string from airflow.configuration import conf from airflow.exceptions import AirflowConfigException @@ -49,30 +49,26 @@ def set(cls, remote_task_log: RemoteLogIO | None, default_remote_conn_id: str | def get_remote_task_log() -> RemoteLogIO | None: if not _ActiveLoggingConfig.logging_config_loaded: - load_logging_config() + _load_logging_config() return _ActiveLoggingConfig.remote_task_log def get_default_remote_conn_id() -> str | None: - from airflow.configuration import conf - if conn_id := conf.get("logging", "remote_log_conn_id", fallback=None): return conn_id if not _ActiveLoggingConfig.logging_config_loaded: - load_logging_config() + _load_logging_config() return _ActiveLoggingConfig.default_remote_conn_id -def load_logging_config() -> tuple[dict[str, Any], str]: - """Configure & Validate Airflow Logging.""" - fallback = "airflow.config_templates.airflow_local_settings.DEFAULT_LOGGING_CONFIG" - logging_class_path = conf.get("logging", "logging_config_class", fallback=fallback) - - # Sometimes we end up with `""` as the value! - logging_class_path = logging_class_path or fallback - - user_defined = logging_class_path != fallback +def _get_logging_config() -> dict[str, Any]: + """Import and validate the ``[logging] logging_config_class`` dict.""" + logging_class_path = ( + conf.get("logging", "logging_config_class", fallback=DEFAULT_LOGGING_CONFIG_PATH) + or DEFAULT_LOGGING_CONFIG_PATH + ) + user_defined = logging_class_path != DEFAULT_LOGGING_CONFIG_PATH try: logging_config = import_string(logging_class_path) @@ -83,27 +79,51 @@ def load_logging_config() -> tuple[dict[str, Any], str]: if user_defined: log.info("Successfully imported user-defined logging config from %s", logging_class_path) - except Exception as err: - # Import default logging configurations. raise ImportError( f"Unable to load {'custom ' if user_defined else ''}logging config from {logging_class_path} due " f"to: {type(err).__name__}:{err}" ) - else: - # Load remote logging configuration using shared discovery logic - remote_task_log, default_remote_conn_id = discover_remote_log_handler( - logging_class_path, fallback, import_string - ) - _ActiveLoggingConfig.set(remote_task_log, default_remote_conn_id) - return logging_config, logging_class_path + return logging_config + + +def _load_logging_config() -> None: + """Load and cache the remote logging configuration from core config.""" + from airflow.providers_manager import ProvidersManager + + remote_task_log, default_remote_conn_id = resolve_remote_task_log( + conf=conf, + providers_manager=ProvidersManager(), + import_string=import_string, + ) + _ActiveLoggingConfig.set(remote_task_log, default_remote_conn_id) + + +def load_logging_config() -> tuple[dict[str, Any], str]: + """ + Import the logging config dict and load the remote logging handler. + + .. deprecated:: + Use :func:`_get_logging_config` for the logging dict and + :func:`_load_logging_config` for remote handler setup. + """ + warnings.warn( + "load_logging_config is deprecated; use _get_logging_config() for the logging dict " + "and _load_logging_config() for remote handler setup.", + DeprecationWarning, + stacklevel=2, + ) + _load_logging_config() + return _get_logging_config(), conf.get( + "logging", "logging_config_class", fallback=DEFAULT_LOGGING_CONFIG_PATH + ) or DEFAULT_LOGGING_CONFIG_PATH def configure_logging(): from airflow._shared.logging import configure_logging, init_log_folder, translate_config_values - logging_config, logging_class_path = load_logging_config() + logging_config = _get_logging_config() try: level: str = getattr( logging_config, "LOG_LEVEL", conf.get("logging", "logging_level", fallback="INFO") diff --git a/airflow-core/tests/unit/logging/test_logging_config.py b/airflow-core/tests/unit/logging/test_logging_config.py new file mode 100644 index 0000000000000..04449a6741978 --- /dev/null +++ b/airflow-core/tests/unit/logging/test_logging_config.py @@ -0,0 +1,210 @@ +# +# 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 + +import sys +from types import SimpleNamespace +from unittest import mock + +import pytest + +from airflow._shared.logging.factory import DEFAULT_LOGGING_CONFIG_PATH +from airflow.logging_config import ( + _ActiveLoggingConfig, + _get_logging_config, + _load_logging_config, + get_default_remote_conn_id, + get_remote_task_log, + load_logging_config, +) + + +@pytest.fixture(autouse=True) +def _reset_active_logging_config(monkeypatch): + monkeypatch.setattr(_ActiveLoggingConfig, "logging_config_loaded", False, raising=False) + monkeypatch.setattr(_ActiveLoggingConfig, "remote_task_log", None, raising=False) + monkeypatch.setattr(_ActiveLoggingConfig, "default_remote_conn_id", None, raising=False) + + +class TestGetLoggingConfig: + def test_returns_default_logging_dict(self): + from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG + + config = _get_logging_config() + assert config == DEFAULT_LOGGING_CONFIG + + def test_user_defined_dict_is_imported(self, monkeypatch): + fake_module = "fake_user_logging_module_for_test" + custom = {"version": 1, "marker": "user-defined"} + monkeypatch.setitem(sys.modules, fake_module, SimpleNamespace(LOGGING_CONFIG=custom)) + with mock.patch("airflow.logging_config.conf") as mocked_conf: + mocked_conf.get.return_value = f"{fake_module}.LOGGING_CONFIG" + config = _get_logging_config() + assert config is custom + + def test_empty_string_falls_back_to_default(self): + from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG + + with mock.patch("airflow.logging_config.conf") as mocked_conf: + mocked_conf.get.return_value = "" + config = _get_logging_config() + assert config == DEFAULT_LOGGING_CONFIG + + @pytest.mark.parametrize( + "logging_class_path", + [pytest.param("", id="empty-string"), pytest.param(None, id="none")], + ) + def test_falsy_logging_class_path_falls_back_to_default(self, logging_class_path): + from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG + + with mock.patch("airflow.logging_config.conf") as mocked_conf: + mocked_conf.get.return_value = logging_class_path + config = _get_logging_config() + assert config == DEFAULT_LOGGING_CONFIG + + def test_raises_import_error_when_path_invalid(self): + with mock.patch("airflow.logging_config.conf") as mocked_conf: + mocked_conf.get.return_value = "nonexistent.module.LOGGING" + with pytest.raises(ImportError, match="Unable to load custom logging config"): + _get_logging_config() + + def test_raises_import_error_when_value_not_dict(self, monkeypatch): + fake_module = "fake_user_logging_module_not_dict" + monkeypatch.setitem(sys.modules, fake_module, SimpleNamespace(LOGGING_CONFIG="not-a-dict")) + with mock.patch("airflow.logging_config.conf") as mocked_conf: + mocked_conf.get.return_value = f"{fake_module}.LOGGING_CONFIG" + with pytest.raises(ImportError, match="Logging Config should be of dict type"): + _get_logging_config() + + +class TestLoadLoggingConfigPrivate: + def test_caches_resolved_remote_handler_and_conn_id(self): + sentinel_handler = object() + with ( + mock.patch("airflow.logging_config.resolve_remote_task_log") as mock_resolve, + mock.patch("airflow.providers_manager.ProvidersManager") as mock_pm, + ): + mock_resolve.return_value = (sentinel_handler, "my_conn") + _load_logging_config() + + assert _ActiveLoggingConfig.logging_config_loaded is True + assert _ActiveLoggingConfig.remote_task_log is sentinel_handler + assert _ActiveLoggingConfig.default_remote_conn_id == "my_conn" + # resolve_remote_task_log must be called with the core providers manager. + mock_resolve.assert_called_once() + assert mock_resolve.call_args.kwargs["providers_manager"] is mock_pm.return_value + + def test_caches_none_when_resolver_returns_nothing(self): + with ( + mock.patch("airflow.logging_config.resolve_remote_task_log") as mock_resolve, + mock.patch("airflow.providers_manager.ProvidersManager"), + ): + mock_resolve.return_value = (None, None) + _load_logging_config() + + assert _ActiveLoggingConfig.logging_config_loaded is True + assert _ActiveLoggingConfig.remote_task_log is None + assert _ActiveLoggingConfig.default_remote_conn_id is None + + +class TestLoadLoggingConfigDeprecated: + def test_emits_deprecation_warning_and_returns_tuple(self): + sentinel_handler = object() + with ( + mock.patch("airflow.logging_config.resolve_remote_task_log") as mock_resolve, + mock.patch("airflow.providers_manager.ProvidersManager"), + ): + mock_resolve.return_value = (sentinel_handler, "my_conn") + with pytest.warns(DeprecationWarning, match="load_logging_config is deprecated"): + logging_config, logging_class_path = load_logging_config() + + assert isinstance(logging_config, dict) + assert logging_class_path == DEFAULT_LOGGING_CONFIG_PATH + # The deprecated wrapper still primes the remote handler cache. + assert _ActiveLoggingConfig.remote_task_log is sentinel_handler + + def test_returns_user_logging_class_path(self, monkeypatch): + fake_module = "fake_user_logging_module_deprecated" + custom = {"version": 1} + custom_path = f"{fake_module}.LOGGING_CONFIG" + monkeypatch.setitem(sys.modules, fake_module, SimpleNamespace(LOGGING_CONFIG=custom)) + + with ( + mock.patch("airflow.logging_config.conf") as mocked_conf, + mock.patch("airflow.logging_config.resolve_remote_task_log") as mock_resolve, + mock.patch("airflow.providers_manager.ProvidersManager"), + ): + mocked_conf.get.return_value = custom_path + mock_resolve.return_value = (None, None) + with pytest.warns(DeprecationWarning, match="load_logging_config is deprecated"): + _, logging_class_path = load_logging_config() + + assert logging_class_path == custom_path + + +class TestGetRemoteTaskLog: + def test_returns_cached_value_without_reload(self): + sentinel = object() + _ActiveLoggingConfig.set(sentinel, None) + with mock.patch("airflow.logging_config._load_logging_config") as mock_load: + result = get_remote_task_log() + assert result is sentinel + mock_load.assert_not_called() + + def test_triggers_load_when_not_loaded(self): + sentinel = object() + + def _fake_load(): + _ActiveLoggingConfig.set(sentinel, None) + + with mock.patch("airflow.logging_config._load_logging_config", side_effect=_fake_load) as mock_load: + result = get_remote_task_log() + mock_load.assert_called_once() + assert result is sentinel + + +class TestGetDefaultRemoteConnId: + def test_prefers_explicit_conf_value(self): + with ( + mock.patch("airflow.logging_config.conf") as mocked_conf, + mock.patch("airflow.logging_config._load_logging_config") as mock_load, + ): + mocked_conf.get.return_value = "explicit_conn" + assert get_default_remote_conn_id() == "explicit_conn" + mock_load.assert_not_called() + + def test_falls_back_to_cached_default(self): + def _fake_load(): + _ActiveLoggingConfig.set(None, "cached_conn") + + with ( + mock.patch("airflow.logging_config.conf") as mocked_conf, + mock.patch("airflow.logging_config._load_logging_config", side_effect=_fake_load), + ): + mocked_conf.get.return_value = None + assert get_default_remote_conn_id() == "cached_conn" + + def test_skips_reload_when_already_loaded(self): + _ActiveLoggingConfig.set(None, "cached_conn") + with ( + mock.patch("airflow.logging_config.conf") as mocked_conf, + mock.patch("airflow.logging_config._load_logging_config") as mock_load, + ): + mocked_conf.get.return_value = None + assert get_default_remote_conn_id() == "cached_conn" + mock_load.assert_not_called() From 61e1d900184e00d846522a19c1a9348f3f817783 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sun, 17 May 2026 14:42:16 +0800 Subject: [PATCH 5/7] Add significant md --- .../newsfragments/67056.significant.rst | 101 ++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 airflow-core/newsfragments/67056.significant.rst diff --git a/airflow-core/newsfragments/67056.significant.rst b/airflow-core/newsfragments/67056.significant.rst new file mode 100644 index 0000000000000..5143aa0641635 --- /dev/null +++ b/airflow-core/newsfragments/67056.significant.rst @@ -0,0 +1,101 @@ +Decouple remote logging resolution from ``airflow.logging_config`` + +Remote task log handler resolution is now owned by the shared +``airflow_shared.logging.factory`` module and applies a single, well-defined +precedence rule. ``airflow.logging_config.load_logging_config`` is deprecated: +loading the ``[logging] logging_config_class`` dict and resolving the remote +handler are now two independent steps. + +**Resolution order (``resolve_remote_task_log``):** + +1. **User-defined ``[logging] logging_config_class``** — if the user has set + ``logging_config_class`` to a custom module path, and that module exports a + ``REMOTE_TASK_LOG`` (or ``DEFAULT_REMOTE_CONN_ID``) attribute, those values + win. Existing custom logging configs keep working unchanged. +2. **ProvidersManager scheme dispatch** — the scheme of ``[logging] + remote_base_log_folder`` (e.g. ``s3``, ``gs``, ``wasb``) is looked up in the + provider yaml ``remote-logging:`` registry. The matching ``RemoteLogIO`` + class is imported and instantiated via its ``from_config()`` classmethod. + The connection id comes from ``[logging] remote_log_conn_id`` (set + explicitly by the user; providers needing a backend default can read it + inside ``from_config``). +3. **Legacy attr-path fallback** — if neither of the above produced a handler, + the resolver imports the default logging module + (``airflow.config_templates.airflow_local_settings``) and reads its + ``REMOTE_TASK_LOG`` / ``DEFAULT_REMOTE_CONN_ID`` attributes. This is the + per-scheme ``if/elif`` chain in ``airflow_local_settings.py`` and is + transitional — it will be removed once every in-tree provider exposes + ``from_config``. + +**``RemoteLogIO.from_config`` contract:** + +Provider remote-log handler classes opting into provider dispatch must expose +a ``from_config`` classmethod. The shape is:: + + class MyRemoteLogIO(LoggingMixin): + @classmethod + def from_config(cls) -> "MyRemoteLogIO": + from airflow.providers.common.compat.sdk import conf + + return cls( + base_log_folder=conf.get("logging", "base_log_folder"), + remote_base=conf.get("logging", "remote_base_log_folder"), + delete_local_copy=conf.getboolean("logging", "delete_local_logs"), + # backend-specific keys live in the provider's own config section + ..., + ) + +Key properties: + +- Takes no arguments — the shared factory calls ``cls.from_config()`` with no + inputs. Providers read ``airflow.providers.common.compat.sdk.conf`` themselves and pick + the keys they care about. +- Returns a fully instantiated ``RemoteLogIO`` (or ``RemoteLogStreamIO``). +- Failures inside ``from_config`` are logged and treated as "no remote + handler" (the factory returns ``None`` and the legacy fallback runs); under + ``PYTEST_CURRENT_TEST`` the exception is re-raised so tests fail loudly. +- Providers that don't yet implement ``from_config`` continue to work via the + legacy ``airflow_local_settings.py`` chain (step 3). + +**``airflow.logging_config`` API changes:** + +- ``_get_logging_config()`` — new private helper that imports and validates + the ``[logging] logging_config_class`` dict only. Does not touch remote + logging state. +- ``_load_logging_config()`` — new private helper that calls + ``resolve_remote_task_log`` and caches the result on + ``_ActiveLoggingConfig``. Used lazily by ``get_remote_task_log`` and + ``get_default_remote_conn_id``. +- ``load_logging_config()`` — deprecated. Emits ``DeprecationWarning`` and + delegates to both helpers; still returns ``(logging_config_dict, + logging_class_path)`` so existing callers keep working. + +**Behaviour changes:** + +- ``configure_logging`` no longer eagerly resolves the remote handler. + Resolution is now lazy and happens on the first call to + ``get_remote_task_log()`` / ``get_default_remote_conn_id()``. +- Providers that registered a ``remote-logging:`` block but did not implement + ``from_config`` will be skipped with a warning; the legacy fallback path + takes over. + +* Types of change + + * [ ] Dag changes + * [ ] Config changes + * [ ] API changes + * [ ] CLI changes + * [x] Behaviour changes + * [ ] Plugin changes + * [ ] Dependency changes + * [x] Code interface changes + +* Migration rules needed + + * Replace direct calls to ``airflow.logging_config.load_logging_config()`` + with ``_get_logging_config()`` (for the logging dict) and/or + ``_load_logging_config()`` (to prime the remote-handler cache). + * Provider remote-log handler classes should implement a no-argument + ``from_config`` classmethod that reads ``airflow.providers.common.compat.sdk.conf`` + and returns a configured instance. Until they do, resolution falls + through to the legacy ``airflow_local_settings.py`` chain. From f2634064a8b5ba2254e5aa9aacf0d0bc4f544286 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Sun, 17 May 2026 16:07:53 +0800 Subject: [PATCH 6/7] Fix: Respct ("logging", "remote_logging") in shared logging factory --- .../src/airflow_shared/logging/factory.py | 19 +++--- shared/logging/tests/logging/test_factory.py | 58 ++++++++++++++++++- 2 files changed, 66 insertions(+), 11 deletions(-) diff --git a/shared/logging/src/airflow_shared/logging/factory.py b/shared/logging/src/airflow_shared/logging/factory.py index 9d4c00654d7e6..acd016d6273ef 100644 --- a/shared/logging/src/airflow_shared/logging/factory.py +++ b/shared/logging/src/airflow_shared/logging/factory.py @@ -23,7 +23,7 @@ :func:`resolve_remote_task_log` with their own ``ProvidersManager`` instance and ``import_string`` implementation; that function in turn delegates to :func:`_build_remote_task_log_from_provider` for the provider-dispatch step and to -:func:`airflow_shared.logging.remote.discover_remote_log_handler` for the +:func:`airflow._shared.logging.remote.discover_remote_log_handler` for the user-defined ``logging_config_class`` and legacy fallbacks. """ @@ -35,7 +35,7 @@ from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit -from airflow_shared.logging.remote import discover_remote_log_handler +from .remote import discover_remote_log_handler if TYPE_CHECKING: from airflow.configuration import AirflowConfigParser @@ -170,12 +170,13 @@ def resolve_remote_task_log( if remote_task_log is not None or default_remote_conn_id is not None: return remote_task_log, default_remote_conn_id - remote_task_log = _build_remote_task_log_from_provider( - remote_base_log_folder=conf.get("logging", "remote_base_log_folder", fallback=None), - providers_manager=providers_manager, - import_string=import_string, - ) - if remote_task_log is not None: - return remote_task_log, conf.get("logging", "remote_log_conn_id", fallback=None) + if conf.getboolean("logging", "remote_logging", fallback=False): + remote_task_log = _build_remote_task_log_from_provider( + remote_base_log_folder=conf.get("logging", "remote_base_log_folder", fallback=None), + providers_manager=providers_manager, + import_string=import_string, + ) + if remote_task_log is not None: + return remote_task_log, conf.get("logging", "remote_log_conn_id", fallback=None) return discover_remote_log_handler(logging_class_path, DEFAULT_LOGGING_CONFIG_PATH, import_string) diff --git a/shared/logging/tests/logging/test_factory.py b/shared/logging/tests/logging/test_factory.py index 03dae31ea5e08..fa81d7fa1c638 100644 --- a/shared/logging/tests/logging/test_factory.py +++ b/shared/logging/tests/logging/test_factory.py @@ -160,6 +160,14 @@ def __init__(self, values: dict[tuple[str, str], Any] | None = None) -> None: def get(self, section, key, **kwargs): return self._values.get((section, key), kwargs.get("fallback")) + def getboolean(self, section, key, **kwargs): + value = self._values.get((section, key), kwargs.get("fallback")) + if isinstance(value, bool): + return value + if isinstance(value, str): + return value.lower() in ("1", "true", "yes", "on") + return bool(value) + def test_resolve_user_defined_module_wins(): """If the user's logging_config_class module exposes REMOTE_TASK_LOG, it wins.""" @@ -195,7 +203,12 @@ def test_resolve_falls_through_to_provider_dispatch(): """Default logging_class_path → user_defined is False → provider dispatch runs.""" info = _info(scheme="fake") handler, conn_id = resolve_remote_task_log( - conf=FakeConf({("logging", "remote_base_log_folder"): "fake://bucket"}), + conf=FakeConf( + { + ("logging", "remote_base_log_folder"): "fake://bucket", + ("logging", "remote_logging"): True, + } + ), providers_manager=FakeProvidersManager({"fake": info}), import_string=_fake_import_string, ) @@ -203,6 +216,42 @@ def test_resolve_falls_through_to_provider_dispatch(): assert conn_id is None +def test_resolve_skips_provider_dispatch_when_remote_logging_disabled(): + """``[logging] remote_logging = False`` must short-circuit provider dispatch.""" + sentinel_remote = object() + + def fake_import_string(path: str): + if path == DEFAULT_LOGGING_CONFIG_PATH: + return _FAKE_LOGGING_DICT + raise AssertionError(f"unexpected import: {path}") + + import sys + + parent_module = DEFAULT_LOGGING_CONFIG_PATH.rsplit(".", 1)[0] + saved = sys.modules.get(parent_module) + sys.modules[parent_module] = SimpleNamespace(REMOTE_TASK_LOG=sentinel_remote, DEFAULT_REMOTE_CONN_ID=None) + try: + # ExplodingProvidersManager fails the test if provider dispatch runs. + handler, conn_id = resolve_remote_task_log( + conf=FakeConf( + { + ("logging", "remote_base_log_folder"): "fake://bucket", + ("logging", "remote_logging"): False, + } + ), + providers_manager=ExplodingProvidersManager(), + import_string=fake_import_string, + ) + finally: + if saved is not None: + sys.modules[parent_module] = saved + else: + sys.modules.pop(parent_module, None) + + assert handler is sentinel_remote + assert conn_id is None + + def test_resolve_falls_back_to_legacy_attr_path_when_dispatch_returns_none(): """When provider dispatch yields nothing, the legacy module-attr path is consulted.""" sentinel_remote = object() @@ -221,7 +270,12 @@ def fake_import_string(path: str): ) try: handler, conn_id = resolve_remote_task_log( - conf=FakeConf({("logging", "remote_base_log_folder"): "unknown://b"}), + conf=FakeConf( + { + ("logging", "remote_base_log_folder"): "unknown://b", + ("logging", "remote_logging"): True, + } + ), providers_manager=FakeProvidersManager(), # empty registry import_string=fake_import_string, ) From 37939dca6bf004b45a8b78d82da0550e48b20b03 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Tue, 19 May 2026 12:32:11 +0800 Subject: [PATCH 7/7] Addressed Jens comments --- airflow-core/newsfragments/67056.significant.rst | 2 +- airflow-core/src/airflow/providers_manager.py | 5 ++--- airflow-core/tests/unit/always/test_providers_manager.py | 2 ++ task-sdk/src/airflow/sdk/providers_manager_runtime.py | 5 ++--- task-sdk/tests/task_sdk/test_providers_manager_runtime.py | 2 ++ 5 files changed, 9 insertions(+), 7 deletions(-) diff --git a/airflow-core/newsfragments/67056.significant.rst b/airflow-core/newsfragments/67056.significant.rst index 5143aa0641635..266cb2c24e7af 100644 --- a/airflow-core/newsfragments/67056.significant.rst +++ b/airflow-core/newsfragments/67056.significant.rst @@ -25,7 +25,7 @@ handler are now two independent steps. ``REMOTE_TASK_LOG`` / ``DEFAULT_REMOTE_CONN_ID`` attributes. This is the per-scheme ``if/elif`` chain in ``airflow_local_settings.py`` and is transitional — it will be removed once every in-tree provider exposes - ``from_config``. + ``from_config`` in Airflow 4.0. **``RemoteLogIO.from_config`` contract:** diff --git a/airflow-core/src/airflow/providers_manager.py b/airflow-core/src/airflow/providers_manager.py index 118ac9d9102d8..8e270209a8b0a 100644 --- a/airflow-core/src/airflow/providers_manager.py +++ b/airflow-core/src/airflow/providers_manager.py @@ -1269,9 +1269,7 @@ def _discover_remote_logging(self) -> None: scheme=entry["scheme"], package_name=provider_package, ) - self._remote_logging_info_list.append(info) - existing = self._remote_logging_by_scheme.get(info.scheme) - if existing is not None: + if (existing := self._remote_logging_by_scheme.get(info.scheme)) is not None: log.warning( "Remote logging scheme '%s' is already registered by %s; ignoring " "duplicate registration from %s.", @@ -1280,6 +1278,7 @@ def _discover_remote_logging(self) -> None: info.package_name, ) continue + self._remote_logging_info_list.append(info) self._remote_logging_by_scheme[info.scheme] = info def _discover_secrets_backends(self) -> None: diff --git a/airflow-core/tests/unit/always/test_providers_manager.py b/airflow-core/tests/unit/always/test_providers_manager.py index 5b915162a09ce..25e3774d4f7e7 100644 --- a/airflow-core/tests/unit/always/test_providers_manager.py +++ b/airflow-core/tests/unit/always/test_providers_manager.py @@ -225,6 +225,8 @@ def test_providers_manager_register_remote_logging_duplicate_scheme_first_wins(s providers_manager._discover_remote_logging() winner = providers_manager._remote_logging_by_scheme["dup"] assert winner.package_name == "fake.remote.logging.first" + assert len(providers_manager._remote_logging_info_list) == 1 + assert providers_manager._remote_logging_info_list[0].package_name == "fake.remote.logging.first" def test_providers_manager_remote_logging_bad_class_filtered(self): providers_manager = ProvidersManager() diff --git a/task-sdk/src/airflow/sdk/providers_manager_runtime.py b/task-sdk/src/airflow/sdk/providers_manager_runtime.py index ba3382b221463..ca4d4a94732fb 100644 --- a/task-sdk/src/airflow/sdk/providers_manager_runtime.py +++ b/task-sdk/src/airflow/sdk/providers_manager_runtime.py @@ -563,9 +563,7 @@ def _discover_remote_logging(self) -> None: scheme=entry["scheme"], package_name=provider_package, ) - self._remote_logging_info_list.append(info) - existing = self._remote_logging_by_scheme.get(info.scheme) - if existing is not None: + if (existing := self._remote_logging_by_scheme.get(info.scheme)) is not None: log.warning( "Remote logging scheme '%s' is already registered by %s; ignoring " "duplicate registration from %s.", @@ -574,6 +572,7 @@ def _discover_remote_logging(self) -> None: info.package_name, ) continue + self._remote_logging_info_list.append(info) self._remote_logging_by_scheme[info.scheme] = info def _discover_taskflow_decorators(self) -> None: diff --git a/task-sdk/tests/task_sdk/test_providers_manager_runtime.py b/task-sdk/tests/task_sdk/test_providers_manager_runtime.py index dfa18c3d9d7e5..1f22ff6e73fc1 100644 --- a/task-sdk/tests/task_sdk/test_providers_manager_runtime.py +++ b/task-sdk/tests/task_sdk/test_providers_manager_runtime.py @@ -325,6 +325,8 @@ def test_register_remote_logging_duplicate_scheme_first_wins(self): assert providers_manager._remote_logging_by_scheme["dup"].package_name == ( "fake.remote.logging.first" ) + assert len(providers_manager._remote_logging_info_list) == 1 + assert providers_manager._remote_logging_info_list[0].package_name == "fake.remote.logging.first" def test_register_remote_logging_bad_class_filtered(self): providers_manager = ProvidersManagerTaskRuntime()