This is an automated email from the ASF dual-hosted git repository.
jason810496 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 3986d828ec3 Prevent malformed Elasticsearch log entries from crashing
task log fetch (#69306)
3986d828ec3 is described below
commit 3986d828ec377d70e5c63252da61ab172916f1e9
Author: Hemkumar Chheda <[email protected]>
AuthorDate: Wed Jul 15 12:46:26 2026 +0530
Prevent malformed Elasticsearch log entries from crashing task log fetch
(#69306)
* Prevent malformed Elasticsearch 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 Elasticsearch
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.
closes: #69304
* Fix Elasticsearch compat tests on Airflow 2.11
The new StructuredLogMessage fallback tests exercise an Airflow 3-only
type, but the provider compatibility matrix still runs this provider against
Airflow 2.11.1. Guarding those tests the same way as the runtime path keeps the
compatibility job focused on behavior that actually exists in that release line.
* Fix Elasticsearch compat test for finished tasks
The malformed log-entry regression test was unintentionally exercising
Airflow 3's live-log delegation path because the test task instance was still
running. Marking the task instance successful keeps the test on the
Elasticsearch-read path it is intended to cover, which matches the provider
compatibility and lowest-dependencies jobs.
* 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.
---
.../providers/elasticsearch/log/es_task_handler.py | 28 +++++++++++++-
.../unit/elasticsearch/log/test_es_task_handler.py | 45 ++++++++++++++++++++++
2 files changed, 72 insertions(+), 1 deletion(-)
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 fa05b2a0b34..2fb610186e9 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
@@ -40,6 +40,7 @@ import elasticsearch
import pendulum
from elasticsearch import helpers
from elasticsearch.exceptions import NotFoundError
+from pydantic import ValidationError
import airflow.logging_config as alc
from airflow.exceptions import AirflowProviderDeprecationWarning
@@ -76,6 +77,8 @@ else:
EsLogMsgType = list[tuple[str, str]] # type: ignore[assignment,misc]
+logger = logging.getLogger(__name__)
+
LOG_LINE_DEFAULTS = {"exc_text": "", "stack_info": ""}
# Elasticsearch hosted log type
@@ -140,6 +143,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 Elasticsearch 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")
+ )
+
+
VALID_ES_CONFIG_KEYS =
set(inspect.signature(elasticsearch.Elasticsearch.__init__).parameters.keys())
# Remove `self` from the valid set of kwargs
VALID_ES_CONFIG_KEYS.remove("self")
@@ -472,7 +498,7 @@ class ElasticsearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMix
# 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/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
index b923d8b7c6f..9f7091d90bb 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
@@ -43,6 +43,7 @@ from airflow.providers.elasticsearch.log.es_task_handler
import (
_clean_date,
_format_error_detail,
_render_log_id,
+ _safe_build_structured_log_message,
_strip_userinfo,
get_es_kwargs_from_config,
getattr_nested,
@@ -388,6 +389,32 @@ class TestElasticsearchTaskHandler:
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_es_response(self.es_task_handler.io,
self.base_log_source, malformed_source)
+
+ with patch.object(self.es_task_handler.io, "_es_read",
return_value=response):
+ with
patch("airflow.providers.elasticsearch.log.es_task_handler.logger") as
mock_logger:
+ logs, metadatas = self.es_task_handler.read(ti, 1)
+
+ metadata = _assert_log_events(
+ logs,
+ metadatas,
+ expected_events=[self.test_message, str(malformed_event)],
+ expected_sources=["http://localhost:9200";],
+ )
+ 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])
@@ -973,3 +1000,21 @@ class TestBuildStructuredLogFields:
hit = {"event": "msg", "error_detail": []}
result = _build_log_fields(hit)
assert "error_detail" not in result
+
+
[email protected](not AIRFLOW_V_3_0_PLUS, reason="StructuredLogMessage
fallback is Airflow 3+ only")
+class TestSafeBuildStructuredLogMessage:
+ def test_string_event_returns_unchanged_and_does_not_log(self):
+ hit = {"event": "hello", "level": "info"}
+ with
patch("airflow.providers.elasticsearch.log.es_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.elasticsearch.log.es_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()
