Initial data stream commit

client/rest-high-level/src/test/java/org/elasticsearch/client/RestHighLevelClientTests.java

@@ -793,7 +793,10 @@ public void testApiNamingConventions() throws Exception {
             "scripts_painless_execute",
             "cluster.put_component_template",
             "cluster.get_component_template",
-            "cluster.delete_component_template"
+            "cluster.delete_component_template",
+            "indices.create_data_stream",
+            "indices.get_data_streams",
+            "indices.delete_data_stream"
         };
         //These API are not required for high-level client feature completeness
         String[] notRequiredApi = new String[] {

distribution/archives/integ-test-zip/build.gradle

@@ -37,5 +37,6 @@ integTest.runner {
 testClusters.integTest {
   if (BuildParams.isSnapshotBuild() == false) {
     systemProperty 'es.itv2_feature_flag_registered', 'true'
+    systemProperty 'es.datastreams_feature_flag_registered', 'true'
   }
 }

qa/smoke-test-multinode/build.gradle

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+import org.elasticsearch.gradle.info.BuildParams
 
 import org.elasticsearch.gradle.info.BuildParams
 
@@ -52,5 +53,6 @@ integTest.runner {
 testClusters.integTest {
   if (BuildParams.isSnapshotBuild() == false) {
     systemProperty 'es.itv2_feature_flag_registered', 'true'
+    systemProperty 'es.datastreams_feature_flag_registered', 'true'
   }
 }

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

@@ -0,0 +1,31 @@
+{
+  "indices.create_data_stream":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/data-streams.html",
+      "description":"Creates or updates a data stream"
+    },
+    "stability":"experimental",
+    "url":{
+      "paths":[
+        {
+          "path":"/_data_stream/{name}",
+          "methods":[
+            "PUT"
+          ],
+          "parts":{
+            "name":{
+              "type":"string",
+              "description":"The name of the data stream"
+            }
+          }
+        }
+      ]
+    },
+    "params":{
+    },
+    "body":{
+      "description":"The data stream definition",
+      "required":true
+    }
+  }
+}

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

@@ -0,0 +1,26 @@
+{
+  "indices.delete_data_stream":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/data-streams.html",
+      "description":"Deletes a data stream."
+    },
+    "stability":"experimental",
+    "url":{
+      "paths":[
+        {
+          "path":"/_data_stream/{name}",
+          "methods":[
+            "DELETE"
+          ],
+          "parts":{
+            "name":{
+              "type":"string",
+              "description":"The name of the data stream"
+            }
+          }
+        }
+      ]
+    },
+    "params":{}
+  }
+}

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

@@ -0,0 +1,33 @@
+{
+  "indices.get_data_streams":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/data-streams.html",
+      "description":"Returns data streams."
+    },
+    "stability":"experimental",
+    "url":{
+      "paths":[
+        {
+          "path":"/_data_streams",
+          "methods":[
+            "GET"
+          ]
+        },
+        {
+          "path":"/_data_streams/{name}",
+          "methods":[
+            "GET"
+          ],
+          "parts":{
+            "name":{
+              "type":"list",
+              "description":"The comma separated names of data streams"
+            }
+          }
+        }
+      ]
+    },
+    "params":{
+    }
+  }
+}

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

@@ -0,0 +1,26 @@
+---
+"Test stubs":
+  - skip:
+      version: " - 7.99.99"
+      reason: not backported yet
+
+  - do:
+      indices.create_data_stream:
+        name: data-stream2
+        body:
+          timestamp_field: "@timestamp"
+  - is_true: acknowledged
+
+  - do:
+      indices.get_data_streams: {}
+  - match: { 0.name: my_data_stream1 }
+  - match: { 0.timestamp_field: '@timestamp' }
+  - match: { 0.indices: ['my_data_stream1-000000'] }
+  - match: { 1.name: my_data_stream2 }
+  - match: { 1.timestamp_field: '@timestamp' }
+  - match: { 1.indices: [] }
+
+  - do:
+      indices.delete_data_stream:
+        name: data-stream2
+  - is_true: acknowledged

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -28,6 +28,9 @@
 import org.elasticsearch.action.admin.cluster.configuration.ClearVotingConfigExclusionsAction;
 import org.elasticsearch.action.admin.cluster.configuration.TransportAddVotingConfigExclusionsAction;
 import org.elasticsearch.action.admin.cluster.configuration.TransportClearVotingConfigExclusionsAction;
+import org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamAction;
+import org.elasticsearch.action.admin.indices.datastream.GetDataStreamsAction;
+import org.elasticsearch.action.admin.indices.datastream.CreateDataStreamAction;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthAction;
 import org.elasticsearch.action.admin.cluster.health.TransportClusterHealthAction;
 import org.elasticsearch.action.admin.cluster.node.hotthreads.NodesHotThreadsAction;
@@ -250,9 +253,11 @@
 import org.elasticsearch.rest.action.admin.cluster.RestClusterStatsAction;
 import org.elasticsearch.rest.action.admin.cluster.RestClusterUpdateSettingsAction;
 import org.elasticsearch.rest.action.admin.cluster.RestCreateSnapshotAction;
+import org.elasticsearch.rest.action.admin.indices.RestDeleteDataStreamAction;
 import org.elasticsearch.rest.action.admin.cluster.RestDeleteRepositoryAction;
 import org.elasticsearch.rest.action.admin.cluster.RestDeleteSnapshotAction;
 import org.elasticsearch.rest.action.admin.cluster.RestDeleteStoredScriptAction;
