Decouple remote logging config from core

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`` in Airflow 4.0.
+
+**``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.

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,25 +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:
+    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)
@@ -78,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")

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",

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",

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,31 @@ 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,
+                )
+                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.",
+                        info.scheme,
+                        existing.package_name,
+                        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:
         """Retrieve all secrets backends defined in the providers."""
         for provider_package, provider in self._provider_dict.items():
@@ -1450,6 +1491,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 +1584,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()