Prevent malformed OpenSearch log entries from crashing task log fetch

providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py

@@ -36,6 +36,7 @@
 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
@@ -71,6 +72,8 @@
 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."""
@@ -120,6 +123,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.
@@ -625,7 +651,7 @@ def _read(
 
                 # 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
                 ]

providers/opensearch/tests/unit/opensearch/log/test_os_task_handler.py

@@ -38,6 +38,7 @@
     _build_log_fields,
     _format_error_detail,
     _render_log_id,
+    _safe_build_structured_log_message,
     _strip_userinfo,
     get_os_kwargs_from_config,
     getattr_nested,
@@ -396,6 +397,32 @@ def test_read_with_custom_offset_and_host_fields(self, ti):
         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
     def test_set_context(self, ti):
         self.os_task_handler.set_context(ti)
@@ -838,3 +865,20 @@ def test_error_detail_dropped_when_empty(self):
         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()