Data Streams Stats action only on searchable nodes (elastic#134737)

Relates ES-12829

Author: kingherc

modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/TransportDataStreamsStatsAction.java

@@ -106,7 +106,7 @@ protected ClusterBlockException checkRequestBlock(
 
     @Override
     protected ShardsIterator shards(ClusterState clusterState, DataStreamsStatsAction.Request request, String[] concreteIndices) {
-        return clusterState.routingTable(projectResolver.getProjectId()).allShards(concreteIndices);
+        return clusterState.routingTable(projectResolver.getProjectId()).allSearchableShards(concreteIndices);
     }
 
     @Override
@@ -117,6 +117,7 @@ protected void shardOperation(
         ActionListener<DataStreamsStatsAction.DataStreamShardStats> listener
     ) {
         ActionListener.completeWith(listener, () -> {
+            assert shardRouting.isSearchable() : "shard routing is not searchable: " + shardRouting;
             IndexService indexService = indicesService.indexServiceSafe(shardRouting.shardId().getIndex());
             IndexShard indexShard = indexService.getShard(shardRouting.shardId().id());
             StoreStats storeStats = indexShard.storeStats();

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/120_data_streams_stats.yml

@@ -8,9 +8,6 @@ setup:
         name: my-template1
         body:
           index_patterns: [simple-data-stream1]
-          template:
-            settings:
-              index.number_of_replicas: 0
           data_stream: {}
   - do:
       allowed_warnings:
@@ -19,9 +16,6 @@ setup:
         name: my-template2
         body:
           index_patterns: [simple-data-stream2]
-          template:
-            settings:
-              index.number_of_replicas: 0
           data_stream: {}
 
 ---
@@ -51,6 +45,10 @@ setup:
       indices.rollover:
         alias: "simple-data-stream1"
 
+  - do:
+      cluster.health:
+        wait_for_no_initializing_shards: true
+
   - do:
       indices.data_streams_stats: {}
   - match: { data_stream_count: 1 }
@@ -195,8 +193,8 @@ setup:
 
   - do:
       indices.data_streams_stats: {}
-  - match: { _shards.total: 3 }
-  - match: { _shards.successful: 3 }
+  - match: { _shards.total: 6 }
+  - match: { _shards.successful: 6 }
   - match: { _shards.failed: 0 }
   - match: { data_stream_count: 2 }
   - match: { backing_indices: 3 }

server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java

@@ -242,6 +242,13 @@ public ShardsIterator allShards(String[] indices) {
         return allShardsSatisfyingPredicate(indices, Predicates.always(), false);
     }
 
+    /**
+     * Returns an iterator over all shard routing entries that are searchable (i.e., on a node that can service searches).
+     */
+    public ShardsIterator allSearchableShards(String[] indices) {
+        return allShardsSatisfyingPredicate(indices, ShardRouting::isSearchable, false);
+    }
+
     public ShardsIterator allActiveShards(String[] indices) {
         return allShardsSatisfyingPredicate(indices, ShardRouting::active, false);
     }