Skip to content

Commit 407088a

Browse files
saketh-pallempatisaketh-pallempati
committed
fix(opensearch): default to OpenSearch if distribution detection fails
1 parent 93e245c commit 407088a

2 files changed

Lines changed: 106 additions & 13 deletions

File tree

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

Lines changed: 78 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,8 @@ public SearchAccessor getSearchAccessor() {
8181

8282
InfoResponse infoResponse = null;
8383

84-
PluginComponentRefresher<ElasticsearchClient, OpenSearchSourceConfiguration> elasticsearchClientRefresher =
85-
null;
84+
PluginComponentRefresher<ElasticsearchClient, OpenSearchSourceConfiguration> elasticsearchClientRefresher = null;
85+
8686
try {
8787
infoResponse = clientRefresher.get().info();
8888
pluginConfigObservable.addPluginConfigObserver(newConfig -> clientRefresher.update(
@@ -135,12 +135,33 @@ public SearchAccessor getSearchAccessor() {
135135
searchContextType = SearchContextType.SCROLL;
136136
}
137137

138-
if (Objects.isNull(elasticsearchClientRefresher)) {
139-
return new OpenSearchAccessor(clientRefresher,
140-
searchContextType);
138+
if (OPENSEARCH_DISTRIBUTION.equals(distribution)) {
139+
LOG.debug("Using OpenSearchAccessor as the determined distribution is OpenSearch.");
140+
return new OpenSearchAccessor(clientRefresher, searchContextType);
141+
} else {
142+
LOG.debug("Using ElasticsearchAccessor as the determined distribution is Elasticsearch.");
143+
PluginComponentRefresher<ElasticsearchClient, OpenSearchSourceConfiguration> currentElasticsearchClientRefresher = elasticsearchClientRefresher;
144+
if (Objects.isNull(currentElasticsearchClientRefresher)) {
145+
LOG.info("Elasticsearch distribution determined, but no existing Elasticsearch client refresher found (e.g., OpenSearch client reported Elasticsearch, or distribution_version was set to elasticsearch). Creating a new Elasticsearch client refresher.");
146+
try {
147+
currentElasticsearchClientRefresher = new ClientRefresher<>(openSearchSourcePluginMetrics,
148+
ElasticsearchClient.class, openSearchClientFactory::provideElasticSearchClient,
149+
openSearchSourceConfiguration);
150+
final PluginComponentRefresher<ElasticsearchClient, OpenSearchSourceConfiguration>
151+
finalNewElasticsearchClientRefresher = currentElasticsearchClientRefresher;
152+
pluginConfigObservable.addPluginConfigObserver(
153+
newConfig -> {
154+
if (finalNewElasticsearchClientRefresher != null) {
155+
finalNewElasticsearchClientRefresher.update((OpenSearchSourceConfiguration) newConfig);
156+
}
157+
});
158+
} catch (final Exception e) {
159+
LOG.error("Failed to create Elasticsearch client for an Elasticsearch distribution when one was not pre-existing.", e);
160+
throw new RuntimeException("Failed to create Elasticsearch client for an Elasticsearch distribution: " + e.getMessage(), e);
161+
}
162+
}
163+
return new ElasticsearchAccessor(currentElasticsearchClientRefresher, searchContextType);
141164
}
142-
143-
return new ElasticsearchAccessor(elasticsearchClientRefresher, searchContextType);
144165
}
145166

146167
private SearchAccessor createSearchAccessorForServerlessCollection(final PluginComponentRefresher clientRefresher) {
@@ -189,14 +210,58 @@ private boolean versionSupportsPointInTime(final String distribution, final Stri
189210
private Pair<String, String> getDistributionAndVersionNumber(final InfoResponse infoResponseOpenSearch,
190211
final PluginComponentRefresher<ElasticsearchClient, OpenSearchSourceConfiguration> elasticsearchClientRefresher) {
191212
if (Objects.nonNull(infoResponseOpenSearch)) {
192-
return Pair.of(infoResponseOpenSearch.version().distribution(), infoResponseOpenSearch.version().number());
213+
final String distribution = infoResponseOpenSearch.version().distribution();
214+
final String versionNumber = infoResponseOpenSearch.version().number();
215+
if (Objects.nonNull(distribution) && Objects.nonNull(versionNumber)) {
216+
LOG.info("Detected OpenSearch distribution '{}' version '{}' from API response.", distribution, versionNumber);
217+
return Pair.of(distribution, versionNumber);
218+
}
219+
LOG.warn("Distribution or version number is null in OpenSearch API response. Proceeding to check Elasticsearch or default.");
193220
}
194221

195-
try {
196-
final co.elastic.clients.elasticsearch.core.InfoResponse infoResponseElasticsearch = elasticsearchClientRefresher.get().info();
197-
return Pair.of(ELASTICSEARCH_DISTRIBUTION + "-" + infoResponseElasticsearch.version().buildFlavor(), infoResponseElasticsearch.version().number());
198-
} catch (final Exception e) {
199-
throw new RuntimeException("Unable to call info API using the elasticsearch client", e);
222+
if (elasticsearchClientRefresher != null) {
223+
try {
224+
final ElasticsearchClient elasticsearchClient = elasticsearchClientRefresher.get();
225+
if (elasticsearchClient == null) {
226+
LOG.info("ElasticsearchClient from refresher is null. Cannot attempt Elasticsearch .info() call. Proceeding to default.");
227+
} else {
228+
final co.elastic.clients.elasticsearch.core.InfoResponse infoResponseElasticsearch = elasticsearchClient.info();
229+
final String esBuildFlavor = infoResponseElasticsearch.version().buildFlavor();
230+
final String esVersionNumber = infoResponseElasticsearch.version().number();
231+
232+
if (Objects.nonNull(esVersionNumber)) {
233+
String effectiveDistribution = ELASTICSEARCH_DISTRIBUTION;
234+
LOG.info("Detected Elasticsearch (flavor '{}') version '{}' from API response.",
235+
Objects.toString(esBuildFlavor, "unknown"), esVersionNumber);
236+
if (ELASTICSEARCH_OSS_BUILD_FLAVOR.equalsIgnoreCase(esBuildFlavor)) {
237+
effectiveDistribution = ELASTICSEARCH_DISTRIBUTION + "-" + ELASTICSEARCH_OSS_BUILD_FLAVOR;
238+
}
239+
return Pair.of(effectiveDistribution, esVersionNumber);
240+
}
241+
LOG.warn("Version number is null in Elasticsearch API response. Proceeding to default.");
242+
}
243+
} catch (final Exception e) {
244+
LOG.warn("Failed to get cluster info using Elasticsearch client. Will proceed with default behavior.", e);
245+
}
246+
} else {
247+
LOG.info("Elasticsearch client refresher is null, skipping Elasticsearch .info() call.");
248+
}
249+
250+
if (openSearchSourceConfiguration.getDistributionVersion() == null) {
251+
LOG.warn("Failed to auto-detect cluster distribution from API response for both OpenSearch and (if attempted) Elasticsearch clients. " +
252+
"Defaulting to OpenSearch distribution. Consider setting 'distribution_version' in the configuration if this is not an OpenSearch cluster.");
253+
return Pair.of(OPENSEARCH_DISTRIBUTION, OPENSEARCH_POINT_IN_TIME_SUPPORT_VERSION_CUTOFF);
254+
} else {
255+
final String configuredDistributionName = openSearchSourceConfiguration.getDistributionVersion().name();
256+
LOG.error("Failed to connect to the cluster and verify its distribution. The 'distribution_version' was configured as '{}', " +
257+
"but communication attempts with the cluster failed or the cluster did not respond as expected for this distribution type. " +
258+
"Please check the endpoint, network connectivity, and the specified 'distribution_version'.",
259+
configuredDistributionName);
260+
throw new InvalidPluginConfigurationException(
261+
String.format("Failed to connect to the cluster and verify its distribution. The 'distribution_version' was configured as '%s', " +
262+
"but communication attempts with the cluster failed or the cluster did not respond as expected for this distribution type. " +
263+
"Please check the endpoint, network connectivity, and the specified 'distribution_version'.",
264+
configuredDistributionName));
200265
}
201266
}
202267

‎data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/SearchAccessStrategyTest.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import static org.hamcrest.MatcherAssert.assertThat;
3333
import static org.hamcrest.Matchers.equalTo;
3434
import static org.hamcrest.Matchers.notNullValue;
35+
import static org.hamcrest.Matchers.instanceOf;
3536
import static org.junit.jupiter.api.Assertions.assertThrows;
3637
import static org.mockito.ArgumentMatchers.any;
3738
import static org.mockito.Mockito.mock;
@@ -55,6 +56,9 @@ public class SearchAccessStrategyTest {
5556
@Mock
5657
private PluginConfigObservable pluginConfigObservable;
5758

59+
@Mock
60+
private SearchConfiguration searchConfiguration;
61+
5862
private SearchAccessorStrategy createObjectUnderTest() {
5963
return SearchAccessorStrategy.create(
6064
openSearchSourcePluginMetrics, openSearchSourceConfiguration, openSearchClientFactory,
@@ -297,4 +301,28 @@ void force_OpenSearch_client_and_uses_search_context_type_override_when_distribu
297301
assertThat(searchAccessor, notNullValue());
298302
assertThat(searchAccessor.getSearchContextType(), equalTo(SearchContextType.POINT_IN_TIME));
299303
}
304+
305+
@Test
306+
void getSearchAccessor_whenDistVersionNullAndAllClientInfoFails_usesDefaultOpenSearchParametersForSearchContext() throws IOException {
307+
when(openSearchSourceConfiguration.getDistributionVersion()).thenReturn(null);
308+
when(openSearchSourceConfiguration.getSearchConfiguration()).thenReturn(searchConfiguration);
309+
when(searchConfiguration.getSearchContextType()).thenReturn(null);
310+
311+
final OpenSearchClient mockOpenSearchClient = mock(OpenSearchClient.class);
312+
when(openSearchClientFactory.provideOpenSearchClient(openSearchSourceConfiguration)).thenReturn(mockOpenSearchClient);
313+
when(mockOpenSearchClient.info()).thenThrow(new IOException("Simulated OpenSearch info failure"));
314+
315+
final ElasticsearchClient mockElasticsearchClient = mock(ElasticsearchClient.class);
316+
when(openSearchClientFactory.provideElasticSearchClient(openSearchSourceConfiguration)).thenReturn(mockElasticsearchClient);
317+
when(mockElasticsearchClient.info()).thenThrow(new IOException("Simulated Elasticsearch info failure"));
318+
319+
final SearchAccessorStrategy strategy = createObjectUnderTest();
320+
321+
final SearchAccessor searchAccessor = strategy.getSearchAccessor();
322+
323+
assertThat(searchAccessor, instanceOf(OpenSearchAccessor.class));
324+
assertThat(searchAccessor.getSearchContextType(), equalTo(SearchContextType.POINT_IN_TIME));
325+
326+
verify(pluginConfigObservable).addPluginConfigObserver(any());
327+
}
300328
}

0 commit comments

Comments
 (0)