opensource/graylog2-server
1
0
Fork 0
mirror of https://github.com/Graylog2/graylog2-server.git synced 2026-03-13 09:32:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

Setting up injection/usage of new OpenSearch Java Client. (#17760)

Browse Source 
* 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 <bernd@graylog.com>
Co-authored-by: Jan Heise <jan.heise@graylog.com>
This commit is contained in:
Dennis Oelkers
2024-01-19 10:37:44 +01:00
committed by GitHub
parent 2749333825
commit aacaa0c4d2
31 changed files with 854 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java
View File

@@ -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<String> 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!");}
);

26
data-node/src/main/java/org/graylog/datanode/management/OpensearchRestClient.java
View File

@@ -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();
}
}

18
graylog-storage-opensearch2/assembly.xml
View File

@@ -14,11 +14,29 @@
<scope>runtime</scope>
<useProjectArtifact>true</useProjectArtifact>
<useTransitiveDependencies>true</useTransitiveDependencies>
<useStrictFiltering>true</useStrictFiltering>
<includes>
<include>org.graylog:graylog-storage-opensearch2</include>
<include>org.graylog.shaded:opensearch2-rest-high-level-client</include>
<include>org.graylog.shaded:opensearch2-rest-client-sniffer</include>
<include>org.opensearch.client:opensearch-java</include>
<include>org.opensearch.client:opensearch-rest-client</include>
<include>org.opensearch.client:opensearch-rest-client-sniffer</include>
<include>org.apache.httpcomponents:httpasyncclient</include>
<include>org.apache.httpcomponents:httpclient</include>
<include>org.apache.httpcomponents:httpcore</include>
<include>org.apache.httpcomponents:httpcore-nio</include>
<include>jakarta.json.bind:jakarta.json.bind-api</include>
<include>jakarta.json:jakarta.json-api</include>
</includes>
<unpackOptions>
<excludes>
<!-- Including module-info.class files would turn the
artifact into a Java module and we would run into
class loader issues. -->
<exclude>module-info.class</exclude>
</excludes>
</unpackOptions>
</dependencySet>
</dependencySets>
</assembly>

33
graylog-storage-opensearch2/pom.xml
View File

@@ -72,6 +72,39 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.opensearch.client</groupId>
<artifactId>opensearch-java</artifactId>
<version>2.8.1</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.opensearch.client</groupId>
<artifactId>opensearch-rest-client</artifactId>
<version>2.11.1</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.opensearch.client</groupId>
<artifactId>opensearch-rest-client-sniffer</artifactId>
<version>2.11.1</version>
<exclusions>
<exclusion>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.jsonwebtoken</groupId>
<artifactId>jjwt-api</artifactId>

2
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSniffer.java
View File

@@ -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;

71
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyFilteredOpenSearchNodesSniffer.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<Node> sniff(List<Node> 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);
}
}

60
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodeListSniffer.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<String> savedNodes = ConcurrentHashMap.newKeySet();
static LegacyNodeListSniffer create() {
return new LegacyNodeListSniffer();
}
@Override
public List<Node> sniff(final List<Node> nodes) {
final Set<String> currentNodes = nodes.stream().map(n -> n.getHost().toURI()).collect(Collectors.toSet());
final Set<String> nodesAdded = Sets.difference(currentNodes, savedNodes);
final Set<String> 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;
}
}

26
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyNodesSniffer.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<Node> sniff(List<Node> nodes) throws IOException;
}

54
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacyOpenSearchFilterDeprecationWarningsInterceptor.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<Header> warnings = Arrays.stream(response.getHeaders("Warning")).filter(header -> !this.isDeprecationMessage(header.getValue())).toList();
response.removeHeaders("Warning");
warnings.forEach(response::addHeader);
}
}

73
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/LegacySnifferWrapper.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<LegacyNodesSniffer> 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<Node> sniff() throws IOException {
List<Node> 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<Sniffer> 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);
}
}

21
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeAdapterOS2.java
View File

@@ -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<SearchVersion> version() {
var info = client.execute(c -> c.info());
final Request request = new Request("GET", "/?filter_path=version.number,version.distribution");
final Optional<JsonNode> resp = Optional.of(jsonApi.perform(request, "Unable to retrieve cluster information"));
final Optional<String> version = Optional.ofNullable(info.version().number());
final Optional<String> 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);

2
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodeListSniffer.java
View File

@@ -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;

3
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/NodesSniffer.java
View File

@@ -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;

9
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearch2Module.java
View File

@@ -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 <T> LinkedBindingBuilder<T> bindForSupportedVersion(Class<T> interfaceClass) {

18
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClient.java
View File

@@ -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<Integer> indexerMaxConcurrentSearches;
private final Optional<Integer> 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> R execute(ThrowingFunction<org.opensearch.client.opensearch.OpenSearchClient, R, IOException> fn, String errorMessage) {
try {
return fn.apply(openSearchClient);
} catch (Exception e) {
throw exceptionFrom(e, errorMessage);
}
}
public <R> R execute(ThrowingFunction<org.opensearch.client.opensearch.OpenSearchClient, R, IOException> fn) {
return execute(fn, "An error occurred: ");
}
@WithSpan
public <R> R executeWithIOException(ThrowingBiFunction<RestHighLevelClient, RequestOptions, R, IOException> fn, String errorMessage) throws IOException {
try {

45
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchClientProvider.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<OpenSearchClient> {
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)));
}
}

