This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8216231fbd3 Enable provider-based GCS remote logging resolution
(#70504)
8216231fbd3 is described below
commit 8216231fbd307be7835705c186a7940f051d3fd2
Author: Aaron Chen <[email protected]>
AuthorDate: Mon Jul 27 20:45:49 2026 +0800
Enable provider-based GCS remote logging resolution (#70504)
---
providers/google/provider.yaml | 4 ++
.../providers/google/cloud/log/gcs_task_handler.py | 27 +++++++
.../airflow/providers/google/get_provider_info.py | 6 ++
.../unit/google/cloud/log/test_gcs_task_handler.py | 83 ++++++++++++++++++++++
4 files changed, 120 insertions(+)
diff --git a/providers/google/provider.yaml b/providers/google/provider.yaml
index b19fe812725..4a1fe1d3589 100644
--- a/providers/google/provider.yaml
+++ b/providers/google/provider.yaml
@@ -1489,5 +1489,9 @@ logging:
- airflow.providers.google.cloud.log.gcs_task_handler.GCSTaskHandler
-
airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverTaskHandler
+remote-logging:
+ - classpath:
airflow.providers.google.cloud.log.gcs_task_handler.GCSRemoteLogIO
+ scheme: gs
+
queues:
-
airflow.providers.google.event_scheduling.events.pubsub.PubSubMessageQueueEventTriggerContainer
diff --git
a/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
b/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
index 42c4ba9a61b..26a83af4190 100644
---
a/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
+++
b/providers/google/src/airflow/providers/google/cloud/log/gcs_task_handler.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import inspect
import logging
import os
import shutil
@@ -70,6 +71,32 @@ class GCSRemoteLogIO(LoggingMixin): # noqa: D101
processors = ()
+ @classmethod
+ def from_config(cls) -> GCSRemoteLogIO:
+ """Build the remote log IO from Airflow logging configuration."""
+ remote_task_handler_kwargs = conf.getjson("logging",
"remote_task_handler_kwargs", fallback={})
+ if not isinstance(remote_task_handler_kwargs, dict):
+ raise ValueError(
+ "logging/remote_task_handler_kwargs must be a JSON object (a
python dict), we got "
+ f"{type(remote_task_handler_kwargs)}"
+ )
+ # remote_task_handler_kwargs mixes FileTaskHandler kwargs with IO
kwargs; only the
+ # latter belong to this class (same split as
airflow_local_settings.py).
+ fth_params =
frozenset(inspect.signature(FileTaskHandler.__init__).parameters) - {
+ "self",
+ "base_log_folder",
+ }
+ io_kwargs = {k: v for k, v in remote_task_handler_kwargs.items() if k
not in fth_params}
+ return cls(
+ **{
+ "base_log_folder":
os.path.expanduser(conf.get_mandatory_value("logging", "base_log_folder")),
+ "remote_base": conf.get_mandatory_value("logging",
"remote_base_log_folder"),
+ "delete_local_copy": conf.getboolean("logging",
"delete_local_logs"),
+ "gcp_key_path": conf.get_mandatory_value("logging",
"google_key_path", fallback=None),
+ }
+ | io_kwargs,
+ )
+
def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None) ->
None:
"""Upload the given log path to the remote storage."""
path = Path(path)
diff --git a/providers/google/src/airflow/providers/google/get_provider_info.py
b/providers/google/src/airflow/providers/google/get_provider_info.py
index 13feaa8d1e7..8806a6a31e0 100644
--- a/providers/google/src/airflow/providers/google/get_provider_info.py
+++ b/providers/google/src/airflow/providers/google/get_provider_info.py
@@ -1705,6 +1705,12 @@ def get_provider_info():
"airflow.providers.google.cloud.log.gcs_task_handler.GCSTaskHandler",
"airflow.providers.google.cloud.log.stackdriver_task_handler.StackdriverTaskHandler",
],
+ "remote-logging": [
+ {
+ "classpath":
"airflow.providers.google.cloud.log.gcs_task_handler.GCSRemoteLogIO",
+ "scheme": "gs",
+ }
+ ],
"queues": [
"airflow.providers.google.event_scheduling.events.pubsub.PubSubMessageQueueEventTriggerContainer"
],
diff --git
a/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
b/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
index aba65728472..08e3a453990 100644
--- a/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
+++ b/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
@@ -20,6 +20,7 @@ import copy
import io
import logging
import os
+import pathlib
from types import GeneratorType
from typing import TYPE_CHECKING
from unittest import mock
@@ -49,6 +50,88 @@ def patch_mock_client_for_list_blobs(mock_client: MagicMock,
blob_names: list[st
mock_client.return_value.list_blobs.return_value = mock_blobs
+class TestGCSRemoteLogIOFromConfig:
+ @conf_vars(
+ {
+ ("logging", "base_log_folder"): "~/airflow/logs",
+ ("logging", "remote_base_log_folder"):
"gs://bucket/remote/log/location",
+ ("logging", "delete_local_logs"): "True",
+ ("logging", "google_key_path"): "/tmp/gcs-logging-key.json",
+ }
+ )
+ def test_from_config(self):
+ subject = GCSRemoteLogIO.from_config()
+
+ assert subject.remote_base == "gs://bucket/remote/log/location"
+ assert subject.base_log_folder ==
pathlib.Path(os.path.expanduser("~/airflow/logs"))
+ assert subject.delete_local_copy is True
+ assert subject.gcp_key_path == "/tmp/gcs-logging-key.json"
+
+ @conf_vars(
+ {
+ ("logging", "base_log_folder"): "/tmp/airflow/logs",
+ ("logging", "remote_base_log_folder"):
"gs://bucket/remote/log/location",
+ ("logging", "delete_local_logs"): "False",
+ ("logging", "remote_task_handler_kwargs"): '{"delete_local_copy":
true, "max_bytes": 1024}',
+ }
+ )
+ def
test_from_config_applies_io_kwargs_and_filters_file_handler_kwargs(self):
+ subject = GCSRemoteLogIO.from_config()
+
+ assert subject.delete_local_copy is True
+ assert not hasattr(subject, "max_bytes")
+
+ @conf_vars({("logging", "remote_task_handler_kwargs"): '["not", "a",
"dict"]'})
+ def test_from_config_rejects_non_dict_remote_task_handler_kwargs(self):
+ with pytest.raises(ValueError, match="remote_task_handler_kwargs"):
+ GCSRemoteLogIO.from_config()
+
+ def test_provider_registers_gs_scheme(self):
+ from airflow.providers_manager import ProvidersManager
+
+ manager = ProvidersManager()
+ if not hasattr(manager, "remote_logging_handler_by_scheme"):
+ pytest.skip("Airflow core does not support remote logging provider
dispatch")
+
+ info = manager.remote_logging_handler_by_scheme("gs")
+
+ assert info is not None
+ assert info.classpath ==
"airflow.providers.google.cloud.log.gcs_task_handler.GCSRemoteLogIO"
+
+ @pytest.mark.parametrize(
+ "manager_classpath",
+ [
+ pytest.param("airflow.providers_manager.ProvidersManager",
id="core"),
+ pytest.param(
+
"airflow.sdk.providers_manager_runtime.ProvidersManagerTaskRuntime",
id="task-runtime"
+ ),
+ ],
+ )
+ @conf_vars(
+ {
+ ("logging", "remote_logging"): "True",
+ ("logging", "remote_base_log_folder"):
"gs://bucket/remote/log/location",
+ ("logging", "remote_log_conn_id"): "google_cloud_default",
+ }
+ )
+ def
test_resolve_remote_task_log_uses_provider_dispatch_not_local_settings(self,
manager_classpath):
+ factory = pytest.importorskip("airflow._shared.logging.factory")
+ from airflow._shared.module_loading import import_string
+ from airflow.configuration import conf
+
+ with mock.patch.object(factory, "discover_remote_log_handler",
autospec=True) as legacy_discover:
+ remote_task_log, conn_id = factory.resolve_remote_task_log(
+ conf=conf,
+ providers_manager=import_string(manager_classpath)(),
+ import_string=import_string,
+ )
+
+ assert isinstance(remote_task_log, GCSRemoteLogIO)
+ assert remote_task_log.remote_base == "gs://bucket/remote/log/location"
+ assert conn_id == "google_cloud_default"
+ legacy_discover.assert_not_called()
+
+
@pytest.mark.db_test
@mock.patch(
"airflow.providers.google.cloud.log.gcs_task_handler.get_credentials_and_project_id",
