Make OpenSearch log handler respect the per-run log ID template

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -435,6 +435,12 @@ class TIRunContext(BaseModel):
     always reflects when the task *first* started, not when it was rescheduled/resumed.
     """
 
+    log_id_template: str | None = None
+    """
+    Elasticsearch/OpenSearch log id template pinned to this Dag run (``LogTemplate.elasticsearch_id``),
+    so remote log writers use the template that was in effect when the run was created.
+    """
+
 
 class PrevSuccessfulDagRunResponse(BaseModel):
     """Schema for response with previous successful DagRun information for Task Template Context."""

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -76,7 +76,7 @@
     require_auth,
 )
 from airflow.configuration import conf
-from airflow.exceptions import InvalidPartitionKeyError, TaskNotFound
+from airflow.exceptions import AirflowException, InvalidPartitionKeyError, TaskNotFound
 from airflow.models.asset import AssetActive
 from airflow.models.base import ID_LEN
 from airflow.models.dag import DagModel
@@ -310,6 +310,9 @@ def ti_run(
             should_retry=_is_eligible_to_retry(previous_state, ti.try_number, ti.max_tries),
         )
 
+        with contextlib.suppress(AirflowException):  # no LogTemplate row: worker falls back to conf
+            context.log_id_template = dr.get_log_template(session=session).elasticsearch_id
+
         # Only set if they are non-null
         if ti.next_method:
             context.next_method = ti.next_method

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -51,9 +51,14 @@
     AddTeamNameField,
     AddVariableKeysEndpoint,
 )
+from airflow.api_fastapi.execution_api.versions.v2026_09_30 import AddLogIdTemplateField
 
 bundle = VersionBundle(
     HeadVersion(),
+    Version(
+        "2026-09-30",
+        AddLogIdTemplateField,
+    ),
     Version(
         "2026-06-30",
         AddVariableKeysEndpoint,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_09_30.py

@@ -0,0 +1,40 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import (
+    ResponseInfo,
+    VersionChange,
+    convert_response_to_previous_version_for,
+    schema,
+)
+
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import TIRunContext
+
+
+class AddLogIdTemplateField(VersionChange):
+    """Add the Dag-run-pinned `log_id_template` field to TIRunContext."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("log_id_template").didnt_exist,)
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def remove_log_id_template_field(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Remove log_id_template field for older API versions."""
+        response.body.pop("log_id_template", None)

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -278,6 +278,7 @@ def test_ti_run_state_to_running(
             "variables": [],
             "connections": [],
             "xcom_keys_to_clear": [],
+            "log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}",
         }
         # upstream_map_indexes is now computed by Task SDK, not returned by the server in HEAD version
         assert "upstream_map_indexes" not in result
@@ -692,6 +693,7 @@ def test_next_kwargs_still_encoded(self, client, session, create_task_instance,
             "next_method": "execute_complete",
             "next_kwargs": expected_next_kwargs,
             "start_date": None,
+            "log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}",
         }
 
     @pytest.mark.parametrize("resume", [True, False])
@@ -766,6 +768,7 @@ def test_next_kwargs_determines_start_date_update(self, client, session, create_
             "xcom_keys_to_clear": [],
             "next_method": "execute_complete",
             "next_kwargs": expected_next_kwargs,
+            "log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}",
         }
         session.expunge_all()
         ti = session.get(TaskInstance, ti.id)
@@ -1030,6 +1033,41 @@ def test_ti_run_populates_team_name(
         assert response.status_code == 200
         assert response.json()["dag_run"]["team_name"] == (team_name if expect_team else None)
 
+    def test_ti_run_populates_log_id_template(
+        self, client, session, create_task_instance, create_log_template, time_machine
+    ):
+        """``log_id_template`` echoes the LogTemplate row pinned to the Dag run."""
+        instant = timezone.parse("2024-09-30T12:00:00Z")
+        time_machine.move_to(instant, tick=False)
+
+        custom_elasticsearch_id = "{dag_id}-{task_id}-{logical_date}-{try_number}"
+        create_log_template("{{ ti.dag_id }}.log", custom_elasticsearch_id)
+
+        # The Dag run pins the newest LogTemplate row at creation time.
+        ti = create_task_instance(
+            task_id="test_ti_run_populates_log_id_template",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=instant,
+            dag_id=str(uuid4()),
+        )
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": instant.isoformat(),
+            },
+        )
+
+        assert response.status_code == 200
+        assert response.json()["log_id_template"] == custom_elasticsearch_id
+
     def test_ti_run_creates_audit_log(self, client, session, create_task_instance, time_machine):
         """Test that transitioning to RUNNING creates an audit log record."""
         instant_str = "2024-09-30T12:00:00Z"

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_09_30/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_09_30/test_task_instances.py

@@ -0,0 +1,85 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow._shared.timezones import timezone
+from airflow.utils.state import DagRunState, State
+
+from tests_common.test_utils.db import clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+TIMESTAMP_STR = "2024-09-30T12:00:00Z"
+TIMESTAMP = timezone.parse(TIMESTAMP_STR)
+
+RUN_PATCH_BODY = {
+    "state": "running",
+    "hostname": "h",
+    "unixname": "u",
+    "pid": 1,
+    "start_date": TIMESTAMP_STR,
+}
+
+
+@pytest.fixture
+def old_ver_client(client):
+    """Execution API version immediately before ``log_id_template`` was added."""
+    client.headers["Airflow-API-Version"] = "2026-06-30"
+    return client
+
+
+class TestLogIdTemplateFieldBackwardCompat:
+    @pytest.fixture(autouse=True)
+    def _freeze_time(self, time_machine):
+        time_machine.move_to(TIMESTAMP_STR, tick=False)
+
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    def test_old_version_strips_log_id_template(self, old_ver_client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_log_id_template_downgrade",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        session.commit()
+
+        response = old_ver_client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        assert response.status_code == 200
+        assert "log_id_template" not in response.json()
+
+    def test_head_version_includes_log_id_template(self, client, session, create_task_instance):
+        ti = create_task_instance(
+            task_id="test_log_id_template_head",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        session.commit()
+
+        response = client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        assert response.status_code == 200
+        assert response.json()["log_id_template"] == "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}"

providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py

@@ -68,6 +68,8 @@
 else:
     from airflow.utils import timezone  # type: ignore[attr-defined,no-redef]
 
+log = logging.getLogger(__name__)
+
 USE_PER_RUN_LOG_ID = hasattr(DagRun, "get_log_template")
 LOG_LINE_DEFAULTS = {"exc_text": "", "stack_info": ""}
 TASK_LOG_FIELDS = ["timestamp", "event", "level", "chan", "logger", "error_detail", "message", "levelname"]
@@ -222,18 +224,73 @@ def _strip_userinfo(url: str) -> str:
     return parsed._replace(netloc=netloc).geturl()
 
 
+def _clean_date(value: datetime | None) -> str:
+    """
+    Clean up a date value so that it is safe to query in opensearch by removing reserved characters.
+
+    https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html#_reserved_characters
+    """
+    if value is None:
+        return ""
+    return value.strftime("%Y_%m_%dT%H_%M_%S_%f")
+
+
 def _render_log_id(
-    log_id_template: str, ti: TaskInstance | TaskInstanceKey | RuntimeTI, try_number: int
+    log_id_template: str,
+    ti: TaskInstance | TaskInstanceKey | RuntimeTI,
+    try_number: int,
+    *,
+    dag_run: Any = None,
+    json_format: bool = False,
 ) -> str:
+    # Templates pinned by older LogTemplate rows may use date placeholders (e.g. the pre-2.3
+    # default `{dag_id}-{task_id}-{logical_date}-{try_number}`), and str.format raises KeyError
+    # on any missing name, so always supply the full set.
+    def format_date(value: datetime | None) -> str:
+        if json_format:
+            return _clean_date(value)
+        return value.isoformat() if value else ""
+
+    logical_date = format_date(getattr(dag_run, "logical_date", None))
     return log_id_template.format(
         dag_id=ti.dag_id,
         task_id=ti.task_id,
         run_id=getattr(ti, "run_id", ""),
         try_number=try_number,
         map_index=getattr(ti, "map_index", ""),
+        logical_date=logical_date,
+        execution_date=logical_date,
+        data_interval_start=format_date(getattr(dag_run, "data_interval_start", None)),
+        data_interval_end=format_date(getattr(dag_run, "data_interval_end", None)),
     )
 
 
+def _resolve_log_id_template(
+    ti: TaskInstance | TaskInstanceKey | RuntimeTI, default_template: str
+) -> tuple[str, DagRun | None]:
+    """
+    Fetch the Dag-run-pinned log id template and the Dag run from the metadata DB.
+
+    Falls back to the configured template when the DB is not reachable (e.g. on workers, which
+    receive the pinned template through ``TIRunContext`` instead).
+    """
+    if not hasattr(ti, "get_dagrun"):
+        # A worker-side RuntimeTI: don't even open a session, the metadata DB is out of reach.
+        return default_template, None
+    try:
+        with create_session() as session:
+            dag_run = ti.get_dagrun(session=session)  # type: ignore[union-attr]
+            if USE_PER_RUN_LOG_ID:
+                return dag_run.get_log_template(session=session).elasticsearch_id, dag_run
+            return default_template, dag_run
+    except Exception:
+        log.warning(
+            "Could not fetch the Dag-run-pinned log id template; falling back to the configured one",
+            exc_info=True,
+        )
+        return default_template, None
+
+
 def _resolve_nested(hit: dict[Any, Any], parent_class=None) -> type[Hit]:
     """
     Resolve nested hits from OpenSearch by iteratively navigating the `_nested` field.
@@ -864,8 +921,13 @@ def __attrs_post_init__(self):
         self._doc_type_map: dict[Any, Any] = {}
         self._doc_type: list[Any] = []
 
-    def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None) -> None:
-        """Emit structured task logs to stdout and/or write them directly to OpenSearch."""
+    def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None, *, ti_context: Any = None) -> None:
+        """
+        Emit structured task logs to stdout and/or write them directly to OpenSearch.
+
+        :param ti_context: the ``TIRunContext`` the supervisor received at task start, carrying the
+            Dag-run-pinned log id template; when absent, fall back to the configured template.
+        """
         if ti is None:
             return
 
