This is an automated email from the ASF dual-hosted git repository.
potiuk 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 cb23988b8bd Make _parse_raw_log resilient to malformed or non-JSON log
lines by introducing best-effort parsing with a fallback structure. Add unit
tests. (#66383)
cb23988b8bd is described below
commit cb23988b8bd8715280e26fe5825738e3324de00d
Author: SameerMesiah97 <[email protected]>
AuthorDate: Sun May 10 03:20:48 2026 +0100
Make _parse_raw_log resilient to malformed or non-JSON log lines by
introducing best-effort parsing with a fallback structure. Add unit tests.
(#66383)
Co-authored-by: Sameer Mesiah <[email protected]>
---
.../providers/elasticsearch/log/es_task_handler.py | 30 ++++++++++++---
.../unit/elasticsearch/log/test_es_task_handler.py | 44 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 6 deletions(-)
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 7e0255677a8..eb531452df8 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
@@ -695,13 +695,31 @@ class ElasticsearchRemoteLogIO(LoggingMixin): # noqa:
D101
offset = 1
for line in logs:
# Make sure line is not empty
- if line.strip():
- # construct log_id which is
{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}
- # also construct the offset field (default is 'offset')
+ if not line.strip():
+ continue
+
+ try:
log_dict = json.loads(line)
- log_dict.update({"log_id": log_id, self.offset_field: offset})
- offset += 1
- parsed_logs.append(log_dict)
+ except json.JSONDecodeError:
+ # Best-effort fallback: preserve the raw line
+ self.log.debug("Failed to parse log line as JSON",
exc_info=True)
+ log_dict = {
+ "message": line,
+ "unparsed": True,
+ }
+
+ # Ensure minimal compatibility with Airflow log expectations
+ if "event" not in log_dict and "message" not in log_dict:
+ log_dict["message"] = str(line)
+
+ log_dict.update(
+ {
+ "log_id": log_id,
+ self.offset_field: offset,
+ }
+ )
+ offset += 1
+ parsed_logs.append(log_dict)
return parsed_logs
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 14a1e951e88..3ac143fa35b 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
@@ -669,6 +669,50 @@ class TestElasticsearchRemoteLogIO:
assert [line["offset"] for line in json_log_lines] == [1, 2, 3]
assert all(line["log_id"] == log_id for line in json_log_lines)
+ def test_raw_log_handles_invalid_json_line(self, ti):
+
+ raw_log = '{"message": "ok"}\nINVALID_JSON\n{"message": "ok2"}\n'
+ log_id = _render_log_id(self.elasticsearch_io.log_id_template, ti,
ti.try_number)
+
+ json_log_lines = self.elasticsearch_io._parse_raw_log(raw_log, log_id)
+
+ assert len(json_log_lines) == 3
+
+ assert json_log_lines[1]["message"] == "INVALID_JSON"
+ assert json_log_lines[1]["unparsed"] is True
+
+ assert [line["offset"] for line in json_log_lines] == [1, 2, 3]
+
+ def test_raw_log_all_plain_text(self, ti):
+ raw_log = "line1\nline2\nline3\n"
+ log_id = _render_log_id(self.elasticsearch_io.log_id_template, ti,
ti.try_number)
+
+ json_log_lines = self.elasticsearch_io._parse_raw_log(raw_log, log_id)
+
+ assert len(json_log_lines) == 3
+ assert all(line["unparsed"] for line in json_log_lines)
+ assert [line["message"] for line in json_log_lines] == ["line1",
"line2", "line3"]
+
+ def test_raw_log_mixed_content(self, ti):
+ raw_log = '{"event": "ok"}\nplain text\n{"message": "done"}\n'
+ log_id = _render_log_id(self.elasticsearch_io.log_id_template, ti,
ti.try_number)
+
+ json_log_lines = self.elasticsearch_io._parse_raw_log(raw_log, log_id)
+
+ assert json_log_lines[0]["event"] == "ok"
+ assert json_log_lines[1]["message"] == "plain text"
+ assert json_log_lines[1]["unparsed"] is True
+ assert json_log_lines[2]["message"] == "done"
+
+ def test_raw_log_ignores_empty_lines(self, ti):
+ raw_log = '\n{"message": "ok"}\n\n'
+ log_id = _render_log_id(self.elasticsearch_io.log_id_template, ti,
ti.try_number)
+
+ json_log_lines = self.elasticsearch_io._parse_raw_log(raw_log, log_id)
+
+ assert len(json_log_lines) == 1
+ assert json_log_lines[0]["message"] == "ok"
+
def test_get_source_includes(self):
assert self.elasticsearch_io._get_source_includes() == [
"@timestamp",
