Add statsName field on stream while constructing PersistedStateStats (o…

…pensearch-project#10964) Signed-off-by: Aman Khare <[email protected]> Co-authored-by: Aman Khare <[email protected]> Signed-off-by: Shivansh Arora <[email protected]>
Gaurav614 · Apr 25, 2024 · 24fd0fa · 24fd0fa
1 parent 8d3eedc
commit 24fd0fa
Show file tree

Hide file tree

Showing 3 changed files with 69 additions and 10 deletions.
diff --git a/...c/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java b/...c/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java
@@ -114,6 +114,31 @@ public void testRemoteStateStats() {
             .addMetric(NodesStatsRequest.Metric.DISCOVERY.metricName())
             .get();
 
+        // assert cluster state stats
+        assertClusterManagerClusterStateStats(nodesStatsResponse);
+
+        NodesStatsResponse nodesStatsResponseDataNode = client().admin()
+            .cluster()
+            .prepareNodesStats(dataNode)
+            .addMetric(NodesStatsRequest.Metric.DISCOVERY.metricName())
+            .get();
+        // assert cluster state stats for data node
+        DiscoveryStats dataNodeDiscoveryStats = nodesStatsResponseDataNode.getNodes().get(0).getDiscoveryStats();
+        assertNotNull(dataNodeDiscoveryStats.getClusterStateStats());
+        assertEquals(0, dataNodeDiscoveryStats.getClusterStateStats().getUpdateSuccess());
+
+        // call nodes/stats with nodeId filter
+        NodesStatsResponse nodesStatsNodeIdFilterResponse = client().admin()
+            .cluster()
+            .prepareNodesStats(dataNode)
+            .addMetric(NodesStatsRequest.Metric.DISCOVERY.metricName())
+            .setNodesIds(clusterManagerNode)
+            .get();
+
+        assertClusterManagerClusterStateStats(nodesStatsNodeIdFilterResponse);
+    }
+
+    private void assertClusterManagerClusterStateStats(NodesStatsResponse nodesStatsResponse) {
         // assert cluster state stats
         DiscoveryStats discoveryStats = nodesStatsResponse.getNodes().get(0).getDiscoveryStats();
 
@@ -125,16 +150,43 @@ public void testRemoteStateStats() {
         assertTrue(discoveryStats.getClusterStateStats().getPersistenceStats().get(0).getSuccessCount() > 1);
         assertEquals(0, discoveryStats.getClusterStateStats().getPersistenceStats().get(0).getFailedCount());
         assertTrue(discoveryStats.getClusterStateStats().getPersistenceStats().get(0).getTotalTimeInMillis() > 0);
+    }
 
-        NodesStatsResponse nodesStatsResponseDataNode = client().admin()
-            .cluster()
-            .prepareNodesStats(dataNode)
-            .addMetric(NodesStatsRequest.Metric.DISCOVERY.metricName())
-            .get();
-        // assert cluster state stats for data node
-        DiscoveryStats dataNodeDiscoveryStats = nodesStatsResponseDataNode.getNodes().get(0).getDiscoveryStats();
-        assertNotNull(dataNodeDiscoveryStats.getClusterStateStats());
-        assertEquals(0, dataNodeDiscoveryStats.getClusterStateStats().getUpdateSuccess());
+    public void testRemoteStateStatsFromAllNodes() {
+        int shardCount = randomIntBetween(1, 5);
+        int replicaCount = 1;
+        int dataNodeCount = shardCount * (replicaCount + 1);
+        int clusterManagerNodeCount = 3;
+        prepareCluster(clusterManagerNodeCount, dataNodeCount, INDEX_NAME, replicaCount, shardCount);
+        String[] allNodes = internalCluster().getNodeNames();
+        // call _nodes/stats/discovery from all the nodes
+        for (String node : allNodes) {
+            NodesStatsResponse nodesStatsResponse = client().admin()
+                .cluster()
+                .prepareNodesStats(node)
+                .addMetric(NodesStatsRequest.Metric.DISCOVERY.metricName())
+                .get();
+            validateNodesStatsResponse(nodesStatsResponse);
+        }
+
+        // call _nodes/stats/discovery from all the nodes with random nodeId filter
+        for (String node : allNodes) {
+            NodesStatsResponse nodesStatsResponse = client().admin()
+                .cluster()
+                .prepareNodesStats(node)
+                .addMetric(NodesStatsRequest.Metric.DISCOVERY.metricName())
+                .setNodesIds(allNodes[randomIntBetween(0, allNodes.length - 1)])
+                .get();
+            validateNodesStatsResponse(nodesStatsResponse);
+        }
+    }
+
+    private void validateNodesStatsResponse(NodesStatsResponse nodesStatsResponse) {
+        // _nodes/stats/discovery must never fail due to any exception
+        assertFalse(nodesStatsResponse.toString().contains("exception"));
+        assertNotNull(nodesStatsResponse.getNodes());
+        assertNotNull(nodesStatsResponse.getNodes().get(0));
+        assertNotNull(nodesStatsResponse.getNodes().get(0).getDiscoveryStats());
     }
 
     private void setReplicaCount(int replicaCount) {

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/PersistedStateStats.java b/server/src/main/java/org/opensearch/cluster/coordination/PersistedStateStats.java
@@ -25,7 +25,7 @@
  * @opensearch.internal
  */
 public class PersistedStateStats implements Writeable, ToXContentObject {
-    private String statsName;
+    private final String statsName;
     private AtomicLong totalTimeInMillis = new AtomicLong(0);
     private AtomicLong failedCount = new AtomicLong(0);
     private AtomicLong successCount = new AtomicLong(0);
@@ -37,6 +37,7 @@ public PersistedStateStats(String statsName) {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(statsName);
         out.writeVLong(successCount.get());
         out.writeVLong(failedCount.get());
         out.writeVLong(totalTimeInMillis.get());
@@ -53,6 +54,7 @@ public void writeTo(StreamOutput out) throws IOException {
     }
 
     public PersistedStateStats(StreamInput in) throws IOException {
+        this.statsName = in.readString();
         this.successCount = new AtomicLong(in.readVLong());
         this.failedCount = new AtomicLong(in.readVLong());
         this.totalTimeInMillis = new AtomicLong(in.readVLong());
@@ -113,6 +115,10 @@ protected void addToExtendedFields(String extendedField, AtomicLong extendedFiel
         this.extendedFields.put(extendedField, extendedFieldValue);
     }
 
+    public String getStatsName() {
+        return statsName;
+    }
+
     /**
      * Fields for parsing and toXContent
      *

diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/node/stats/NodeStatsTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/node/stats/NodeStatsTests.java
@@ -368,6 +368,7 @@ public void testSerialization() throws IOException {
                             .getPersistenceStats()
                             .get(0);
                         PersistedStateStats remoteStateStats = stateStats.getPersistenceStats().get(0);
+                        assertEquals(remoteStateStats.getStatsName(), deserializedRemoteStateStats.getStatsName());
                         assertEquals(remoteStateStats.getFailedCount(), deserializedRemoteStateStats.getFailedCount());
                         assertEquals(remoteStateStats.getSuccessCount(), deserializedRemoteStateStats.getSuccessCount());
                         assertEquals(remoteStateStats.getTotalTimeInMillis(), deserializedRemoteStateStats.getTotalTimeInMillis());