Data stream admin actions are now index-level actions (#59095)

build.gradle

@@ -174,8 +174,8 @@ tasks.register("verifyVersions") {
  * after the backport of the backcompat code is complete.
  */
 
-boolean bwc_tests_enabled = true
-final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
+boolean bwc_tests_enabled = false
+final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/pull/59095" /* place a PR link here when committing bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_data_stream.json

@@ -20,8 +20,8 @@
           ],
           "parts":{
             "name":{
-              "type":"string",
-              "description":"The name or wildcard expression of the requested data streams"
+              "type":"list",
+              "description":"A comma-separated list of data streams to get; use `*` to get all data streams"
             }
           }
         }

rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/10_basic.yml

@@ -157,7 +157,7 @@ setup:
       catch: missing
 
   - match: { status: 404 }
-  - match: { error.root_cause.0.type: "resource_not_found_exception" }
+  - match: { error.root_cause.0.type: "index_not_found_exception" }
 
   - do:
       indices.get_data_stream:

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java

@@ -257,7 +257,7 @@ public class BulkIntegrationIT extends ESIntegTestCase {
         bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
 
-        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
+        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
         GetDataStreamAction.Response getDataStreamsResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
         assertThat(getDataStreamsResponse.getDataStreams(), hasSize(4));
         getDataStreamsResponse.getDataStreams().sort(Comparator.comparing(dataStreamInfo -> dataStreamInfo.getDataStream().getName()));
@@ -293,7 +293,7 @@ public class BulkIntegrationIT extends ESIntegTestCase {
         BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
 
-        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
+        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
         GetDataStreamAction.Response getDataStreamsResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
         assertThat(getDataStreamsResponse.getDataStreams(), hasSize(0));
 

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

@@ -116,7 +116,7 @@ public class DataStreamIT extends ESIntegTestCase {
         createDataStreamRequest = new CreateDataStreamAction.Request("metrics-bar");
         client().admin().indices().createDataStream(createDataStreamRequest).get();
 
-        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
+        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
         GetDataStreamAction.Response getDataStreamResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
         getDataStreamResponse.getDataStreams().sort(Comparator.comparing(dataStreamInfo -> dataStreamInfo.getDataStream().getName()));
         assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(2));
@@ -278,7 +278,7 @@ public class DataStreamIT extends ESIntegTestCase {
         verifyDocs(dataStreamName, numDocs, 1, 1);
 
         String backingIndex = DataStream.getDefaultBackingIndexName(dataStreamName, 1);
-        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("*");
+        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"*"});
         GetDataStreamAction.Response getDataStreamResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
         assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
         assertThat(getDataStreamResponse.getDataStreams().get(0).getDataStream().getName(), equalTo(dataStreamName));
@@ -506,7 +506,7 @@ public class DataStreamIT extends ESIntegTestCase {
 
         CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request("logs-foobar");
         client().admin().indices().createDataStream(createDataStreamRequest).get();
-        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request("logs-foobar");
+        GetDataStreamAction.Request getDataStreamRequest = new GetDataStreamAction.Request(new String[]{"logs-foobar"});
         GetDataStreamAction.Response getDataStreamResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
         assertThat(getDataStreamResponse.getDataStreams().size(), equalTo(1));
         assertThat(getDataStreamResponse.getDataStreams().get(0).getDataStream().getName(), equalTo("logs-foobar"));
@@ -658,7 +658,7 @@ public class DataStreamIT extends ESIntegTestCase {
         indexDocs("metrics-foo", "@timestamp", numDocsFoo);
 
         GetDataStreamAction.Response response =
-            client().admin().indices().getDataStreams(new GetDataStreamAction.Request("metrics-foo")).actionGet();
+            client().admin().indices().getDataStreams(new GetDataStreamAction.Request(new String[]{"metrics-foo"})).actionGet();
         assertThat(response.getDataStreams().size(), is(1));
         GetDataStreamAction.Response.DataStreamInfo metricsFooDataStream = response.getDataStreams().get(0);
         assertThat(metricsFooDataStream.getDataStream().getName(), is("metrics-foo"));

server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java

@@ -116,7 +116,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
         assertEquals(1, hits.length);
         assertEquals(DOCUMENT_SOURCE, hits[0].getSourceAsMap());
 
-        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds")).get();
+        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
+            new GetDataStreamAction.Request(new String[]{"ds"})).get();
         assertEquals(1, ds.getDataStreams().size());
         assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
         assertEquals(DS_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@@ -154,7 +155,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
         assertEquals(1, hits.length);
         assertEquals(DOCUMENT_SOURCE, hits[0].getSourceAsMap());
 
-        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds")).get();
+        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
+            new GetDataStreamAction.Request(new String[]{"ds"})).get();
         assertEquals(1, ds.getDataStreams().size());
         assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
         assertEquals(DS_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@@ -187,7 +189,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
             .setRenameReplacement("ds2")
             .get();
 
-        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds2")).get();
+        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
+            new GetDataStreamAction.Request(new String[]{"ds2"})).get();
         assertEquals(1, ds.getDataStreams().size());
         assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
         assertEquals(DS2_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@@ -226,7 +229,7 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
 
         assertThat(restoreSnapshotResponse.status(), is(RestStatus.OK));
 
-        GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request("ds");
+        GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request(new String[]{"ds"});
         GetDataStreamAction.Response response = client.admin().indices().getDataStreams(getDSRequest).actionGet();
         assertThat(response.getDataStreams().get(0).getDataStream().getIndices().get(0).getName(), is(DS_BACKING_INDEX_NAME));
     }
@@ -260,13 +263,13 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
         assertThat(restoreSnapshotResponse.status(), is(RestStatus.OK));
 
         // assert "ds" was restored as "test-ds" and the backing index has a valid name
-        GetDataStreamAction.Request getRenamedDS = new GetDataStreamAction.Request("test-ds");
+        GetDataStreamAction.Request getRenamedDS = new GetDataStreamAction.Request(new String[]{"test-ds"});
         GetDataStreamAction.Response response = client.admin().indices().getDataStreams(getRenamedDS).actionGet();
         assertThat(response.getDataStreams().get(0).getDataStream().getIndices().get(0).getName(),
             is(DataStream.getDefaultBackingIndexName("test-ds", 1L)));
 
         // data stream "ds" should still exist in the system
-        GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request("ds");
+        GetDataStreamAction.Request getDSRequest = new GetDataStreamAction.Request(new String[]{"ds"});
         response = client.admin().indices().getDataStreams(getDSRequest).actionGet();
         assertThat(response.getDataStreams().get(0).getDataStream().getIndices().get(0).getName(), is(DS_BACKING_INDEX_NAME));
     }
@@ -292,7 +295,8 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
 
         assertEquals(RestStatus.OK, restoreSnapshotResponse.status());
 
-        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(new GetDataStreamAction.Request("ds2")).get();
+        GetDataStreamAction.Response ds = client.admin().indices().getDataStreams(
+            new GetDataStreamAction.Request(new String[]{"ds2"})).get();
         assertEquals(1, ds.getDataStreams().size());
         assertEquals(1, ds.getDataStreams().get(0).getDataStream().getIndices().size());
         assertEquals(DS2_BACKING_INDEX_NAME, ds.getDataStreams().get(0).getDataStream().getIndices().get(0).getName());
@@ -339,7 +343,7 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
 
         assertEquals(RestStatus.OK, restoreSnapshotResponse.status());
 
-        GetDataStreamAction.Request getRequest = new GetDataStreamAction.Request("ds");
+        GetDataStreamAction.Request getRequest = new GetDataStreamAction.Request(new String[]{"ds"});
         expectThrows(ResourceNotFoundException.class, () -> client.admin().indices().getDataStreams(getRequest).actionGet());
     }
 

