From aacaa0c4d2005e4e47311b988d4e84cecabcd486 Mon Sep 17 00:00:00 2001 From: Dennis Oelkers Date: Fri, 19 Jan 2024 10:37:44 +0100 Subject: [PATCH] Setting up injection/usage of new OpenSearch Java Client. (#17760) * Adding new additional dependencies. * Setting up injection infrastructure for new client in parallel. * Adding license headers. * Including new deps in test jar. * Removing jakarta deps from assembly. * Building separate jars for integration tests & distribution. * Fix class loader issue Exclude "module-info.class" files from the graylog-storage-opensearch JAR file to avoid class loader issues. Revert previous assembly and dependency changes that worked around the issue in tests. * Adjusting to Jakarta and recent data node changes. * fixing wrong builder usage --------- Co-authored-by: Bernd Ahlers Co-authored-by: Jan Heise --- .../management/OpensearchProcessImpl.java | 11 +- .../management/OpensearchRestClient.java | 26 +++ graylog-storage-opensearch2/assembly.xml | 18 ++ graylog-storage-opensearch2/pom.xml | 33 ++++ .../FilteredOpenSearchNodesSniffer.java | 2 +- .../LegacyFilteredOpenSearchNodesSniffer.java | 71 +++++++ .../opensearch2/LegacyNodeListSniffer.java | 60 ++++++ .../opensearch2/LegacyNodesSniffer.java | 26 +++ ...hFilterDeprecationWarningsInterceptor.java | 54 ++++++ .../opensearch2/LegacySnifferWrapper.java | 73 ++++++++ .../storage/opensearch2/NodeAdapterOS2.java | 21 +-- .../storage/opensearch2/NodeListSniffer.java | 2 +- .../storage/opensearch2/NodesSniffer.java | 3 +- .../opensearch2/OpenSearch2Module.java | 9 +- .../storage/opensearch2/OpenSearchClient.java | 18 +- .../opensearch2/OpenSearchClientProvider.java | 45 +++++ ...hFilterDeprecationWarningsInterceptor.java | 21 +-- .../opensearch2/RestClientProvider.java | 174 ++++++++++++++++++ .../RestHighLevelClientProvider.java | 11 +- .../storage/opensearch2/SnifferWrapper.java | 18 +- .../storage/opensearch2/ThrowingFunction.java | 21 +++ .../client/LegacyOSCredentialsProvider.java | 81 ++++++++ .../client/OSCredentialsProvider.java | 8 +- .../opensearch2/ClusterAdapterOS2Test.java | 2 +- .../FilteredOpenSearchNodesSnifferTest.java | 16 +- .../opensearch2/NodeAdapterOS2Test.java | 69 ++++--- .../opensearch2/OpenSearchExceptionTest.java | 5 +- ...terDeprecationWarningsInterceptorTest.java | 21 +-- .../storage/opensearch2/cat/CatApiTest.java | 3 +- .../opensearch2/testing/AdaptersOS2.java | 2 +- .../testing/OpenSearchInstance.java | 37 +++- 31 files changed, 854 insertions(+), 107 deletions(-) create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyFilteredOpenSearchNodesSniffer.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodeListSniffer.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodesSniffer.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyOpenSearchFilterDeprecationWarningsInterceptor.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacySnifferWrapper.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClientProvider.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestClientProvider.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ThrowingFunction.java create mode 100644 graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/LegacyOSCredentialsProvider.java diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java index beee2e0652..58d0f66e67 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java @@ -35,6 +35,9 @@ import org.graylog.storage.opensearch2.OpenSearchClient; import org.graylog2.cluster.nodes.DataNodeDto; import org.graylog2.cluster.nodes.NodeService; import org.graylog2.security.CustomCAX509TrustManager; +import org.opensearch.client.RestClient; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; +import org.opensearch.client.transport.rest_client.RestClientTransport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -97,6 +100,10 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { return OpensearchRestClient.build(configuration, datanodeConfiguration, trustManager); } + private RestClient createNewRestClient(OpensearchConfiguration configuration) { + return OpensearchRestClient.buildNewClient(configuration, datanodeConfiguration, trustManager); + } + @Override public List stdOutLogs() { return stdout.stream().toList(); @@ -195,7 +202,9 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { commandLineProcess = new OpensearchCommandLineProcess(config, this); commandLineProcess.start(); restClient = Optional.of(createRestClient(config)); - openSearchClient = restClient.map(c -> new OpenSearchClient(c, objectMapper)); + final var transport = new RestClientTransport(createNewRestClient(config), new JacksonJsonpMapper(objectMapper)); + final var client = new org.opensearch.client.opensearch.OpenSearchClient(transport); + openSearchClient = restClient.map(c -> new OpenSearchClient(c, client, objectMapper)); }), () -> {throw new IllegalArgumentException("Opensearch configuration required but not supplied!");} ); diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java index ba75092d25..0ccb2f9b70 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java @@ -22,6 +22,7 @@ import org.graylog.shaded.opensearch2.org.apache.http.HttpHost; import org.graylog.shaded.opensearch2.org.apache.http.HttpRequestInterceptor; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; import org.graylog2.security.CustomCAX509TrustManager; +import org.opensearch.client.RestClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,4 +59,29 @@ public class OpensearchRestClient { } return new RestHighLevelClient(builder); } + + public static RestClient buildNewClient(OpensearchConfiguration configuration, DatanodeConfiguration datanodeConfiguration, CustomCAX509TrustManager tm) { + final HttpHost host = configuration.getRestBaseUrl(); + + final var builder = org.opensearch.client.RestClient.builder(new org.apache.http.HttpHost(host.getHostName(), host.getPort(), host.getSchemeName())); + if ("https".equals(host.getSchemeName())) { + + try { + final var sslContext = SSLContext.getInstance("TLS"); + sslContext.init(null, new TrustManager[]{tm}, new SecureRandom()); + + builder.setHttpClientConfigCallback(httpClientBuilder -> { + httpClientBuilder.addInterceptorLast((org.apache.http.HttpRequestInterceptor) (request, context) -> { + final String jwtToken = datanodeConfiguration.indexerJwtAuthTokenProvider().get(); + request.addHeader("Authorization", jwtToken); + }); + httpClientBuilder.setSSLContext(sslContext); + return httpClientBuilder; + }); + } catch (NoSuchAlgorithmException | KeyManagementException ex) { + LOG.error("Could not initialize SSL correctly: {}", ex.getMessage(), ex); + } + } + return builder.build(); + } } diff --git a/graylog-storage-opensearch2/assembly.xml b/graylog-storage-opensearch2/assembly.xml index a7d3b08106..2df9da6d8d 100644 --- a/graylog-storage-opensearch2/assembly.xml +++ b/graylog-storage-opensearch2/assembly.xml @@ -14,11 +14,29 @@ runtime true true + true org.graylog:graylog-storage-opensearch2 org.graylog.shaded:opensearch2-rest-high-level-client org.graylog.shaded:opensearch2-rest-client-sniffer + org.opensearch.client:opensearch-java + org.opensearch.client:opensearch-rest-client + org.opensearch.client:opensearch-rest-client-sniffer + org.apache.httpcomponents:httpasyncclient + org.apache.httpcomponents:httpclient + org.apache.httpcomponents:httpcore + org.apache.httpcomponents:httpcore-nio + jakarta.json.bind:jakarta.json.bind-api + jakarta.json:jakarta.json-api + + + + module-info.class + + diff --git a/graylog-storage-opensearch2/pom.xml b/graylog-storage-opensearch2/pom.xml index 65101e9e21..ce62a6822e 100644 --- a/graylog-storage-opensearch2/pom.xml +++ b/graylog-storage-opensearch2/pom.xml @@ -72,6 +72,39 @@ + + org.opensearch.client + opensearch-java + 2.8.1 + + + commons-logging + commons-logging + + + + + org.opensearch.client + opensearch-rest-client + 2.11.1 + + + commons-logging + commons-logging + + + + + org.opensearch.client + opensearch-rest-client-sniffer + 2.11.1 + + + commons-logging + commons-logging + + + io.jsonwebtoken jjwt-api diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSniffer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSniffer.java index b37604540e..3b6d221987 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSniffer.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSniffer.java @@ -18,7 +18,7 @@ package org.graylog.storage.opensearch2; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Strings; -import org.graylog.shaded.opensearch2.org.opensearch.client.Node; +import org.opensearch.client.Node; import java.util.Collections; import java.util.List; diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyFilteredOpenSearchNodesSniffer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyFilteredOpenSearchNodesSniffer.java new file mode 100644 index 0000000000..73492a6f5a --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyFilteredOpenSearchNodesSniffer.java @@ -0,0 +1,71 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Strings; +import org.graylog.shaded.opensearch2.org.opensearch.client.Node; + +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +class LegacyFilteredOpenSearchNodesSniffer implements LegacyNodesSniffer { + private final String attribute; + private final String value; + + static LegacyFilteredOpenSearchNodesSniffer create(String filter) { + final String attribute; + final String value; + if (!Strings.isNullOrEmpty(filter)) { + final String[] conditions = filter.split(":"); + if (conditions.length < 2) { + throw new IllegalArgumentException("Invalid filter specified for ES node discovery: " + filter); + } + attribute = conditions[0].trim(); + value = conditions[1].trim(); + } else { + attribute = null; + value = null; + } + + return new LegacyFilteredOpenSearchNodesSniffer(attribute, value); + } + + @VisibleForTesting + LegacyFilteredOpenSearchNodesSniffer(String attribute, String value) { + this.attribute = attribute; + this.value = value; + } + + @Override + public List sniff(List nodes) { + if (attribute == null || value == null) { + return nodes; + } + + return nodes.stream() + .filter(node -> nodeMatchesFilter(node, attribute, value)) + .collect(Collectors.toList()); + } + + private boolean nodeMatchesFilter(Node node, String attribute, String value) { + return node.getAttributes() + .getOrDefault(attribute, Collections.emptyList()) + .contains(value); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodeListSniffer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodeListSniffer.java new file mode 100644 index 0000000000..344ad531d4 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodeListSniffer.java @@ -0,0 +1,60 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import com.google.common.collect.Sets; +import org.graylog.shaded.opensearch2.org.opensearch.client.Node; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +public class LegacyNodeListSniffer implements LegacyNodesSniffer { + private static final Logger LOG = LoggerFactory.getLogger(LegacyNodeListSniffer.class); + private static final Set savedNodes = ConcurrentHashMap.newKeySet(); + + static LegacyNodeListSniffer create() { + return new LegacyNodeListSniffer(); + } + + @Override + public List sniff(final List nodes) { + final Set currentNodes = nodes.stream().map(n -> n.getHost().toURI()).collect(Collectors.toSet()); + + final Set nodesAdded = Sets.difference(currentNodes, savedNodes); + final Set nodesDropped = Sets.difference(savedNodes, currentNodes); + + if(!nodesAdded.isEmpty()) { + LOG.info("Added node(s): {}", nodesAdded); + } + if(!nodesDropped.isEmpty()) { + LOG.info("Dropped node(s): {}", nodesDropped); + } + if(!nodesAdded.isEmpty() || !nodesDropped.isEmpty()) { + LOG.info("Current node list: {}", currentNodes); + } + + savedNodes.clear(); + savedNodes.addAll(currentNodes); + + return nodes; + } +} + diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodesSniffer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodesSniffer.java new file mode 100644 index 0000000000..384c931427 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodesSniffer.java @@ -0,0 +1,26 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import org.graylog.shaded.opensearch2.org.opensearch.client.Node; + +import java.io.IOException; +import java.util.List; + +public interface LegacyNodesSniffer { + List sniff(List nodes) throws IOException; +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyOpenSearchFilterDeprecationWarningsInterceptor.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyOpenSearchFilterDeprecationWarningsInterceptor.java new file mode 100644 index 0000000000..f6b98db24d --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyOpenSearchFilterDeprecationWarningsInterceptor.java @@ -0,0 +1,54 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import org.graylog.shaded.opensearch2.org.apache.http.Header; +import org.graylog.shaded.opensearch2.org.apache.http.HttpException; +import org.graylog.shaded.opensearch2.org.apache.http.HttpResponse; +import org.graylog.shaded.opensearch2.org.apache.http.HttpResponseInterceptor; +import org.graylog.shaded.opensearch2.org.apache.http.protocol.HttpContext; +import org.graylog.shaded.opensearch2.org.opensearch.common.joda.JodaDeprecationPatterns; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; + +// TODO: check, if these messages still need filtering +public class LegacyOpenSearchFilterDeprecationWarningsInterceptor implements HttpResponseInterceptor { + private final String[] messagesToFilter = { + "setting was deprecated in OpenSearch", + "but in a future major version, direct access to system indices and their aliases will not be allowed", + "but in a future major version, direct access to system indices will be prevented by default", + "in epoch time formats is deprecated and will not be supported in the next major version of OpenSearch", + JodaDeprecationPatterns.USE_NEW_FORMAT_SPECIFIERS + }; + + private boolean isDeprecationMessage(final String message) { + for(String msg: messagesToFilter) { + if(message.contains(msg)) return true; + } + return false; + } + + @Override + public void process(HttpResponse response, HttpContext context) throws + HttpException, IOException { + List
warnings = Arrays.stream(response.getHeaders("Warning")).filter(header -> !this.isDeprecationMessage(header.getValue())).toList(); + response.removeHeaders("Warning"); + warnings.forEach(response::addHeader); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacySnifferWrapper.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacySnifferWrapper.java new file mode 100644 index 0000000000..6dc0d9e079 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacySnifferWrapper.java @@ -0,0 +1,73 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import com.github.joschi.jadconfig.util.Duration; +import org.graylog.shaded.opensearch2.org.opensearch.client.Node; +import org.graylog.shaded.opensearch2.org.opensearch.client.RestClient; +import org.graylog.shaded.opensearch2.org.opensearch.client.sniff.OpenSearchNodesSniffer; +import org.graylog.shaded.opensearch2.org.opensearch.client.sniff.Sniffer; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CopyOnWriteArrayList; + +public class LegacySnifferWrapper implements org.graylog.shaded.opensearch2.org.opensearch.client.sniff.NodesSniffer { + private final List sniffers = new CopyOnWriteArrayList<>(); + private final RestClient restClient; + private final long sniffRequestTimeoutMillis; + private final Duration discoveryFrequency; + private final OpenSearchNodesSniffer.Scheme scheme; + private org.graylog.shaded.opensearch2.org.opensearch.client.sniff.NodesSniffer nodesSniffer; + + private LegacySnifferWrapper(RestClient restClient, long sniffRequestTimeoutMillis, Duration discoveryFrequency, OpenSearchNodesSniffer.Scheme scheme) { + this.restClient = restClient; + this.sniffRequestTimeoutMillis = sniffRequestTimeoutMillis; + this.discoveryFrequency = discoveryFrequency; + this.scheme = scheme; + } + + @Override + public List sniff() throws IOException { + List nodes = this.nodesSniffer.sniff(); + for (LegacyNodesSniffer sniffer : sniffers) { + nodes = sniffer.sniff(nodes); + } + return nodes; + } + + public static LegacySnifferWrapper create(RestClient restClient, long sniffRequestTimeoutMillis, Duration discoveryFrequency, OpenSearchNodesSniffer.Scheme scheme) { + return new LegacySnifferWrapper(restClient, sniffRequestTimeoutMillis, discoveryFrequency, scheme); + } + + public Optional build() { + if(sniffers.isEmpty()) { + return Optional.empty(); + } + + this.nodesSniffer = new OpenSearchNodesSniffer(restClient, sniffRequestTimeoutMillis, scheme); + return Optional.of(Sniffer.builder(restClient) + .setSniffIntervalMillis(Math.toIntExact(discoveryFrequency.toMilliseconds())) + .setNodesSniffer(this) + .build()); + } + + public void add(LegacyNodesSniffer sniffer) { + this.sniffers.add(sniffer); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeAdapterOS2.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeAdapterOS2.java index 0f9339f0a0..67bcd9f790 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeAdapterOS2.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeAdapterOS2.java @@ -16,9 +16,6 @@ */ package org.graylog.storage.opensearch2; -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.graylog.shaded.opensearch2.org.opensearch.client.Request; import org.graylog2.indexer.cluster.NodeAdapter; import org.graylog2.shared.utilities.StringUtils; import org.graylog2.storage.SearchVersion; @@ -28,26 +25,20 @@ import jakarta.inject.Inject; import java.util.Optional; public class NodeAdapterOS2 implements NodeAdapter { - private final PlainJsonApi jsonApi; + private final OpenSearchClient client; @Inject - public NodeAdapterOS2(OpenSearchClient client, ObjectMapper objectMapper) { - this.jsonApi = new PlainJsonApi(objectMapper, client); - } - - NodeAdapterOS2(final PlainJsonApi jsonApi) { - this.jsonApi = jsonApi; + public NodeAdapterOS2(OpenSearchClient client) { + this.client = client; } @Override public Optional version() { + var info = client.execute(c -> c.info()); - final Request request = new Request("GET", "/?filter_path=version.number,version.distribution"); - final Optional resp = Optional.of(jsonApi.perform(request, "Unable to retrieve cluster information")); + final Optional version = Optional.ofNullable(info.version().number()); - final Optional version = resp.map(r -> r.path("version")).map(r -> r.path("number")).map(JsonNode::textValue); - - final SearchVersion.Distribution distribution = resp.map(r -> r.path("version")).map(r -> r.path("distribution")).map(JsonNode::textValue) + final SearchVersion.Distribution distribution = Optional.ofNullable(info.version().distribution()) .map(StringUtils::toUpperCase) .map(SearchVersion.Distribution::valueOf) .orElse(SearchVersion.Distribution.ELASTICSEARCH); diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeListSniffer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeListSniffer.java index fc9aaddb32..4611b1f8c1 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeListSniffer.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeListSniffer.java @@ -17,7 +17,7 @@ package org.graylog.storage.opensearch2; import com.google.common.collect.Sets; -import org.graylog.shaded.opensearch2.org.opensearch.client.Node; +import org.opensearch.client.Node; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodesSniffer.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodesSniffer.java index e755cf55c9..6629521a7f 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodesSniffer.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodesSniffer.java @@ -16,7 +16,8 @@ */ package org.graylog.storage.opensearch2; -import org.graylog.shaded.opensearch2.org.opensearch.client.Node; + +import org.opensearch.client.Node; import java.io.IOException; import java.util.List; diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java index b92543ac83..fee50fa41d 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java @@ -23,6 +23,7 @@ import org.graylog.plugins.views.migrations.V20200730000000_AddGl2MessageIdField import org.graylog.plugins.views.search.engine.QuerySuggestionsService; import org.graylog.shaded.opensearch2.org.apache.http.client.CredentialsProvider; import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient; +import org.graylog.storage.opensearch2.client.LegacyOSCredentialsProvider; import org.graylog.storage.opensearch2.client.OSCredentialsProvider; import org.graylog.storage.opensearch2.fieldtypes.streams.StreamsForFieldRetrieverOS2; import org.graylog.storage.opensearch2.migrations.V20170607164210_MigrateReopenedIndicesToAliasesClusterStateOS2; @@ -43,6 +44,8 @@ import org.graylog2.indexer.searches.SearchesAdapter; import org.graylog2.migrations.V20170607164210_MigrateReopenedIndicesToAliases; import org.graylog2.plugin.VersionAwareModule; import org.graylog2.storage.SearchVersion; +import org.opensearch.client.RestClient; +import org.opensearch.client.opensearch.OpenSearchClient; public class OpenSearch2Module extends VersionAwareModule { private final SearchVersion supportedVersion; @@ -85,7 +88,11 @@ public class OpenSearch2Module extends VersionAwareModule { install(new FactoryModuleBuilder().build(ScrollResultOS2.Factory.class)); bind(RestHighLevelClient.class).toProvider(RestHighLevelClientProvider.class); - bind(CredentialsProvider.class).toProvider(OSCredentialsProvider.class); + bind(CredentialsProvider.class).toProvider(LegacyOSCredentialsProvider.class); + + bind(RestClient.class).toProvider(RestClientProvider.class); + bind(OpenSearchClient.class).toProvider(OpenSearchClientProvider.class); + bind(org.apache.http.client.CredentialsProvider.class).toProvider(OSCredentialsProvider.class); } private LinkedBindingBuilder bindForSupportedVersion(Class interfaceClass) { diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClient.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClient.java index 3f18fad9f3..7a36f6a329 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClient.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClient.java @@ -60,6 +60,7 @@ public class OpenSearchClient { private static final Logger LOG = LoggerFactory.getLogger(OpenSearchClient.class); private final RestHighLevelClient client; + private final org.opensearch.client.opensearch.OpenSearchClient openSearchClient; private final boolean compressionEnabled; private final Optional indexerMaxConcurrentSearches; private final Optional indexerMaxConcurrentShardRequests; @@ -67,11 +68,13 @@ public class OpenSearchClient { @Inject public OpenSearchClient(RestHighLevelClient client, + org.opensearch.client.opensearch.OpenSearchClient openSearchClient, @Named("elasticsearch_compression_enabled") boolean compressionEnabled, @Named("indexer_max_concurrent_searches") @Nullable Integer indexerMaxConcurrentSearches, @Named("indexer_max_concurrent_shard_requests") @Nullable Integer indexerMaxConcurrentShardRequests, ObjectMapper objectMapper) { this.client = client; + this.openSearchClient = openSearchClient; this.compressionEnabled = compressionEnabled; this.indexerMaxConcurrentSearches = Optional.ofNullable(indexerMaxConcurrentSearches); this.indexerMaxConcurrentShardRequests = Optional.ofNullable(indexerMaxConcurrentShardRequests); @@ -79,8 +82,8 @@ public class OpenSearchClient { } @VisibleForTesting - public OpenSearchClient(RestHighLevelClient client, ObjectMapper objectMapper) { - this(client, false, null, null, objectMapper); + public OpenSearchClient(RestHighLevelClient client, org.opensearch.client.opensearch.OpenSearchClient openSearchClient, ObjectMapper objectMapper) { + this(client, openSearchClient, false, null, null, objectMapper); } public SearchResponse search(SearchRequest searchRequest, String errorMessage) { @@ -136,6 +139,17 @@ public class OpenSearchClient { } } + public R execute(ThrowingFunction fn, String errorMessage) { + try { + return fn.apply(openSearchClient); + } catch (Exception e) { + throw exceptionFrom(e, errorMessage); + } + } + + public R execute(ThrowingFunction fn) { + return execute(fn, "An error occurred: "); + } @WithSpan public R executeWithIOException(ThrowingBiFunction fn, String errorMessage) throws IOException { try { diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClientProvider.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClientProvider.java new file mode 100644 index 0000000000..077233784f --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClientProvider.java @@ -0,0 +1,45 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.inject.Inject; +import jakarta.inject.Provider; +import org.opensearch.client.RequestOptions; +import org.opensearch.client.RestClient; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; +import org.opensearch.client.opensearch.OpenSearchClient; +import org.opensearch.client.transport.rest_client.RestClientOptions; +import org.opensearch.client.transport.rest_client.RestClientTransport; + +public class OpenSearchClientProvider implements Provider { + private final RestClientTransport transport; + + @Inject + public OpenSearchClientProvider(RestClient restClient, ObjectMapper objectMapper) { + this.transport = new RestClientTransport(restClient, new JacksonJsonpMapper(objectMapper)); + } + + @Override + public OpenSearchClient get() { + return new OpenSearchClient(transport); + } + + public OpenSearchClient getWithRequestOptions(RequestOptions requestOptions) { + return new OpenSearchClient(transport.withRequestOptions(new RestClientOptions(requestOptions))); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptor.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptor.java index 4baa94062e..301bbb1d7d 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptor.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptor.java @@ -16,26 +16,21 @@ */ package org.graylog.storage.opensearch2; -import org.graylog.shaded.opensearch2.org.apache.http.Header; -import org.graylog.shaded.opensearch2.org.apache.http.HttpException; -import org.graylog.shaded.opensearch2.org.apache.http.HttpResponse; -import org.graylog.shaded.opensearch2.org.apache.http.HttpResponseInterceptor; -import org.graylog.shaded.opensearch2.org.apache.http.protocol.HttpContext; -import org.graylog.shaded.opensearch2.org.opensearch.common.joda.JodaDeprecationPatterns; +import org.apache.http.HttpException; +import org.apache.http.HttpResponse; +import org.apache.http.HttpResponseInterceptor; +import org.apache.http.protocol.HttpContext; import java.io.IOException; import java.util.Arrays; -import java.util.List; -import java.util.stream.Collectors; // TODO: check, if these messages still need filtering public class OpenSearchFilterDeprecationWarningsInterceptor implements HttpResponseInterceptor { - private String[] messagesToFilter = { + private final String[] messagesToFilter = { "setting was deprecated in OpenSearch", "but in a future major version, direct access to system indices and their aliases will not be allowed", "but in a future major version, direct access to system indices will be prevented by default", - "in epoch time formats is deprecated and will not be supported in the next major version of OpenSearch", - JodaDeprecationPatterns.USE_NEW_FORMAT_SPECIFIERS + "in epoch time formats is deprecated and will not be supported in the next major version of OpenSearch" }; private boolean isDeprecationMessage(final String message) { @@ -48,8 +43,8 @@ public class OpenSearchFilterDeprecationWarningsInterceptor implements HttpRespo @Override public void process(HttpResponse response, HttpContext context) throws HttpException, IOException { - List
warnings = Arrays.stream(response.getHeaders("Warning")).filter(header -> !this.isDeprecationMessage(header.getValue())).collect(Collectors.toList()); + final var warnings = Arrays.stream(response.getHeaders("Warning")).filter(header -> !this.isDeprecationMessage(header.getValue())).toList(); response.removeHeaders("Warning"); - warnings.stream().forEach(header -> response.addHeader(header)); + warnings.forEach(response::addHeader); } } diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestClientProvider.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestClientProvider.java new file mode 100644 index 0000000000..39fe4f7371 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestClientProvider.java @@ -0,0 +1,174 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +import com.github.joschi.jadconfig.util.Duration; +import com.google.common.base.Suppliers; +import jakarta.inject.Inject; +import jakarta.inject.Named; +import jakarta.inject.Provider; +import jakarta.inject.Singleton; +import org.apache.http.HttpHost; +import org.apache.http.HttpRequestInterceptor; +import org.apache.http.client.CredentialsProvider; +import org.graylog2.configuration.IndexerHosts; +import org.graylog2.configuration.RunsWithDataNode; +import org.graylog2.security.IndexerJwtAuthTokenProvider; +import org.graylog2.security.TrustManagerAndSocketFactoryProvider; +import org.graylog2.system.shutdown.GracefulShutdownService; +import org.opensearch.client.RestClient; +import org.opensearch.client.sniff.OpenSearchNodesSniffer; + +import javax.annotation.Nullable; +import java.net.URI; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +@Singleton +public class RestClientProvider implements Provider { + private final Supplier clientSupplier; + private final TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider; + + @SuppressWarnings("unused") + @Inject + public RestClientProvider( + GracefulShutdownService shutdownService, + @IndexerHosts List hosts, + @Named("elasticsearch_connect_timeout") Duration connectTimeout, + @Named("elasticsearch_socket_timeout") Duration socketTimeout, + @Named("elasticsearch_idle_timeout") Duration elasticsearchIdleTimeout, + @Named("elasticsearch_max_total_connections") int maxTotalConnections, + @Named("elasticsearch_max_total_connections_per_route") int maxTotalConnectionsPerRoute, + @Named("elasticsearch_max_retries") int elasticsearchMaxRetries, + @Named("elasticsearch_discovery_enabled") boolean discoveryEnabled, + @Named("elasticsearch_node_activity_logger_enabled") boolean nodeActivity, + @Named("elasticsearch_discovery_filter") @Nullable String discoveryFilter, + @Named("elasticsearch_discovery_frequency") Duration discoveryFrequency, + @Named("elasticsearch_discovery_default_scheme") String defaultSchemeForDiscoveredNodes, + @Named("elasticsearch_use_expect_continue") boolean useExpectContinue, + @Named("elasticsearch_mute_deprecation_warnings") boolean muteOpenSearchDeprecationWarnings, + @Named("elasticsearch_compression_enabled") boolean compressionEnabled, + CredentialsProvider credentialsProvider, + TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider, + @RunsWithDataNode Boolean runsWithDataNode, + @Named("indexer_use_jwt_authentication") boolean indexerUseJwtAuthentication, + IndexerJwtAuthTokenProvider indexerJwtAuthTokenProvider) { + + this.trustManagerAndSocketFactoryProvider = trustManagerAndSocketFactoryProvider; + + clientSupplier = Suppliers.memoize(() -> { + final var client = buildClient(hosts, + connectTimeout, + socketTimeout, + maxTotalConnections, + maxTotalConnectionsPerRoute, + useExpectContinue, + muteOpenSearchDeprecationWarnings, + credentialsProvider, + runsWithDataNode || indexerUseJwtAuthentication, + indexerJwtAuthTokenProvider, + compressionEnabled); + + var sniffer = SnifferWrapper.create( + client, + TimeUnit.SECONDS.toMillis(5), + discoveryFrequency, + mapDefaultScheme(defaultSchemeForDiscoveredNodes) + ); + + if (discoveryEnabled) { + sniffer.add(FilteredOpenSearchNodesSniffer.create(discoveryFilter)); + } + if (nodeActivity) { + sniffer.add(NodeListSniffer.create()); + } + + sniffer.build().ifPresent(s -> shutdownService.register(s::close)); + + return client; + }); + } + + private OpenSearchNodesSniffer.Scheme mapDefaultScheme(String defaultSchemeForDiscoveredNodes) { + switch (defaultSchemeForDiscoveredNodes.toUpperCase(Locale.ENGLISH)) { + case "HTTP": + return OpenSearchNodesSniffer.Scheme.HTTP; + case "HTTPS": + return OpenSearchNodesSniffer.Scheme.HTTPS; + default: + throw new IllegalArgumentException("Invalid default scheme for discovered OS nodes: " + defaultSchemeForDiscoveredNodes); + } + } + + @Override + public RestClient get() { + return this.clientSupplier.get(); + } + + private RestClient buildClient( + List hosts, + Duration connectTimeout, + Duration socketTimeout, + int maxTotalConnections, + int maxTotalConnectionsPerRoute, + boolean useExpectContinue, + boolean muteElasticsearchDeprecationWarnings, + CredentialsProvider credentialsProvider, + boolean isJwtAuthentication, + final IndexerJwtAuthTokenProvider indexerJwtAuthTokenProvider, + boolean compressionEnabled) { + final HttpHost[] esHosts = hosts.stream().map(uri -> new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme())).toArray(HttpHost[]::new); + final var restClientBuilder = RestClient.builder(esHosts) + .setRequestConfigCallback(requestConfig -> { + requestConfig + .setConnectTimeout(Math.toIntExact(connectTimeout.toMilliseconds())) + .setSocketTimeout(Math.toIntExact(socketTimeout.toMilliseconds())) + .setExpectContinueEnabled(useExpectContinue); + // manually handle Auth if we use JWT + if (!isJwtAuthentication) { + requestConfig.setAuthenticationEnabled(true); + } + return requestConfig; + } + ) + .setHttpClientConfigCallback(httpClientConfig -> { + httpClientConfig + .setMaxConnTotal(maxTotalConnections) + .setMaxConnPerRoute(maxTotalConnectionsPerRoute); + + if (isJwtAuthentication) { + httpClientConfig.addInterceptorLast((HttpRequestInterceptor) (request, context) -> request.addHeader("Authorization", indexerJwtAuthTokenProvider.get())); + } else { + httpClientConfig.setDefaultCredentialsProvider(credentialsProvider); + } + + if (muteElasticsearchDeprecationWarnings) { + httpClientConfig.addInterceptorFirst(new OpenSearchFilterDeprecationWarningsInterceptor()); + } + + if (hosts.stream().anyMatch(host -> host.getScheme().equalsIgnoreCase("https"))) { + httpClientConfig.setSSLContext(trustManagerAndSocketFactoryProvider.getSslContext()); + } + return httpClientConfig; + }) + .setChunkedEnabled(compressionEnabled); + + return restClientBuilder.build(); + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestHighLevelClientProvider.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestHighLevelClientProvider.java index 494414cafe..5a6f9f6612 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestHighLevelClientProvider.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestHighLevelClientProvider.java @@ -30,8 +30,6 @@ import org.graylog2.configuration.RunsWithDataNode; import org.graylog2.security.IndexerJwtAuthTokenProvider; import org.graylog2.security.TrustManagerAndSocketFactoryProvider; import org.graylog2.system.shutdown.GracefulShutdownService; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nullable; @@ -48,7 +46,6 @@ import java.util.function.Supplier; @Singleton public class RestHighLevelClientProvider implements Provider { - private static final Logger LOG = LoggerFactory.getLogger(RestHighLevelClientProvider.class); private final Supplier clientSupplier; private final TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider; @@ -90,7 +87,7 @@ public class RestHighLevelClientProvider implements Provider shutdownService.register(s::close)); @@ -163,7 +160,7 @@ public class RestHighLevelClientProvider implements Provider host.getScheme().equalsIgnoreCase("https"))) { diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/SnifferWrapper.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/SnifferWrapper.java index b1485f5f05..843aab66b7 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/SnifferWrapper.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/SnifferWrapper.java @@ -17,23 +17,23 @@ package org.graylog.storage.opensearch2; import com.github.joschi.jadconfig.util.Duration; -import org.graylog.shaded.opensearch2.org.opensearch.client.Node; -import org.graylog.shaded.opensearch2.org.opensearch.client.RestClient; -import org.graylog.shaded.opensearch2.org.opensearch.client.sniff.OpenSearchNodesSniffer; -import org.graylog.shaded.opensearch2.org.opensearch.client.sniff.Sniffer; +import org.opensearch.client.Node; +import org.opensearch.client.RestClient; +import org.opensearch.client.sniff.OpenSearchNodesSniffer; +import org.opensearch.client.sniff.Sniffer; import java.io.IOException; import java.util.List; import java.util.Optional; import java.util.concurrent.CopyOnWriteArrayList; -public class SnifferWrapper implements org.graylog.shaded.opensearch2.org.opensearch.client.sniff.NodesSniffer { +public class SnifferWrapper implements org.opensearch.client.sniff.NodesSniffer { private final List sniffers = new CopyOnWriteArrayList(); private final RestClient restClient; - private final long sniffRequestTimeoutMillis; - private final Duration discoveryFrequency; + private final long sniffRequestTimeoutMillis; + private final Duration discoveryFrequency; private final OpenSearchNodesSniffer.Scheme scheme; - private org.graylog.shaded.opensearch2.org.opensearch.client.sniff.NodesSniffer nodesSniffer; + private org.opensearch.client.sniff.NodesSniffer nodesSniffer; private SnifferWrapper(RestClient restClient, long sniffRequestTimeoutMillis, Duration discoveryFrequency, OpenSearchNodesSniffer.Scheme scheme) { this.restClient = restClient; @@ -56,7 +56,7 @@ public class SnifferWrapper implements org.graylog.shaded.opensearch2.org.opense } public Optional build() { - if(sniffers.isEmpty()) { + if (sniffers.isEmpty()) { return Optional.empty(); } diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ThrowingFunction.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ThrowingFunction.java new file mode 100644 index 0000000000..8d1a898451 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ThrowingFunction.java @@ -0,0 +1,21 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2; + +public interface ThrowingFunction { + R apply(A a) throws E; +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/LegacyOSCredentialsProvider.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/LegacyOSCredentialsProvider.java new file mode 100644 index 0000000000..322b1f1651 --- /dev/null +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/LegacyOSCredentialsProvider.java @@ -0,0 +1,81 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch2.client; + +import com.google.common.base.Splitter; +import com.google.common.base.Strings; +import jakarta.inject.Inject; +import jakarta.inject.Named; +import jakarta.inject.Provider; +import org.graylog.shaded.opensearch2.org.apache.http.auth.AuthScope; +import org.graylog.shaded.opensearch2.org.apache.http.auth.UsernamePasswordCredentials; +import org.graylog.shaded.opensearch2.org.apache.http.client.CredentialsProvider; +import org.graylog.shaded.opensearch2.org.apache.http.impl.client.BasicCredentialsProvider; +import org.graylog2.configuration.IndexerHosts; + +import javax.annotation.Nullable; +import java.net.URI; +import java.util.Iterator; +import java.util.List; + +public class LegacyOSCredentialsProvider implements Provider { + private final List openSearchHosts; + private final String defaultUserForDiscoveredNodes; + private final String defaultPasswordForDiscoveredNodes; + private final boolean discoveryEnabled; + + @Inject + public LegacyOSCredentialsProvider(@IndexerHosts List openSearchHosts, + @Named("elasticsearch_discovery_default_user") @Nullable String defaultUserForDiscoveredNodes, + @Named("elasticsearch_discovery_default_password") @Nullable String defaultPasswordForDiscoveredNodes, + @Named("elasticsearch_discovery_enabled") boolean discoveryEnabled) { + this.openSearchHosts = openSearchHosts; + this.defaultUserForDiscoveredNodes = defaultUserForDiscoveredNodes; + this.defaultPasswordForDiscoveredNodes = defaultPasswordForDiscoveredNodes; + this.discoveryEnabled = discoveryEnabled; + } + + @Override + public CredentialsProvider get() { + final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); + openSearchHosts + .forEach(hostUri -> { + if (!Strings.isNullOrEmpty(hostUri.getUserInfo())) { + final Iterator splittedUserInfo = Splitter.on(":") + .split(hostUri.getUserInfo()) + .iterator(); + if (splittedUserInfo.hasNext()) { + final String username = splittedUserInfo.next(); + final String password = splittedUserInfo.hasNext() ? splittedUserInfo.next() : null; + credentialsProvider.setCredentials( + new AuthScope(hostUri.getHost(), hostUri.getPort(), AuthScope.ANY_REALM, AuthScope.ANY_SCHEME), + new UsernamePasswordCredentials(username, password) + ); + } + } + }); + + if (discoveryEnabled && !Strings.isNullOrEmpty(defaultUserForDiscoveredNodes) && !Strings.isNullOrEmpty(defaultPasswordForDiscoveredNodes)) { + credentialsProvider.setCredentials( + new AuthScope(AuthScope.ANY_HOST, AuthScope.ANY_PORT, AuthScope.ANY_REALM, AuthScope.ANY_SCHEME), + new UsernamePasswordCredentials(defaultUserForDiscoveredNodes, defaultPasswordForDiscoveredNodes) + ); + } + + return credentialsProvider; + } +} diff --git a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/OSCredentialsProvider.java b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/OSCredentialsProvider.java index 5b48f5e45a..99891aba18 100644 --- a/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/OSCredentialsProvider.java +++ b/graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/OSCredentialsProvider.java @@ -18,10 +18,10 @@ package org.graylog.storage.opensearch2.client; import com.google.common.base.Splitter; import com.google.common.base.Strings; -import org.graylog.shaded.opensearch2.org.apache.http.auth.AuthScope; -import org.graylog.shaded.opensearch2.org.apache.http.auth.UsernamePasswordCredentials; -import org.graylog.shaded.opensearch2.org.apache.http.client.CredentialsProvider; -import org.graylog.shaded.opensearch2.org.apache.http.impl.client.BasicCredentialsProvider; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.impl.client.BasicCredentialsProvider; import org.graylog2.configuration.IndexerHosts; import javax.annotation.Nullable; diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java index 2efc24b649..d2396ebda5 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java @@ -107,7 +107,7 @@ class ClusterAdapterOS2Test { @Test void returnsEmptyOptionalForHealthWhenElasticsearchExceptionThrown() { - when(client.execute(any())).thenThrow(new OpenSearchException("Exception")); + when(client.execute(any(ThrowingBiFunction.class))).thenThrow(new OpenSearchException("Exception")); final Optional healthStatus = clusterAdapter.health(); assertThat(healthStatus).isEmpty(); } diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSnifferTest.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSnifferTest.java index 698280716a..fa916866b0 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSnifferTest.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSnifferTest.java @@ -39,7 +39,7 @@ class FilteredOpenSearchNodesSnifferTest { void doesNotFilterNodesIfNoFilterIsSet() throws Exception { final List nodes = mockNodes(); - final NodesSniffer nodesSniffer = new FilteredOpenSearchNodesSniffer(null, null); + final LegacyNodesSniffer nodesSniffer = new LegacyFilteredOpenSearchNodesSniffer(null, null); assertThat(nodesSniffer.sniff(nodes)).isEqualTo(nodes); } @@ -48,7 +48,7 @@ class FilteredOpenSearchNodesSnifferTest { void worksWithEmptyNodesListIfFilterIsSet() throws Exception { final List nodes = Collections.emptyList(); - final NodesSniffer nodesSniffer = new FilteredOpenSearchNodesSniffer("rack", "42"); + final LegacyNodesSniffer nodesSniffer = new LegacyFilteredOpenSearchNodesSniffer("rack", "42"); assertThat(nodesSniffer.sniff(nodes)).isEqualTo(nodes); } @@ -57,7 +57,7 @@ class FilteredOpenSearchNodesSnifferTest { void returnsNodesMatchingGivenFilter() throws Exception { final List nodes = mockNodes(); - final NodesSniffer nodesSniffer = new FilteredOpenSearchNodesSniffer("rack", "42"); + final LegacyNodesSniffer nodesSniffer = new LegacyFilteredOpenSearchNodesSniffer("rack", "42"); assertThat(nodesSniffer.sniff(nodes)).containsExactly(nodeOnRack42); } @@ -66,7 +66,7 @@ class FilteredOpenSearchNodesSnifferTest { void returnsNoNodesIfFilterDoesNotMatch() throws Exception { final List nodes = mockNodes(); - final NodesSniffer nodesSniffer = new FilteredOpenSearchNodesSniffer("location", "alaska"); + final LegacyNodesSniffer nodesSniffer = new LegacyFilteredOpenSearchNodesSniffer("location", "alaska"); assertThat(nodesSniffer.sniff(nodes)).isEmpty(); } @@ -75,7 +75,7 @@ class FilteredOpenSearchNodesSnifferTest { void returnsAllNodesIfFilterMatchesAll() throws Exception { final List nodes = mockNodes(); - final NodesSniffer nodesSniffer = new FilteredOpenSearchNodesSniffer("always", "true"); + final LegacyNodesSniffer nodesSniffer = new LegacyFilteredOpenSearchNodesSniffer("always", "true"); assertThat(nodesSniffer.sniff(nodes)).isEqualTo(nodes); } @@ -87,7 +87,7 @@ class FilteredOpenSearchNodesSnifferTest { )); final List nodes = Collections.singletonList(matchingNode); - final NodesSniffer nodesSniffer = new FilteredOpenSearchNodesSniffer("something", "42"); + final LegacyNodesSniffer nodesSniffer = new LegacyFilteredOpenSearchNodesSniffer("something", "42"); assertThat(nodesSniffer.sniff(nodes)).isEqualTo(nodes); } @@ -112,8 +112,8 @@ class FilteredOpenSearchNodesSnifferTest { return ImmutableList.of(nodeOnRack42, nodeOnRack23, nodeWithNoAttributes); } - private NodesSniffer mockSniffer(List nodes) throws IOException { - final NodesSniffer mockSniffer = mock(NodesSniffer.class); + private LegacyNodesSniffer mockSniffer(List nodes) throws IOException { + final LegacyNodesSniffer mockSniffer = mock(LegacyNodesSniffer.class); when(mockSniffer.sniff(nodes)).thenReturn(nodes); return mockSniffer; } diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/NodeAdapterOS2Test.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/NodeAdapterOS2Test.java index c80245dabe..85aeae0376 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/NodeAdapterOS2Test.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/NodeAdapterOS2Test.java @@ -16,43 +16,42 @@ */ package org.graylog.storage.opensearch2; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.github.zafarkhaja.semver.Version; import org.graylog.shaded.opensearch2.org.opensearch.client.Request; +import org.graylog2.shared.bindings.providers.ObjectMapperProvider; import org.graylog2.storage.SearchVersion; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; - -import java.io.IOException; +import org.opensearch.client.opensearch._types.OpenSearchVersionInfo; +import org.opensearch.client.opensearch.core.InfoResponse; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.doReturn; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; class NodeAdapterOS2Test { private NodeAdapterOS2 toTest; - private PlainJsonApi jsonApiMock; + private OpenSearchClient openSearchClient; private Request request; - private final ObjectMapper objectMapper = new ObjectMapper(); + private final ObjectMapper objectMapper = new ObjectMapperProvider().get(); @BeforeEach void setUp() { - jsonApiMock = mock(PlainJsonApi.class); - toTest = new NodeAdapterOS2(jsonApiMock); + openSearchClient = mock(OpenSearchClient.class); + toTest = new NodeAdapterOS2(openSearchClient); request = new Request("GET", "/?filter_path=version.number,version.distribution"); } @Test - void testElasticsearchVersionFetching() throws IOException { - mockResponse("{\"version\" : " + - " {" + - " \"number\" : \"7.10.2\"" + - " }" + - "}"); + void testElasticsearchVersionFetching() { + mockResponse(responseBuilder() + .version(versionBuilder() + .distribution("Elasticsearch") + .number("7.10.2") + .build()).build()); assertThat(toTest.version()) .isNotEmpty() @@ -61,13 +60,13 @@ class NodeAdapterOS2Test { } @Test - void testOpensearchVersionFetching() throws IOException { - mockResponse("{\"version\" : " + - " {" + - " \"distribution\" : \"opensearch\"," + - " \"number\" : \"1.3.1\"" + - " }" + - "}"); + void testOpensearchVersionFetching() { + mockResponse(responseBuilder() + .version(versionBuilder() + .distribution("opensearch") + .number("1.3.1") + .build()) + .build()); assertThat(toTest.version()) .isNotEmpty() @@ -75,8 +74,26 @@ class NodeAdapterOS2Test { } - private void mockResponse(final String jsonResponseWithVersion) throws IOException { - JsonNode jsonNode = objectMapper.readTree(jsonResponseWithVersion); - doReturn(jsonNode).when(jsonApiMock).perform(eq(request), anyString()); + private InfoResponse.Builder responseBuilder() { + return new InfoResponse.Builder() + .name("node01") + .clusterName("testcluster") + .clusterUuid("deadbeef") + .tagline("The best product for search"); + } + + private OpenSearchVersionInfo.Builder versionBuilder() { + return new OpenSearchVersionInfo.Builder() + .buildDate("2021-12-01") + .buildHash("deadbeef") + .buildSnapshot(false) + .buildType("release") + .luceneVersion("9.0.0") + .minimumWireCompatibilityVersion("7.0.0") + .minimumIndexCompatibilityVersion("7.0.0"); + } + + private void mockResponse(final InfoResponse response) { + when(openSearchClient.execute(any(ThrowingFunction.class))).thenReturn(response); } } diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchExceptionTest.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchExceptionTest.java index 66b25fb0e8..bc70855b36 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchExceptionTest.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchExceptionTest.java @@ -50,6 +50,9 @@ public class OpenSearchExceptionTest { @Mock RestHighLevelClient restHighLevelClient; + @Mock + org.opensearch.client.opensearch.OpenSearchClient openSearchClient; + // Verify that an OpenSearchStatusException is translated to a BatchSizeTooLargeException. @Test public void handle429() throws IOException { @@ -143,7 +146,7 @@ public class OpenSearchExceptionTest { RestStatus restStatus = RestStatus.BAD_REQUEST; OpenSearchStatusException statusException = new OpenSearchStatusException( "status msg", restStatus, responseException); - final OpenSearchClient openSearchClient = new OpenSearchClient(restHighLevelClient, new ObjectMapper()); + final var openSearchClient = new OpenSearchClient(restHighLevelClient, this.openSearchClient, new ObjectMapper()); Exception exception = assertThrows(BatchSizeTooLargeException.class, () -> { openSearchClient.execute((a, b) -> {throw statusException;}); diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptorTest.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptorTest.java index 3d9eeaa885..c0ae8672c3 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptorTest.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptorTest.java @@ -16,11 +16,10 @@ */ package org.graylog.storage.opensearch2; -import org.graylog.shaded.opensearch2.org.apache.http.HttpException; -import org.graylog.shaded.opensearch2.org.apache.http.HttpResponse; -import org.graylog.shaded.opensearch2.org.apache.http.ProtocolVersion; -import org.graylog.shaded.opensearch2.org.apache.http.message.BasicHttpResponse; -import org.graylog.shaded.opensearch2.org.apache.http.message.BasicStatusLine; +import org.apache.http.HttpException; +import org.apache.http.ProtocolVersion; +import org.apache.http.message.BasicHttpResponse; +import org.apache.http.message.BasicStatusLine; import org.junit.Test; import java.io.IOException; @@ -32,7 +31,7 @@ public class OpenSearchFilterDeprecationWarningsInterceptorTest { public void testInterceptorNoHeader() throws IOException, HttpException { OpenSearchFilterDeprecationWarningsInterceptor interceptor = new OpenSearchFilterDeprecationWarningsInterceptor(); - HttpResponse response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); + final var response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); interceptor.process(response, null); assertThat(response.getAllHeaders()) @@ -44,7 +43,7 @@ public class OpenSearchFilterDeprecationWarningsInterceptorTest { public void testInterceptorSingleHeader() throws IOException, HttpException { OpenSearchFilterDeprecationWarningsInterceptor interceptor = new OpenSearchFilterDeprecationWarningsInterceptor(); - HttpResponse response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); + final var response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); response.addHeader("Test", "This header should not trigger the interceptor."); interceptor.process(response, null); @@ -61,7 +60,7 @@ public class OpenSearchFilterDeprecationWarningsInterceptorTest { public void testInterceptorMultipleHeaderIgnoredWarning() throws IOException, HttpException { OpenSearchFilterDeprecationWarningsInterceptor interceptor = new OpenSearchFilterDeprecationWarningsInterceptor(); - HttpResponse response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); + final var response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); response.addHeader("Test", "This header should not trigger the interceptor."); response.addHeader("Warning", "This warning should not trigger the interceptor."); interceptor.process(response, null); @@ -75,7 +74,7 @@ public class OpenSearchFilterDeprecationWarningsInterceptorTest { public void testInterceptorMultipleHeaderFilteredWarning() throws IOException, HttpException { OpenSearchFilterDeprecationWarningsInterceptor interceptor = new OpenSearchFilterDeprecationWarningsInterceptor(); - HttpResponse response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); + final var response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); response.addHeader("Test", "This header should not trigger the interceptor."); response.addHeader("Warning", "This warning should not trigger the interceptor."); response.addHeader("Warning", "This text contains the trigger: setting was deprecated in OpenSearch - and should be filtered out"); @@ -95,7 +94,7 @@ public class OpenSearchFilterDeprecationWarningsInterceptorTest { public void testInterceptorMultipleHeaderFilteredWarning2() throws IOException, HttpException { OpenSearchFilterDeprecationWarningsInterceptor interceptor = new OpenSearchFilterDeprecationWarningsInterceptor(); - HttpResponse response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); + final var response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); response.addHeader("Test", "This header should not trigger the interceptor."); response.addHeader("Warning", "This warning should not trigger the interceptor."); response.addHeader("Warning", "This text contains the trigger: but in a future major version, direct access to system indices and their aliases will not be allowed - and should be filtered out"); @@ -115,7 +114,7 @@ public class OpenSearchFilterDeprecationWarningsInterceptorTest { public void testInterceptorMultipleHeaderFilteredWarningAndMultipleTriggers() throws IOException, HttpException { OpenSearchFilterDeprecationWarningsInterceptor interceptor = new OpenSearchFilterDeprecationWarningsInterceptor(); - HttpResponse response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); + final var response = new BasicHttpResponse(new BasicStatusLine(new ProtocolVersion("HTTP", 0, 0), 0, null)); response.addHeader("Test", "This header should not trigger the interceptor."); response.addHeader("Warning", "This warning should not trigger the interceptor."); response.addHeader("Warning", "This text contains the trigger: but in a future major version, direct access to system indices and their aliases will not be allowed - and should be filtered out"); diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/cat/CatApiTest.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/cat/CatApiTest.java index 3c18e69329..d7e6a9d720 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/cat/CatApiTest.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/cat/CatApiTest.java @@ -19,6 +19,7 @@ package org.graylog.storage.opensearch2.cat; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import org.graylog.storage.opensearch2.OpenSearchClient; +import org.graylog.storage.opensearch2.ThrowingBiFunction; import org.graylog2.shared.bindings.providers.ObjectMapperProvider; import org.junit.jupiter.api.Test; @@ -57,7 +58,7 @@ class CatApiTest { final OpenSearchClient client = mock(OpenSearchClient.class); final CatApi toTest = new CatApi(objectMapper, client); - when(client.execute(any(), anyString())) + when(client.execute(any(ThrowingBiFunction.class), anyString())) .thenReturn(objectMapper.readValue(SAMPLE_CAT_NODES_RESPONSE, new TypeReference>() {})); final List nodes = toTest.nodes(); diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/AdaptersOS2.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/AdaptersOS2.java index 9293ad83f5..17fb5d9d32 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/AdaptersOS2.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/AdaptersOS2.java @@ -72,7 +72,7 @@ public class AdaptersOS2 implements Adapters { @Override public NodeAdapter nodeAdapter() { - return new NodeAdapterOS2(client, objectMapper); + return new NodeAdapterOS2(client); } @Override diff --git a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java index 82d5059adf..9083a96a80 100644 --- a/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java +++ b/graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java @@ -34,6 +34,8 @@ import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.Composable import org.graylog.shaded.opensearch2.org.opensearch.cluster.metadata.Template; import org.graylog.shaded.opensearch2.org.opensearch.common.settings.Settings; import org.graylog.storage.opensearch2.OpenSearchClient; +import org.graylog.storage.opensearch2.OpenSearchClientProvider; +import org.graylog.storage.opensearch2.RestClientProvider; import org.graylog.storage.opensearch2.RestHighLevelClientProvider; import org.graylog.testing.containermatrix.SearchServer; import org.graylog.testing.elasticsearch.Adapters; @@ -43,6 +45,7 @@ import org.graylog.testing.elasticsearch.TestableSearchServerInstance; import org.graylog2.shared.bindings.providers.ObjectMapperProvider; import org.graylog2.storage.SearchVersion; import org.graylog2.system.shutdown.GracefulShutdownService; +import org.opensearch.client.RestClient; import org.opensearch.testcontainers.OpensearchContainer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -80,8 +83,10 @@ public class OpenSearchInstance extends TestableSearchServerInstance { @Override public OpenSearchInstance init() { super.init(); - RestHighLevelClient restHighLevelClient = buildRestClient(); - this.openSearchClient = new OpenSearchClient(restHighLevelClient, new ObjectMapperProvider().get()); + RestHighLevelClient restHighLevelClient = buildRestHighLevelClient(); + final var objectMapper = new ObjectMapperProvider().get(); + final var restClient = buildRestClient(); + this.openSearchClient = new OpenSearchClient(restHighLevelClient, new OpenSearchClientProvider(restClient, objectMapper).get(), objectMapper); this.client = new ClientOS2(this.openSearchClient, featureFlags); this.fixtureImporter = new FixtureImporterOS2(this.openSearchClient); adapters = new AdaptersOS2(openSearchClient); @@ -92,6 +97,32 @@ public class OpenSearchInstance extends TestableSearchServerInstance { return this; } + private RestClient buildRestClient() { + return new RestClientProvider( + new GracefulShutdownService(), + ImmutableList.of(URI.create("http://" + this.getHttpHostAddress())), + Duration.seconds(60), + Duration.seconds(60), + Duration.seconds(60), + 1, + 1, + 1, + false, + false, + null, + Duration.seconds(60), + "http", + false, + false, + false, + new org.apache.http.impl.client.BasicCredentialsProvider(), + null, + false, + false, + null) + .get(); + } + public static OpenSearchInstance create() { return OpenSearchInstanceBuilder.builder().instantiate(); } @@ -160,7 +191,7 @@ public class OpenSearchInstance extends TestableSearchServerInstance { return OPENSEARCH_VERSION; } - private RestHighLevelClient buildRestClient() { + private RestHighLevelClient buildRestHighLevelClient() { return new RestHighLevelClientProvider( new GracefulShutdownService(), ImmutableList.of(URI.create("http://" + this.getHttpHostAddress())),