Allow a number of broadcast transport actions to resolve data streams (#55726) (#56502)

Change TransportBroadcastByNodeAction and TransportBroadcastReplicationAction
to be able to resolve data streams by default. Implementations can change this ability.

This change allows to following APIs to resolve data streams: flush,
refresh (already supported data streams), force merge, clear indices cache,
indices stats (already supported data streams), segments, upgrade stats, 
upgrade, validate query, searchable snapshots stats, clear searchable snapshots cache and
reload analyzers APIs.

Relates to #53100

server/src/internalClusterTest/java/org/elasticsearch/indices/IndicesOptionsIntegrationIT.java

@@ -87,7 +87,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1", "test2"), true);
         verify(_flush("test1", "test2"),true);
         verify(segments("test1", "test2"), true);
-        verify(stats("test1", "test2"), true);
+        verify(indicesStats("test1", "test2"), true);
         verify(forceMerge("test1", "test2"), true);
         verify(refreshBuilder("test1", "test2"), true);
         verify(validateQuery("test1", "test2"), true);
@@ -104,7 +104,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1", "test2").setIndicesOptions(options), true);
         verify(_flush("test1", "test2").setIndicesOptions(options),true);
         verify(segments("test1", "test2").setIndicesOptions(options), true);
-        verify(stats("test1", "test2").setIndicesOptions(options), true);
+        verify(indicesStats("test1", "test2").setIndicesOptions(options), true);
         verify(forceMerge("test1", "test2").setIndicesOptions(options), true);
         verify(refreshBuilder("test1", "test2").setIndicesOptions(options), true);
         verify(validateQuery("test1", "test2").setIndicesOptions(options), true);
@@ -121,7 +121,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1", "test2").setIndicesOptions(options), false);
         verify(_flush("test1", "test2").setIndicesOptions(options), false);
         verify(segments("test1", "test2").setIndicesOptions(options), false);
-        verify(stats("test1", "test2").setIndicesOptions(options), false);
+        verify(indicesStats("test1", "test2").setIndicesOptions(options), false);
         verify(forceMerge("test1", "test2").setIndicesOptions(options), false);
         verify(refreshBuilder("test1", "test2").setIndicesOptions(options), false);
         verify(validateQuery("test1", "test2").setIndicesOptions(options), false);
@@ -139,7 +139,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1", "test2").setIndicesOptions(options), false);
         verify(_flush("test1", "test2").setIndicesOptions(options),false);
         verify(segments("test1", "test2").setIndicesOptions(options), false);
-        verify(stats("test1", "test2").setIndicesOptions(options), false);
+        verify(indicesStats("test1", "test2").setIndicesOptions(options), false);
         verify(forceMerge("test1", "test2").setIndicesOptions(options), false);
         verify(refreshBuilder("test1", "test2").setIndicesOptions(options), false);
         verify(validateQuery("test1", "test2").setIndicesOptions(options), false);
@@ -166,7 +166,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1").setIndicesOptions(options), true);
         verify(_flush("test1").setIndicesOptions(options),true);
         verify(segments("test1").setIndicesOptions(options), true);
-        verify(stats("test1").setIndicesOptions(options), true);
+        verify(indicesStats("test1").setIndicesOptions(options), true);
         verify(forceMerge("test1").setIndicesOptions(options), true);
         verify(refreshBuilder("test1").setIndicesOptions(options), true);
         verify(validateQuery("test1").setIndicesOptions(options), true);
@@ -184,7 +184,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1").setIndicesOptions(options), false);
         verify(_flush("test1").setIndicesOptions(options),false);
         verify(segments("test1").setIndicesOptions(options), false);
-        verify(stats("test1").setIndicesOptions(options), false);
+        verify(indicesStats("test1").setIndicesOptions(options), false);
         verify(forceMerge("test1").setIndicesOptions(options), false);
         verify(refreshBuilder("test1").setIndicesOptions(options), false);
         verify(validateQuery("test1").setIndicesOptions(options), false);
@@ -204,7 +204,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1").setIndicesOptions(options), false);
         verify(_flush("test1").setIndicesOptions(options),false);
         verify(segments("test1").setIndicesOptions(options), false);
-        verify(stats("test1").setIndicesOptions(options), false);
+        verify(indicesStats("test1").setIndicesOptions(options), false);
         verify(forceMerge("test1").setIndicesOptions(options), false);
         verify(refreshBuilder("test1").setIndicesOptions(options), false);
         verify(validateQuery("test1").setIndicesOptions(options), false);
@@ -223,7 +223,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1").setIndicesOptions(options), true);
         verify(_flush("test1").setIndicesOptions(options),true);
         verify(segments("test1").setIndicesOptions(options), true);
-        verify(stats("test1").setIndicesOptions(options), true);
+        verify(indicesStats("test1").setIndicesOptions(options), true);
         verify(forceMerge("test1").setIndicesOptions(options), true);
         verify(refreshBuilder("test1").setIndicesOptions(options), true);
         verify(validateQuery("test1").setIndicesOptions(options), true);
@@ -241,7 +241,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1").setIndicesOptions(options), false);
         verify(_flush("test1").setIndicesOptions(options),false);
         verify(segments("test1").setIndicesOptions(options), false);
-        verify(stats("test1").setIndicesOptions(options), false);
+        verify(indicesStats("test1").setIndicesOptions(options), false);
         verify(forceMerge("test1").setIndicesOptions(options), false);
         verify(refreshBuilder("test1").setIndicesOptions(options), false);
         verify(validateQuery("test1").setIndicesOptions(options), false);
@@ -260,7 +260,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache("test1").setIndicesOptions(options), false);
         verify(_flush("test1").setIndicesOptions(options),false);
         verify(segments("test1").setIndicesOptions(options), false);