server/src/main/java/org/elasticsearch/action/admin/indices/datastream/CreateDataStreamAction.java

@@ -21,8 +21,10 @@ package org.elasticsearch.action.admin.indices.datastream;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.ValidateActions;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
@@ -53,7 +55,7 @@ public class CreateDataStreamAction extends ActionType<AcknowledgedResponse> {
         super(NAME, AcknowledgedResponse::new);
     }
 
-    public static class Request extends AcknowledgedRequest<Request> {
+    public static class Request extends AcknowledgedRequest<Request> implements IndicesRequest {
 
         private final String name;
 
@@ -93,6 +95,16 @@ public class CreateDataStreamAction extends ActionType<AcknowledgedResponse> {
         public int hashCode() {
             return Objects.hash(name);
         }
+
+        @Override
+        public String[] indices() {
+            return new String[]{name};
+        }
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
+        }
     }
 
     public static class TransportAction extends TransportMasterNodeAction<Request, AcknowledgedResponse> {

server/src/main/java/org/elasticsearch/action/admin/indices/datastream/DeleteDataStreamAction.java

@@ -20,11 +20,12 @@ package org.elasticsearch.action.admin.indices.datastream;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
@@ -71,9 +72,9 @@ public class DeleteDataStreamAction extends ActionType<AcknowledgedResponse> {
         super(NAME, AcknowledgedResponse::new);
     }
 
-    public static class Request extends MasterNodeRequest<Request> {
+    public static class Request extends MasterNodeRequest<Request> implements IndicesRequest.Replaceable {
 
-        private final String[] names;
+        private String[] names;
 
         public Request(String[] names) {
             this.names = Objects.requireNonNull(names);
@@ -111,6 +112,29 @@ public class DeleteDataStreamAction extends ActionType<AcknowledgedResponse> {
         public int hashCode() {
             return Arrays.hashCode(names);
         }
+
+        @Override
+        public String[] indices() {
+            return names;
+        }
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            // this doesn't really matter since data stream name resolution isn't affected by IndicesOptions and
+            // a data stream's backing indices are retrieved from its metadata
+            return IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
+        }
+
+        @Override
+        public boolean includeDataStreams() {
+            return true;
+        }
+
+        @Override
+        public IndicesRequest indices(String... indices) {
+            this.names = indices;
+            return this;
+        }
     }
 
     public static class TransportAction extends TransportMasterNodeAction<Request, AcknowledgedResponse> {
@@ -176,16 +200,6 @@ public class DeleteDataStreamAction extends ActionType<AcknowledgedResponse> {
                 snapshottingDataStreams.addAll(SnapshotsService.snapshottingDataStreams(currentState, dataStreams));
             }
 
-            if (dataStreams.isEmpty()) {
-                // if only a match-all pattern was specified and no data streams were found because none exist, do not
-                // fail with data stream missing exception
-                if (request.names.length == 1 && Regex.isMatchAllPattern(request.names[0])) {
-                    return currentState;
-                }
-                throw new ResourceNotFoundException("data_streams matching [" + Strings.arrayToCommaDelimitedString(request.names) +
-                    "] not found");
-            }
-
             if (snapshottingDataStreams.isEmpty() == false) {
                 throw new SnapshotInProgressException("Cannot delete data streams that are being snapshotted: " + snapshottingDataStreams +
                     ". Try again after snapshot finishes or cancel the currently running snapshot.");

server/src/main/java/org/elasticsearch/action/admin/indices/datastream/GetDataStreamAction.java

@@ -20,12 +20,13 @@ package org.elasticsearch.action.admin.indices.datastream;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.MasterNodeReadRequest;
 import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
 import org.elasticsearch.cluster.AbstractDiffable;
@@ -43,7 +44,6 @@ import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -54,10 +54,12 @@ import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response> {
 
@@ -68,12 +70,12 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
         super(NAME, Response::new);
     }
 
-    public static class Request extends MasterNodeReadRequest<Request> {
+    public static class Request extends MasterNodeReadRequest<Request> implements IndicesRequest.Replaceable {
 
-        private final String name;
+        private String[] names;
 
-        public Request(String name) {
-            this.name = name;
+        public Request(String[] names) {
+            this.names = names;
         }
 
         @Override
@@ -83,13 +85,13 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
 
         public Request(StreamInput in) throws IOException {
             super(in);
-            this.name = in.readOptionalString();
+            this.names = in.readOptionalStringArray();
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
-            out.writeOptionalString(name);
+            out.writeOptionalStringArray(names);
         }
 
         @Override
@@ -97,12 +99,35 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
-            return Objects.equals(name, request.name);
+            return Arrays.equals(names, request.names);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(name);
+            return Arrays.hashCode(names);
+        }
+
+        @Override
+        public String[] indices() {
+            return names;
+        }
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            // this doesn't really matter since data stream name resolution isn't affected by IndicesOptions and
+            // a data stream's backing indices are retrieved from its metadata
+            return IndicesOptions.fromOptions(false, true, true, true, false, false, true, false);
+        }
+
+        @Override
+        public boolean includeDataStreams() {
+            return true;
+        }
+
+        @Override
+        public IndicesRequest indices(String... indices) {
+            this.names = indices;
+            return this;
         }
     }
 
@@ -261,7 +286,7 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
         @Override
         protected void masterOperation(Task task, Request request, ClusterState state,
                                        ActionListener<Response> listener) throws Exception {
-            List<DataStream> dataStreams = getDataStreams(state, request);
+            List<DataStream> dataStreams = getDataStreams(state, indexNameExpressionResolver, request);
             List<Response.DataStreamInfo> dataStreamInfos = new ArrayList<>(dataStreams.size());
             for (DataStream dataStream : dataStreams) {
                 String indexTemplate = MetadataIndexTemplateService.findV2Template(state.metadata(), dataStream.getName(), false);
@@ -280,26 +305,14 @@ public class GetDataStreamAction extends ActionType<GetDataStreamAction.Response
             listener.onResponse(new Response(dataStreamInfos));
         }
 
-        static List<DataStream> getDataStreams(ClusterState clusterState, Request request) {
+        static List<DataStream> getDataStreams(ClusterState clusterState, IndexNameExpressionResolver iner, Request request) {
+            List<String> results = iner.dataStreamNames(clusterState, request.indicesOptions(), request.names);
             Map<String, DataStream> dataStreams = clusterState.metadata().dataStreams();
 
-            // return all data streams if no name was specified
-            final String requestedName = request.name == null ? "*" : request.name;
-
-            final List<DataStream> results = new ArrayList<>();
-                if (Regex.isSimpleMatchPattern(requestedName)) {
-                    for (Map.Entry<String, DataStream> entry : dataStreams.entrySet()) {
-                        if (Regex.simpleMatch(requestedName, entry.getKey())) {
-                            results.add(entry.getValue());
-                        }
-                    }
-                } else if (dataStreams.containsKey(request.name)) {
-                    results.add(dataStreams.get(request.name));
-                } else {
-                    throw new ResourceNotFoundException("data_stream matching [" + request.name + "] not found");
-                }
-            results.sort(Comparator.comparing(DataStream::getName));
-            return results;
+            return results.stream()
+                .map(dataStreams::get)
+                .sorted(Comparator.comparing(DataStream::getName))
+                .collect(Collectors.toList());
         }
 
         @Override

server/src/main/java/org/elasticsearch/rest/action/admin/indices/RestGetDataStreamsAction.java

@@ -20,6 +20,7 @@ package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
@@ -44,7 +45,8 @@ public class RestGetDataStreamsAction extends BaseRestHandler {
 
     @Override
     protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
-        GetDataStreamAction.Request getDataStreamsRequest = new GetDataStreamAction.Request(request.param("name"));
+        GetDataStreamAction.Request getDataStreamsRequest = new GetDataStreamAction.Request(
+            Strings.splitStringByCommaToArray(request.param("name")));
         return channel -> client.admin().indices().getDataStreams(getDataStreamsRequest, new RestToXContentListener<>(channel));
     }
 }

server/src/test/java/org/elasticsearch/action/admin/indices/datastream/DeleteDataStreamRequestTests.java

@@ -18,7 +18,6 @@
  */
 package org.elasticsearch.action.admin.indices.datastream;
 
-import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamAction.Request;
@@ -29,6 +28,7 @@ import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.MetadataDeleteIndexService;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -46,6 +46,7 @@ import java.util.Set;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Matchers.any;
@@ -140,11 +141,18 @@ public class DeleteDataStreamRequestTests extends AbstractWireSerializingTestCas
 
     public void testDeleteNonexistentDataStream() {
         final String dataStreamName = "my-data-stream";
-        ClusterState cs = ClusterState.builder(new ClusterName("_name")).build();
+        String[] dataStreamNames = {"foo", "bar", "baz", "eggplant"};
+        ClusterState cs = getClusterStateWithDataStreams(List.of(
+            new Tuple<>(dataStreamNames[0], randomIntBetween(1, 3)),
+            new Tuple<>(dataStreamNames[1], randomIntBetween(1, 3)),
+            new Tuple<>(dataStreamNames[2], randomIntBetween(1, 3)),
+            new Tuple<>(dataStreamNames[3], randomIntBetween(1, 3))
+        ), List.of());
         DeleteDataStreamAction.Request req = new DeleteDataStreamAction.Request(new String[]{dataStreamName});
-        ResourceNotFoundException e = expectThrows(ResourceNotFoundException.class,
-            () -> DeleteDataStreamAction.TransportAction.removeDataStream(getMetadataDeleteIndexService(), cs, req));
-        assertThat(e.getMessage(), containsString("data_streams matching [" + dataStreamName + "] not found"));
+        ClusterState newState = DeleteDataStreamAction.TransportAction.removeDataStream(getMetadataDeleteIndexService(), cs, req);
+        assertThat(newState.metadata().dataStreams().size(), equalTo(cs.metadata().dataStreams().size()));
+        assertThat(newState.metadata().dataStreams().keySet(),
+            containsInAnyOrder(cs.metadata().dataStreams().keySet().toArray(Strings.EMPTY_ARRAY)));
     }
 
     @SuppressWarnings("unchecked")

server/src/test/java/org/elasticsearch/action/admin/indices/datastream/GetDataStreamsRequestTests.java

@@ -18,21 +18,19 @@
  */
 package org.elasticsearch.action.admin.indices.datastream;
 
-import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction.Request;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.DataStreamTestHelper;
 import org.elasticsearch.cluster.metadata.DataStream;
-import org.elasticsearch.cluster.metadata.IndexMetadata;
-import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 
 import java.util.List;
-import java.util.Map;
 
-import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
+import static org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamRequestTests.getClusterStateWithDataStreams;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -45,16 +43,20 @@ public class GetDataStreamsRequestTests extends AbstractWireSerializingTestCase<
 
     @Override
     protected Request createTestInstance() {
-        final String searchParameter;
+        final String[] searchParameter;
         switch (randomIntBetween(1, 4)) {
             case 1:
-                searchParameter = randomAlphaOfLength(8);
+                searchParameter = generateRandomStringArray(3, 8, false, false);
                 break;
             case 2:
-                searchParameter = randomAlphaOfLength(8) + "*";
+                String[] parameters = generateRandomStringArray(3, 8, false, false);
+                for (int k = 0; k < parameters.length; k++) {
+                    parameters[k] = parameters[k] + "*";
+                }
+                searchParameter = parameters;
                 break;
             case 3:
-                searchParameter = "*";
+                searchParameter = new String[]{"*"};
                 break;
             default:
                 searchParameter = null;
@@ -65,58 +67,75 @@ public class GetDataStreamsRequestTests extends AbstractWireSerializingTestCase<
 
     public void testGetDataStream() {
         final String dataStreamName = "my-data-stream";
-        IndexMetadata idx = DataStreamTestHelper.createFirstBackingIndex(dataStreamName).build();
-        DataStream existingDataStream =
-            new DataStream(dataStreamName, createTimestampField("@timestamp"), List.of(idx.getIndex()));
-        ClusterState cs = ClusterState.builder(new ClusterName("_name"))
-            .metadata(Metadata.builder().dataStreams(Map.of(dataStreamName, existingDataStream)).build()).build();
-        GetDataStreamAction.Request req = new GetDataStreamAction.Request(dataStreamName);
-        List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
+        ClusterState cs = getClusterStateWithDataStreams(
+            List.of(new Tuple<>(dataStreamName, 1)), List.of());
+        GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamName});
+        List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
         assertThat(dataStreams.size(), equalTo(1));
         assertThat(dataStreams.get(0).getName(), equalTo(dataStreamName));
     }
 
     public void testGetDataStreamsWithWildcards() {
         final String[] dataStreamNames = {"my-data-stream", "another-data-stream"};
-        IndexMetadata idx1 = DataStreamTestHelper.createFirstBackingIndex(dataStreamNames[0]).build();
-        IndexMetadata idx2 = DataStreamTestHelper.createFirstBackingIndex(dataStreamNames[1]).build();
-
-        DataStream ds1 = new DataStream(dataStreamNames[0], createTimestampField("@timestamp"), List.of(idx1.getIndex()));
-        DataStream ds2 = new DataStream(dataStreamNames[1], createTimestampField("@timestamp"), List.of(idx2.getIndex()));
-        ClusterState cs = ClusterState.builder(new ClusterName("_name"))
-            .metadata(Metadata.builder().dataStreams(
-                Map.of(dataStreamNames[0], ds1, dataStreamNames[1], ds2)).build())
-            .build();
-
-        GetDataStreamAction.Request req = new GetDataStreamAction.Request(dataStreamNames[1].substring(0, 5) + "*");
-        List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
+        ClusterState cs = getClusterStateWithDataStreams(
+            List.of(new Tuple<>(dataStreamNames[0], 1), new Tuple<>(dataStreamNames[1], 1)), List.of());
+
+        GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamNames[1].substring(0, 5) + "*"});
+        List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
         assertThat(dataStreams.size(), equalTo(1));
         assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
 
-        req = new GetDataStreamAction.Request("*");
-        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
+        req = new GetDataStreamAction.Request(new String[]{"*"});
+        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
         assertThat(dataStreams.size(), equalTo(2));
         assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
         assertThat(dataStreams.get(1).getName(), equalTo(dataStreamNames[0]));
 
-        req = new GetDataStreamAction.Request((String) null);
-        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
+        req = new GetDataStreamAction.Request((String[]) null);
+        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
         assertThat(dataStreams.size(), equalTo(2));
         assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
         assertThat(dataStreams.get(1).getName(), equalTo(dataStreamNames[0]));
 
-        req = new GetDataStreamAction.Request("matches-none*");
-        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, req);
+        req = new GetDataStreamAction.Request(new String[]{"matches-none*"});
+        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
         assertThat(dataStreams.size(), equalTo(0));
     }
 
+    public void testGetDataStreamsWithoutWildcards() {
+        final String[] dataStreamNames = {"my-data-stream", "another-data-stream"};
+        ClusterState cs = getClusterStateWithDataStreams(
+            List.of(new Tuple<>(dataStreamNames[0], 1), new Tuple<>(dataStreamNames[1], 1)), List.of());
+
+        GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamNames[0], dataStreamNames[1]});
+        List<DataStream> dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
+        assertThat(dataStreams.size(), equalTo(2));
+        assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
+        assertThat(dataStreams.get(1).getName(), equalTo(dataStreamNames[0]));
+
+        req = new GetDataStreamAction.Request(new String[]{dataStreamNames[1]});
+        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
+        assertThat(dataStreams.size(), equalTo(1));
+        assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[1]));
+
+        req = new GetDataStreamAction.Request(new String[]{dataStreamNames[0]});
+        dataStreams = GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req);
+        assertThat(dataStreams.size(), equalTo(1));
+        assertThat(dataStreams.get(0).getName(), equalTo(dataStreamNames[0]));
+
+        GetDataStreamAction.Request req2 = new GetDataStreamAction.Request(new String[]{"foo"});
+        IndexNotFoundException e = expectThrows(IndexNotFoundException.class,
+            () -> GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req2));
+        assertThat(e.getMessage(), containsString("no such index [foo]"));
+    }
+
     public void testGetNonexistentDataStream() {
         final String dataStreamName = "my-data-stream";
         ClusterState cs = ClusterState.builder(new ClusterName("_name")).build();
-        GetDataStreamAction.Request req = new GetDataStreamAction.Request(dataStreamName);
-        ResourceNotFoundException e = expectThrows(ResourceNotFoundException.class,
-            () -> GetDataStreamAction.TransportAction.getDataStreams(cs, req));
-        assertThat(e.getMessage(), containsString("data_stream matching [" + dataStreamName + "] not found"));
+        GetDataStreamAction.Request req = new GetDataStreamAction.Request(new String[]{dataStreamName});
+        IndexNotFoundException e = expectThrows(IndexNotFoundException.class,
+            () -> GetDataStreamAction.TransportAction.getDataStreams(cs, new IndexNameExpressionResolver(), req));
+        assertThat(e.getMessage(), containsString("no such index [" + dataStreamName + "]"));
     }
 
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolver.java