21
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptor.java
View File

@@ -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<Header> 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);
}
}

174
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestClientProvider.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<RestClient> {
private final Supplier<RestClient> clientSupplier;
private final TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider;
@SuppressWarnings("unused")
@Inject
public RestClientProvider(
GracefulShutdownService shutdownService,
@IndexerHosts List<URI> 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<URI> 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();
}
}

11
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/RestHighLevelClientProvider.java
View File

@@ -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<RestHighLevelClient> {
private static final Logger LOG = LoggerFactory.getLogger(RestHighLevelClientProvider.class);
private final Supplier<RestHighLevelClient> clientSupplier;
private final TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider;
@@ -90,7 +87,7 @@ public class RestHighLevelClientProvider implements Provider<RestHighLevelClient
runsWithDataNode || indexerUseJwtAuthentication,
indexerJwtAuthTokenProvider);
var sniffer = SnifferWrapper.create(
var sniffer = LegacySnifferWrapper.create(
client.getLowLevelClient(),
TimeUnit.SECONDS.toMillis(5),
discoveryFrequency,
@@ -98,10 +95,10 @@ public class RestHighLevelClientProvider implements Provider<RestHighLevelClient
);
if (discoveryEnabled) {
sniffer.add(FilteredOpenSearchNodesSniffer.create(discoveryFilter));
sniffer.add(LegacyFilteredOpenSearchNodesSniffer.create(discoveryFilter));
}
if (nodeActivity) {
sniffer.add(NodeListSniffer.create());
sniffer.add(LegacyNodeListSniffer.create());
}
sniffer.build().ifPresent(s -> shutdownService.register(s::close));
@@ -163,7 +160,7 @@ public class RestHighLevelClientProvider implements Provider<RestHighLevelClient
}
if (muteElasticsearchDeprecationWarnings) {
httpClientConfig.addInterceptorFirst(new OpenSearchFilterDeprecationWarningsInterceptor());
httpClientConfig.addInterceptorFirst(new LegacyOpenSearchFilterDeprecationWarningsInterceptor());
}
if (hosts.stream().anyMatch(host -> host.getScheme().equalsIgnoreCase("https"))) {

18
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/SnifferWrapper.java
View File

@@ -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<NodesSniffer> 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<Sniffer> build() {
if(sniffers.isEmpty()) {
if (sniffers.isEmpty()) {
return Optional.empty();
}

21
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ThrowingFunction.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.storage.opensearch2;
public interface ThrowingFunction<A, R, E extends Exception> {
R apply(A a) throws E;
}

81
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/LegacyOSCredentialsProvider.java Normal file
View File

@@ -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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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<CredentialsProvider> {
private final List<URI> openSearchHosts;
private final String defaultUserForDiscoveredNodes;
private final String defaultPasswordForDiscoveredNodes;
private final boolean discoveryEnabled;
@Inject
public LegacyOSCredentialsProvider(@IndexerHosts List<URI> 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<String> 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;
}
}

8
graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/client/OSCredentialsProvider.java
View File

@@ -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;

2
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/ClusterAdapterOS2Test.java
View File

@@ -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> healthStatus = clusterAdapter.health();
assertThat(healthStatus).isEmpty();
}

16
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/FilteredOpenSearchNodesSnifferTest.java
View File

@@ -39,7 +39,7 @@ class FilteredOpenSearchNodesSnifferTest {
void doesNotFilterNodesIfNoFilterIsSet() throws Exception {
final List<Node> 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<Node> 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<Node> 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<Node> 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<Node> 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<Node> 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<Node> nodes) throws IOException {
final NodesSniffer mockSniffer = mock(NodesSniffer.class);
private LegacyNodesSniffer mockSniffer(List<Node> nodes) throws IOException {
final LegacyNodesSniffer mockSniffer = mock(LegacyNodesSniffer.class);
when(mockSniffer.sniff(nodes)).thenReturn(nodes);
return mockSniffer;
}

69
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/NodeAdapterOS2Test.java
View File

@@ -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);
}
}

5
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchExceptionTest.java
View File

@@ -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;});

21
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/OpenSearchFilterDeprecationWarningsInterceptorTest.java
View File

@@ -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");

3
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/cat/CatApiTest.java
View File

@@ -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<List<NodeResponse>>() {}));
final List<NodeResponse> nodes = toTest.nodes();

2
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/AdaptersOS2.java
View File

@@ -72,7 +72,7 @@ public class AdaptersOS2 implements Adapters {
@Override
public NodeAdapter nodeAdapter() {
return new NodeAdapterOS2(client, objectMapper);
return new NodeAdapterOS2(client);
}
@Override

37
graylog-storage-opensearch2/src/test/java/org/graylog/storage/opensearch2/testing/OpenSearchInstance.java
View File

@@ -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())),