This is an automated email from the ASF dual-hosted git repository.
eladkal 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 7b3389664a2 Show running task logs in the UI with Elasticsearch remote
logging (#69090)
7b3389664a2 is described below
commit 7b3389664a298bf2e3e5eb13e00173e417cf5508
Author: Jorge Rocamora <[email protected]>
AuthorDate: Sat Jul 4 17:20:28 2026 +0200
Show running task logs in the UI with Elasticsearch remote logging (#69090)
---
.../providers/elasticsearch/log/es_task_handler.py | 11 ++++++
.../unit/elasticsearch/log/test_es_task_handler.py | 39 ++++++++++++++++++++++
2 files changed, 50 insertions(+)
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 3e2fdcc366d..fa05b2a0b34 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
@@ -51,6 +51,7 @@ from airflow.providers.elasticsearch.log.es_response import
ElasticSearchRespons
from airflow.providers.elasticsearch.version_compat import AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_2_PLUS
from airflow.utils.log.file_task_handler import FileTaskHandler
from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
+from airflow.utils.state import TaskInstanceState
if AIRFLOW_V_3_2_PLUS:
from airflow._shared.module_loading import import_string
@@ -388,6 +389,16 @@ class ElasticsearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMix
can be used for steaming log reading and auto-tailing.
:return: a list of tuple with host and log documents, metadata.
"""
+ # In Airflow 3 logs reach Elasticsearch only after the task finishes,
so a running task
+ # has nothing to read there yet. Defer to the base handler for live
worker/executor logs,
+ # as S3/GCS do. Airflow 2 has no remote-log-IO read path, so it keeps
the ES-only path below.
+ if (
+ AIRFLOW_V_3_0_PLUS
+ and ti.try_number == try_number
+ and ti.state in (TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED)
+ ):
+ return super()._read(ti, try_number, metadata) # type:
ignore[return-value]
+
if not metadata:
# LogMetadata(TypedDict) is used as type annotation for
log_reader; added ignore to suppress mypy error
metadata = {"offset": 0} # type: ignore[assignment]
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 56c60214089..b923d8b7c6f 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
@@ -48,6 +48,7 @@ from airflow.providers.elasticsearch.log.es_task_handler
import (
getattr_nested,
)
from airflow.utils import timezone
+from airflow.utils.log.file_task_handler import FileTaskHandler
from airflow.utils.state import DagRunState, TaskInstanceState
from airflow.utils.timezone import datetime
@@ -270,6 +271,9 @@ class TestElasticsearchTaskHandler:
@pytest.mark.db_test
@pytest.mark.parametrize("metadata_mode", ["provided", "none", "empty"])
def test_read(self, ti, metadata_mode):
+ # A finished task reads from Elasticsearch directly. A running task is
delegated to the
+ # base handler (covered by
test_read_running_task_delegates_to_base_handler).
+ ti.state = TaskInstanceState.SUCCESS
start_time = pendulum.now()
response = _make_es_response(self.es_task_handler.io,
self.base_log_source)
@@ -298,6 +302,7 @@ class TestElasticsearchTaskHandler:
@pytest.mark.db_test
def test_read_defaults_offset_when_missing_from_metadata(self, ti):
+ ti.state = TaskInstanceState.SUCCESS
start_time = pendulum.now()
with patch.object(self.es_task_handler.io, "_es_read",
return_value=None):
logs, metadatas = self.es_task_handler.read(ti, 1, {"end_of_log":
False})
@@ -310,6 +315,7 @@ class TestElasticsearchTaskHandler:
@pytest.mark.db_test
@pytest.mark.parametrize("seconds", [3, 6])
def test_read_missing_logs(self, ti, seconds):
+ ti.state = TaskInstanceState.SUCCESS
start_time = pendulum.now().add(seconds=-seconds)
with patch.object(self.es_task_handler.io, "_es_read",
return_value=None):
logs, metadatas = self.es_task_handler.read(
@@ -330,6 +336,7 @@ class TestElasticsearchTaskHandler:
@pytest.mark.db_test
def test_read_timeout(self, ti):
+ ti.state = TaskInstanceState.SUCCESS
start_time = pendulum.now().subtract(minutes=5)
with patch.object(self.es_task_handler.io, "_es_read",
return_value=None):
logs, metadatas = self.es_task_handler.read(
@@ -349,6 +356,7 @@ class TestElasticsearchTaskHandler:
@pytest.mark.db_test
def test_read_with_custom_offset_and_host_fields(self, ti):
+ ti.state = TaskInstanceState.SUCCESS
self.es_task_handler.host_field = "host.name"
self.es_task_handler.offset_field = "log.offset"
self.es_task_handler.io.host_field = "host.name"
@@ -380,6 +388,37 @@ class TestElasticsearchTaskHandler:
assert metadata["offset"] == "1"
assert not metadata["end_of_log"]
+ @pytest.mark.db_test
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Live-log delegation
only applies to Airflow 3")
+ @pytest.mark.parametrize("state", [TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED])
+ def test_read_running_task_delegates_to_base_handler(self, ti, state):
+ ti.state = state
+ base_result = (["live log line"], {"end_of_log": False})
+ with (
+ patch.object(FileTaskHandler, "_read", return_value=base_result)
as base_read,
+ patch.object(self.es_task_handler.io, "_es_read") as es_read,
+ ):
+ result = self.es_task_handler._read(ti, 1, {})
+
+ assert result == base_result
+ base_read.assert_called_once()
+ es_read.assert_not_called()
+
+ @pytest.mark.db_test
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Live-log delegation
only applies to Airflow 3")
+ def test_read_old_try_of_running_task_does_not_delegate(self, ti):
+ ti.state = TaskInstanceState.RUNNING
+ ti.try_number = 2
+ response = _make_es_response(self.es_task_handler.io,
self.base_log_source)
+ with (
+ patch.object(FileTaskHandler, "_read") as base_read,
+ patch.object(self.es_task_handler.io, "_es_read",
return_value=response) as es_read,
+ ):
+ self.es_task_handler.read(ti, 1, {"offset": 0})
+
+ base_read.assert_not_called()
+ es_read.assert_called_once()
+
@pytest.mark.db_test
def test_set_context(self, ti):
self.es_task_handler.set_context(ti)
