[BUG] Can't connect to OpenSearch if Build Flavor is not returned

[kclinden]

Describe the bug
Pipeline failing to connect to OpenSearch when BuildFlavor is not returned from root OpenSearch api. Source seems to point to this line - https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/SearchAccessorStrategy.java#L198

Do we need to get buildFlavor?

To Reproduce
Steps to reproduce the behavior:
Pipeline:

version: '2'
opensearch-source-pipeline:
  source:
    opensearch:
      hosts: ['https://192.168.1.100:9200']
      username: 'admin'
      password: 'somepass'
      indices:
        include:
          - index_name_regex: 'wazuh-alerts-4.x*'
      scheduling:
        interval: 'PT5M'
      connection:
        insecure: true
  sink:
    - stdout:

Expected behavior
Pipeline connects

Additional context
OpenSearch API Return Data:

{
  "name": "node-1",
  "cluster_name": "wazuh-cluster",
  "cluster_uuid": "EzI18fqRRsqyp_bqBT0xyQ",
  "version": {
    "number": "7.10.2",
    "build_type": "rpm",
    "build_hash": "eee49cb340edc6c4d489bcd9324dda571fc8dc03",
    "build_date": "2023-09-20T23:54:29.889267151Z",
    "build_snapshot": false,
    "lucene_version": "9.7.0",
    "minimum_wire_compatibility_version": "7.10.0",
    "minimum_index_compatibility_version": "7.0.0"
  },
  "tagline": "The OpenSearch Project: https://opensearch.org/"
}

Error:

2024-06-21T18:14:35,773 [opensearch-source-pipeline-sink-worker-2-thread-1] ERROR org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor - Pipeline [opensearch-source-pipeline] process worker encountered a fatal exception, cannot proceed further
java.util.concurrent.ExecutionException: java.lang.RuntimeException: Unable to call info API using the elasticsearch client
	at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122) ~[?:?]
	at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191) ~[?:?]
	at org.opensearch.dataprepper.pipeline.common.PipelineThreadPoolExecutor.afterExecute(PipelineThreadPoolExecutor.java:70) [data-prepper-core-2.8.0.jar:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1137) [?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
Caused by: java.lang.RuntimeException: Unable to call info API using the elasticsearch client
	at org.opensearch.dataprepper.plugins.source.opensearch.worker.client.SearchAccessorStrategy.getDistributionAndVersionNumber(SearchAccessorStrategy.java:199) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.worker.client.SearchAccessorStrategy.getSearchAccessor(SearchAccessorStrategy.java:115) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSource.startProcess(OpenSearchSource.java:75) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSource.start(OpenSearchSource.java:65) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.startSourceAndProcessors(Pipeline.java:215) ~[data-prepper-core-2.8.0.jar:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.lambda$execute$2(Pipeline.java:260) ~[data-prepper-core-2.8.0.jar:?]
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	... 2 more
Caused by: co.elastic.clients.util.MissingRequiredPropertyException: Missing required property 'ElasticsearchVersionInfo.buildFlavor'
	at co.elastic.clients.util.ApiTypeHelper.requireNonNull(ApiTypeHelper.java:76) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.elasticsearch._types.ElasticsearchVersionInfo.<init>(ElasticsearchVersionInfo.java:74) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.elasticsearch._types.ElasticsearchVersionInfo.<init>(ElasticsearchVersionInfo.java:50) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.elasticsearch._types.ElasticsearchVersionInfo$Builder.build(ElasticsearchVersionInfo.java:300) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.elasticsearch._types.ElasticsearchVersionInfo$Builder.build(ElasticsearchVersionInfo.java:200) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.ObjectBuilderDeserializer.deserialize(ObjectBuilderDeserializer.java:80) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.DelegatingDeserializer$SameType.deserialize(DelegatingDeserializer.java:43) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.ObjectDeserializer$FieldObjectDeserializer.deserialize(ObjectDeserializer.java:72) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.ObjectDeserializer.deserialize(ObjectDeserializer.java:176) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.ObjectDeserializer.deserialize(ObjectDeserializer.java:137) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.JsonpDeserializer.deserialize(JsonpDeserializer.java:75) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.ObjectBuilderDeserializer.deserialize(ObjectBuilderDeserializer.java:79) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.json.DelegatingDeserializer$SameType.deserialize(DelegatingDeserializer.java:43) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.transport.rest_client.RestClientTransport.decodeResponse(RestClientTransport.java:328) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.transport.rest_client.RestClientTransport.getHighLevelResponse(RestClientTransport.java:294) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.transport.rest_client.RestClientTransport.performRequest(RestClientTransport.java:147) ~[elasticsearch-java-7.17.0.jar:?]
	at co.elastic.clients.elasticsearch.ElasticsearchClient.info(ElasticsearchClient.java:983) ~[elasticsearch-java-7.17.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.worker.client.SearchAccessorStrategy.getDistributionAndVersionNumber(SearchAccessorStrategy.java:196) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.worker.client.SearchAccessorStrategy.getSearchAccessor(SearchAccessorStrategy.java:115) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSource.startProcess(OpenSearchSource.java:75) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSource.start(OpenSearchSource.java:65) ~[opensearch-2.8.0.jar:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.startSourceAndProcessors(Pipeline.java:215) ~[data-prepper-core-2.8.0.jar:?]
	at org.opensearch.dataprepper.pipeline.Pipeline.lambda$execute$2(Pipeline.java:260) ~[data-prepper-core-2.8.0.jar:?]
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
	... 2 more

[kclinden]

Looks like elasticsearch returns this value but opensearch does not.
https://www.elastic.co/guide/en/elasticsearch/reference/current/rest-api-root.html

Elasticsearch Source: https://github.com/elastic/elasticsearch/blob/e47ec4efc99ca4e643b7e6195b86c1ecdcdfac45/modules/rest-root/src/main/java/org/elasticsearch/rest/root/MainResponse.java#L71

OpenSearch Source:
https://github.com/opensearch-project/OpenSearch/blob/f5dbbb04b96bb8866255e4f967241db8bed16141/server/src/main/java/org/opensearch/action/main/MainResponse.java#L121

[kclinden]

Similar to - #3640

[dlvenable]

@kclinden , We do allow for configuring the distribution_version as a fallback when the cluster does not provide it. You can see the logic here.

data-prepper/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/SearchAccessorStrategy.java

Line 92 in 980b0ec

if (DistributionVersion.OPENSEARCH.equals(openSearchSourceConfiguration.getDistributionVersion())) {

Here is the configuration parameter:

data-prepper/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceConfiguration.java

Lines 66 to 67 in 980b0ec

	@JsonProperty("distribution_version")
	private String distributionVersion;

And this shows the valid values:

data-prepper/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceConfiguration.java

Line 141 in 980b0ec

@AssertTrue(message = "distribution_version must be one of [ 'es7', 'opensearch' ]")

For example:

source:
  opensearch:
    distribution_version: opensearch

[dlvenable]

We need to add this to the documentation. It is not present now.

[kclinden]

@dlvenable thanks. That gets met further. When I set that though my pipeline hangs at the following

dataprepper-1  | 2024-07-02T15:41:36,559 [opensearch-source-pipeline-sink-worker-2-thread-1] INFO  org.opensearch.dataprepper.pipeline.Pipeline - Pipeline [opensearch-source-pipeline] - Submitting request to initiate the pipeline processing