Backport: auto create data streams using index templates v2 (#56596)

Backport: #55377

This commit adds the ability to auto create data streams using index templates v2.
Index templates (v2) now have a data_steam field that includes a timestamp field,
if provided and index name matches with that template then a data stream
(plus first backing index) is auto created.

Relates to #53100

Author: martijnvg

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

@@ -2039,7 +2039,8 @@ public void testIndexTemplates() throws Exception {
         AliasMetadata alias = AliasMetadata.builder("alias").writeIndex(true).build();
         Template template = new Template(settings, mappings, Collections.singletonMap("alias", alias));
         List<String> pattern = Collections.singletonList("pattern");
-        IndexTemplateV2 indexTemplate = new IndexTemplateV2(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>());
+        IndexTemplateV2 indexTemplate =
+            new IndexTemplateV2(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>(), null);
         PutIndexTemplateV2Request putIndexTemplateV2Request =
             new PutIndexTemplateV2Request().name(templateName).create(true).indexTemplate(indexTemplate);
 
@@ -2085,7 +2086,8 @@ public void testSimulateIndexTemplate() throws Exception {
         AliasMetadata alias = AliasMetadata.builder("alias").writeIndex(true).build();
         Template template = new Template(settings, mappings, org.elasticsearch.common.collect.Map.of("alias", alias));
         List<String> pattern = org.elasticsearch.common.collect.List.of("pattern");
-        IndexTemplateV2 indexTemplate = new IndexTemplateV2(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>());
+        IndexTemplateV2 indexTemplate =
+            new IndexTemplateV2(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>(), null);
         PutIndexTemplateV2Request putIndexTemplateV2Request =
             new PutIndexTemplateV2Request().name(templateName).create(true).indexTemplate(indexTemplate);
 
@@ -2097,7 +2099,7 @@ public void testSimulateIndexTemplate() throws Exception {
         AliasMetadata simulationAlias = AliasMetadata.builder("simulation-alias").writeIndex(true).build();
         IndexTemplateV2 simulationTemplate = new IndexTemplateV2(pattern, new Template(null, null,
             org.elasticsearch.common.collect.Map.of("simulation-alias", simulationAlias)), Collections.emptyList(), 2L, 1L,
-            new HashMap<>());
+            new HashMap<>(), null);
         PutIndexTemplateV2Request newIndexTemplateReq =
             new PutIndexTemplateV2Request().name("used-for-simulation").create(true).indexTemplate(indexTemplate);
         newIndexTemplateReq.indexTemplate(simulationTemplate);

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

@@ -75,6 +75,7 @@ private static IndexTemplateV2 randomIndexTemplate() {
         List<String> patterns = Arrays.asList(generateRandomStringArray(10, 10, false, false));
         List<String> composedOf = null;
         Map<String, Object> meta = null;
+        IndexTemplateV2.DataStreamTemplate dataStreamTemplate = null;
         if (randomBoolean()) {
             composedOf = Arrays.asList(generateRandomStringArray(10, 10, false, false));
         }
@@ -84,6 +85,9 @@ private static IndexTemplateV2 randomIndexTemplate() {
 
         Long priority = randomBoolean() ? null : randomNonNegativeLong();
         Long version = randomBoolean() ? null : randomNonNegativeLong();
-        return new IndexTemplateV2(patterns, randomTemplate(), composedOf, priority, version, meta);
+        if (randomBoolean()) {
+            dataStreamTemplate = new IndexTemplateV2.DataStreamTemplate(randomAlphaOfLength(8));
+        }
+        return new IndexTemplateV2(patterns, randomTemplate(), composedOf, priority, version, meta, dataStreamTemplate);
     }
 }

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

@@ -1,8 +1,8 @@
 ---
 "Create data stream":
   - skip:
-      version: " - 7.7.99"
-      reason: available only in 7.8+
+      version: " - 7.8.99"
+      reason: "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:
@@ -62,8 +62,8 @@
 ---
 "Create data stream with invalid name":
   - skip:
-      version: " - 7.7.99"
-      reason: available only in 7.8+
+      version: " - 7.8.99"
+      reason: "data streams only supported in 7.9+"
 
   - do:
       catch: bad_request
@@ -78,8 +78,8 @@
 ---
 "Get data stream":
   - skip:
-      version: " - 7.7.99"
-      reason: available only in 7.8+
+      version: " - 7.8.99"
+      reason: "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:
@@ -147,8 +147,8 @@
 ---
 "Delete data stream with backing indices":
   - skip:
-      version: " - 7.7.99"
-      reason: available only in 7.8+
+      version: " - 7.8.99"
+      reason: data streams only supported in 7.9+
 
   - do:
       indices.create_data_stream:

rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml

@@ -1,8 +1,8 @@
 ---
 "Test apis that do not supported data streams":
   - skip:
-      version: " - 7.7.99"
-      reason: available only in 7.8+
+      version: " - 7.8.99"
+      reason: "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:

rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/30_auto_create_data_stream.yml

@@ -0,0 +1,51 @@
+---
+"Put index template":
+  - skip:
+      version: " - 7.8.99"
+      reason: "data streams supported from 7.9"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [test] has index patterns [test-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [test] will take precedence during new index creation"
+      indices.put_index_template:
+        name: generic_logs_template
+        body:
+          index_patterns: logs-*
+          data_stream:
+            timestamp_field: timestamp
+          template:
+            settings:
+              number_of_shards:   1
+              number_of_replicas: 0
+            mappings:
+              properties:
+                timestamp:
+                  type: date
+
+  - do:
+      index:
+        index:  logs-foobar
+        refresh: true
+        body:   { foo: bar }
+
+  - do:
+      search:
+        index: logs-foobar
+        body: { query: { match_all: {} } }
+  - length:   { hits.hits: 1  }
+  - match: { hits.hits.0._index: logs-foobar-000001 }
+  - match: { hits.hits.0._source.foo: 'bar' }
+
+  - do:
+      indices.get_data_streams:
+        name: logs-foobar
+  - match: { 0.name: logs-foobar }
+  - match: { 0.timestamp_field: 'timestamp' }
+  - length: { 0.indices: 1 }
+  - match: { 0.indices.0.index_name: 'logs-foobar-000001' }
+
+  - do:
+      indices.delete_data_stream:
+        name: logs-foobar
+  - is_true: acknowledged

rest-api-spec/src/main/resources/rest-api-spec/test/indices.delete/20_backing_indices.yml

@@ -1,8 +1,8 @@
 ---
 "Delete backing index on data stream":
   - skip:
-      version: " - 7.99.99"
-      reason:  "enable in 7.8+ after backporting"
+      version: " - 7.8.99"
+      reason:  "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:
@@ -55,8 +55,8 @@
 ---
 "Attempt to delete write index on data stream is rejected":
   - skip:
-      version: " - 7.99.99"
-      reason:  "enable in 7.8+ after backporting"
+      version: " - 7.8.99"
+      reason:  "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:

rest-api-spec/src/main/resources/rest-api-spec/test/indices.get/20_backing_indices.yml

@@ -1,8 +1,8 @@
 ---
 "Get backing indices for data stream":
   - skip:
-      version: " - 7.99.99"
-      reason:  "enable in 7.8+ after backporting"
+      version: " - 7.8.99"
+      reason:  "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:

rest-api-spec/src/main/resources/rest-api-spec/test/indices.rollover/50_data_streams.yml

@@ -1,8 +1,8 @@
 ---
 "Roll over a data stream":
   - skip:
-      version: " - 7.99.99"
-      reason:  "enable in 7.8+ after backporting"
+      version: " - 7.8.99"
+      reason:  "data streams only supported in 7.9+"
 
   - do:
       indices.create_data_stream:

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

@@ -24,14 +24,25 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.admin.indices.alias.Alias;
+import org.elasticsearch.action.admin.indices.datastream.GetDataStreamsAction;
+import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
+import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
 import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
+import org.elasticsearch.action.admin.indices.template.delete.DeleteIndexTemplateV2Action;
+import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequest;
+import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateV2Action;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.replication.ReplicationRequest;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexTemplateV2;
+import org.elasticsearch.cluster.metadata.Template;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.ingest.IngestTestPlugin;
@@ -44,20 +55,25 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.elasticsearch.action.DocWriteRequest.OpType.CREATE;
 import static org.elasticsearch.action.DocWriteResponse.Result.CREATED;
 import static org.elasticsearch.action.DocWriteResponse.Result.UPDATED;
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.test.StreamsUtils.copyToStringFromClasspath;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasItemInArray;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.oneOf;
 
@@ -202,4 +218,87 @@ public void testDeleteIndexWhileIndexing() throws Exception {
             assertFalse(thread.isAlive());
         }
     }
+
+    public void testMixedAutoCreate() {
+        Settings settings = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0).build();
+
+        PutIndexTemplateV2Action.Request createTemplateRequest = new PutIndexTemplateV2Action.Request("logs-foo");
+        createTemplateRequest.indexTemplate(
+            new IndexTemplateV2(
+                Collections.singletonList("logs-foo*"),
+                new Template(settings, null, null),
+                null, null, null, null,
+                new IndexTemplateV2.DataStreamTemplate("@timestamp"))
+        );
+        client().execute(PutIndexTemplateV2Action.INSTANCE, createTemplateRequest).actionGet();
+
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo").opType(CREATE).source("{}", XContentType.JSON));
+        BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
+        assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
+
+        bulkRequest = new BulkRequest();
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz2").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo2").opType(CREATE).source("{}", XContentType.JSON));
+        bulkResponse = client().bulk(bulkRequest).actionGet();
+        assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
+
+        bulkRequest = new BulkRequest();
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz2").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz3").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo2").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo3").opType(CREATE).source("{}", XContentType.JSON));
+        bulkResponse = client().bulk(bulkRequest).actionGet();
+        assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
+
+        GetDataStreamsAction.Request getDataStreamRequest = new GetDataStreamsAction.Request("*");
+        GetDataStreamsAction.Response getDataStreamsResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
+        assertThat(getDataStreamsResponse.getDataStreams(), hasSize(4));
+        getDataStreamsResponse.getDataStreams().sort(Comparator.comparing(DataStream::getName));
+        assertThat(getDataStreamsResponse.getDataStreams().get(0).getName(), equalTo("logs-foobar"));
+        assertThat(getDataStreamsResponse.getDataStreams().get(1).getName(), equalTo("logs-foobaz"));
+        assertThat(getDataStreamsResponse.getDataStreams().get(2).getName(), equalTo("logs-foobaz2"));
+        assertThat(getDataStreamsResponse.getDataStreams().get(3).getName(), equalTo("logs-foobaz3"));
+
+        GetIndexResponse getIndexResponse = client().admin().indices().getIndex(new GetIndexRequest().indices("logs-bar*")).actionGet();
+        assertThat(getIndexResponse.getIndices(), arrayWithSize(4));
+        assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-barbaz"));
+        assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-barfoo"));
+        assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-barfoo2"));
+        assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-barfoo3"));
+
+        DeleteIndexTemplateV2Action.Request deleteTemplateRequest = new DeleteIndexTemplateV2Action.Request("*");
+        client().execute(DeleteIndexTemplateV2Action.INSTANCE, deleteTemplateRequest).actionGet();
+    }
+
+    public void testAutoCreateV1TemplateNoDataStream() {
+        Settings settings = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0).build();
+
+        PutIndexTemplateRequest v1Request = new PutIndexTemplateRequest("logs-foo");
+        v1Request.patterns(Collections.singletonList("logs-foo*"));
+        v1Request.settings(settings);
+        v1Request.order(Integer.MAX_VALUE); // in order to avoid number_of_replicas being overwritten by random_template
+        client().admin().indices().putTemplate(v1Request).actionGet();
+
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
+        BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
+        assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
+
+        GetDataStreamsAction.Request getDataStreamRequest = new GetDataStreamsAction.Request("*");
+        GetDataStreamsAction.Response getDataStreamsResponse = client().admin().indices().getDataStreams(getDataStreamRequest).actionGet();
+        assertThat(getDataStreamsResponse.getDataStreams(), hasSize(0));
+
+        GetIndexResponse getIndexResponse = client().admin().indices().getIndex(new GetIndexRequest().indices("logs-foobar")).actionGet();
+        assertThat(getIndexResponse.getIndices(), arrayWithSize(1));
+        assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-foobar"));
+        assertThat(getIndexResponse.getSettings().get("logs-foobar").get(IndexMetadata.SETTING_NUMBER_OF_REPLICAS), equalTo("0"));
+    }
 }