Reset discovery nodes in most transport node actions request. (opense…

…arch-project#15131) Signed-off-by: Swetha Guptha <[email protected]> Co-authored-by: Swetha Guptha <[email protected]>
akolarkunnu · Sep 10, 2024 · 1eae5b1 · 1eae5b1
1 parent 9a3fdbc
commit 1eae5b1
Show file tree

Hide file tree

Showing 25 changed files with 87 additions and 112 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -51,6 +51,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Remote Publication] Added checksum validation for cluster state behind a cluster setting ([#15218](https://github.com/opensearch-project/OpenSearch/pull/15218))
 - Add canRemain method to TargetPoolAllocationDecider to move shards from local to remote pool for hot to warm tiering ([#15010](https://github.com/opensearch-project/OpenSearch/pull/15010))
 - ClusterManagerTaskThrottler Improvements ([#15508](https://github.com/opensearch-project/OpenSearch/pull/15508))
+- Reset DiscoveryNodes in all transport node actions request ([#15131](https://github.com/opensearch-project/OpenSearch/pull/15131))
 
 ### Dependencies
 - Bump `netty` from 4.1.111.Final to 4.1.112.Final ([#15081](https://github.com/opensearch-project/OpenSearch/pull/15081))

diff --git a/...main/java/org/opensearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequest.java b/...main/java/org/opensearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequest.java
@@ -70,7 +70,7 @@ public NodesHotThreadsRequest(StreamInput in) throws IOException {
      * threads for all nodes is used.
      */
     public NodesHotThreadsRequest(String... nodesIds) {
-        super(nodesIds);
+        super(false, nodesIds);
     }
 
     public int threads() {

diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/node/info/NodesInfoRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/node/info/NodesInfoRequest.java
@@ -72,7 +72,7 @@ public NodesInfoRequest(StreamInput in) throws IOException {
      * for all nodes will be returned.
      */
     public NodesInfoRequest(String... nodesIds) {
-        super(nodesIds);
+        super(false, nodesIds);
         defaultMetrics();
     }
 

diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodesStatsRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/node/stats/NodesStatsRequest.java
@@ -58,7 +58,7 @@ public class NodesStatsRequest extends BaseNodesRequest<NodesStatsRequest> {
     private final Set<String> requestedMetrics = new HashSet<>();
 
     public NodesStatsRequest() {
-        super((String[]) null);
+        super(false, (String[]) null);
     }
 
     public NodesStatsRequest(StreamInput in) throws IOException {
@@ -74,7 +74,7 @@ public NodesStatsRequest(StreamInput in) throws IOException {
      * for all nodes will be returned.
      */
     public NodesStatsRequest(String... nodesIds) {
-        super(nodesIds);
+        super(false, nodesIds);
     }
 
     /**

diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/node/usage/NodesUsageRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/node/usage/NodesUsageRequest.java
@@ -61,7 +61,7 @@ public NodesUsageRequest(StreamInput in) throws IOException {
      * passed, usage for all nodes will be returned.
      */
     public NodesUsageRequest(String... nodesIds) {
-        super(nodesIds);
+        super(false, nodesIds);
     }
 
     /**

diff --git a/...a/org/opensearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java b/...a/org/opensearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java
@@ -161,7 +161,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(String[] nodesIds) {
-            super(nodesIds);
+            super(false, nodesIds);
         }
 
         public Request snapshots(Snapshot[] snapshots) {

diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/stats/ClusterStatsRequest.java
@@ -62,7 +62,7 @@ public ClusterStatsRequest(StreamInput in) throws IOException {
      * based on all nodes will be returned.
      */
     public ClusterStatsRequest(String... nodesIds) {
-        super(nodesIds);
+        super(false, nodesIds);
     }
 
     public boolean useAggregatedNodeLevelResponses() {

diff --git a/...main/java/org/opensearch/action/admin/indices/dangling/find/FindDanglingIndexRequest.java b/...main/java/org/opensearch/action/admin/indices/dangling/find/FindDanglingIndexRequest.java
@@ -53,7 +53,7 @@ public FindDanglingIndexRequest(StreamInput in) throws IOException {
     }
 
     public FindDanglingIndexRequest(String indexUUID) {
-        super(Strings.EMPTY_ARRAY);
+        super(false, Strings.EMPTY_ARRAY);
         this.indexUUID = indexUUID;
     }
 

diff --git a/...in/java/org/opensearch/action/admin/indices/dangling/list/ListDanglingIndicesRequest.java b/...in/java/org/opensearch/action/admin/indices/dangling/list/ListDanglingIndicesRequest.java
@@ -58,12 +58,12 @@ public ListDanglingIndicesRequest(StreamInput in) throws IOException {
     }
 
     public ListDanglingIndicesRequest() {
-        super(Strings.EMPTY_ARRAY);
+        super(false, Strings.EMPTY_ARRAY);
         this.indexUUID = null;
     }
 
     public ListDanglingIndicesRequest(String indexUUID) {
-        super(Strings.EMPTY_ARRAY);
+        super(false, Strings.EMPTY_ARRAY);
         this.indexUUID = indexUUID;
     }
 

diff --git a/server/src/main/java/org/opensearch/action/search/GetAllPitNodesRequest.java b/server/src/main/java/org/opensearch/action/search/GetAllPitNodesRequest.java
@@ -27,7 +27,7 @@ public class GetAllPitNodesRequest extends BaseNodesRequest<GetAllPitNodesReques
 
     @Inject
     public GetAllPitNodesRequest(DiscoveryNode... concreteNodes) {
-        super(concreteNodes);
+        super(false, concreteNodes);
     }
 
     public GetAllPitNodesRequest(StreamInput in) throws IOException {

diff --git a/server/src/main/java/org/opensearch/action/support/nodes/BaseNodesRequest.java b/server/src/main/java/org/opensearch/action/support/nodes/BaseNodesRequest.java
@@ -73,6 +73,7 @@ public abstract class BaseNodesRequest<Request extends BaseNodesRequest<Request>
      * Setting default behavior as `true` but can be explicitly changed in requests that do not require.
      */
     private boolean includeDiscoveryNodes = true;
+
     private final TimeValue DEFAULT_TIMEOUT_SECS = TimeValue.timeValueSeconds(30);
 
     private TimeValue timeout;
@@ -88,11 +89,22 @@ protected BaseNodesRequest(String... nodesIds) {
         this.nodesIds = nodesIds;
     }
 
+    protected BaseNodesRequest(boolean includeDiscoveryNodes, String... nodesIds) {
+        this.nodesIds = nodesIds;
+        this.includeDiscoveryNodes = includeDiscoveryNodes;
+    }
+
     protected BaseNodesRequest(DiscoveryNode... concreteNodes) {
         this.nodesIds = null;
         this.concreteNodes = concreteNodes;
     }
 
+    protected BaseNodesRequest(boolean includeDiscoveryNodes, DiscoveryNode... concreteNodes) {
+        this.nodesIds = null;
+        this.concreteNodes = concreteNodes;
+        this.includeDiscoveryNodes = includeDiscoveryNodes;
+    }
+
     public final String[] nodesIds() {
         return nodesIds;
     }
@@ -127,10 +139,6 @@ public void setConcreteNodes(DiscoveryNode[] concreteNodes) {
         this.concreteNodes = concreteNodes;
     }
 
-    public void setIncludeDiscoveryNodes(boolean value) {
-        includeDiscoveryNodes = value;
-    }
-
     public boolean getIncludeDiscoveryNodes() {
         return includeDiscoveryNodes;
     }

diff --git a/server/src/main/java/org/opensearch/action/support/nodes/TransportNodesAction.java b/server/src/main/java/org/opensearch/action/support/nodes/TransportNodesAction.java
@@ -240,18 +240,16 @@ class AsyncAction {
             }
             this.responses = new AtomicReferenceArray<>(request.concreteNodes().length);
             this.concreteNodes = request.concreteNodes();
-
             if (request.getIncludeDiscoveryNodes() == false) {
-                // As we transfer the ownership of discovery nodes to route the request to into the AsyncAction class, we
-                // remove the list of DiscoveryNodes from the request. This reduces the payload of the request and improves
+                // As we transfer the ownership of discovery nodes to route the request to into the AsyncAction class,
+                // we remove the list of DiscoveryNodes from the request. This reduces the payload of the request and improves
                 // the number of concrete nodes in the memory.
                 request.setConcreteNodes(null);
             }
         }
 
         void start() {
-            final DiscoveryNode[] nodes = this.concreteNodes;
-            if (nodes.length == 0) {
+            if (this.concreteNodes.length == 0) {
                 // nothing to notify
                 threadPool.generic().execute(() -> listener.onResponse(newResponse(request, responses)));
                 return;
@@ -260,9 +258,9 @@ void start() {
             if (request.timeout() != null) {
                 builder.withTimeout(request.timeout());
             }
-            for (int i = 0; i < nodes.length; i++) {
+            for (int i = 0; i < this.concreteNodes.length; i++) {
                 final int idx = i;
-                final DiscoveryNode node = nodes[i];
+                final DiscoveryNode node = this.concreteNodes[i];
                 final String nodeId = node.getId();
                 try {
                     TransportRequest nodeRequest = newNodeRequest(request);

diff --git a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayMetaState.java b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayMetaState.java
@@ -133,7 +133,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(String... nodesIds) {
-            super(nodesIds);
+            super(false, nodesIds);
         }
     }
 

diff --git a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShards.java
@@ -197,7 +197,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(ShardId shardId, String customDataPath, DiscoveryNode[] nodes) {
-            super(nodes);
+            super(false, nodes);
             this.shardId = Objects.requireNonNull(shardId);
             this.customDataPath = Objects.requireNonNull(customDataPath);
         }

diff --git a/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShardsBatch.java b/server/src/main/java/org/opensearch/gateway/TransportNodesListGatewayStartedShardsBatch.java
@@ -182,7 +182,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(DiscoveryNode[] nodes, Map<ShardId, ShardAttributes> shardAttributes) {
-            super(nodes);
+            super(false, nodes);
             this.shardAttributes = Objects.requireNonNull(shardAttributes);
         }
 

diff --git a/server/src/main/java/org/opensearch/indices/store/TransportNodesListShardStoreMetadata.java b/server/src/main/java/org/opensearch/indices/store/TransportNodesListShardStoreMetadata.java
@@ -176,7 +176,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(ShardId shardId, String customDataPath, DiscoveryNode[] nodes) {
-            super(nodes);
+            super(false, nodes);
             this.shardId = Objects.requireNonNull(shardId);
             this.customDataPath = Objects.requireNonNull(customDataPath);
         }

diff --git a/...src/main/java/org/opensearch/indices/store/TransportNodesListShardStoreMetadataBatch.java b/...src/main/java/org/opensearch/indices/store/TransportNodesListShardStoreMetadataBatch.java
@@ -188,7 +188,7 @@ public Request(StreamInput in) throws IOException {
         }
 
         public Request(Map<ShardId, ShardAttributes> shardAttributes, DiscoveryNode[] nodes) {
-            super(nodes);
+            super(false, nodes);
             this.shardAttributes = Objects.requireNonNull(shardAttributes);
         }
 

diff --git a/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsAction.java b/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterStatsAction.java
@@ -66,7 +66,6 @@ public String getName() {
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ClusterStatsRequest clusterStatsRequest = new ClusterStatsRequest().nodesIds(request.paramAsStringArray("nodeId", null));
         clusterStatsRequest.timeout(request.param("timeout"));
-        clusterStatsRequest.setIncludeDiscoveryNodes(false);
         clusterStatsRequest.useAggregatedNodeLevelResponses(true);
         return channel -> client.admin().cluster().clusterStats(clusterStatsRequest, new NodesResponseRestListener<>(channel));
     }

diff --git a/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestNodesInfoAction.java b/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestNodesInfoAction.java
@@ -88,7 +88,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final NodesInfoRequest nodesInfoRequest = prepareRequest(request);
         nodesInfoRequest.timeout(request.param("timeout"));
         settingsFilter.addFilterSettingParams(request);
-        nodesInfoRequest.setIncludeDiscoveryNodes(false);
         return channel -> client.admin().cluster().nodesInfo(nodesInfoRequest, new NodesResponseRestListener<>(channel));
     }
 

diff --git a/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestNodesStatsAction.java b/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestNodesStatsAction.java
@@ -232,7 +232,6 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         // If no levels are passed in this results in an empty array.
         String[] levels = Strings.splitStringByCommaToArray(request.param("level"));
         nodesStatsRequest.indices().setLevels(levels);
-        nodesStatsRequest.setIncludeDiscoveryNodes(false);
         nodesStatsRequest.indices().setIncludeIndicesStatsByLevel(true);
 
         return channel -> client.admin().cluster().nodesStats(nodesStatsRequest, new NodesResponseRestListener<>(channel));

diff --git a/server/src/main/java/org/opensearch/rest/action/cat/RestNodesAction.java b/server/src/main/java/org/opensearch/rest/action/cat/RestNodesAction.java
@@ -125,7 +125,6 @@ public RestChannelConsumer doCatRequest(final RestRequest request, final NodeCli
             public void processResponse(final ClusterStateResponse clusterStateResponse) {
                 NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
                 nodesInfoRequest.timeout(request.param("timeout"));
-                nodesInfoRequest.setIncludeDiscoveryNodes(false);
                 nodesInfoRequest.clear()
                     .addMetrics(
                         NodesInfoRequest.Metric.JVM.metricName(),
@@ -138,7 +137,6 @@ public void processResponse(final ClusterStateResponse clusterStateResponse) {
                     public void processResponse(final NodesInfoResponse nodesInfoResponse) {
                         NodesStatsRequest nodesStatsRequest = new NodesStatsRequest();
                         nodesStatsRequest.timeout(request.param("timeout"));
-                        nodesStatsRequest.setIncludeDiscoveryNodes(false);
                         nodesStatsRequest.clear()
                             .indices(true)
                             .addMetrics(

diff --git a/...r/src/test/java/org/opensearch/action/support/nodes/TransportClusterStatsActionTests.java b/...r/src/test/java/org/opensearch/action/support/nodes/TransportClusterStatsActionTests.java
@@ -33,48 +33,12 @@
 
 public class TransportClusterStatsActionTests extends TransportNodesActionTests {
 
-    /**
-     * By default, we send discovery nodes list to each request that is sent across from the coordinator node. This
-     * behavior is asserted in this test.
-     */
-    public void testClusterStatsActionWithRetentionOfDiscoveryNodesList() {
-        ClusterStatsRequest request = new ClusterStatsRequest();
-        request.setIncludeDiscoveryNodes(true);
-        Map<String, List<MockClusterStatsNodeRequest>> combinedSentRequest = performNodesInfoAction(request);
-
-        assertNotNull(combinedSentRequest);
-        combinedSentRequest.forEach((node, capturedRequestList) -> {
-            assertNotNull(capturedRequestList);
-            capturedRequestList.forEach(sentRequest -> {
-                assertNotNull(sentRequest.getDiscoveryNodes());
-                assertEquals(sentRequest.getDiscoveryNodes().length, clusterService.state().nodes().getSize());
-            });
-        });
-    }
-
-    public void testClusterStatsActionWithPreFilledConcreteNodesAndWithRetentionOfDiscoveryNodesList() {
-        ClusterStatsRequest request = new ClusterStatsRequest();
-        Collection<DiscoveryNode> discoveryNodes = clusterService.state().getNodes().getNodes().values();
-        request.setConcreteNodes(discoveryNodes.toArray(DiscoveryNode[]::new));
-        Map<String, List<MockClusterStatsNodeRequest>> combinedSentRequest = performNodesInfoAction(request);
-
-        assertNotNull(combinedSentRequest);
-        combinedSentRequest.forEach((node, capturedRequestList) -> {
-            assertNotNull(capturedRequestList);
-            capturedRequestList.forEach(sentRequest -> {
-                assertNotNull(sentRequest.getDiscoveryNodes());
-                assertEquals(sentRequest.getDiscoveryNodes().length, clusterService.state().nodes().getSize());
-            });
-        });
-    }
-
     /**
      * In the optimized ClusterStats Request, we do not send the DiscoveryNodes List to each node. This behavior is
      * asserted in this test.
      */
     public void testClusterStatsActionWithoutRetentionOfDiscoveryNodesList() {
         ClusterStatsRequest request = new ClusterStatsRequest();
-        request.setIncludeDiscoveryNodes(false);
         Map<String, List<MockClusterStatsNodeRequest>> combinedSentRequest = performNodesInfoAction(request);
 
         assertNotNull(combinedSentRequest);
@@ -88,7 +52,6 @@ public void testClusterStatsActionWithPreFilledConcreteNodesAndWithoutRetentionO
         ClusterStatsRequest request = new ClusterStatsRequest();
         Collection<DiscoveryNode> discoveryNodes = clusterService.state().getNodes().getNodes().values();
         request.setConcreteNodes(discoveryNodes.toArray(DiscoveryNode[]::new));
-        request.setIncludeDiscoveryNodes(false);
         Map<String, List<MockClusterStatsNodeRequest>> combinedSentRequest = performNodesInfoAction(request);
 
         assertNotNull(combinedSentRequest);