hkc-8010 commented on code in PR #69306:
URL: https://github.com/apache/airflow/pull/69306#discussion_r3570667592
##########
providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py:
##########
@@ -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.warning(
+ "Failed to parse stored log entry into StructuredLogMessage;
falling back to "
+ "stringified event. Offending fields: %s",
+ fields,
+ )
Review Comment:
Good point, lowered it to `debug`. Since this runs per hit inside `_read`, a
single request with many malformed entries would have logged one warning per
line. Pushed the change, mind taking another look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
