Add ElasticsearchRemoteLogIO.from_config and register elasticsearch scheme

providers/elasticsearch/docs/logging/index.rst

@@ -37,6 +37,19 @@ First, to use the handler, ``airflow.cfg`` must be configured as follows:
     [elasticsearch]
     host = <host>:<port>
 
+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 = <host>:<port>
+
 To output task logs to stdout in JSON format, the following config could be used:
 
 .. code-block:: ini

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: ~

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,

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))

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"
+        )