Report shard count per node in _nodes/stats (#75760)

Backporting _nodes/stats feature

This commit cherry-picks the commit from #75433, and changes it to apply to 7.15 and later.

Author: masseyke

docs/reference/cluster/nodes-stats.asciidoc

@@ -959,6 +959,18 @@ Time by which recovery operations were delayed due to throttling.
 Time in milliseconds
 recovery operations were delayed due to throttling.
 =======
+
+`shards`::
+(object)
+Contains statistics about all shards assigned to the node.
++
+.Properties of `shards`
+[%collapsible%open]
+=======
+`total_count`::
+(integer)
+The total number of shards assigned to the node.
+=======
 ======
 
 [[cluster-nodes-stats-api-response-body-os]]

rest-api-spec/src/main/resources/rest-api-spec/api/nodes.stats.json

@@ -127,7 +127,8 @@
                 "segments",
                 "store",
                 "warmer",
-                "suggest"
+                "suggest",
+                "shards"
               ],
               "description":"Limit the information returned for `indices` metric to the specific index metrics. Isn't used if `indices` (or `all`) metric isn't specified."
             }
@@ -175,7 +176,8 @@
                 "segments",
                 "store",
                 "warmer",
-                "suggest"
+                "suggest",
+                "shards"
               ],
               "description":"Limit the information returned for `indices` metric to the specific index metrics. Isn't used if `indices` (or `all`) metric isn't specified."
             },

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/nodes.stats/11_indices_metrics.yml

@@ -109,6 +109,7 @@
   - is_false:  nodes.$node_id.indices.segments
   - is_false:  nodes.$node_id.indices.translog
   - is_false:  nodes.$node_id.indices.recovery
+  - is_false:  nodes.$node_id.indices.shards
 
 ---
 "Metric - multi":
@@ -166,6 +167,7 @@
   - is_false:  nodes.$node_id.indices.segments
   - is_false:  nodes.$node_id.indices.translog
   - is_true:   nodes.$node_id.indices.recovery
+  - is_false:  nodes.$node_id.indices.shards
 
 ---
 "Metric - _all include_segment_file_sizes":
@@ -223,6 +225,7 @@
   - is_true:   nodes.$node_id.indices.segments
   - is_false:  nodes.$node_id.indices.translog
   - is_false:  nodes.$node_id.indices.recovery
+  - is_false:  nodes.$node_id.indices.shards
   - is_true:   nodes.$node_id.indices.segments.file_sizes
 
 ---
@@ -254,6 +257,7 @@
   - is_true:   nodes.$node_id.indices.segments
   - is_false:  nodes.$node_id.indices.translog
   - is_false:  nodes.$node_id.indices.recovery
+  - is_false:  nodes.$node_id.indices.shards
 
 ---
 "Metric - _all include_unloaded_segments":
@@ -314,3 +318,72 @@
   - gte: { nodes.$node_id.http.clients.0.request_size_bytes: 0 }
   # values for clients.0.closed_time_millis, clients.0.x_forwarded_for, and clients.0.x_opaque_id are often
   # null and cannot be tested here
+
+---
+"Metric - blank for indices shards":
+  - skip:
+      features: [arbitrary_key]
+      version: " - 7.99.99"
+      reason:  "total shard count added in version 8.0"
+  - do:
+      nodes.info: {}
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      nodes.stats: {}
+
+  - is_true:  nodes.$node_id.indices.shards
+  - match: { nodes.$node_id.indices.shards.total_count: 0 }
+
+---
+"Metric - _all for indices shards":
+  - skip:
+      features: [arbitrary_key]
+      version: " - 7.99.99"
+      reason:  "total shard count added in version 8.0"
+  - do:
+      nodes.info: {}
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      nodes.stats: { metric: _all }
+
+  - is_true:  nodes.$node_id.indices.shards
+  - match: { nodes.$node_id.indices.shards.total_count: 0 }
+
+
+---
+"indices shards total count test":
+
+  - skip:
+      features: ["allowed_warnings", arbitrary_key]
+      version: " - 7.99.99"
+      reason:  "total shard count added in version 8.0"
+
+  - do:
+      indices.create:
+        index: index1
+        body:
+          settings:
+            number_of_shards: "5"
+            number_of_replicas: "0"
+
+  - do:
+      indices.create:
+        index: index2
+        body:
+          settings:
+            number_of_shards: "3"
+            number_of_replicas: "1"
+
+  - do:
+      nodes.info: {}
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      nodes.stats: { metric: _all }
+
+  - gte: { nodes.$node_id.indices.shards.total_count: 1 }