+import org.elasticsearch.rest.action.admin.indices.RestGetDataStreamsAction;
 import org.elasticsearch.rest.action.admin.cluster.RestGetRepositoriesAction;
 import org.elasticsearch.rest.action.admin.cluster.RestGetScriptContextAction;
 import org.elasticsearch.rest.action.admin.cluster.RestGetScriptLanguageAction;
@@ -265,6 +270,7 @@
 import org.elasticsearch.rest.action.admin.cluster.RestNodesStatsAction;
 import org.elasticsearch.rest.action.admin.cluster.RestNodesUsageAction;
 import org.elasticsearch.rest.action.admin.cluster.RestPendingClusterTasksAction;
+import org.elasticsearch.rest.action.admin.indices.RestCreateDataStreamAction;
 import org.elasticsearch.rest.action.admin.cluster.RestPutRepositoryAction;
 import org.elasticsearch.rest.action.admin.cluster.RestPutStoredScriptAction;
 import org.elasticsearch.rest.action.admin.cluster.RestReloadSecureSettingsAction;
@@ -388,6 +394,24 @@ public class ActionModule extends AbstractModule {
         }
     }
 
+    private static final boolean DATASTREAMS_FEATURE_FLAG_REGISTERED;
+
+    static {
+        final String property = System.getProperty("es.datastreams_feature_flag_registered");
+        if (Build.CURRENT.isSnapshot() && property != null) {
+            throw new IllegalArgumentException("es.datastreams_feature_flag_registered is only supported in non-snapshot builds");
+        }
+        if (Build.CURRENT.isSnapshot() || "true".equals(property)) {
+            DATASTREAMS_FEATURE_FLAG_REGISTERED = true;
+        } else if ("false".equals(property) || property == null) {
+            DATASTREAMS_FEATURE_FLAG_REGISTERED = false;
+        } else {
+            throw new IllegalArgumentException(
+                "expected es.datastreams_feature_flag_registered to be unset or [true|false] but was [" + property + "]"
+            );
+        }
+    }
+
     private final Settings settings;
     private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final IndexScopedSettings indexScopedSettings;
@@ -565,6 +589,13 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
 
         actionPlugins.stream().flatMap(p -> p.getActions().stream()).forEach(actions::register);
 
+        // Data streams:
+        if (DATASTREAMS_FEATURE_FLAG_REGISTERED) {
+            actions.register(CreateDataStreamAction.INSTANCE, CreateDataStreamAction.TransportAction.class);
+            actions.register(DeleteDataStreamAction.INSTANCE, DeleteDataStreamAction.TransportAction.class);
+            actions.register(GetDataStreamsAction.INSTANCE, GetDataStreamsAction.TransportAction.class);
+        }
+
         // Persistent tasks:
         actions.register(StartPersistentTaskAction.INSTANCE, StartPersistentTaskAction.TransportAction.class);
         actions.register(UpdatePersistentTaskStatusAction.INSTANCE, UpdatePersistentTaskStatusAction.TransportAction.class);
@@ -717,6 +748,13 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestDeletePipelineAction());
         registerHandler.accept(new RestSimulatePipelineAction());
 
+        // Data Stream API
+        if (DATASTREAMS_FEATURE_FLAG_REGISTERED) {
+            registerHandler.accept(new RestCreateDataStreamAction());
+            registerHandler.accept(new RestDeleteDataStreamAction());
+            registerHandler.accept(new RestGetDataStreamsAction());
+        }
+
         // CAT API
         registerHandler.accept(new RestAllocationAction());
         registerHandler.accept(new RestShardsAction());

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

@@ -0,0 +1,128 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+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.support.ActionFilters;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.MasterNodeRequest;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class CreateDataStreamAction extends ActionType<AcknowledgedResponse> {
+
+    public static final CreateDataStreamAction INSTANCE = new CreateDataStreamAction();
+    public static final String NAME = "indices:admin/data_stream/create";
+
+    private CreateDataStreamAction() {
+        super(NAME, AcknowledgedResponse::new);
+    }
+
+    public static class Request extends MasterNodeRequest<Request> {
+
+        private final String name;
+        private String timestampFieldName;
+
+        public Request(String name) {
+            this.name = name;
+        }
+
+        public void setTimestampFieldName(String timestampFieldName) {
+            this.timestampFieldName = timestampFieldName;
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            this.name = in.readString();
+            this.timestampFieldName = in.readString();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(name);
+            out.writeString(timestampFieldName);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return name.equals(request.name) &&
+                timestampFieldName.equals(request.timestampFieldName);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(name, timestampFieldName);
+        }
+    }
+
+    public static class TransportAction extends TransportMasterNodeAction<Request, AcknowledgedResponse> {
+
+        @Inject
+        public TransportAction(TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
+                               ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver) {
+            super(NAME, transportService, clusterService, threadPool, actionFilters, Request::new, indexNameExpressionResolver);
+        }
+
+        @Override
+        protected String executor() {
+            return ThreadPool.Names.SAME;
+        }
+
+        @Override
+        protected AcknowledgedResponse read(StreamInput in) throws IOException {
+            return new AcknowledgedResponse(in);
+        }
+
+        @Override
+        protected void masterOperation(Task task, Request request, ClusterState state,
+                                       ActionListener<AcknowledgedResponse> listener) throws Exception {
+            listener.onResponse(new AcknowledgedResponse(true));
+        }
+
+        @Override
+        protected ClusterBlockException checkBlock(Request request, ClusterState state) {
+            return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+        }
+    }
+
+}