Fix the write-to-es feature

airflow-core/src/airflow/config_templates/airflow_local_settings.py

@@ -279,35 +279,29 @@ def _default_conn_name_from(mod_path, hook_name):
         )
         remote_task_handler_kwargs = {}
     elif ELASTICSEARCH_HOST:
-        ELASTICSEARCH_END_OF_LOG_MARK: str = conf.get_mandatory_value("elasticsearch", "END_OF_LOG_MARK")
-        ELASTICSEARCH_FRONTEND: str = conf.get_mandatory_value("elasticsearch", "frontend")
+        from airflow.providers.elasticsearch.log.es_task_handler import ElasticsearchRemoteLogIO
+
         ELASTICSEARCH_WRITE_STDOUT: bool = conf.getboolean("elasticsearch", "WRITE_STDOUT")
         ELASTICSEARCH_WRITE_TO_ES: bool = conf.getboolean("elasticsearch", "WRITE_TO_ES")
         ELASTICSEARCH_JSON_FORMAT: bool = conf.getboolean("elasticsearch", "JSON_FORMAT")
-        ELASTICSEARCH_JSON_FIELDS: str = conf.get_mandatory_value("elasticsearch", "JSON_FIELDS")
         ELASTICSEARCH_TARGET_INDEX: str = conf.get_mandatory_value("elasticsearch", "TARGET_INDEX")
         ELASTICSEARCH_HOST_FIELD: str = conf.get_mandatory_value("elasticsearch", "HOST_FIELD")
         ELASTICSEARCH_OFFSET_FIELD: str = conf.get_mandatory_value("elasticsearch", "OFFSET_FIELD")
+        ELASTICSEARCH_LOG_ID_TEMPLATE: str = conf.get_mandatory_value("elasticsearch", "LOG_ID_TEMPLATE")
+
+        REMOTE_TASK_LOG = ElasticsearchRemoteLogIO(
+            host=ELASTICSEARCH_HOST,
+            target_index=ELASTICSEARCH_TARGET_INDEX,
+            write_stdout=ELASTICSEARCH_WRITE_STDOUT,
+            write_to_es=ELASTICSEARCH_WRITE_TO_ES,
+            offset_field=ELASTICSEARCH_OFFSET_FIELD,
+            host_field=ELASTICSEARCH_HOST_FIELD,
+            base_log_folder=BASE_LOG_FOLDER,
+            delete_local_copy=delete_local_copy,
+            json_format=ELASTICSEARCH_JSON_FORMAT,
+            log_id_template=ELASTICSEARCH_LOG_ID_TEMPLATE,
+        )
 
-        ELASTIC_REMOTE_HANDLERS: dict[str, dict[str, str | bool | None]] = {
-            "task": {
-                "class": "airflow.providers.elasticsearch.log.es_task_handler.ElasticsearchTaskHandler",
-                "formatter": "airflow",
-                "base_log_folder": BASE_LOG_FOLDER,
-                "end_of_log_mark": ELASTICSEARCH_END_OF_LOG_MARK,
-                "host": ELASTICSEARCH_HOST,
-                "frontend": ELASTICSEARCH_FRONTEND,
-                "write_stdout": ELASTICSEARCH_WRITE_STDOUT,
-                "write_to_es": ELASTICSEARCH_WRITE_TO_ES,
-                "target_index": ELASTICSEARCH_TARGET_INDEX,
-                "json_format": ELASTICSEARCH_JSON_FORMAT,
-                "json_fields": ELASTICSEARCH_JSON_FIELDS,
-                "host_field": ELASTICSEARCH_HOST_FIELD,
-                "offset_field": ELASTICSEARCH_OFFSET_FIELD,
-            },
-        }
-
-        DEFAULT_LOGGING_CONFIG["handlers"].update(ELASTIC_REMOTE_HANDLERS)
     elif OPENSEARCH_HOST:
         OPENSEARCH_END_OF_LOG_MARK: str = conf.get_mandatory_value("opensearch", "END_OF_LOG_MARK")
         OPENSEARCH_PORT: str = conf.get_mandatory_value("opensearch", "PORT")

chart/values.yaml

@@ -3023,7 +3023,7 @@ config:
     run_duration: 41460
   elasticsearch:
     json_format: 'True'
-    log_id_template: "{dag_id}_{task_id}_{execution_date}_{try_number}"
+    log_id_template: "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}"
   elasticsearch_configs:
     max_retries: 3
     timeout: 30

devel-common/pyproject.toml

@@ -139,6 +139,7 @@ dependencies = [
     "pytest-unordered>=0.6.1",
     "pytest-xdist>=3.5.0",
     "pytest>=9.0.0",
+    "testcontainers>=4.12.0",
 ]
 "sentry" = [
     "blinker>=1.7.0",

providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_response.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterator
+from typing import Any
 
 
 def _wrap(val):
@@ -25,6 +26,33 @@ def _wrap(val):
     return val
 
 
+def resolve_nested(self, hit: dict[Any, Any], parent_class=None) -> type[Hit]:
+    """
+    Resolve nested hits from Elasticsearch by iteratively navigating the `_nested` field.
+
+    The result is used to fetch the appropriate document class to handle the hit.
+
+    This method can be used with nested Elasticsearch fields which are structured
+    as dictionaries with "field" and "_nested" keys.
+    """
+    doc_class = Hit
+
+    nested_path: list[str] = []
+    nesting = hit["_nested"]
+    while nesting and "field" in nesting:
+        nested_path.append(nesting["field"])
+        nesting = nesting.get("_nested")
+    nested_path_str = ".".join(nested_path)
+
+    if hasattr(parent_class, "_index"):
+        nested_field = parent_class._index.resolve_field(nested_path_str)
+
+    if nested_field is not None:
+        return nested_field._doc_class
+
+    return doc_class
+
+
 class AttributeList:
     """Helper class to provide attribute like access to List objects."""