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 e4fdf692ece Show running task logs in the UI with OpenSearch remote
logging (#69407)
e4fdf692ece is described below
commit e4fdf692ece6a906626caf6b4bfd09f67ed3b247
Author: Jorge Rocamora <[email protected]>
AuthorDate: Mon Jul 6 17:42:30 2026 +0200
Show running task logs in the UI with OpenSearch remote logging (#69407)
---
.../providers/opensearch/log/os_task_handler.py | 11 ++++++
.../unit/opensearch/log/test_os_task_handler.py | 39 ++++++++++++++++++++++
2 files changed, 50 insertions(+)
diff --git
a/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
b/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
index 348795bde9c..77ed6830e8b 100644
---
a/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
+++
b/providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py
@@ -49,6 +49,7 @@ from airflow.providers.opensearch.version_compat import
AIRFLOW_V_3_0_PLUS, AIRF
from airflow.utils.log.file_task_handler import FileTaskHandler
from airflow.utils.log.logging_mixin import ExternalLoggingMixin, LoggingMixin
from airflow.utils.session import create_session
+from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
@@ -550,6 +551,16 @@ class OpensearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMixin)
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 OpenSearch 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.
+ 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/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
b/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
index 46a0e9805cd..f9d9fa89426 100644
--- a/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
+++ b/providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py
@@ -43,6 +43,7 @@ from airflow.providers.opensearch.log.os_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
@@ -290,6 +291,9 @@ class TestOpensearchTaskHandler:
@pytest.mark.db_test
@pytest.mark.parametrize("metadata_mode", ["provided", "none", "empty"])
def test_read(self, ti, metadata_mode):
+ # A finished task reads from OpenSearch 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_os_response(self.os_task_handler.io,
self.base_log_source)
@@ -318,6 +322,7 @@ class TestOpensearchTaskHandler:
@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.os_task_handler.io, "_os_read",
return_value=None):
logs, metadatas = self.os_task_handler.read(ti, 1, {"end_of_log":
False})
@@ -330,6 +335,7 @@ class TestOpensearchTaskHandler:
@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.os_task_handler.io, "_os_read",
return_value=None):
logs, metadatas = self.os_task_handler.read(
@@ -350,6 +356,7 @@ class TestOpensearchTaskHandler:
@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.os_task_handler.io, "_os_read",
return_value=None):
logs, metadatas = self.os_task_handler.read(
@@ -365,6 +372,7 @@ class TestOpensearchTaskHandler:
@pytest.mark.db_test
def test_read_with_custom_offset_and_host_fields(self, ti):
+ ti.state = TaskInstanceState.SUCCESS
self.os_task_handler.host_field = "host.name"
self.os_task_handler.offset_field = "log.offset"
self.os_task_handler.io.host_field = "host.name"
@@ -396,6 +404,37 @@ class TestOpensearchTaskHandler:
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.os_task_handler.io, "_os_read") as os_read,
+ ):
+ result = self.os_task_handler._read(ti, 1, {})
+
+ assert result == base_result
+ base_read.assert_called_once()
+ os_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_os_response(self.os_task_handler.io,
self.base_log_source)
+ with (
+ patch.object(FileTaskHandler, "_read") as base_read,
+ patch.object(self.os_task_handler.io, "_os_read",
return_value=response) as os_read,
+ ):
+ self.os_task_handler.read(ti, 1, {"offset": 0})
+
+ base_read.assert_not_called()
+ os_read.assert_called_once()
+
@pytest.mark.db_test
def test_set_context(self, ti):
self.os_task_handler.set_context(ti)
