Add auto create action

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

@@ -28,6 +28,10 @@
 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.create.AutoCreateAction;
+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;
@@ -584,6 +588,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(ClearScrollAction.INSTANCE, TransportClearScrollAction.class);
         actions.register(RecoveryAction.INSTANCE, TransportRecoveryAction.class);
         actions.register(NodesReloadSecureSettingsAction.INSTANCE, TransportNodesReloadSecureSettingsAction.class);
+        actions.register(AutoCreateAction.INSTANCE, AutoCreateAction.TransportAction.class);
 
         //Indexed scripts
         actions.register(PutStoredScriptAction.INSTANCE, TransportPutStoredScriptAction.class);

server/src/main/java/org/elasticsearch/action/admin/indices/create/AutoCreateAction.java

@@ -0,0 +1,87 @@
+/*
+ * 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.create;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.ActionFilters;
+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.metadata.MetadataCreateIndexService;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+
+/**
+ * Api that auto creates an index that originate from requests that write into an index that doesn't yet exist.
+ */
+public final class AutoCreateAction extends ActionType<CreateIndexResponse> {
+
+    public static final AutoCreateAction INSTANCE = new AutoCreateAction();
+    public static final String NAME = "indices:admin/auto_create";
+
+    private AutoCreateAction() {
+        super(NAME, CreateIndexResponse::new);
+    }
+
+    public static final class TransportAction extends TransportMasterNodeAction<CreateIndexRequest, CreateIndexResponse> {
+
+        private final MetadataCreateIndexService createIndexService;
+
+        @Inject
+        public TransportAction(TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
+                               ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
+                               MetadataCreateIndexService createIndexService) {
+            super(NAME, transportService, clusterService, threadPool, actionFilters, CreateIndexRequest::new, indexNameExpressionResolver);
+            this.createIndexService = createIndexService;
+        }
+
+        @Override
+        protected String executor() {
+            return ThreadPool.Names.SAME;
+        }
+
+        @Override
+        protected CreateIndexResponse read(StreamInput in) throws IOException {
+            return new CreateIndexResponse(in);
+        }
+
+        @Override
+        protected void masterOperation(Task task,
+                                       CreateIndexRequest request,
+                                       ClusterState state,
+                                       ActionListener<CreateIndexResponse> listener) {
+            TransportCreateIndexAction.innerCreateIndex(request, listener, indexNameExpressionResolver, createIndexService);
+        }
+
+        @Override
+        protected ClusterBlockException checkBlock(CreateIndexRequest request, ClusterState state) {
+            return state.blocks().indexBlockedException(ClusterBlockLevel.METADATA_WRITE, request.index());
+        }
+    }
+
+}

server/src/main/java/org/elasticsearch/action/admin/indices/create/TransportCreateIndexAction.java

