Prevent malformed OpenSearch 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 OpenSearch 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. The identical pattern was
  independently fixed in the Elasticsearch provider in Prevent malformed Elasticsearch log entries from crashing task log fetch #69306.

Changes

• providers/opensearch/src/airflow/providers/opensearch/log/os_task_handler.py:
  wrap StructuredLogMessage construction in a try/except ValidationError with a
  stringified-event fallback and a warning log.
• providers/opensearch/tests/unit/opensearch/log/test_os_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

Related: #69304

[eladkal]

cc @Owen-CH-Leung

[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

[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 OpensearchJSONFormatter.formatTime() (Elasticsearch 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.