server/src/internalClusterTest/java/org/elasticsearch/indices/stats/IndexStatsIT.java

@@ -729,7 +729,7 @@ public void testEncodeDecodeCommonStats() throws IOException {
     public void testFlagOrdinalOrder() {
         Flag[] flags = new Flag[]{Flag.Store, Flag.Indexing, Flag.Get, Flag.Search, Flag.Merge, Flag.Flush, Flag.Refresh,
                 Flag.QueryCache, Flag.FieldData, Flag.Docs, Flag.Warmer, Flag.Completion, Flag.Segments,
-                Flag.Translog, Flag.RequestCache, Flag.Recovery};
+                Flag.Translog, Flag.RequestCache, Flag.Recovery, Flag.Shards};
 
         assertThat(flags.length, equalTo(Flag.values().length));
         for (int i = 0; i < flags.length; i++) {
@@ -941,6 +941,10 @@ private static void set(Flag flag, IndicesStatsRequestBuilder builder, boolean s
             case Recovery:
                 builder.setRecovery(set);
                 break;
+            case Shards:
+                // We don't actually expose shards in IndexStats, but this test fails if it isn't handled
+                builder.request().flags().set(Flag.Shards, set);
+                break;
             default:
                 fail("new flag? " + flag);
                 break;
@@ -981,6 +985,8 @@ private static boolean isSet(Flag flag, CommonStats response) {
                 return response.getRequestCache() != null;
             case Recovery:
                 return response.getRecoveryStats() != null;
+            case Shards:
+                return response.getShards() != null;
             default:
                 fail("new flag? " + flag);
                 return false;

server/src/main/java/org/elasticsearch/action/admin/indices/stats/CommonStats.java

@@ -9,6 +9,7 @@
 package org.elasticsearch.action.admin.indices.stats;
 
 import org.apache.lucene.store.AlreadyClosedException;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -30,6 +31,7 @@
 import org.elasticsearch.index.shard.DocsStats;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.IndexingStats;
+import org.elasticsearch.index.shard.ShardCountStats;
 import org.elasticsearch.index.store.StoreStats;
 import org.elasticsearch.index.translog.TranslogStats;
 import org.elasticsearch.index.warmer.WarmerStats;
@@ -91,6 +93,9 @@ public class CommonStats implements Writeable, ToXContentFragment {
     @Nullable
     public RecoveryStats recoveryStats;
 
+    @Nullable
+    public ShardCountStats shards;
+
     public CommonStats() {
         this(CommonStatsFlags.NONE);
     }
@@ -148,6 +153,9 @@ public CommonStats(CommonStatsFlags flags) {
                 case Recovery:
                     recoveryStats = new RecoveryStats();
                     break;
+                case Shards:
+                    shards = new ShardCountStats();
+                    break;
                 default:
                     throw new IllegalStateException("Unknown Flag: " + flag);
             }
@@ -207,6 +215,10 @@ public CommonStats(IndicesQueryCache indicesQueryCache, IndexShard indexShard, C
                     case Recovery:
                         recoveryStats = indexShard.recoveryStats();
                         break;
+                    case Shards:
+                        // Setting to 1 because the single IndexShard passed to this method implies 1 shard
+                        shards = new ShardCountStats(1);
+                        break;
                     default:
                         throw new IllegalStateException("Unknown Flag: " + flag);
                 }
@@ -233,6 +245,9 @@ public CommonStats(StreamInput in) throws IOException {
         translog = in.readOptionalWriteable(TranslogStats::new);
         requestCache = in.readOptionalWriteable(RequestCacheStats::new);
         recoveryStats = in.readOptionalWriteable(RecoveryStats::new);
+        if (in.getVersion().onOrAfter(Version.V_7_15_0)) {
+            shards = in.readOptionalWriteable(ShardCountStats::new);
+        }
     }
 
     @Override
@@ -253,6 +268,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalWriteable(translog);
         out.writeOptionalWriteable(requestCache);
         out.writeOptionalWriteable(recoveryStats);
+        if (out.getVersion().onOrAfter(Version.V_7_15_0)) {
+            out.writeOptionalWriteable(shards);
+        }
     }
 
     public void add(CommonStats stats) {
@@ -385,6 +403,14 @@ public void add(CommonStats stats) {
         } else {
             recoveryStats.add(stats.getRecoveryStats());
         }
+        if (stats.shards != null) {
+            if (shards == null) {
+                shards = stats.shards;
+            }
+            else {
+                shards = shards.add(stats.shards);
+            }
+        }
     }
 
     @Nullable
@@ -467,6 +493,11 @@ public RecoveryStats getRecoveryStats() {
         return recoveryStats;
     }
 
+    @Nullable
+    public ShardCountStats getShards() {
+        return shards;
+    }
+
     /**
      * Utility method which computes total memory by adding
      * FieldData, PercolatorCache, Segments (memory, index writer, version map)
@@ -492,7 +523,7 @@ public ByteSizeValue getTotalMemory() {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         final Stream<ToXContent> stream = Arrays.stream(new ToXContent[] {
-            docs, store, indexing, get, search, merge, refresh, flush, warmer, queryCache,
+            docs, shards, store, indexing, get, search, merge, refresh, flush, warmer, queryCache,
             fieldData, completion, segments, translog, requestCache, recoveryStats})
             .filter(Objects::nonNull);
         for (ToXContent toXContent : ((Iterable<ToXContent>)stream::iterator)) {

server/src/main/java/org/elasticsearch/action/admin/indices/stats/CommonStatsFlags.java

@@ -231,7 +231,8 @@ public enum Flag {
         Translog("translog", 13),
         // 14 was previously used for Suggest
         RequestCache("request_cache", 15),
-        Recovery("recovery", 16);
+        Recovery("recovery", 16),
+        Shards("shards", 18);
 
         private final String restName;
         private final int index;

server/src/main/java/org/elasticsearch/index/shard/ShardCountStats.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.index.shard;
+
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ToXContentFragment;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.core.Nullable;
+
+import java.io.IOException;
+
+public class ShardCountStats implements Writeable, ToXContentFragment {
+
+    private final long totalCount;
+
+    public ShardCountStats() {
+        totalCount = 0;
+    }
+
+    public ShardCountStats(StreamInput in) throws IOException {
+        totalCount = in.readVLong();
+    }
+
+    public ShardCountStats(long totalCount) {
+        this.totalCount = totalCount;
+    }
+
+    public long getTotalCount() {
+        return this.totalCount;
+    }
+
+    public ShardCountStats add(@Nullable ShardCountStats other) {
+        return new ShardCountStats(this.totalCount + (other == null ? 0 : other.totalCount));
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(Fields.SHARDS);
+        builder.field(Fields.TOTAL_COUNT, totalCount);
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeVLong(totalCount);
+    }
+
+    static final class Fields {
+        static final String SHARDS = "shards";
+        static final String TOTAL_COUNT = "total_count";
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        return (o instanceof ShardCountStats) && totalCount == ((ShardCountStats) o).totalCount;
+    }
+
+    @Override
+    public int hashCode() {
+        return Long.hashCode(totalCount);
+    }
+}