Prevent malformed Elasticsearch log entries from crashing task log fetch

[hkc-8010]

Summary

• 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 change 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.

Changes

• providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py:
  wrap StructuredLogMessage construction in a try/except ValidationError with a
  stringified-event fallback and a warning log.
• providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py: add
  coverage for malformed event handling and for read() surviving one bad hit among
  otherwise-valid hits.

PR Checklist

• My PR is targeted at the main branch
• Tests added/updated
• prek hooks all pass
• Unit tests pass locally

closes: #69304

[Vamsi-klu]

Could the fallback avoid reusing the original timestamp when validation failed? Right now any ValidationError falls into the except block, but then it builds StructuredLogMessage(..., timestamp=fields.get("timestamp")). If the malformed field was timestamp rather than event, the fallback can raise the same validation error again and still fail the whole log fetch.

---

Drafted-by: Codex (GPT-5); reviewed by @Vamsi-klu before posting

[jason810496]

LGTM except the logging level, thanks.

[potiuk]

@hkc-8010 — Some review feedback from @jason810496 is waiting on you: 1 unresolved review thread(s). Could you either push a fix or reply in each thread explaining why the feedback doesn't apply? When you believe the feedback is addressed, please mark the threads as resolved and ping the reviewer (@jason810496) for a final look. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[hkc-8010]

@Vamsi-klu Good catch on the code path. I left the timestamp as-is on purpose here: in both handlers the timestamp is not user data. It is written by ElasticsearchJSONFormatter.formatTime() (OpenSearch uses the same pattern) from record.created, so it is always a valid ISO-8601 string. The only field a user can put arbitrary values in is event, for example logging a list or dict, which is what the fallback and the test cover. A timestamp that fails validation would have to come from documents written into the index by something outside Airflow's own formatter. That is out of scope for this fix, so I would rather keep the fallback narrow. If we want to harden against externally ingested log docs, I am happy to do that as a follow-up.