Store error message for streaming job execution in Flint metadata log

[dai-chen]

Description

This PR enhances the Flint Spark Index monitor by storing the error messages in the Flint index metadata log. Previously, error messages were not captured in the metadata log, which made it difficult to diagnose issues when they occurred. Specifically, this PR converts the ErrorMessages utility class to Java code and introduces a new method to extract, redact, and truncate error messages.

TODO

1. Consider merging with the logic below (this may happen after FlintSparkIndexMonitor is moved out of Flint core, or the logic below is moved to core):

   opensearch-spark/spark-sql-application/src/main/scala/org/apache/spark/sql/FlintJobExecutor.scala

   Line 454 in f7aaa41

   def processQueryException(ex: Exception): String = {

2. Currently, there is a limitation where no error message is stored if the streaming job exits early. In Spark 4.0, StreamingQueryListener provides exceptions, which can help remove this limitation.

Example

Create a Flint index:

CREATE INDEX streaming_error ON glue.default.http_logs (
  clientip
)
WITH (
  auto_refresh = true,
  checkpoint_location = 's3://checkpoint/streaming-error'
);

Change index setting to block write for the index:

PUT flint_glue_default_http_logs_streaming_error_index/_block/write

Check error message in metadata log (empty before the changes)

  ...
        "_source": {
          "version": "1.0",
          "latestId": "ZmxpbnRfZ2x1ZV9kZWZhdWx0X2h0dHBfbG9nc19zdHJlYW1pbmdfZXJyb3JfM19pbmRleA==",
          "type": "flintindexstate",
          "state": "failed",
          "applicationId": "XXXXXX",
          "jobId": "XXXXXX",
          "dataSourceName": "glue",
          "jobStartTime": 1721150179877,
          "lastUpdateTime": 1721150641171,
          "error": """failure in bulk execution:
[0]: index [flint_glue_default_http_logs_streaming_error_index], id [kqSSvJAB69S8YQePXjzu], message [OpenSearchException[OpenSearch exception [type=cluster_block_exception, reason=index [flint_glue_default_http_logs_streaming_error_index] blocked by: [FORBIDDEN/8/index write (api)];]]]
[1]: index [flint_glue_default_http_logs_streaming_error_index], id [k6SSvJAB69S8YQePXjzu], message [OpenSearchException[OpenSearch exception [type=cluster_block_exception, reason=index [flint_glue_default_http_logs_streaming_error_index] blocked by: [FORBIDDEN/8/index write (api)];]]]
[2]: index [flint_glue_default_http_logs_streaming_error_index], id [lKSSvJAB69S8YQePXjzu], message [OpenSearchException[OpenSearch exception [type=cluster_block_exception, reason=index [flint_glue_default_http_logs_streaming_error_index] blocked by: [FORBIDDEN/8/index write (api)];]]]
[3]: index [flint_glue_default_http_logs_streaming_error_index], id [laSSvJAB69S8YQePXjzu], messag..."""
        }
  ...

Issues Resolved

#405

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[noCharger]

Just curious, is it possible to extend https://github.com/opensearch-project/opensearch-spark/blob/main/flint-core/src/main/java/org/opensearch/flint/core/logging/OperationMessage.java to support redaction?

[dai-chen]

Will have a look.

[dai-chen]

I don't see this is relevant directly and it seems binding to Log4j API. May reconsider when we refactor the entire extractRootCause and processQueryException after #435. Thanks!