@@ -874,7 +936,14 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI | None = None) -> None:
         if not local_loc.is_file():
             return
 
-        log_id = _render_log_id(self.log_id_template, ti, ti.try_number)  # type: ignore[arg-type]
+        log_id_template = getattr(ti_context, "log_id_template", None) or self.log_id_template
+        log_id = _render_log_id(
+            log_id_template,
+            ti,
+            ti.try_number,
+            dag_run=getattr(ti_context, "dag_run", None),
+            json_format=self.json_format,
+        )
         if self.write_stdout or self.write_to_opensearch:
             log_lines = self._parse_raw_log(local_loc.read_text(), log_id)
 
@@ -932,7 +1001,14 @@ def _write_to_opensearch(self, log_lines: list[dict[str, Any]]) -> bool:
             return False
 
     def read(self, _relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages]:
-        log_id = _render_log_id(self.log_id_template, ti, ti.try_number)  # type: ignore[arg-type]
+        log_id_template, dag_run = _resolve_log_id_template(ti, self.log_id_template)
+        log_id = _render_log_id(
+            log_id_template,
+            ti,
+            ti.try_number,
+            dag_run=dag_run,
+            json_format=self.json_format,
+        )
         self.log.info("Reading log %s from Opensearch", log_id)
         response = self._os_read(log_id, 0, ti)
         if response is not None and response.hits: