This is an automated email from the ASF dual-hosted git repository.
jason810496 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 3160334d0ae Add HdfsRemoteLogIO.from_config and register hdfs remote
logging scheme (#71278)
3160334d0ae is described below
commit 3160334d0ae3213a40c8ae89330ef9c4022ab790
Author: Baha Bouali <[email protected]>
AuthorDate: Mon Aug 10 10:54:18 2026 +0100
Add HdfsRemoteLogIO.from_config and register hdfs remote logging scheme
(#71278)
Signed-off-by: baha-bouali <[email protected]>
---
providers/apache/hdfs/provider.yaml | 4 ++
.../providers/apache/hdfs/get_provider_info.py | 6 ++
.../providers/apache/hdfs/log/hdfs_task_handler.py | 26 +++++++
.../unit/apache/hdfs/log/test_hdfs_task_handler.py | 82 +++++++++++++++++++++-
4 files changed, 117 insertions(+), 1 deletion(-)
diff --git a/providers/apache/hdfs/provider.yaml
b/providers/apache/hdfs/provider.yaml
index 155137037af..d6f92d70137 100644
--- a/providers/apache/hdfs/provider.yaml
+++ b/providers/apache/hdfs/provider.yaml
@@ -117,6 +117,10 @@ hooks:
logging:
- airflow.providers.apache.hdfs.log.hdfs_task_handler.HdfsTaskHandler
+remote-logging:
+ - classpath:
airflow.providers.apache.hdfs.log.hdfs_task_handler.HdfsRemoteLogIO
+ scheme: hdfs
+
connection-types:
- hook-class-name: airflow.providers.apache.hdfs.hooks.webhdfs.WebHDFSHook
hook-name: "Apache WebHDFS"
diff --git
a/providers/apache/hdfs/src/airflow/providers/apache/hdfs/get_provider_info.py
b/providers/apache/hdfs/src/airflow/providers/apache/hdfs/get_provider_info.py
index 381cc777b45..98a0936e426 100644
---
a/providers/apache/hdfs/src/airflow/providers/apache/hdfs/get_provider_info.py
+++
b/providers/apache/hdfs/src/airflow/providers/apache/hdfs/get_provider_info.py
@@ -66,6 +66,12 @@ def get_provider_info():
{"integration-name": "WebHDFS", "python-modules":
["airflow.providers.apache.hdfs.hooks.webhdfs"]}
],
"logging":
["airflow.providers.apache.hdfs.log.hdfs_task_handler.HdfsTaskHandler"],
+ "remote-logging": [
+ {
+ "classpath":
"airflow.providers.apache.hdfs.log.hdfs_task_handler.HdfsRemoteLogIO",
+ "scheme": "hdfs",
+ }
+ ],
"connection-types": [
{
"hook-class-name":
"airflow.providers.apache.hdfs.hooks.webhdfs.WebHDFSHook",
diff --git
a/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
b/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
index 22304637db2..5694c62602a 100644
---
a/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
+++
b/providers/apache/hdfs/src/airflow/providers/apache/hdfs/log/hdfs_task_handler.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import inspect
import logging
import os
import shutil
@@ -46,6 +47,31 @@ class HdfsRemoteLogIO(LoggingMixin): # noqa: D101
processors = ()
+ @classmethod
+ def from_config(cls) -> HdfsRemoteLogIO:
+ """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": urlsplit(conf.get_mandatory_value("logging",
"remote_base_log_folder")).path,
+ "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/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py
b/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py
index b5e2adb7577..bd82513cbff 100644
--- a/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py
+++ b/providers/apache/hdfs/tests/unit/apache/hdfs/log/test_hdfs_task_handler.py
@@ -25,7 +25,7 @@ from unittest.mock import PropertyMock
import pytest
from airflow.providers.apache.hdfs.hooks.webhdfs import WebHDFSHook
-from airflow.providers.apache.hdfs.log.hdfs_task_handler import HdfsTaskHandler
+from airflow.providers.apache.hdfs.log.hdfs_task_handler import
HdfsRemoteLogIO, HdfsTaskHandler
from airflow.utils.state import TaskInstanceState
from airflow.utils.timezone import datetime
@@ -37,6 +37,86 @@ pytestmark = pytest.mark.db_test
DEFAULT_DATE = datetime(2020, 8, 10)
+class TestHdfsRemoteLogIOFromConfig:
+ @conf_vars(
+ {
+ ("logging", "base_log_folder"): "~/airflow/logs",
+ ("logging", "remote_base_log_folder"):
"hdfs://namenode/remote/log/location",
+ ("logging", "delete_local_logs"): "True",
+ }
+ )
+ def test_from_config(self):
+ subject = HdfsRemoteLogIO.from_config()
+
+ assert subject.remote_base == "/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"):
"hdfs://namenode/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 = HdfsRemoteLogIO.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"):
+ HdfsRemoteLogIO.from_config()
+
+ def test_provider_registers_hdfs_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("hdfs")
+
+ assert info is not None
+ assert info.classpath ==
"airflow.providers.apache.hdfs.log.hdfs_task_handler.HdfsRemoteLogIO"
+
+ @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"):
"hdfs://namenode/remote/log/location",
+ ("logging", "remote_log_conn_id"): "webhdfs_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, HdfsRemoteLogIO)
+ assert remote_task_log.remote_base == "/remote/log/location"
+ assert conn_id == "webhdfs_default"
+ legacy_discover.assert_not_called()
+
+
class TestHdfsTaskHandler:
@pytest.fixture(autouse=True)
def ti(self, create_task_instance, create_log_template):
