This is an automated email from the ASF dual-hosted git repository.
potiuk 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 1a3f32aa9e3 Add OSSRemoteLogIO.from_config and register oss remote
logging scheme (#70682)
1a3f32aa9e3 is described below
commit 1a3f32aa9e3ae205bef2d6cecb758550dbdb396f
Author: Baha Bouali <[email protected]>
AuthorDate: Sat Aug 1 06:54:55 2026 +0100
Add OSSRemoteLogIO.from_config and register oss remote logging scheme
(#70682)
Signed-off-by: baha-bouali <[email protected]>
---
providers/alibaba/provider.yaml | 4 ++
.../alibaba/cloud/log/oss_task_handler.py | 26 +++++++
.../airflow/providers/alibaba/get_provider_info.py | 6 ++
.../alibaba/cloud/log/test_oss_task_handler.py | 83 +++++++++++++++++++++-
4 files changed, 118 insertions(+), 1 deletion(-)
diff --git a/providers/alibaba/provider.yaml b/providers/alibaba/provider.yaml
index acc7bfa8a4a..988bf96a269 100644
--- a/providers/alibaba/provider.yaml
+++ b/providers/alibaba/provider.yaml
@@ -193,5 +193,9 @@ connection-types:
logging:
- airflow.providers.alibaba.cloud.log.oss_task_handler.OSSTaskHandler
+remote-logging:
+ - classpath:
airflow.providers.alibaba.cloud.log.oss_task_handler.OSSRemoteLogIO
+ scheme: oss
+
extra-links:
- airflow.providers.alibaba.cloud.links.maxcompute.MaxComputeLogViewLink
diff --git
a/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
b/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
index a000c6ff01f..d679f5cf251 100644
---
a/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
+++
b/providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py
@@ -18,6 +18,7 @@
from __future__ import annotations
import contextlib
+import inspect
import os
import shutil
from functools import cached_property
@@ -44,6 +45,31 @@ class OSSRemoteLogIO(LoggingMixin): # noqa: D101
processors = ()
+ @classmethod
+ def from_config(cls) -> OSSRemoteLogIO:
+ """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 belongs to this (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"),
+ }
+ | 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/alibaba/src/airflow/providers/alibaba/get_provider_info.py
b/providers/alibaba/src/airflow/providers/alibaba/get_provider_info.py
index 27e01cd0e74..29cbc14aae5 100644
--- a/providers/alibaba/src/airflow/providers/alibaba/get_provider_info.py
+++ b/providers/alibaba/src/airflow/providers/alibaba/get_provider_info.py
@@ -137,5 +137,11 @@ def get_provider_info():
},
],
"logging":
["airflow.providers.alibaba.cloud.log.oss_task_handler.OSSTaskHandler"],
+ "remote-logging": [
+ {
+ "classpath":
"airflow.providers.alibaba.cloud.log.oss_task_handler.OSSRemoteLogIO",
+ "scheme": "oss",
+ }
+ ],
"extra-links":
["airflow.providers.alibaba.cloud.links.maxcompute.MaxComputeLogViewLink"],
}
diff --git
a/providers/alibaba/tests/unit/alibaba/cloud/log/test_oss_task_handler.py
b/providers/alibaba/tests/unit/alibaba/cloud/log/test_oss_task_handler.py
index 24131d44346..677f1c0a2a4 100644
--- a/providers/alibaba/tests/unit/alibaba/cloud/log/test_oss_task_handler.py
+++ b/providers/alibaba/tests/unit/alibaba/cloud/log/test_oss_task_handler.py
@@ -18,13 +18,14 @@
from __future__ import annotations
import os
+from pathlib import Path
from typing import TYPE_CHECKING
from unittest import mock
from unittest.mock import PropertyMock
import pytest
-from airflow.providers.alibaba.cloud.log.oss_task_handler import OSSTaskHandler
+from airflow.providers.alibaba.cloud.log.oss_task_handler import
OSSRemoteLogIO, OSSTaskHandler
from airflow.utils.state import TaskInstanceState
from airflow.utils.timezone import datetime
@@ -45,6 +46,86 @@ MOCK_CONTENT = "mock_content"
MOCK_FILE_PATH = "mock_file_path"
+class TestOSSRemoteLogIOFromConfig:
+ @conf_vars(
+ {
+ ("logging", "base_log_folder"): "~/airflow/logs",
+ ("logging", "remote_base_log_folder"):
"oss://bucket/remote/log/location",
+ ("logging", "delete_local_logs"): "True",
+ }
+ )
+ def test_from_config(self):
+ subject = OSSRemoteLogIO.from_config()
+
+ assert subject.remote_base == "oss://bucket/remote/log/location"
+ assert subject.base_log_folder ==
Path(os.path.expanduser("~/airflow/logs"))
+ assert subject.delete_local_copy is True
+
+ @conf_vars(
+ {
+ ("logging", "base_log_folder"): "/tmp/airflow/logs",
+ ("logging", "remote_base_log_folder"):
"oss://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 = OSSRemoteLogIO.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"):
+ OSSRemoteLogIO.from_config()
+
+ def test_provider_registers_oss_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("oss")
+
+ assert info is not None
+ assert info.classpath ==
"airflow.providers.alibaba.cloud.log.oss_task_handler.OSSRemoteLogIO"
+
+ @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"):
"oss://bucket/remote/log/location",
+ ("logging", "remote_log_conn_id"): "oss_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, OSSRemoteLogIO)
+ assert remote_task_log.remote_base ==
"oss://bucket/remote/log/location"
+ assert conn_id == "oss_default"
+ legacy_discover.assert_not_called()
+
+
class TestOSSTaskHandler:
def setup_method(self):
self.base_log_folder = "local/airflow/logs/1.log"