@@ -71,14 +71,20 @@ protected ClusterBlockException checkBlock(CreateIndexRequest request, ClusterSt
     @Override
     protected void masterOperation(Task task, final CreateIndexRequest request, final ClusterState state,
                                    final ActionListener<CreateIndexResponse> listener) {
-        String cause = request.cause();
-        if (cause.length() == 0) {
-            cause = "api";
+        if (request.cause().length() == 0) {
+            request.cause("api");
         }
 
+        innerCreateIndex(request, listener, indexNameExpressionResolver, createIndexService);
+    }
+
+    static void innerCreateIndex(CreateIndexRequest request,
+                                 ActionListener<CreateIndexResponse> listener,
+                                 IndexNameExpressionResolver indexNameExpressionResolver,
+                                 MetadataCreateIndexService createIndexService) {
         final String indexName = indexNameExpressionResolver.resolveDateMathExpression(request.index());
         final CreateIndexClusterStateUpdateRequest updateRequest =
-            new CreateIndexClusterStateUpdateRequest(cause, indexName, request.index())
+            new CreateIndexClusterStateUpdateRequest(request.cause(), indexName, request.index())
                 .ackTimeout(request.timeout()).masterNodeTimeout(request.masterNodeTimeout())
                 .settings(request.settings()).mappings(request.mappings())
                 .aliases(request.aliases())

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.RoutingMissingException;
+import org.elasticsearch.action.admin.indices.create.AutoCreateAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.index.IndexRequest;
@@ -237,7 +238,7 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk
             } else {
                 final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size());
                 for (String index : autoCreateIndices) {
-                    createIndex(index, bulkRequest.preferV2Templates(), bulkRequest.timeout(), new ActionListener<>() {
+                    createIndex(index, bulkRequest.preferV2Templates(), bulkRequest.timeout(), minNodeVersion, new ActionListener<>() {
                         @Override
                         public void onResponse(CreateIndexResponse result) {
                             if (counter.decrementAndGet() == 0) {
@@ -381,13 +382,21 @@ boolean shouldAutoCreate(String index, ClusterState state) {
         return autoCreateIndex.shouldAutoCreate(index, state);
     }
 
-    void createIndex(String index, Boolean preferV2Templates, TimeValue timeout, ActionListener<CreateIndexResponse> listener) {
+    void createIndex(String index,
+                     Boolean preferV2Templates,
+                     TimeValue timeout,
+                     Version minNodeVersion,
+                     ActionListener<CreateIndexResponse> listener) {
         CreateIndexRequest createIndexRequest = new CreateIndexRequest();
         createIndexRequest.index(index);
         createIndexRequest.cause("auto(bulk api)");
         createIndexRequest.masterNodeTimeout(timeout);
         createIndexRequest.preferV2Templates(preferV2Templates);
-        client.admin().indices().create(createIndexRequest, listener);
+        if (minNodeVersion.onOrAfter(Version.V_8_0_0)) {
+            client.execute(AutoCreateAction.INSTANCE, createIndexRequest, listener);
+        } else {
+            client.admin().indices().create(createIndexRequest, listener);
+        }
     }
 
     private boolean setResponseFailureIfIndexMatches(AtomicArray<BulkItemResponse> responses, int idx, DocWriteRequest<?> request,

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIndicesThatCannotBeCreatedTests.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
@@ -111,7 +112,7 @@ private void indicesThatCannotBeCreatedTestCase(Set<String> expected,
         when(clusterService.state()).thenReturn(state);
         DiscoveryNodes discoveryNodes = mock(DiscoveryNodes.class);
         when(state.getNodes()).thenReturn(discoveryNodes);
-        when(discoveryNodes.getMinNodeVersion()).thenReturn(Version.CURRENT);
+        when(discoveryNodes.getMinNodeVersion()).thenReturn(VersionUtils.randomCompatibleVersion(random(), Version.CURRENT));
         DiscoveryNode localNode = mock(DiscoveryNode.class);
         when(clusterService.localNode()).thenReturn(localNode);
         when(localNode.isIngestNode()).thenReturn(randomBoolean());
@@ -138,7 +139,7 @@ boolean shouldAutoCreate(String index, ClusterState state) {
 
             @Override
             void createIndex(String index, Boolean preferV2Templates,
-                             TimeValue timeout, ActionListener<CreateIndexResponse> listener) {
+                             TimeValue timeout, Version minNodeVersion, ActionListener<CreateIndexResponse> listener) {
                 // If we try to create an index just immediately assume it worked
                 listener.onResponse(new CreateIndexResponse(true, true, index) {});
             }

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java

@@ -55,6 +55,7 @@
 import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
@@ -158,7 +159,8 @@ void executeBulk(Task task, final BulkRequest bulkRequest, final long startTimeN
 
         @Override
         void createIndex(String index, Boolean preferV2Templates,
-                         TimeValue timeout, ActionListener<CreateIndexResponse> listener) {
+                         TimeValue timeout, Version minNodeVersion,
+                         ActionListener<CreateIndexResponse> listener) {
             indexCreated = true;
             listener.onResponse(null);
         }
@@ -193,7 +195,7 @@ public void setupAction() {
         ImmutableOpenMap<String, DiscoveryNode> ingestNodes = ImmutableOpenMap.<String, DiscoveryNode>builder(2)
             .fPut("node1", remoteNode1).fPut("node2", remoteNode2).build();
         when(nodes.getIngestNodes()).thenReturn(ingestNodes);
-        when(nodes.getMinNodeVersion()).thenReturn(Version.CURRENT);
+        when(nodes.getMinNodeVersion()).thenReturn(VersionUtils.randomCompatibleVersion(random(), Version.CURRENT));
         ClusterState state = mock(ClusterState.class);
         when(state.getNodes()).thenReturn(nodes);
         Metadata metadata = Metadata.builder().indices(ImmutableOpenMap.<String, IndexMetadata>builder()

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTests.java

@@ -33,6 +33,8 @@
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -41,6 +43,7 @@
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.test.transport.CapturingTransport;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -82,7 +85,7 @@ protected boolean needToCheck() {
 
         @Override
         void createIndex(String index, Boolean preferV2Templates,
-                         TimeValue timeout, ActionListener<CreateIndexResponse> listener) {
+                         TimeValue timeout, Version minNodeVersion, ActionListener<CreateIndexResponse> listener) {
             indexCreated = true;
             listener.onResponse(null);
         }
@@ -92,7 +95,9 @@ void createIndex(String index, Boolean preferV2Templates,
     public void setUp() throws Exception {
         super.setUp();
         threadPool = new TestThreadPool(getClass().getName());
-        clusterService = createClusterService(threadPool);
+        DiscoveryNode discoveryNode = new DiscoveryNode("node", ESTestCase.buildNewFakeTransportAddress(), Collections.emptyMap(),
+            DiscoveryNodeRole.BUILT_IN_ROLES, VersionUtils.randomCompatibleVersion(random(), Version.CURRENT));
+        clusterService = createClusterService(threadPool, discoveryNode);
         CapturingTransport capturingTransport = new CapturingTransport();
         transportService = capturingTransport.createTransportService(clusterService.getSettings(), threadPool,
             TransportService.NOOP_TRANSPORT_INTERCEPTOR,

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTookTests.java

@@ -21,6 +21,7 @@
 package org.elasticsearch.action.bulk;
 
 import org.apache.lucene.util.Constants;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
@@ -32,6 +33,8 @@
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
@@ -40,6 +43,7 @@
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.test.transport.CapturingTransport;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -81,7 +85,9 @@ public static void afterClass() {
     @Before
     public void setUp() throws Exception {
         super.setUp();
-        clusterService = createClusterService(threadPool);
+        DiscoveryNode discoveryNode = new DiscoveryNode("node", ESTestCase.buildNewFakeTransportAddress(), Collections.emptyMap(),
+            DiscoveryNodeRole.BUILT_IN_ROLES, VersionUtils.randomCompatibleVersion(random(), Version.CURRENT));
+        clusterService = createClusterService(threadPool, discoveryNode);
     }
 
     @After

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

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsAction;
 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.delete.DeleteIndexAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexAction;
@@ -57,7 +58,7 @@ 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);
+    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,

x-pack/plugin/ml/qa/ml-with-security/roles.yml

@@ -9,7 +9,7 @@ minimal:
     # non-ML indices
     - names: [ 'airline-data', 'index-*', 'unavailable-data', 'utopia' ]
       privileges:
-        - indices:admin/create
+        - create_index
         - indices:admin/refresh
         - read
         - index

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/WriteActionsTests.java

@@ -36,7 +36,8 @@ protected String configRoles() {
                 "  cluster: [ ALL ]\n" +
                 "  indices:\n" +
                 "    - names: 'missing'\n" +
-                "      privileges: [ 'indices:admin/create', 'indices:admin/delete' ]\n" +
+                "      privileges: [ 'indices:admin/create', 'indices:admin/auto_create', " +
+                                    "'indices:admin/delete' ]\n" +
                 "    - names: ['/index.*/']\n" +
                 "      privileges: [ manage ]\n" +
                 "    - names: ['/test.*/']\n" +