diff --git a/providers/elasticsearch/docs/logging/index.rst b/providers/elasticsearch/docs/logging/index.rst index df2a6e6be71f2..f3fd5486ff0e2 100644 --- a/providers/elasticsearch/docs/logging/index.rst +++ b/providers/elasticsearch/docs/logging/index.rst @@ -37,6 +37,19 @@ First, to use the handler, ``airflow.cfg`` must be configured as follows: [elasticsearch] host = : +On Airflow 3.x you can also route remote logging to Elasticsearch through the provider +dispatch mechanism by adding an ``elasticsearch://`` scheme to +``[logging] remote_base_log_folder``: + +.. code-block:: ini + + [logging] + remote_logging = True + remote_base_log_folder = elasticsearch:// + + [elasticsearch] + host = : + To output task logs to stdout in JSON format, the following config could be used: .. code-block:: ini diff --git a/providers/elasticsearch/provider.yaml b/providers/elasticsearch/provider.yaml index 11de3dc9891b5..3e9d1933a13ba 100644 --- a/providers/elasticsearch/provider.yaml +++ b/providers/elasticsearch/provider.yaml @@ -116,6 +116,10 @@ connection-types: logging: - airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchTaskHandler +remote-logging: + - classpath: airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchRemoteLogIO + scheme: elasticsearch + config: elasticsearch: description: ~ diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py index b0853a98580cd..fbd155a4afd95 100644 --- a/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py +++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py @@ -48,6 +48,12 @@ def get_provider_info(): } ], "logging": ["airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchTaskHandler"], + "remote-logging": [ + { + "classpath": "airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchRemoteLogIO", + "scheme": "elasticsearch", + } + ], "config": { "elasticsearch": { "description": None, diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py index 2fb610186e9d9..b4ebe096139bd 100644 --- a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py +++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py @@ -712,6 +712,28 @@ class ElasticsearchRemoteLogIO(LoggingMixin): # noqa: D101 processors = () + @classmethod + def from_config(cls) -> ElasticsearchRemoteLogIO: + """ + Build the remote log IO from Airflow logging and ``[elasticsearch]`` configuration. + + Mirrors the legacy branch in ``airflow_local_settings.py``. Unlike the object-storage + backends, this does not merge ``[logging] remote_task_handler_kwargs`` IO-kwargs, matching + the legacy behavior for Elasticsearch. + """ + return cls( + base_log_folder=os.path.expanduser(conf.get_mandatory_value("logging", "base_log_folder")), + delete_local_copy=conf.getboolean("logging", "delete_local_logs"), + host=conf.get("elasticsearch", "host") or "http://localhost:9200", + target_index=conf.get_mandatory_value("elasticsearch", "target_index"), + write_stdout=conf.getboolean("elasticsearch", "write_stdout"), + write_to_es=conf.getboolean("elasticsearch", "write_to_es"), + json_format=conf.getboolean("elasticsearch", "json_format"), + host_field=conf.get_mandatory_value("elasticsearch", "host_field"), + offset_field=conf.get_mandatory_value("elasticsearch", "offset_field"), + log_id_template=conf.get_mandatory_value("elasticsearch", "log_id_template"), + ) + def __attrs_post_init__(self): es_kwargs = get_es_kwargs_from_config() self.client = apply_compat_with(elasticsearch.Elasticsearch(self.host, **es_kwargs)) diff --git a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py index a2cb4c315d892..1233c17ea9d80 100644 --- a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py +++ b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py @@ -20,6 +20,7 @@ import dataclasses import json import logging +import os import re from io import StringIO from pathlib import Path @@ -1016,3 +1017,64 @@ def test_non_string_event_falls_back_to_stringified_event(self): assert result.event == str(["a", "b"]) assert result.timestamp is not None mock_logger.debug.assert_called_once() + + +class TestElasticsearchRemoteLogIOFromConfig: + @conf_vars( + { + ("logging", "base_log_folder"): "~/airflow/logs", + ("logging", "delete_local_logs"): "True", + ("elasticsearch", "host"): "http://elasticsearch.example.com:9200", + ("elasticsearch", "target_index"): "my-logs", + ("elasticsearch", "write_stdout"): "True", + ("elasticsearch", "write_to_es"): "True", + ("elasticsearch", "json_format"): "True", + ("elasticsearch", "host_field"): "host.name", + ("elasticsearch", "offset_field"): "log.offset", + ("elasticsearch", "log_id_template"): "{dag_id}-{task_id}-{run_id}", + } + ) + def test_from_config(self): + subject = ElasticsearchRemoteLogIO.from_config() + + assert subject.base_log_folder == Path(os.path.expanduser("~/airflow/logs")) + assert subject.delete_local_copy is True + assert subject.host == "http://elasticsearch.example.com:9200" + assert subject.target_index == "my-logs" + assert subject.write_stdout is True + assert subject.write_to_es is True + assert subject.json_format is True + assert subject.host_field == "host.name" + assert subject.offset_field == "log.offset" + assert subject.log_id_template == "{dag_id}-{task_id}-{run_id}" + + @conf_vars( + { + ("logging", "base_log_folder"): "~/airflow/logs", + ("elasticsearch", "host"): "", + ("elasticsearch", "target_index"): "my-logs", + ("elasticsearch", "host_field"): "host", + ("elasticsearch", "offset_field"): "offset", + ("elasticsearch", "log_id_template"): "{dag_id}-{task_id}-{run_id}", + } + ) + def test_from_config_missing_host_keeps_class_default(self): + # An empty [elasticsearch] host must not override the class default with "", which would + # make elasticsearch.Elasticsearch("") raise and silently disable remote logging. + subject = ElasticsearchRemoteLogIO.from_config() + + assert subject.host == "http://localhost:9200" + + def test_provider_registers_elasticsearch_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("elasticsearch") + + assert info is not None + assert ( + info.classpath == "airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchRemoteLogIO" + )