@@ -201,8 +201,6 @@ public class ClusterPrivilegeResolver {
     public static boolean isClusterAction(String actionName) {
         return actionName.startsWith("cluster:") ||
             actionName.startsWith("indices:admin/template/") ||
-            // todo: hack until we implement security of data_streams
-            actionName.startsWith("indices:admin/data_stream/") ||
             actionName.startsWith("indices:admin/index_template/");
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/IndexPrivilege.java

@@ -13,6 +13,9 @@ import org.elasticsearch.action.admin.indices.alias.get.GetAliasesAction;
 import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
 import org.elasticsearch.action.admin.indices.create.AutoCreateAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
+import org.elasticsearch.action.admin.indices.datastream.CreateDataStreamAction;
+import org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamAction;
+import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexAction;
 import org.elasticsearch.action.admin.indices.mapping.get.GetFieldMappingsAction;
@@ -60,11 +63,12 @@ public final class IndexPrivilege extends Privilege {
     private static final Automaton MONITOR_AUTOMATON = patterns("indices:monitor/*");
     private static final Automaton MANAGE_AUTOMATON =
             unionAndMinimize(Arrays.asList(MONITOR_AUTOMATON, patterns("indices:admin/*")));
-    private static final Automaton CREATE_INDEX_AUTOMATON = patterns(CreateIndexAction.NAME, AutoCreateAction.NAME);
-    private static final Automaton DELETE_INDEX_AUTOMATON = patterns(DeleteIndexAction.NAME);
-    private static final Automaton VIEW_METADATA_AUTOMATON = patterns(GetAliasesAction.NAME,
-            GetIndexAction.NAME, GetFieldMappingsAction.NAME + "*", GetMappingsAction.NAME,
-            ClusterSearchShardsAction.NAME, ValidateQueryAction.NAME + "*", GetSettingsAction.NAME, ExplainLifecycleAction.NAME);
+    private static final Automaton CREATE_INDEX_AUTOMATON = patterns(CreateIndexAction.NAME, AutoCreateAction.NAME,
+            CreateDataStreamAction.NAME);
+    private static final Automaton DELETE_INDEX_AUTOMATON = patterns(DeleteIndexAction.NAME, DeleteDataStreamAction.NAME);
+    private static final Automaton VIEW_METADATA_AUTOMATON = patterns(GetAliasesAction.NAME, GetIndexAction.NAME,
+            GetFieldMappingsAction.NAME + "*", GetMappingsAction.NAME, ClusterSearchShardsAction.NAME, ValidateQueryAction.NAME + "*",
+            GetSettingsAction.NAME, ExplainLifecycleAction.NAME, GetDataStreamAction.NAME);
     private static final Automaton MANAGE_FOLLOW_INDEX_AUTOMATON = patterns(PutFollowAction.NAME, UnfollowAction.NAME,
         CloseIndexAction.NAME + "*");
     private static final Automaton MANAGE_LEADER_INDEX_AUTOMATON = patterns(ForgetFollowerAction.NAME + "*");

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java

@@ -15,6 +15,7 @@ import org.elasticsearch.action.StepListener;
 import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.action.admin.indices.datastream.CreateDataStreamAction;
 import org.elasticsearch.action.bulk.BulkItemRequest;
 import org.elasticsearch.action.bulk.BulkShardRequest;
 import org.elasticsearch.action.bulk.TransportShardBulkAction;
@@ -294,11 +295,11 @@ public class AuthorizationService {
         }
         //if we are creating an index we need to authorize potential aliases created at the same time
         if (IndexPrivilege.CREATE_INDEX_MATCHER.test(action)) {
-            assert request instanceof CreateIndexRequest;
-            Set<Alias> aliases = ((CreateIndexRequest) request).aliases();
-            if (aliases.isEmpty()) {
+            assert (request instanceof CreateIndexRequest) || (request instanceof CreateDataStreamAction.Request);
+            if (request instanceof CreateDataStreamAction.Request || ((CreateIndexRequest) request).aliases().isEmpty()) {
                 runRequestInterceptors(requestInfo, authzInfo, authorizationEngine, listener);
             } else {
+                Set<Alias> aliases = ((CreateIndexRequest) request).aliases();
                 final RequestInfo aliasesRequestInfo = new RequestInfo(authentication, request, IndicesAliasesAction.NAME);
                 authzEngine.authorizeIndexAction(aliasesRequestInfo, authzInfo,
                     ril -> {

x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/50_data_streams.yml

@@ -15,7 +15,17 @@ setup:
         body:  >
           {
             "indices": [
-              { "names": ["simple*"], "privileges": ["read", "write", "view_index_metadata"] }
+              { "names": ["simple*"], "privileges": ["read", "write", "create_index", "view_index_metadata", "delete_index"] }
+            ]
+          }
+
+  - do:
+      security.put_role:
+        name: "data_stream_role2"
+        body:  >
+          {
+            "indices": [
+              { "names": ["matches_none"], "privileges": ["read", "write", "create_index", "view_index_metadata", "delete_index"] }
             ]
           }
 
@@ -26,16 +36,26 @@ setup:
           {
             "password" : "x-pack-test-password",
             "roles" : [ "data_stream_role" ],
-            "full_name" : "user with privileges on data streams but not backing indices"
+            "full_name" : "user with privileges on some data streams"
+          }
+
+  - do:
+      security.put_user:
+        username: "no_authz_user"
+        body:  >
+          {
+            "password" : "x-pack-test-password",
+            "roles" : [ "data_stream_role2" ],
+            "full_name" : "user with privileges on no data streams"
           }
 
   - do:
       allowed_warnings:
-        - "index template [my-template1] has index patterns [simple-data-stream1] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
+        - "index template [my-template1] has index patterns [s*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
       indices.put_index_template:
         name: my-template1
         body:
-          index_patterns: [simple-data-stream1]
+          index_patterns: [s*]
           template:
             mappings:
               properties:
@@ -51,11 +71,21 @@ teardown:
         username: "test_user"
         ignore: 404
 
+  - do:
+      security.delete_user:
+        username: "test_user2"
+        ignore: 404
+
   - do:
       security.delete_role:
         name: "data_stream_role"
         ignore: 404
 
+  - do:
+      security.delete_role:
+        name: "data_stream_role2"
+        ignore: 404
+
 ---
 "Test backing indices inherit parent data stream privileges":
   - skip:
@@ -147,3 +177,130 @@ teardown:
       indices.delete_data_stream:
         name: simple-data-stream1
   - is_true: acknowledged
+
+---
+"Test that create data stream is limited to authorized namespace":
+  - skip:
+      version: " - 7.99.99"
+      reason: "change to 7.8.99 after backport"
+
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.create_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do: # superuser
+      indices.delete_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do:
+      catch: forbidden
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.create_data_stream:
+        name: outside_of_namespace
+
+---
+"Test that get data stream is limited to authorized namespace":
+  - skip:
+      version: " - 7.99.99"
+      reason: "change to 7.8.99 after backport"
+
+  - do: # superuser
+      indices.create_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do: # superuser
+      indices.create_data_stream:
+        name: s-outside-of-authed-namespace
+  - is_true: acknowledged
+
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.get_data_stream:
+        name: simple-data-stream1
+
+  - length: { data_streams: 1 }
+  - match: { data_streams.0.name: simple-data-stream1 }
+
+  - do: # superuser
+      indices.get_data_stream:
+        name: "*"
+
+  # superuser should be authorized for both data streams
+  - length: { data_streams: 2 }
+  - match: { data_streams.0.name: s-outside-of-authed-namespace }
+  - match: { data_streams.1.name: simple-data-stream1 }
+
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.get_data_stream:
+        name: "*"
+
+  # test_user should be authorized for only one data stream
+  - length: { data_streams: 1 }
+  - match: { data_streams.0.name: simple-data-stream1 }
+
+  - do:
+      catch: forbidden
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.get_data_stream:
+        name: outside_of_namespace
+
+  - do:
+      headers: { Authorization: "Basic bm9fYXV0aHpfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" } # no_authz_user
+      indices.get_data_stream: {}
+
+  # no_authz_user should not be authorized for any data streams
+  - length: { data_streams: 0 }
+
+  - do: # superuser
+      indices.delete_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do: # superuser
+      indices.delete_data_stream:
+        name: s-outside-of-authed-namespace
+  - is_true: acknowledged
+
+---
+"Test that delete data stream is limited to authorized namespace":
+  - skip:
+      version: " - 7.99.99"
+      reason: "change to 7.8.99 after backport"
+
+  - do: # superuser
+      indices.create_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do: # superuser
+      indices.create_data_stream:
+        name: s-outside-of-authed-namespace
+  - is_true: acknowledged
+
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.delete_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do:
+      catch: forbidden
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.delete_data_stream:
+        name: s-outside-of-authed-namespace
+
+  - do:
+      catch: forbidden
+      headers: { Authorization: "Basic bm9fYXV0aHpfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" } # no_authz_user
+      indices.delete_data_stream:
+        name: simple-data-stream1
+
+  - do: # superuser
+      indices.delete_data_stream:
+        name: s-outside-of-authed-namespace
+  - is_true: acknowledged