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 348795bde9cf1..da2c22a229a8d 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 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 ] 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 46a0e9805cd2f..df24545783410 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 @@ _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()