This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 4bc20e6e938 [v3-3-test] Clarify ``logging_config_class`` contract and
document ``REMOTE_TASK_LOG`` (#67104) (#70592)
4bc20e6e938 is described below
commit 4bc20e6e938c4b28e76811323c08c42348e1b9c3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Jul 29 11:40:28 2026 +0800
[v3-3-test] Clarify ``logging_config_class`` contract and document
``REMOTE_TASK_LOG`` (#67104) (#70592)
---
.../advanced-logging-configuration.rst | 53 ++++++++++++++++--
.../config_templates/airflow_local_settings.py | 8 +++
.../src/airflow/config_templates/config.yml | 9 ++--
airflow-core/src/airflow/logging_config.py | 50 ++++++++++++++++-
.../tests/unit/logging/test_logging_config.py | 63 ++++++++++++++++++++++
pyproject.toml | 8 +--
scripts/ci/prek/update_airflow_pyproject_toml.py | 4 +-
.../logging/src/airflow_shared/logging/remote.py | 20 ++++++-
task-sdk/src/airflow/sdk/log.py | 7 ++-
9 files changed, 205 insertions(+), 17 deletions(-)
diff --git
a/airflow-core/docs/administration-and-deployment/logging-monitoring/advanced-logging-configuration.rst
b/airflow-core/docs/administration-and-deployment/logging-monitoring/advanced-logging-configuration.rst
index fa34e74f314..629cb506551 100644
---
a/airflow-core/docs/administration-and-deployment/logging-monitoring/advanced-logging-configuration.rst
+++
b/airflow-core/docs/administration-and-deployment/logging-monitoring/advanced-logging-configuration.rst
@@ -46,13 +46,14 @@ that Python objects log to loggers that follow naming
convention of ``<package>.
You can read more about standard python logging classes (Loggers, Handlers,
Formatters) in the
`Python logging documentation <https://docs.python.org/library/logging.html>`_.
-Create a custom logging class
------------------------------
+Create a custom logging config
+------------------------------
Configuring your logging classes can be done via the ``logging_config_class``
option in ``airflow.cfg`` file.
-This configuration should specify the import path to a configuration
compatible with
-:func:`logging.config.dictConfig`. If your file is a standard import location,
then you should set a
-:envvar:`PYTHONPATH` environment variable.
+Despite the option name the value is a dotted import path to a ``dict`` that
satisfies
+:func:`logging.config.dictConfig` — not a Python class. The ``_class`` suffix
is
+historical and kept for backwards compatibility. If your file is a standard
import
+location, then you should set a :envvar:`PYTHONPATH` environment variable.
Follow the steps below to enable custom logging config class:
@@ -102,6 +103,48 @@ See :doc:`../modules_management` for details on how Python
and Airflow manage mo
You can override the way both standard logs of the components and "task"
logs are handled.
+Custom logging configs and remote logging
+-----------------------------------------
+
+When ``[logging] remote_logging = True`` and you point ``logging_config_class``
+at your own module, define two module-level attributes in that module:
+
+* ``REMOTE_TASK_LOG`` — an instance of
+ :class:`~airflow.logging.remote.RemoteLogIO` (or
+ :class:`~airflow.logging.remote.RemoteLogStreamIO`) that uploads task logs
+ and reads them back for the UI.
+* ``DEFAULT_REMOTE_CONN_ID`` — default Airflow connection id used when
+ ``[logging] remote_log_conn_id`` is unset.
+
+If ``REMOTE_TASK_LOG`` is missing, Airflow emits one ``WARNING`` at startup
+and the UI cannot read task logs back from the remote backend.
+
+ .. code-block:: python
+
+ # ~/airflow/config/log_config.py
+ from airflow.logging.remote import RemoteLogIO
+
+
+ class MyRemoteLogIO:
+ @property
+ def processors(self):
+ return ()
+
+ def upload(self, path, ti): ... # upload local log file at ``path``
to your backend
+
+ def read(self, relative_path, ti): ... # return (source_info,
log_messages) for the UI
+
+
+ REMOTE_TASK_LOG: RemoteLogIO | None = MyRemoteLogIO()
+ DEFAULT_REMOTE_CONN_ID: str | None = "my_remote_conn"
+
+.. note::
+
+ Define ``REMOTE_TASK_LOG`` in your own module rather than re-exporting it
+ from ``airflow.config_templates.airflow_local_settings``, which is planned
+ for deprecation.
+
+
Custom logger for Operators, Hooks and Tasks
--------------------------------------------
diff --git
a/airflow-core/src/airflow/config_templates/airflow_local_settings.py
b/airflow-core/src/airflow/config_templates/airflow_local_settings.py
index e30e68fb179..c5bb91b682f 100644
--- a/airflow-core/src/airflow/config_templates/airflow_local_settings.py
+++ b/airflow-core/src/airflow/config_templates/airflow_local_settings.py
@@ -50,6 +50,9 @@ DAG_PROCESSOR_LOG_TARGET: str =
conf.get_mandatory_value("logging", "DAG_PROCESS
BASE_LOG_FOLDER: str = os.path.expanduser(conf.get_mandatory_value("logging",
"BASE_LOG_FOLDER"))
# This isn't used anymore, but kept for compat of people who might have
imported it
+# Default value for the ``[logging] logging_config_class`` option. Plain
+# ``logging.config.dictConfig`` dict; the ``_class`` suffix on the config
option
+# is historical.
DEFAULT_LOGGING_CONFIG: dict[str, Any] = {
"version": 1,
"disable_existing_loggers": False,
@@ -121,6 +124,11 @@ if EXTRA_LOGGER_NAMES:
##################
REMOTE_LOGGING: bool = conf.getboolean("logging", "remote_logging")
+
+# Side-channel attributes read by ``discover_remote_log_handler`` from
whichever
+# module ``[logging] logging_config_class`` resolves through. Custom modules
that
+# override that option should define both at module scope to enable remote
+# task-log read-back.
REMOTE_TASK_LOG: RemoteLogIO | RemoteLogStreamIO | None = None
DEFAULT_REMOTE_CONN_ID: str | None = None
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 975e6ba1121..27d674b5802 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -921,9 +921,12 @@ logging:
default: "INFO"
logging_config_class:
description: |
- Logging class
- Specify the class that will specify the logging configuration
- This class has to be on the python classpath
+ Dotted import path to a ``logging.config.dictConfig`` dict. The
+ ``_class`` suffix is historical — the target is a dict, not a class.
+
+ Airflow also reads two optional module-level attributes from the
+ enclosing module: ``REMOTE_TASK_LOG`` and ``DEFAULT_REMOTE_CONN_ID``,
+ used to drive remote task-log read-back. See
:ref:`write-logs-advanced`.
version_added: 2.0.0
type: string
example: "my.path.default_local_settings.LOGGING_CONFIG"
diff --git a/airflow-core/src/airflow/logging_config.py
b/airflow-core/src/airflow/logging_config.py
index 7f4ae605cb2..da337997de8 100644
--- a/airflow-core/src/airflow/logging_config.py
+++ b/airflow-core/src/airflow/logging_config.py
@@ -36,7 +36,7 @@ class _ActiveLoggingConfig:
"""Private class to hold active logging config variables."""
logging_config_loaded: bool = False
- remote_task_log: RemoteLogIO | None
+ remote_task_log: RemoteLogIO | None = None
default_remote_conn_id: str | None = None
@classmethod
@@ -120,6 +120,44 @@ def load_logging_config() -> tuple[dict[str, Any], str]:
) or DEFAULT_LOGGING_CONFIG_PATH
+def _warn_if_missing_remote_task_log(logging_class_path: str) -> None:
+ """
+ Warn if ``[logging] remote_logging`` is on but the user module exposes no
remote IO.
+
+ Runs *after* ``dictConfig`` has constructed handlers, so deprecated
+ self-registration in provider task handlers (Elasticsearch, OpenSearch) has
+ already had its chance to populate
``_ActiveLoggingConfig.remote_task_log``.
+ Only fires for user-defined ``logging_config_class`` values; the stock
+ fallback is exempt.
+
+ :param logging_class_path: the resolved ``[logging] logging_config_class``
+ dotted path (already defaulted to :data:`DEFAULT_LOGGING_CONFIG_PATH`).
+ """
+ # An empty path is not a meaningful override -- ``_get_logging_config()``
treats it the
+ # same as unset and falls back to ``DEFAULT_LOGGING_CONFIG_PATH``, so it
must not be
+ # treated as a user-defined logging config class here either.
+ has_user_defined_logging_config_class = (
+ bool(logging_class_path) and logging_class_path !=
DEFAULT_LOGGING_CONFIG_PATH
+ )
+ remote_logging_enabled = conf.getboolean("logging", "remote_logging",
fallback=False)
+ if not (has_user_defined_logging_config_class and remote_logging_enabled):
+ return
+ if get_remote_task_log() is not None:
+ return
+ # Strip the trailing ``.<config_attr>`` to leave the enclosing module path.
+ # ``logging_class_path`` should always be dotted since ``import_string``
+ # would have raised otherwise, but guard the access defensively.
+ parts = logging_class_path.rsplit(".", 1)
+ modpath = parts[0] if len(parts) == 2 else logging_class_path
+ log.warning(
+ "[logging] remote_logging is enabled but the user-defined logging
module %r "
+ "does not expose a REMOTE_TASK_LOG attribute, so remote task-log
read-back is "
+ "disabled. Define REMOTE_TASK_LOG (a RemoteLogIO instance) at module
scope "
+ "to enable it.",
+ modpath,
+ )
+
+
def configure_logging():
from airflow._shared.logging import configure_logging, init_log_folder,
translate_config_values
@@ -171,6 +209,16 @@ def configure_logging():
# otherwise Airflow would silently fall back on the default config
raise e
+ # Runs after dictConfig so deprecated handler self-registration (ES/OS) has
+ # had its chance to populate _ActiveLoggingConfig.remote_task_log.
+ # The trailing `or DEFAULT_LOGGING_CONFIG_PATH` also covers an explicit
+ # `logging_config_class = ""`, which conf.get() would otherwise return
as-is.
+ logging_class_path = (
+ conf.get("logging", "logging_config_class",
fallback=DEFAULT_LOGGING_CONFIG_PATH)
+ or DEFAULT_LOGGING_CONFIG_PATH
+ )
+ _warn_if_missing_remote_task_log(logging_class_path)
+
validate_logging_config()
new_folder_permissions = int(
diff --git a/airflow-core/tests/unit/logging/test_logging_config.py
b/airflow-core/tests/unit/logging/test_logging_config.py
index 3fffb376425..f2a715f80a5 100644
--- a/airflow-core/tests/unit/logging/test_logging_config.py
+++ b/airflow-core/tests/unit/logging/test_logging_config.py
@@ -28,11 +28,14 @@ from airflow.logging_config import (
_ActiveLoggingConfig,
_get_logging_config,
_load_logging_config,
+ _warn_if_missing_remote_task_log,
get_default_remote_conn_id,
get_remote_task_log,
load_logging_config,
)
+from tests_common.test_utils.config import conf_vars
+
@pytest.fixture(autouse=True)
def _reset_active_logging_config(monkeypatch):
@@ -200,3 +203,63 @@ class TestGetDefaultRemoteConnId:
mocked_conf.get.return_value = None
assert get_default_remote_conn_id() == "cached_conn"
mock_load.assert_not_called()
+
+
+class TestWarnIfMissingRemoteTaskLog:
+ @pytest.fixture(autouse=True)
+ def _reset_active_logging_config(self, monkeypatch):
+ monkeypatch.setattr(_ActiveLoggingConfig, "remote_task_log", None,
raising=False)
+ monkeypatch.setattr(_ActiveLoggingConfig, "logging_config_loaded",
True, raising=False)
+
+ @pytest.mark.parametrize(
+ ("remote_logging", "logging_class_path",
"remote_task_log_already_set", "expected_module"),
+ [
+ pytest.param(
+ True,
+ "my_pkg.custom_settings.LOGGING_CONFIG",
+ False,
+ "my_pkg.custom_settings",
+ id="user_module_missing_remote_task_log",
+ ),
+ pytest.param(True, DEFAULT_LOGGING_CONFIG_PATH, False, None,
id="fallback_path"),
+ pytest.param(
+ False, "my_pkg.custom_settings.LOGGING_CONFIG", False, None,
id="remote_logging_disabled"
+ ),
+ pytest.param(
+ True, "my_pkg.custom_settings.LOGGING_CONFIG", True, None,
id="remote_task_log_already_set"
+ ),
+ pytest.param(True, "", False, None, id="empty_logging_class_path"),
+ ],
+ )
+ def test_warn_if_missing_remote_task_log(
+ self, monkeypatch, remote_logging, logging_class_path,
remote_task_log_already_set, expected_module
+ ):
+ if remote_task_log_already_set:
+ monkeypatch.setattr(_ActiveLoggingConfig, "remote_task_log",
object(), raising=False)
+ with conf_vars({("logging", "remote_logging"): str(remote_logging)}):
+ with mock.patch("airflow.logging_config.log") as mock_log:
+ _warn_if_missing_remote_task_log(logging_class_path)
+ if expected_module is None:
+ mock_log.warning.assert_not_called()
+ else:
+ mock_log.warning.assert_called_once()
+ assert expected_module in mock_log.warning.call_args.args
+
+ def test_skips_warning_when_resolved_lazily_via_provider_dispatch(self,
monkeypatch):
+ """The cache may still be cold (no ES/OS handler self-registered
during dictConfig),
+ but a working setup resolved through ProvidersManager dispatch must
still suppress
+ the warning -- so the check has to trigger the real resolution via
+ get_remote_task_log() rather than only inspecting the cached field."""
+ monkeypatch.setattr(_ActiveLoggingConfig, "logging_config_loaded",
False, raising=False)
+ sentinel_handler = object()
+ with (
+ conf_vars({("logging", "remote_logging"): "True"}),
+ mock.patch("airflow.logging_config.resolve_remote_task_log") as
mock_resolve,
+ mock.patch("airflow.providers_manager.ProvidersManager"),
+ mock.patch("airflow.logging_config.log") as mock_log,
+ ):
+ mock_resolve.return_value = (sentinel_handler, None)
+
_warn_if_missing_remote_task_log("my_pkg.custom_settings.LOGGING_CONFIG")
+
+ mock_log.warning.assert_not_called()
+ assert _ActiveLoggingConfig.remote_task_log is sentinel_handler
diff --git a/pyproject.toml b/pyproject.toml
index 7eceb2eb5bc..19b006906ca 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -219,7 +219,7 @@ apache-airflow = "airflow.__main__:main"
"apache-airflow-providers-edge3>=1.0.0"
]
"elasticsearch" = [
- "apache-airflow-providers-elasticsearch>=6.5.0" # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
+ "apache-airflow-providers-elasticsearch>=6.6.0" # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
]
"exasol" = [
"apache-airflow-providers-exasol>=4.6.1"
@@ -306,7 +306,7 @@ apache-airflow = "airflow.__main__:main"
"apache-airflow-providers-openlineage>=2.3.0" # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
]
"opensearch" = [
- "apache-airflow-providers-opensearch>=1.9.0" # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
+ "apache-airflow-providers-opensearch>=1.9.3" # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
]
"opsgenie" = [
"apache-airflow-providers-opsgenie>=5.8.0"
@@ -447,7 +447,7 @@ apache-airflow = "airflow.__main__:main"
"apache-airflow-providers-discord>=3.9.0",
"apache-airflow-providers-docker>=3.14.1",
"apache-airflow-providers-edge3>=1.0.0",
- "apache-airflow-providers-elasticsearch>=6.5.0", # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
+ "apache-airflow-providers-elasticsearch>=6.6.0", # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
"apache-airflow-providers-exasol>=4.6.1",
"apache-airflow-providers-fab>=3.6.0", # Set from MIN_VERSION_OVERRIDE in
update_airflow_pyproject_toml.py
"apache-airflow-providers-facebook>=3.7.0",
@@ -476,7 +476,7 @@ apache-airflow = "airflow.__main__:main"
"apache-airflow-providers-openai>=1.5.0",
"apache-airflow-providers-openfaas>=3.7.0",
"apache-airflow-providers-openlineage>=2.3.0", # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
- "apache-airflow-providers-opensearch>=1.9.0", # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
+ "apache-airflow-providers-opensearch>=1.9.3", # Set from
MIN_VERSION_OVERRIDE in update_airflow_pyproject_toml.py
"apache-airflow-providers-opsgenie>=5.8.0",
"apache-airflow-providers-oracle>=3.12.0",
"apache-airflow-providers-pagerduty>=3.8.1",
diff --git a/scripts/ci/prek/update_airflow_pyproject_toml.py
b/scripts/ci/prek/update_airflow_pyproject_toml.py
index a7d65331185..2313790e0a5 100755
--- a/scripts/ci/prek/update_airflow_pyproject_toml.py
+++ b/scripts/ci/prek/update_airflow_pyproject_toml.py
@@ -94,8 +94,8 @@ MIN_VERSION_OVERRIDE: dict[str, Version] = {
"openlineage": parse_version("2.3.0"),
"git": parse_version("0.0.2"),
"common.messaging": parse_version("2.0.0"),
- "elasticsearch": parse_version("6.5.0"),
- "opensearch": parse_version("1.9.0"),
+ "elasticsearch": parse_version("6.6.0"),
+ "opensearch": parse_version("1.9.3"),
}
diff --git a/shared/logging/src/airflow_shared/logging/remote.py
b/shared/logging/src/airflow_shared/logging/remote.py
index 59af50be6bc..b80a3ceb9cd 100644
--- a/shared/logging/src/airflow_shared/logging/remote.py
+++ b/shared/logging/src/airflow_shared/logging/remote.py
@@ -79,7 +79,25 @@ def discover_remote_log_handler(
fallback_path: str,
import_string: Callable[[str], Any],
) -> tuple[RemoteLogIO | None, str | None]:
- """Discover and load the remote log handler from a logging config
module."""
+ """
+ Look up the optional remote-log handler alongside a logging dictConfig.
+
+ ``[logging] logging_config_class`` is a dotted path to a
+ ``logging.config.dictConfig`` dict. After importing the dict, this helper
+ re-imports the enclosing module and reads two optional module-level
+ attributes via ``getattr``:
+
+ * ``REMOTE_TASK_LOG`` — :class:`RemoteLogIO` / :class:`RemoteLogStreamIO`
+ instance that uploads and reads task logs.
+ * ``DEFAULT_REMOTE_CONN_ID`` — default Airflow connection id for that
+ backend.
+
+ Either may be ``None`` immediately after this call; provider task handlers
+ can still populate ``REMOTE_TASK_LOG`` from inside ``__init__`` when
+ ``dictConfig`` instantiates them (deprecated path). Callers that want to
+ warn on missing remote-log configuration should re-check
+ ``_ActiveLoggingConfig.remote_task_log`` *after* ``dictConfig`` has run.
+ """
# Sometimes we end up with `""` as the value!
logging_class_path = logging_class_path or fallback_path
diff --git a/task-sdk/src/airflow/sdk/log.py b/task-sdk/src/airflow/sdk/log.py
index 5e775cbcac4..e105fd70d76 100644
--- a/task-sdk/src/airflow/sdk/log.py
+++ b/task-sdk/src/airflow/sdk/log.py
@@ -173,7 +173,12 @@ def init_log_file(local_relative_path: str) -> Path:
def _load_logging_config() -> None:
- """Load and cache the remote logging configuration from SDK config."""
+ """
+ Load and cache the remote logging configuration from SDK config.
+
+ SDK mirror of :func:`airflow.logging_config.load_logging_config` — see that
+ function for the ``logging_config_class`` / ``REMOTE_TASK_LOG`` contract.
+ """
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
