Simplified plain json call to opensearch, refactored cluster adapter for os3 (#24948)

Author: todvora

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/ClusterAdapterOS.java

@@ -76,19 +76,12 @@ public class ClusterAdapterOS implements ClusterAdapter {
     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdapterOS.class);
     private final Duration requestTimeout;
     private final OfficialOpensearchClient opensearchClient;
-    private final PlainJsonApi jsonApi;
-    private final OpenSearchCatClient catClient;
-    private final OpenSearchClusterClient clusterClient;
 
     @Inject
     public ClusterAdapterOS(OfficialOpensearchClient opensearchClient,
-                            @Named("elasticsearch_socket_timeout") Duration requestTimeout,
-                            PlainJsonApi jsonApi) {
+                            @Named("elasticsearch_socket_timeout") Duration requestTimeout) {
         this.requestTimeout = requestTimeout;
-        this.jsonApi = jsonApi;
         this.opensearchClient = opensearchClient;
-        this.catClient = opensearchClient.sync().cat();
-        this.clusterClient = opensearchClient.sync().cluster();
     }
 
     @Override
@@ -117,8 +110,8 @@ public Set<NodeFileDescriptorStats> fileDescriptorStats() {
     }
 
     List<NodesRecord> nodes() {
-        List<NodesRecord> allNodes = opensearchClient.execute(() ->
-                        catClient.nodes(NodesRequest.builder()
+        List<NodesRecord> allNodes = opensearchClient.sync(c ->
+                        c.cat().nodes(NodesRequest.builder()
                                 .fullId(true)
                                 .headers("id", "name", "node.role", "ip", "version",
                                         "file_desc.max", "disk.used", "disk.total", "disk.used_percent")
@@ -153,8 +146,8 @@ public ClusterShardAllocation clusterShardAllocation() {
             LOG.warn("Could not retrieve max_shards_per_node setting from cluster settings. Threshold warnings disabled.", e);
         }
 
-        List<NodeShardAllocation> nodeShardAllocations = opensearchClient.execute(() ->
-                        catClient.allocation().valueBody().stream().map(this::toNodeShardAllocation).toList(),
+        List<NodeShardAllocation> nodeShardAllocations = opensearchClient.sync(c ->
+                        c.cat().allocation().valueBody().stream().map(this::toNodeShardAllocation).toList(),
                 "Unable to retrieve node shard allocation"
         );
         return new ClusterShardAllocation(maxShardsPerNode, nodeShardAllocations);
@@ -216,7 +209,7 @@ private Optional<JsonNode> nodeById(String nodeId) {
                 .endpoint("/_nodes/" + nodeId)
                 .method("GET")
                 .build();
-        return Optional.of(jsonApi.performRequest(request, "Couldn't read Opensearch nodes data!"))
+        return Optional.of(opensearchClient.performRequest(request, "Couldn't read Opensearch nodes data!"))
                 .map(jsonNode -> jsonNode.path("nodes").path(nodeId))
                 .filter(node -> !node.isMissingNode());
     }
@@ -250,8 +243,8 @@ private ClusterHealth clusterHealthFrom(HealthResponse response) {
 
     @Override
     public PendingTasksStats pendingTasks() {
-        PendingTasksResponse pendingTasks = opensearchClient.execute(
-                clusterClient::pendingTasks,
+        PendingTasksResponse pendingTasks = opensearchClient.sync(c ->
+                c.cluster().pendingTasks(),
                 "Unable to retrieve pending tasks"
         );
 
@@ -302,7 +295,7 @@ public JsonNode rawClusterStats() {
                 .endpoint("/_cluster/stats/nodes/*")
                 .method("GET")
                 .build();
-        return jsonApi.performRequest(request, "Couldn't read Elasticsearch cluster stats");
+        return opensearchClient.performRequest(request, "Couldn't read Elasticsearch cluster stats");
     }
 
     @Override
@@ -311,7 +304,7 @@ public Map<String, org.graylog2.system.stats.elasticsearch.NodeInfo> nodesInfo()
                 .endpoint("/_nodes")
                 .method("GET")
                 .build();
-        JsonNode json = jsonApi.performRequest(request, "Couldn't read Opensearch nodes data!");
+        JsonNode json = opensearchClient.performRequest(request, "Couldn't read Opensearch nodes data!");
         JsonNode nodes = json.at("/nodes");
         return toStream(nodes.fieldNames()).collect(Collectors.toMap(
                 n -> n,
@@ -334,7 +327,7 @@ public Map<String, NodeOSInfo> nodesHostInfo() {
                 .endpoint("/_nodes/stats/os")
                 .method("GET")
                 .build();
-        JsonNode json = jsonApi.performRequest(request, "Couldn't read Opensearch nodes os data!");
+        JsonNode json = opensearchClient.performRequest(request, "Couldn't read Opensearch nodes os data!");
         JsonNode nodes = json.at("/nodes");
         return toStream(nodes.fieldNames())
                 .collect(Collectors.toMap(name -> name, name -> createNodeHostInfo(nodes.get(name))));
@@ -370,7 +363,7 @@ public ShardStats shardStats() {
     private Optional<HealthResponse> clusterHealth() {
         final Time timeout = new Time.Builder().time(requestTimeout.toSeconds() + "s").build();
         try {
-            HealthResponse health = clusterClient.health(HealthRequest.builder()
+            HealthResponse health = opensearchClient.syncWithoutErrorMapping().cluster().health(HealthRequest.builder()
                     .timeout(timeout)
                     .build()
             );
@@ -392,7 +385,7 @@ public Optional<HealthStatus> deflectorHealth(Collection<String> indices) {
         }
 
         final Map<String, String> aliasMapping;
-        aliasMapping = opensearchClient.execute(() -> catClient.aliases().valueBody()
+        aliasMapping = opensearchClient.sync(c -> c.cat().aliases().valueBody()
                         .stream()
                         .filter(alias -> Objects.nonNull(alias.index()))
                         .collect(Collectors.toMap(AliasesRecord::alias, AliasesRecord::index)),
@@ -404,8 +397,8 @@ public Optional<HealthStatus> deflectorHealth(Collection<String> indices) {
                 .map(index -> aliasMapping.getOrDefault(index, index))
                 .collect(Collectors.toSet());
 
-        final Set<IndicesRecord> indexSummaries = opensearchClient.execute(() ->
-                        catClient.indices().valueBody()
+        final Set<IndicesRecord> indexSummaries = opensearchClient.sync(client ->
+                        client.cat().indices().valueBody()
                                 .stream()
                                 .filter(indexSummary -> mappedIndices.contains(indexSummary.index()))
                                 .collect(Collectors.toSet()),
@@ -423,7 +416,7 @@ public Optional<HealthStatus> deflectorHealth(Collection<String> indices) {
     }
 
     private GetClusterSettingsResponse getClusterSettings() {
-        return opensearchClient.execute(() -> clusterClient.getSettings(
+        return opensearchClient.sync(c -> c.cluster().getSettings(
                 GetClusterSettingsRequest.builder()
                         .includeDefaults(true)
                         .flatSettings(true)

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/OfficialOpensearchClient.java

@@ -16,6 +16,9 @@
  */
 package org.graylog.storage.opensearch3;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.joschi.jadconfig.util.Duration;
 import org.apache.hc.core5.http.ContentTooLongException;
 import org.graylog.storage.exceptions.ParsedOpenSearchException;
@@ -30,22 +33,36 @@
 import org.opensearch.client.opensearch.OpenSearchClient;
 import org.opensearch.client.opensearch._types.ErrorCause;
 import org.opensearch.client.opensearch._types.OpenSearchException;
+import org.opensearch.client.opensearch.generic.Body;
+import org.opensearch.client.opensearch.generic.Request;
+import org.opensearch.client.opensearch.generic.Response;
 import org.opensearch.client.transport.httpclient5.ResponseException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-public record OfficialOpensearchClient(OpenSearchClient sync, OpenSearchAsyncClient async) {
+public final class OfficialOpensearchClient {
 
     private static final Logger LOG = LoggerFactory.getLogger(OfficialOpensearchClient.class);
     private static final Pattern invalidWriteTarget = Pattern.compile("no write index is defined for alias \\[(?<target>[\\w_]+)\\]");
 
+    private final OpenSearchClient sync;
+    private final OpenSearchAsyncClient async;
+    private final ObjectMapper objectMapper;
+
+    public OfficialOpensearchClient(OpenSearchClient sync, OpenSearchAsyncClient async, ObjectMapper objectMapper) {
+        this.sync = sync;
+        this.async = async;
+        this.objectMapper = objectMapper;
+    }
+
     public <T> T execute(ThrowingSupplier<T> operation, String errorMessage) {
         try {
             return operation.get();
@@ -95,6 +112,19 @@ public <T> CompletableFuture<T> async(ThrowingAsyncFunction<CompletableFuture<T>
         }
     }
 
+    /**
+     * This shouldn't be used unless you have very specific needs and require JsonNode as a response from plain json API
+     */
+    public JsonNode performRequest(Request request, String errorMessage) {
+        String rawJson;
+        try (Response response = sync.generic().execute(request)) {
+            rawJson = response.getBody().map(Body::bodyAsString).orElse("");
+            return objectMapper.readTree(rawJson);
+        } catch (Exception e) {
+            throw mapException(e, errorMessage);
+        }
+    }
+
     public void close() {
         try {
             sync()._transport().close();
@@ -220,4 +250,22 @@ private static boolean isParentCircuitBreakingException(OpenSearchException open
         }
         return false;
     }
+
+    @Deprecated
+    public OpenSearchClient sync() {
+        return sync;
+    }
+
+    @Deprecated
+    public OpenSearchAsyncClient async() {
+        return async;
+    }
+
+    public OpenSearchClient syncWithoutErrorMapping() {
+        return sync;
+    }
+
+    public OpenSearchAsyncClient asyncWithoutErrorMapping() {
+        return async;
+    }
 }

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/OfficialOpensearchClientProvider.java

@@ -16,6 +16,7 @@
  */
 package org.graylog.storage.opensearch3;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.joschi.jadconfig.util.Duration;
 import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
@@ -63,7 +64,6 @@
  * along with this program. If not, see
  * <http://www.mongodb.com/licensing/server-side-public-license>.
  */
-
 public class OfficialOpensearchClientProvider implements Provider<OfficialOpensearchClient> {
 
     private static Logger log = LoggerFactory.getLogger(OfficialOpensearchClientProvider.class);
@@ -76,8 +76,10 @@ public OfficialOpensearchClientProvider(
             IndexerJwtAuthToken indexerJwtAuthToken,
             CredentialsProvider credentialsProvider,
             ElasticsearchClientConfiguration clientConfiguration,
-            @Nullable TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider) {
-        clientCache = Suppliers.memoize(() -> createClient(hosts, indexerJwtAuthToken, credentialsProvider, clientConfiguration, trustManagerAndSocketFactoryProvider));
+            ObjectMapper objectMapper,
+            @Nullable TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider
+    ) {
+        clientCache = Suppliers.memoize(() -> createClient(hosts, indexerJwtAuthToken, credentialsProvider, clientConfiguration, trustManagerAndSocketFactoryProvider, objectMapper));
     }
 
     @Override
@@ -86,7 +88,7 @@ public OfficialOpensearchClient get() {
     }
 
     @Nonnull
-    private static OfficialOpensearchClient createClient(List<URI> uris, IndexerJwtAuthToken indexerJwtAuthToken, CredentialsProvider credentialsProvider, ElasticsearchClientConfiguration clientConfiguration, TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider) {
+    private static OfficialOpensearchClient createClient(List<URI> uris, IndexerJwtAuthToken indexerJwtAuthToken, CredentialsProvider credentialsProvider, ElasticsearchClientConfiguration clientConfiguration, TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider, ObjectMapper objectMapper) {
 
         log.info("Initializing OpenSearch client");
 
@@ -134,7 +136,7 @@ private static OfficialOpensearchClient createClient(List<URI> uris, IndexerJwtA
         });
 
         final OpenSearchTransport transport = builder.build();
-        return new OfficialOpensearchClient(new CustomOpenSearchClient(transport), new CustomAsyncOpenSearchClient(transport));
+        return new OfficialOpensearchClient(new CustomOpenSearchClient(transport), new CustomAsyncOpenSearchClient(transport), objectMapper);
     }
 
     private static Optional<TlsStrategy> tlsStrategy(@Nullable TrustManagerAndSocketFactoryProvider trustManagerAndSocketFactoryProvider) {

graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/PlainJsonApi.java

@@ -25,6 +25,7 @@
 
 import static org.graylog.storage.opensearch3.OfficialOpensearchClient.mapException;
 
+@Deprecated
 public class PlainJsonApi {
     private final ObjectMapper objectMapper;
     private final OfficialOpensearchClient client;
@@ -40,10 +41,6 @@ public PlainJsonApi(ObjectMapper objectMapper,
     }
 
     @Deprecated
-    public PlainJsonApi(ObjectMapper objectMapper, OpenSearchClient client) {
-        this(objectMapper, client, null);
-    }
-
     public JsonNode performRequest(Request request, String errorMessage) {
         try {
             Response response = client.sync().generic().execute(request);