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 781b6b9f85e Prevent malformed OpenSearch log entries from crashing
task log fetch (#69307)
781b6b9f85e is described below
commit 781b6b9f85e4413f5291ac3acd1c87707eadfee0
Author: Hemkumar Chheda <[email protected]>
AuthorDate: Wed Jul 15 20:03:34 2026 +0530
Prevent malformed OpenSearch log entries from crashing task log fetch
(#69307)
* Prevent malformed OpenSearch log entries from crashing task log fetch
A single stored log entry with a non-string `event` field (for example,
a task that logs a list or dict as the sole message argument) currently
crashes the entire task-log-fetch request with an unhandled
pydantic.ValidationError, instead of degrading gracefully.
_read() built StructuredLogMessage objects from stored OpenSearch hits
without catching validation failures. This catches ValidationError per
hit and falls back to a stringified event, matching the existing
fallback pattern in file_task_handler.py's _log_stream_to_parsed_log_stream.
Related: #69304
* Lower malformed-log fallback logging to debug level
_safe_build_structured_log_message runs once per stored hit inside the
_read comprehension, so logging the fallback at warning level fires once
per malformed line within a single log-fetch request. Drop it to debug
so a request with many malformed entries does not flood the logs.
* Set task state to SUCCESS in malformed-event read test
The shared ti fixture defaults to RUNNING, and _read() now delegates
running/deferred tasks to the base handler's remote-log path, which is
not wired up in unit tests. Match the sibling read tests by setting the
task to SUCCESS so the test exercises the stored-log fallback path.
---
.../providers/opensearch/log/os_task_handler.py | 28 +++++++++++++-
.../unit/opensearch/log/test_os_task_handler.py | 44 ++++++++++++++++++++++
2 files changed, 71 insertions(+), 1 deletion(-)
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 77ed6830e8b..a6fee7ba601 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
@@ -36,6 +36,7 @@ import attrs
import pendulum
from opensearchpy import OpenSearch, helpers
from opensearchpy.exceptions import NotFoundError
+from pydantic import ValidationError
from sqlalchemy import select
import airflow.logging_config as alc
@@ -72,6 +73,8 @@ 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"]
+logger = logging.getLogger(__name__)
+
def _format_error_detail(error_detail: Any) -> str | None:
"""Render the structured ``error_detail`` written by the Airflow 3
supervisor as a traceback string."""
@@ -121,6 +124,29 @@ def _build_log_fields(hit_dict: dict[str, Any]) ->
dict[str, Any]:
return fields
+def _safe_build_structured_log_message(hit_dict: dict[str, Any]) ->
StructuredLogMessage:
+ """
+ Build a StructuredLogMessage from a stored OpenSearch hit, tolerating
malformed fields.
+
+ A single malformed stored log entry (for example a non-string ``event``
produced by
+ logging a list or dict as the sole message argument) must not fail the
entire
+ log-fetch request. Fall back to a stringified event, mirroring the
fallback used for
+ unparsable raw log lines in ``_log_stream_to_parsed_log_stream``.
+ """
+ fields = _build_log_fields(hit_dict)
+ try:
+ return StructuredLogMessage(**fields)
+ except ValidationError:
+ logger.debug(
+ "Failed to parse stored log entry into StructuredLogMessage;
falling back to "
+ "stringified event. Offending fields: %s",
+ fields,
+ )
+ return StructuredLogMessage(
+ event=str(fields.get("event", hit_dict)),
timestamp=fields.get("timestamp")
+ )
+
+
def getattr_nested(obj, item, default):
"""
Get item from obj but return default if not found.
@@ -636,7 +662,7 @@ class OpensearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMixin)
# Flatten all hits, filter to only desired fields, and
construct StructuredLogMessage objects
message = header + [
- StructuredLogMessage(**_build_log_fields(hit.to_dict()))
+ _safe_build_structured_log_message(hit.to_dict())
for hits in logs_by_host.values()
for hit in hits
]
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 f9d9fa89426..e9464e53e3d 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
@@ -38,6 +38,7 @@ from airflow.providers.opensearch.log.os_task_handler import (
_build_log_fields,
_format_error_detail,
_render_log_id,
+ _safe_build_structured_log_message,
_strip_userinfo,
get_os_kwargs_from_config,
getattr_nested,
@@ -404,6 +405,32 @@ class TestOpensearchTaskHandler:
assert metadata["offset"] == "1"
assert not metadata["end_of_log"]
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="StructuredLogMessage
fallback is Airflow 3+ only")
+ @pytest.mark.db_test
+ def test_read_with_malformed_event_falls_back_to_stringified_event(self,
ti):
+ ti.state = TaskInstanceState.SUCCESS
+ malformed_event = ["not", "a", "string"]
+ malformed_source = {
+ "message": self.test_message,
+ "event": malformed_event,
+ "log_id": self.LOG_ID,
+ "offset": 2,
+ }
+ response = _make_os_response(self.os_task_handler.io,
self.base_log_source, malformed_source)
+
+ with patch.object(self.os_task_handler.io, "_os_read",
return_value=response):
+ with
patch("airflow.providers.opensearch.log.os_task_handler.logger") as mock_logger:
+ logs, metadatas = self.os_task_handler.read(ti, 1)
+
+ metadata = _assert_log_events(
+ logs,
+ metadatas,
+ expected_events=[self.test_message, str(malformed_event)],
+ expected_sources=["http://localhost";],
+ )
+ assert not metadata["end_of_log"]
+ mock_logger.debug.assert_called_once()
+
@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])
@@ -877,3 +904,20 @@ class TestBuildStructuredLogFields:
hit = {"event": "msg", "error_detail": []}
result = _build_log_fields(hit)
assert "error_detail" not in result
+
+
+class TestSafeBuildStructuredLogMessage:
+ def test_string_event_returns_unchanged_and_does_not_log(self):
+ hit = {"event": "hello", "level": "info"}
+ with patch("airflow.providers.opensearch.log.os_task_handler.logger")
as mock_logger:
+ result = _safe_build_structured_log_message(hit)
+ assert result.event == "hello"
+ mock_logger.debug.assert_not_called()
+
+ def test_non_string_event_falls_back_to_stringified_event(self):
+ hit = {"event": ["a", "b"], "timestamp": "2024-01-01T00:00:00Z"}
+ with patch("airflow.providers.opensearch.log.os_task_handler.logger")
as mock_logger:
+ result = _safe_build_structured_log_message(hit)
+ assert result.event == str(["a", "b"])
+ assert result.timestamp is not None
+ mock_logger.debug.assert_called_once()
