Add require_data_stream feature

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java

@@ -54,6 +54,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         String defaultRouting = request.param("routing");
         String defaultPipeline = request.param("pipeline");
         Boolean defaultRequireAlias = request.paramAsBoolean("require_alias", null);
+        Boolean defaultRequireDataStream = request.paramAsBoolean("require_data_stream", null);
         Boolean defaultListExecutedPipelines = request.paramAsBoolean("list_executed_pipelines", null);
 
         String waitForActiveShards = request.param("wait_for_active_shards");
@@ -69,6 +70,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             null,
             defaultPipeline,
             defaultRequireAlias,
+            defaultRequireDataStream,
             defaultListExecutedPipelines,
             true,
             request.getXContentType(),

client/rest/src/main/java/org/elasticsearch/client/Request.java

@@ -67,7 +67,7 @@ public String getEndpoint() {
     /**
      * Add a query string parameter.
      * @param name the name of the url parameter. Must not be null.
-     * @param value the value of the url url parameter. If {@code null} then
+     * @param value the value of the url parameter. If {@code null} then
      *      the parameter is sent as {@code name} rather than {@code name=value}
      * @throws IllegalArgumentException if a parameter with that name has
      *      already been set

docs/changelog/101872.yaml

@@ -0,0 +1,6 @@
+pr: 101872
+summary: "Add `require_data_stream` parameter to indexing requests to enforce indexing operations target a data stream"
+area: Data streams
+type: feature
+issues:
+ - 97032

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/AutoCreateDataStreamIT.java

@@ -7,6 +7,7 @@
  */
 package org.elasticsearch.datastreams;
 
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
@@ -23,6 +24,7 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.hamcrest.Matchers.containsString;
 
+@SuppressWarnings("resource")
 public class AutoCreateDataStreamIT extends DisabledSecurityDataStreamTestCase {
 
     /**
@@ -31,7 +33,7 @@ public class AutoCreateDataStreamIT extends DisabledSecurityDataStreamTestCase {
      */
     public void testCanAutoCreateDataStreamWhenAutoCreateIndexDisabled() throws IOException {
         configureAutoCreateIndex(false);
-        createTemplateWithAllowAutoCreate(null);
+        createTemplate(null, true);
         assertOK(this.indexDocument());
     }
 
@@ -40,7 +42,7 @@ public void testCanAutoCreateDataStreamWhenAutoCreateIndexDisabled() throws IOEx
      * and that template has <code>allow_auto_create</code> set to <code>true</code>.
      */
     public void testCanAutoCreateDataStreamWhenExplicitlyAllowedByTemplate() throws IOException {
-        createTemplateWithAllowAutoCreate(true);
+        createTemplate(true, true);
 
         // Attempt to add a document to a non-existing index. Auto-creating the index should succeed because the index name
         // matches the template pattern
@@ -52,15 +54,60 @@ public void testCanAutoCreateDataStreamWhenExplicitlyAllowedByTemplate() throws
      * <code>allow_auto_create</code> explicitly to <code>false</code>.
      */
     public void testCannotAutoCreateDataStreamWhenDisallowedByTemplate() throws IOException {
-        createTemplateWithAllowAutoCreate(false);
+        createTemplate(false, true);
 
-        // Attempt to add a document to a non-existing index. Auto-creating the index should succeed because the index name
-        // matches the template pattern
+        // Auto-creating the index should fail when the template disallows that
         final ResponseException responseException = expectThrows(ResponseException.class, this::indexDocument);
 
         assertThat(
             Streams.copyToString(new InputStreamReader(responseException.getResponse().getEntity().getContent(), UTF_8)),
-            containsString("no such index [composable template [recipe*] forbids index auto creation]")
+            containsString("no such index [recipe_kr] and composable template [recipe*] forbids index auto creation")
+        );
+    }
+
+    /**
+     * Check that if <code>require_data_stream</code> is set to <code>true</code>, automatically creating an index is allowed only
+     * if its name matches an index template AND it contains a data-stream template
+     */
+    public void testCannotAutoCreateDataStreamWhenNoDataStreamTemplateMatch() throws IOException {
+        createTemplate(true, true);
+
+        final Request request = prepareIndexRequest("ingredients_kr");
+        request.addParameter(DocWriteRequest.REQUIRE_DATA_STREAM, Boolean.TRUE.toString());
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should fail because the index name doesn't
+        // match the template pattern and the request requires a data stream template
+        final ResponseException responseException = expectThrows(ResponseException.class, () -> client().performRequest(request));
+
+        assertThat(
+            Streams.copyToString(new InputStreamReader(responseException.getResponse().getEntity().getContent(), UTF_8)),
+            containsString(
+                "no such index [ingredients_kr] and the index creation request requires a data stream, "
+                    + "but no matching index template with data stream template was found for it"
+            )
+        );
+    }
+
+    /**
+     * Check that if <code>require_data_stream</code> is set to <code>true</code>, automatically creating an index is allowed only
+     * if its name matches an index template AND it contains a data-stream template
+     */
+    public void testCannotAutoCreateDataStreamWhenMatchingTemplateIsNotDataStream() throws IOException {
+        createTemplate(true, false);
+
+        final Request request = prepareIndexRequest("recipe_kr");
+        request.addParameter(DocWriteRequest.REQUIRE_DATA_STREAM, Boolean.TRUE.toString());
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should fail because the index name doesn't
+        // match the template pattern and the request requires a data stream template
+        final ResponseException responseException = expectThrows(ResponseException.class, () -> client().performRequest(request));
+
+        assertThat(
+            Streams.copyToString(new InputStreamReader(responseException.getResponse().getEntity().getContent(), UTF_8)),
+            containsString(
+                "no such index [recipe_kr] and the index creation request requires a data stream, "
+                    + "but no matching index template with data stream template was found for it"
+            )
         );
     }
 
@@ -78,16 +125,18 @@ private void configureAutoCreateIndex(boolean value) throws IOException {
         assertOK(settingsResponse);
     }
 
-    private void createTemplateWithAllowAutoCreate(Boolean allowAutoCreate) throws IOException {
+    private void createTemplate(Boolean allowAutoCreate, boolean addDataStreamTemplate) throws IOException {
         XContentBuilder b = JsonXContent.contentBuilder();
         b.startObject();
         {
             b.array("index_patterns", "recipe*");
             if (allowAutoCreate != null) {
                 b.field("allow_auto_create", allowAutoCreate);
             }
-            b.startObject("data_stream");
-            b.endObject();
+            if (addDataStreamTemplate) {
+                b.startObject("data_stream");
+                b.endObject();
+            }
         }
         b.endObject();
 
@@ -98,8 +147,13 @@ private void createTemplateWithAllowAutoCreate(Boolean allowAutoCreate) throws I
     }
 
     private Response indexDocument() throws IOException {
-        final Request indexDocumentRequest = new Request("POST", "recipe_kr/_doc");
-        indexDocumentRequest.setJsonEntity("{ \"@timestamp\": \"" + Instant.now() + "\", \"name\": \"Kimchi\" }");
+        final Request indexDocumentRequest = prepareIndexRequest("recipe_kr");
         return client().performRequest(indexDocumentRequest);
     }
+
+    private Request prepareIndexRequest(String indexName) {
+        final Request indexDocumentRequest = new Request("POST", indexName + "/_doc");
+        indexDocumentRequest.setJsonEntity("{ \"@timestamp\": \"" + Instant.now() + "\", \"name\": \"Kimchi\" }");
+        return indexDocumentRequest;
+    }
 }

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/190_require_data_stream.yml

@@ -0,0 +1,138 @@
+---
+"Testing require_data_stream in index creation":
+  - skip:
+      version: " - 8.12.99"
+      reason: "require_data_stream was introduced in 8.13.0"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [ds-template] has index patterns [ds-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [ds-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: ds-template
+        body:
+          index_patterns: ds-*
+          template:
+            settings:
+              number_of_shards:   1
+              number_of_replicas: 0
+            mappings:
+              properties:
+                field:
+                  type: keyword
+          data_stream: {}
+          allow_auto_create: true
+
+  - do:
+      index:
+        index:  ds-test
+        require_data_stream: true
+        body:
+          '@timestamp': '2022-12-12'
+          foo: bar
+
+  - do:
+      catch: /no matching index template with data stream template was found for it/
+      index:
+        index:  index-test
+        require_data_stream: true
+        body:
+          '@timestamp': '2022-12-12'
+          foo: bar
+
+  - do:
+      index:
+        index:  other-index
+        require_data_stream: false
+        body:
+          '@timestamp': '2022-12-12'
+          foo: bar
+
+  - do:
+      catch: /is not a data stream/
+      index:
+        index:  other-index
+        require_data_stream: true
+        body:
+          '@timestamp': '2022-12-12'
+          foo: bar
+
+---
+"Testing require_data_stream in bulk requests":
+  - skip:
+      version: " - 8.12.99"
+      reason: "require_data_stream was introduced in 8.13.0"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [ds-template] has index patterns [ds-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [ds-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: ds-template
+        body:
+          index_patterns: ds-*
+          template:
+            settings:
+              number_of_shards:   1
+              number_of_replicas: 0
+            mappings:
+              properties:
+                field:
+                  type: keyword
+          data_stream: {}
+          allow_auto_create: true
+
+  - do:
+      bulk:
+        refresh: true
+        require_data_stream: true
+        body:
+          - index:
+              _index: new_index_not_created
+          - f: 1
+          - index:
+              _index: new_index_created
+              require_data_stream: false
+          - f: 2
+          - index:
+              _index: ds-other
+              op_type: create
+          - "@timestamp": "2024-01-01"
+  - match: { errors: true }
+  - match: { items.0.index.status: 404 }
+  - match: { items.0.index.error.type: index_not_found_exception }
+  - match: { items.0.index.error.reason: "no such index [new_index_not_created] and the index creation request requires a data stream, but no matching index template with data stream template was found for it" }
+  - match: { items.1.index.result: created }
+  - match: { items.2.create.result: created }
+
+  - do:
+      allowed_warnings:
+        - "index template [other-template] has index patterns [ds-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [other-template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: other-template
+        body:
+          index_patterns: other-*
+          template:
+            settings:
+              number_of_shards:   1
+              number_of_replicas: 0
+            mappings:
+              properties:
+                field:
+                  type: keyword
+          allow_auto_create: true
+
+  - do:
+      bulk:
+        refresh: true
+        require_data_stream: false
+        body:
+          - index:
+              _index: other-myindex
+              require_data_stream: true
+              op_type: create
+          - "@timestamp": "2024-01-01"
+  - match: { errors: true }
+  - match: { items.0.create.status: 404 }
+  - match: { items.0.create.error.type: index_not_found_exception }
+  - match: { items.0.create.error.reason: "no such index [other-myindex] and the index creation request requires a data stream, but no matching index template with data stream template was found for it" }

rest-api-spec/src/main/resources/rest-api-spec/api/bulk.json

@@ -80,6 +80,10 @@
         "type": "boolean",
         "description": "Sets require_alias for all incoming documents. Defaults to unset (false)"
       },
+      "require_data_stream": {
+        "type": "boolean",
+        "description": "When true, requires the destination to be a data stream (existing or to-be-created). Default is false"
+      },
       "list_executed_pipelines": {
         "type": "boolean",
         "description": "Sets list_executed_pipelines for all incoming documents. Defaults to unset (false)"

rest-api-spec/src/main/resources/rest-api-spec/api/index.json

@@ -101,6 +101,10 @@
       "require_alias": {
         "type": "boolean",
         "description": "When true, requires destination to be an alias. Default is false"
+      },
+      "require_data_stream": {
+        "type": "boolean",
+        "description": "When true, requires the destination to be a data stream (existing or to-be-created). Default is false"
       }
     },
     "body":{

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -188,6 +188,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion PEERFINDER_REPORTS_PEERS_MASTERS = def(8_575_00_0);
     public static final TransportVersion ESQL_MULTI_CLUSTERS_ENRICH = def(8_576_00_0);
     public static final TransportVersion NESTED_KNN_MORE_INNER_HITS = def(8_577_00_0);
+    public static final TransportVersion REQUIRE_DATA_STREAM_ADDED = def(8_578_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -41,6 +41,9 @@ public interface DocWriteRequest<T> extends IndicesRequest, Accountable {
     // Flag set for disallowing index auto creation for an individual write request.
     String REQUIRE_ALIAS = "require_alias";
 
+    // Flag set for disallowing index auto creation if no matching data-stream index template is available.
+    String REQUIRE_DATA_STREAM = "require_data_stream";
+
     // Flag indicating that the list of executed pipelines should be returned in the request
     String LIST_EXECUTED_PIPELINES = "list_executed_pipelines";
 
@@ -149,6 +152,12 @@ public interface DocWriteRequest<T> extends IndicesRequest, Accountable {
      */
     boolean isRequireAlias();
 
+    /**
+     * Should this request override specifically require the destination to be a data stream?
+     * @return boolean flag, when true specifically requires a data stream
+     */
+    boolean isRequireDataStream();
+
     /**
      * Finalize the request before executing or routing it.
      */

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

@@ -244,7 +244,8 @@ ClusterState execute(
                     // This expression only evaluates to true when the argument is non-null and false
                     if (isSystemDataStream == false && Boolean.FALSE.equals(template.getAllowAutoCreate())) {
                         throw new IndexNotFoundException(
-                            "composable template " + template.indexPatterns() + " forbids index auto creation"
+                            "composable template " + template.indexPatterns() + " forbids index auto creation",
+                            request.index()
                         );
                     }
 
@@ -272,6 +273,13 @@ ClusterState execute(
                     successfulRequests.put(request, indexNames);
                     return clusterState;
                 } else {
+                    if (request.isRequireDataStream()) {
+                        throw new IndexNotFoundException(
+                            "the index creation request requires a data stream, "
+                                + "but no matching index template with data stream template was found for it",
+                            request.index()
+                        );
+                    }
                     final var indexName = IndexNameExpressionResolver.resolveDateMathExpression(request.index());
                     if (isSystemIndex) {
                         if (indexName.equals(request.index()) == false) {