-        verify(stats("test1").setIndicesOptions(options), false);
+        verify(indicesStats("test1").setIndicesOptions(options), false);
         verify(forceMerge("test1").setIndicesOptions(options), false);
         verify(refreshBuilder("test1").setIndicesOptions(options), false);
         verify(validateQuery("test1").setIndicesOptions(options), false);
@@ -310,7 +310,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache(indices), false);
         verify(_flush(indices),false);
         verify(segments(indices), false);
-        verify(stats(indices), false);
+        verify(indicesStats(indices), false);
         verify(forceMerge(indices), false);
         verify(refreshBuilder(indices), false);
         verify(validateQuery(indices), true);
@@ -328,7 +328,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache(indices).setIndicesOptions(options), false);
         verify(_flush(indices).setIndicesOptions(options),false);
         verify(segments(indices).setIndicesOptions(options), false);
-        verify(stats(indices).setIndicesOptions(options), false);
+        verify(indicesStats(indices).setIndicesOptions(options), false);
         verify(forceMerge(indices).setIndicesOptions(options), false);
         verify(refreshBuilder(indices).setIndicesOptions(options), false);
         verify(validateQuery(indices).setIndicesOptions(options), false);
@@ -349,7 +349,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache(indices), false);
         verify(_flush(indices),false);
         verify(segments(indices), false);
-        verify(stats(indices), false);
+        verify(indicesStats(indices), false);
         verify(forceMerge(indices), false);
         verify(refreshBuilder(indices), false);
         verify(validateQuery(indices), false);
@@ -367,7 +367,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache(indices), false);
         verify(_flush(indices),false);
         verify(segments(indices), false);
-        verify(stats(indices), false);
+        verify(indicesStats(indices), false);
         verify(forceMerge(indices), false);
         verify(refreshBuilder(indices), false);
         verify(validateQuery(indices), true);
@@ -385,7 +385,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verify(clearCache(indices).setIndicesOptions(options), false);
         verify(_flush(indices).setIndicesOptions(options),false);
         verify(segments(indices).setIndicesOptions(options), false);
-        verify(stats(indices).setIndicesOptions(options), false);
+        verify(indicesStats(indices).setIndicesOptions(options), false);
         verify(forceMerge(indices).setIndicesOptions(options), false);
         verify(refreshBuilder(indices).setIndicesOptions(options), false);
         verify(validateQuery(indices).setIndicesOptions(options), false);
@@ -661,12 +661,15 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verifyResolvability(dataStreamName, refreshBuilder(dataStreamName), false);
         verifyResolvability(dataStreamName, search(dataStreamName), false, 1);
         verifyResolvability(dataStreamName, msearch(null, dataStreamName), false);
-        verifyResolvability(dataStreamName, clearCache(dataStreamName), true);
-        verifyResolvability(dataStreamName, _flush(dataStreamName),true);
-        verifyResolvability(dataStreamName, segments(dataStreamName), true);
-        verifyResolvability(dataStreamName, stats(dataStreamName), false);
-        verifyResolvability(dataStreamName, forceMerge(dataStreamName), true);
-        verifyResolvability(dataStreamName, validateQuery(dataStreamName), true);
+        verifyResolvability(dataStreamName, clearCache(dataStreamName), false);
+        verifyResolvability(dataStreamName, _flush(dataStreamName),false);
+        verifyResolvability(dataStreamName, segments(dataStreamName), false);
+        verifyResolvability(dataStreamName, indicesStats(dataStreamName), false);
+        verifyResolvability(dataStreamName, forceMerge(dataStreamName), false);
+        verifyResolvability(dataStreamName, validateQuery(dataStreamName), false);
+        verifyResolvability(dataStreamName, client().admin().indices().prepareUpgrade(dataStreamName), false);
+        verifyResolvability(dataStreamName, client().admin().indices().prepareRecoveries(dataStreamName), false);
+        verifyResolvability(dataStreamName, client().admin().indices().prepareUpgradeStatus(dataStreamName), false);
         verifyResolvability(dataStreamName, getAliases(dataStreamName), true);
         verifyResolvability(dataStreamName, getFieldMapping(dataStreamName), true);
         verifyResolvability(dataStreamName, getMapping(dataStreamName), true);
@@ -685,12 +688,15 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         verifyResolvability(wildcardExpression, refreshBuilder(wildcardExpression), false);
         verifyResolvability(wildcardExpression, search(wildcardExpression), false, 2);
         verifyResolvability(wildcardExpression, msearch(null, wildcardExpression), false);
-        verifyResolvability(wildcardExpression, clearCache(wildcardExpression), true);
-        verifyResolvability(wildcardExpression, _flush(wildcardExpression),true);
-        verifyResolvability(wildcardExpression, segments(wildcardExpression), true);
-        verifyResolvability(wildcardExpression, stats(wildcardExpression), false);
-        verifyResolvability(wildcardExpression, forceMerge(wildcardExpression), true);
-        verifyResolvability(wildcardExpression, validateQuery(wildcardExpression), true);
+        verifyResolvability(wildcardExpression, clearCache(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, _flush(wildcardExpression),false);
+        verifyResolvability(wildcardExpression, segments(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, indicesStats(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, forceMerge(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, validateQuery(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, client().admin().indices().prepareUpgrade(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, client().admin().indices().prepareRecoveries(wildcardExpression), false);
+        verifyResolvability(wildcardExpression, client().admin().indices().prepareUpgradeStatus(wildcardExpression), false);
         verifyResolvability(wildcardExpression, getAliases(wildcardExpression), true);
         verifyResolvability(wildcardExpression, getFieldMapping(wildcardExpression), true);
         verifyResolvability(wildcardExpression, getMapping(wildcardExpression), true);
@@ -756,7 +762,7 @@ public class IndicesOptionsIntegrationIT extends ESIntegTestCase {
         return client().admin().indices().prepareSegments(indices);
     }
 
-    private static IndicesStatsRequestBuilder stats(String... indices) {
+    private static IndicesStatsRequestBuilder indicesStats(String... indices) {
         return client().admin().indices().prepareStats(indices);
     }
 

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java

@@ -64,9 +64,4 @@ public class TransportRefreshAction
                                           List<DefaultShardOperationFailedException> shardFailures) {
         return new RefreshResponse(totalNumCopies, successfulShards, failedShards, shardFailures);
     }
-
-    @Override
-    protected boolean shouldIncludeDataStreams() {
-        return true;
-    }
 }

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

@@ -57,11 +57,6 @@ public class TransportIndicesStatsAction extends TransportBroadcastByNodeAction<
         this.indicesService = indicesService;
     }
 
-    @Override
-    protected boolean shouldIncludeDataStreams() {
-        return true;
-    }
-
     /**
      * Status goes across *all* shards.
      */

server/src/main/java/org/elasticsearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java

@@ -226,7 +226,7 @@ public abstract class TransportBroadcastByNodeAction<Request extends BroadcastRe
     }
 
     protected boolean shouldIncludeDataStreams() {
-        return false;
+        return true;
     }
 
     protected class AsyncAction {

server/src/main/java/org/elasticsearch/action/support/replication/TransportBroadcastReplicationAction.java

@@ -137,7 +137,7 @@ public abstract class TransportBroadcastReplicationAction<Request extends Broadc
     }
 
     protected boolean shouldIncludeDataStreams() {
-        return false;
+        return true;
     }
 
     protected abstract ShardResponse newShardResponse();

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportForgetFollowerAction.java

@@ -148,4 +148,8 @@ public class TransportForgetFollowerAction extends TransportBroadcastByNodeActio
         return null;
     }
 
+    @Override
+    protected boolean shouldIncludeDataStreams() {
+        return false;
+    }
 }

x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml

@@ -0,0 +1,38 @@
+---
+"Verify data stream resolvability for xpack apis":
+  - skip:
+      version: " - 7.99.99"
+      reason: skip untill backported
+
+  - do:
+      indices.create_data_stream:
+        name: logs-foobar
+        body:
+          timestamp_field: "@timestamp"
+  - is_true: acknowledged
+
+  - do:
+      indices.reload_search_analyzers:
+        index: logs-foobar
+  - length: { reload_details: 1}
+
+  # this 404 is expected, because there are no snapshot indices
+  # but the data stream was resolved, which is what is being tested here
+  - do:
+      catch: missing
+      searchable_snapshots.clear_cache:
+        index: logs-foobar
+  - match: { error.reason: "No searchable snapshots indices found"}
+
+  # this 404 is expected, because there are no snapshot indices
+  # but the data stream was resolved, which is what is being tested here
+  - do:
+      catch: missing
+      searchable_snapshots.stats:
+        index: logs-foobar
+  - match: { error.reason: "No searchable snapshots indices found"}
+
+  - do:
+      indices.delete_data_stream:
+        name: logs-foobar
+  - is_true: acknowledged