Automatically setup time series start and end settings when creating a data stream (#81532)

If template has a `data_stream` snippet and `index.mode` setting is set to `time_series` then
automatically set time series start and end time settings upon creating of a new data stream.
These settings will then be applied to the backing index that gets created as part of creating
the new data stream. The start and end time settings is based on the `index.look_ahead_time`
setting in the composable index template that matches.

Also tweaked the logic that creates composable templates to automatically set time series
start and end time settings if the template being stored has a `data_stream` snippet and
`index.mode` setting is set to `time_series`. Otherwise the time series index mode validation
fails with the fact that the start and end time settings are missing, whilst in reality these
settings will be added automatically by the create data stream api logic.

Relates #74660

server/src/main/java/org/elasticsearch/action/admin/indices/template/post/TransportSimulateIndexTemplateAction.java

@@ -31,6 +31,7 @@ import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.shard.IndexSettingProvider;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.tasks.Task;
@@ -61,6 +62,7 @@ public class TransportSimulateIndexTemplateAction extends TransportMasterNodeRea
     private final IndicesService indicesService;
     private final AliasValidator aliasValidator;
     private final SystemIndices systemIndices;
+    private final Set<IndexSettingProvider> indexSettingProviders;
 
     @Inject
     public TransportSimulateIndexTemplateAction(
@@ -72,7 +74,8 @@ public class TransportSimulateIndexTemplateAction extends TransportMasterNodeRea
         IndexNameExpressionResolver indexNameExpressionResolver,
         NamedXContentRegistry xContentRegistry,
         IndicesService indicesService,
-        SystemIndices systemIndices
+        SystemIndices systemIndices,
+        MetadataCreateIndexService metadataCreateIndexService
     ) {
         super(
             SimulateIndexTemplateAction.NAME,
@@ -90,6 +93,7 @@ public class TransportSimulateIndexTemplateAction extends TransportMasterNodeRea
         this.indicesService = indicesService;
         this.aliasValidator = new AliasValidator();
         this.systemIndices = systemIndices;
+        this.indexSettingProviders = metadataCreateIndexService.getIndexSettingProviders();
     }
 
     @Override
@@ -136,7 +140,8 @@ public class TransportSimulateIndexTemplateAction extends TransportMasterNodeRea
             xContentRegistry,
             indicesService,
             aliasValidator,
-            systemIndices
+            systemIndices,
+            indexSettingProviders
         );
 
         final Map<String, List<String>> overlapping = new HashMap<>();
@@ -188,7 +193,8 @@ public class TransportSimulateIndexTemplateAction extends TransportMasterNodeRea
         final NamedXContentRegistry xContentRegistry,
         final IndicesService indicesService,
         final AliasValidator aliasValidator,
-        final SystemIndices systemIndices
+        final SystemIndices systemIndices,
+        Set<IndexSettingProvider> indexSettingProviders
     ) throws Exception {
         Settings settings = resolveSettings(simulatedState.metadata(), matchingTemplate);
 
@@ -197,14 +203,29 @@ public class TransportSimulateIndexTemplateAction extends TransportMasterNodeRea
             matchingTemplate
         );
 
+        ComposableIndexTemplate template = simulatedState.metadata().templatesV2().get(matchingTemplate);
         // create the index with dummy settings in the cluster state so we can parse and validate the aliases
-        Settings dummySettings = Settings.builder()
+        Settings.Builder dummySettings = Settings.builder()
             .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT)
-            .put(settings)
             .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
             .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
-            .put(IndexMetadata.SETTING_INDEX_UUID, UUIDs.randomBase64UUID())
-            .build();
+            .put(IndexMetadata.SETTING_INDEX_UUID, UUIDs.randomBase64UUID());
+
+        // First apply settings sourced from index settings providers
+        for (var provider : indexSettingProviders) {
+            dummySettings.put(
+                provider.getAdditionalIndexSettings(
+                    indexName,
+                    template.getDataStreamTemplate() != null ? indexName : null,
+                    true,
+                    System.currentTimeMillis(),
+                    settings
+                )
+            );
+        }
+        // Then apply settings resolved from templates:
+        dummySettings.put(settings);
+
         final IndexMetadata indexMetadata = IndexMetadata.builder(indexName).settings(dummySettings).build();
 
         final ClusterState tempClusterState = ClusterState.builder(simulatedState)

server/src/main/java/org/elasticsearch/action/admin/indices/template/post/TransportSimulateTemplateAction.java

@@ -17,11 +17,13 @@ import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.AliasValidator;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
 import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.index.shard.IndexSettingProvider;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.tasks.Task;
@@ -33,6 +35,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
 
 import static org.elasticsearch.cluster.metadata.MetadataIndexTemplateService.findConflictingV1Templates;
 import static org.elasticsearch.cluster.metadata.MetadataIndexTemplateService.findConflictingV2Templates;
@@ -48,8 +51,9 @@ public class TransportSimulateTemplateAction extends TransportMasterNodeReadActi
     private final MetadataIndexTemplateService indexTemplateService;
     private final NamedXContentRegistry xContentRegistry;
     private final IndicesService indicesService;
-    private AliasValidator aliasValidator;
+    private final AliasValidator aliasValidator;
     private final SystemIndices systemIndices;
+    private final Set<IndexSettingProvider> indexSettingProviders;
 
     @Inject
     public TransportSimulateTemplateAction(
@@ -61,7 +65,8 @@ public class TransportSimulateTemplateAction extends TransportMasterNodeReadActi
         IndexNameExpressionResolver indexNameExpressionResolver,
         NamedXContentRegistry xContentRegistry,
         IndicesService indicesService,
-        SystemIndices systemIndices
+        SystemIndices systemIndices,
+        MetadataCreateIndexService service
     ) {
         super(
             SimulateTemplateAction.NAME,
@@ -79,6 +84,7 @@ public class TransportSimulateTemplateAction extends TransportMasterNodeReadActi
         this.indicesService = indicesService;
         this.aliasValidator = new AliasValidator();
         this.systemIndices = systemIndices;
+        this.indexSettingProviders = service.getIndexSettingProviders();
     }
 
     @Override
@@ -157,7 +163,8 @@ public class TransportSimulateTemplateAction extends TransportMasterNodeReadActi
             xContentRegistry,
             indicesService,
             aliasValidator,
-            systemIndices
+            systemIndices,
+            indexSettingProviders
         );
         listener.onResponse(new SimulateIndexTemplateResponse(template, overlapping));
     }

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java

@@ -201,7 +201,7 @@ public class MetadataCreateDataStreamService {
                 "initialize_data_stream",
                 firstBackingIndexName,
                 firstBackingIndexName
-            ).dataStreamName(dataStreamName).systemDataStreamDescriptor(systemDataStreamDescriptor);
+            ).dataStreamName(dataStreamName).systemDataStreamDescriptor(systemDataStreamDescriptor).nameResolvedInstant(request.startTime);
 
             if (isSystem) {
                 createIndexRequest.settings(SystemIndexDescriptor.DEFAULT_SETTINGS);

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java

@@ -170,6 +170,10 @@ public class MetadataCreateIndexService {
         this.indexSettingProviders.add(provider);
     }
 
+    public Set<IndexSettingProvider> getIndexSettingProviders() {
+        return indexSettingProviders;
+    }
+
     /**
      * Validate the name for an index against some static rules and a cluster state.
      */
@@ -863,12 +867,20 @@ public class MetadataCreateIndexService {
         if (sourceMetadata == null) {
             final Settings.Builder additionalIndexSettings = Settings.builder();
             final Settings templateAndRequestSettings = Settings.builder().put(combinedTemplateSettings).put(request.settings()).build();
+            final boolean newDataStream = isDataStreamIndex
+                && currentState.getMetadata().dataStreams().containsKey(request.dataStreamName()) == false;
 
             // Loop through all the explicit index setting providers, adding them to the
             // additionalIndexSettings map
             for (IndexSettingProvider provider : indexSettingProviders) {
                 additionalIndexSettings.put(
-                    provider.getAdditionalIndexSettings(request.index(), isDataStreamIndex, templateAndRequestSettings)
+                    provider.getAdditionalIndexSettings(
+                        request.index(),
+                        request.dataStreamName(),
+                        newDataStream,
+                        request.getNameResolvedAt(),
+                        templateAndRequestSettings
+                    )
                 );
             }
 

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java

@@ -604,32 +604,69 @@ public class MetadataIndexTemplateService {
             return currentState;
         }
 
-        validate(name, finalIndexTemplate);
-        validateDataStreamsStillReferenced(currentState, name, finalIndexTemplate);
+        validateIndexTemplateV2(name, finalIndexTemplate, currentState);
+        logger.info(
+            "{} index template [{}] for index patterns {}",
+            existing == null ? "adding" : "updating",
+            name,
+            template.indexPatterns()
+        );
+        return ClusterState.builder(currentState).metadata(Metadata.builder(currentState.metadata()).put(name, finalIndexTemplate)).build();
+    }
+
+    private void validateIndexTemplateV2(String name, ComposableIndexTemplate indexTemplate, ClusterState currentState) {
+        // Workaround for the fact that start_time and end_time are injected by the MetadataCreateDataStreamService upon creation,
+        // but when validating templates that create data streams the MetadataCreateDataStreamService isn't used.
+        var finalTemplate = Optional.ofNullable(indexTemplate.template());
+        var finalSettings = Settings.builder();
+
+        // First apply settings sourced from index setting providers:
+        for (var provider : metadataCreateIndexService.getIndexSettingProviders()) {
+            finalSettings.put(
+                provider.getAdditionalIndexSettings(
+                    "validate-index-name",
+                    indexTemplate.getDataStreamTemplate() != null ? "validate-data-stream-name" : null,
+                    true,
+                    System.currentTimeMillis(),
+                    finalTemplate.map(Template::settings).orElse(Settings.EMPTY)
+                )
+            );
+        }
+        // Then apply settings resolved from templates:
+        finalSettings.put(finalTemplate.map(Template::settings).orElse(Settings.EMPTY));
+
+        var templateToValidate = new ComposableIndexTemplate(
+            indexTemplate.indexPatterns(),
+            new Template(
+                finalSettings.build(),
+                finalTemplate.map(Template::mappings).orElse(null),
+                finalTemplate.map(Template::aliases).orElse(null)
+            ),
+            indexTemplate.composedOf(),
+            indexTemplate.priority(),
+            indexTemplate.version(),
+            indexTemplate.metadata(),
+            indexTemplate.getDataStreamTemplate(),
+            indexTemplate.getAllowAutoCreate()
+        );
+
+        validate(name, templateToValidate);
+        validateDataStreamsStillReferenced(currentState, name, templateToValidate);
 
         // Finally, right before adding the template, we need to ensure that the composite settings,
         // mappings, and aliases are valid after it's been composed with the component templates
         try {
-            validateCompositeTemplate(currentState, name, finalIndexTemplate, indicesService, xContentRegistry, systemIndices);
+            validateCompositeTemplate(currentState, name, templateToValidate, indicesService, xContentRegistry, systemIndices);
         } catch (Exception e) {
             throw new IllegalArgumentException(
                 "composable template ["
                     + name
                     + "] template after composition "
-                    + (finalIndexTemplate.composedOf().size() > 0
-                        ? "with component templates " + finalIndexTemplate.composedOf() + " "
-                        : "")
+                    + (indexTemplate.composedOf().size() > 0 ? "with component templates " + indexTemplate.composedOf() + " " : "")
                     + "is invalid",
                 e
             );
         }
-        logger.info(
-            "{} index template [{}] for index patterns {}",
-            existing == null ? "adding" : "updating",
-            name,
-            template.indexPatterns()
-        );
-        return ClusterState.builder(currentState).metadata(Metadata.builder(currentState.metadata()).put(name, finalIndexTemplate)).build();
     }
 
     /**

server/src/main/java/org/elasticsearch/cluster/routing/allocation/DataTier.java

@@ -199,7 +199,13 @@ public class DataTier {
         private static final Logger logger = LogManager.getLogger(DefaultHotAllocationSettingProvider.class);
 
         @Override
-        public Settings getAdditionalIndexSettings(String indexName, boolean isDataStreamIndex, Settings indexSettings) {
+        public Settings getAdditionalIndexSettings(
+            String indexName,
+            String dataStreamName,
+            boolean newDataStream,
+            long resolvedAt,
+            Settings indexSettings
+        ) {
             Set<String> settings = indexSettings.keySet();
             if (settings.contains(TIER_PREFERENCE)) {
                 // just a marker -- this null value will be removed or overridden by the template/request settings
@@ -214,7 +220,7 @@ public class DataTier {
                     // Otherwise, put the setting in place by default, the "hot"
                     // tier if the index is part of a data stream, the "content"
                     // tier if it is not.
-                    if (isDataStreamIndex) {
+                    if (dataStreamName != null) {
                         return DATA_HOT_TIER_PREFERENCE_SETTINGS;
                     } else {
                         return DATA_CONTENT_TIER_PREFERENCE_SETTINGS;

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -186,6 +186,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         result.add(IndexMetadata.INDEX_ROUTING_PATH);
         result.add(IndexSettings.TIME_SERIES_START_TIME);
         result.add(IndexSettings.TIME_SERIES_END_TIME);
+        result.add(IndexSettings.LOOK_AHEAD_TIME);
         return Set.copyOf(result);
     }
 

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -541,6 +541,15 @@ public final class IndexSettings {
         Property.Final
     );
 
+    public static final Setting<TimeValue> LOOK_AHEAD_TIME = Setting.timeSetting(
+        "index.look_ahead_time",
+        TimeValue.timeValueHours(2),
+        TimeValue.timeValueMinutes(1),
+        TimeValue.timeValueDays(7),
+        Property.IndexScope,
+        Property.Final
+    );
+
     private final Index index;
     private final Version version;
     private final Logger logger;

server/src/main/java/org/elasticsearch/index/shard/IndexSettingProvider.java

@@ -19,7 +19,13 @@ public interface IndexSettingProvider {
      * Returns explicitly set default index {@link Settings} for the given index. This should not
      * return null.
      */
-    default Settings getAdditionalIndexSettings(String indexName, boolean isDataStreamIndex, Settings templateAndRequestSettings) {
+    default Settings getAdditionalIndexSettings(
+        String indexName,
+        String dataStreamName,
+        boolean newDataStream,
+        long resolvedAt,
+        Settings templateAndRequestSettings
+    ) {
         return Settings.EMPTY;
     }
 }

server/src/test/java/org/elasticsearch/cluster/metadata/HumanReadableIndexSettingsTests.java

@@ -17,6 +17,7 @@ import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 
 import static org.elasticsearch.test.VersionUtils.randomVersion;
+import static org.hamcrest.Matchers.equalTo;
 
 public class HumanReadableIndexSettingsTests extends ESTestCase {
     public void testHumanReadableSettings() {
@@ -28,7 +29,7 @@ public class HumanReadableIndexSettingsTests extends ESTestCase {
             .build();
 
         Settings humanSettings = IndexMetadata.addHumanReadableSettings(testSettings);
-
+        assertThat(humanSettings.size(), equalTo(4));
         assertEquals(versionCreated.toString(), humanSettings.get(IndexMetadata.SETTING_VERSION_CREATED_STRING, null));
         ZonedDateTime creationDate = ZonedDateTime.ofInstant(Instant.ofEpochMilli(created), ZoneOffset.UTC);
         assertEquals(creationDate.toString(), humanSettings.get(IndexMetadata.SETTING_CREATION_DATE_STRING, null));

x-pack/plugin/data-streams/qa/rest/src/javaRestTest/java/org/elasticsearch/xpack/datastreams/TsdbDataStreamRestIT.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.datastreams;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.common.time.DateFormatter;
+import org.elasticsearch.common.time.FormatNames;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.test.rest.yaml.ObjectPath;
+
+import java.time.Instant;
+
+import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.backingIndexEqualTo;
+import static org.hamcrest.Matchers.aMapWithSize;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.notNullValue;
+
+public class TsdbDataStreamRestIT extends ESRestTestCase {
+
+    private static final String TEMPLATE = """
+        {
+            "index_patterns": ["k8s*"],
+            "template": {
+                "settings":{
+                    "index": {
+                        "number_of_replicas": 0,
+                        "number_of_shards": 2,
+                        "mode": "time_series",
+                        "routing_path": ["metricset", "time_series_dimension"]
+                    }
+                },
+                "mappings":{
+                    "properties": {
+                        "@timestamp" : {
+                            "type": "date"
+                        },
+                        "metricset": {
+                            "type": "keyword",
+                            "time_series_dimension": true
+                        },
+                        "k8s": {
+                            "properties": {
+                                "pod": {
+                                    "properties": {
+                                        "uid": {
+                                            "type": "keyword",
+                                            "time_series_dimension": true
+                                        },
+                                        "name": {
+                                            "type": "keyword"
+                                        },
+                                        "ip": {
+                                            "type": "ip"
+                                        },
+                                        "network": {
+                                            "properties": {
+                                                "tx": {
+                                                    "type": "long"
+                                                },
+                                                "rx": {
+                                                    "type": "long"
+                                                }
+                                            }
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "data_stream": {}
+        }""";
+
+    public void testTsdbDataStreams() throws Exception {
+        // Create a template
+        var putComposableIndexTemplateRequest = new Request("POST", "/_index_template/1");
+        putComposableIndexTemplateRequest.setJsonEntity(TEMPLATE);
+        assertOK(client().performRequest(putComposableIndexTemplateRequest));
+
+        Instant now = Instant.now();
+        String nowAsString = DateFormatter.forPattern(FormatNames.STRICT_DATE_OPTIONAL_TIME.getName()).format(now);
+        var bulkRequest = new Request("POST", "/k8s/_bulk");
+        bulkRequest.setJsonEntity(
+            """
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.1", "network": {"tx": 2001818691, "rx": 802133794}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.1", "network": {"tx": 2005177954, "rx": 801479970}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.1", "network": {"tx": 2006223737, "rx": 802337279}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "cat", "uid":"947e4ced-1786-4e53-9e0c-5c447e959507", "ip": "10.10.55.2", "network": {"tx": 2012916202, "rx": 803685721}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434521831, "rx": 530575198}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434577921, "rx": 530600088}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434587694, "rx": 530604797}}}}
+                {"create": {}}
+                {"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "dog", "uid":"df3145b3-0563-4d3b-a0f7-897eb2876ea9", "ip": "10.10.55.3", "network": {"tx": 1434595272, "rx": 530605511}}}}
+                """
+                .replace("$now", nowAsString)
+        );
+        bulkRequest.addParameter("refresh", "true");
+        assertOK(client().performRequest(bulkRequest));
+
+        var getDataStreamsRequest = new Request("GET", "/_data_stream");
+        var response = client().performRequest(getDataStreamsRequest);
+        assertOK(response);
+        var dataStreams = entityAsMap(response);
+        assertThat(ObjectPath.evaluate(dataStreams, "data_streams"), hasSize(1));
+        assertThat(ObjectPath.evaluate(dataStreams, "data_streams.0.name"), equalTo("k8s"));
+        assertThat(ObjectPath.evaluate(dataStreams, "data_streams.0.generation"), equalTo(1));
+        assertThat(ObjectPath.evaluate(dataStreams, "data_streams.0.template"), equalTo("1"));
+        assertThat(ObjectPath.evaluate(dataStreams, "data_streams.0.indices"), hasSize(1));
+        String backingIndex = ObjectPath.evaluate(dataStreams, "data_streams.0.indices.0.index_name");
+        assertThat(backingIndex, backingIndexEqualTo("k8s", 1));
+
+        var getIndexRequest = new Request("GET", "/" + backingIndex + "?human");
+        response = client().performRequest(getIndexRequest);
+        assertOK(response);
+        var indices = entityAsMap(response);
+        var escapedBackingIndex = backingIndex.replace(".", "\\.");
+        assertThat(ObjectPath.evaluate(indices, escapedBackingIndex + ".data_stream"), equalTo("k8s"));
+        assertThat(ObjectPath.evaluate(indices, escapedBackingIndex + ".settings.index.mode"), equalTo("time_series"));
+        assertThat(ObjectPath.evaluate(indices, escapedBackingIndex + ".settings.index.time_series.start_time"), notNullValue());
+        assertThat(ObjectPath.evaluate(indices, escapedBackingIndex + ".settings.index.time_series.end_time"), notNullValue());
+    }
+
+    public void testSimulateTsdbDataStreamTemplate() throws Exception {
+        var putComposableIndexTemplateRequest = new Request("POST", "/_index_template/1");
+        putComposableIndexTemplateRequest.setJsonEntity(TEMPLATE);
+        assertOK(client().performRequest(putComposableIndexTemplateRequest));
+
+        var simulateIndexTemplateRequest = new Request("POST", "/_index_template/_simulate_index/k8s");
+        var response = client().performRequest(simulateIndexTemplateRequest);
+        assertOK(response);
+        var responseBody = entityAsMap(response);
+        assertThat(ObjectPath.evaluate(responseBody, "template.settings.index"), aMapWithSize(4));
+        assertThat(ObjectPath.evaluate(responseBody, "template.settings.index.number_of_shards"), equalTo("2"));
+        assertThat(ObjectPath.evaluate(responseBody, "template.settings.index.number_of_replicas"), equalTo("0"));
+        assertThat(ObjectPath.evaluate(responseBody, "template.settings.index.mode"), equalTo("time_series"));
+        assertThat(
+            ObjectPath.evaluate(responseBody, "template.settings.index.routing_path"),
+            contains("metricset", "time_series_dimension")
+        );
+        assertThat(ObjectPath.evaluate(responseBody, "overlapping"), empty());
+    }
+
+}

x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/DataStreamIndexSettingsProvider.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.datastreams;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.time.DateFormatter;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.shard.IndexSettingProvider;
+
+import java.time.Instant;
+import java.util.Locale;
+import java.util.Optional;
+
+public class DataStreamIndexSettingsProvider implements IndexSettingProvider {
+
+    private static final DateFormatter FORMATTER = DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER;
+
+    @Override
+    public Settings getAdditionalIndexSettings(
+        String indexName,
+        String dataStreamName,
+        boolean newDataStream,
+        long resolvedAt,
+        Settings templateAndRequestSettings
+    ) {
+        if (dataStreamName != null && newDataStream) {
+            IndexMode indexMode = Optional.ofNullable(templateAndRequestSettings.get(IndexSettings.MODE.getKey()))
+                .map(value -> IndexMode.valueOf(value.toUpperCase(Locale.ROOT)))
+                .orElse(IndexMode.STANDARD);
+            TimeValue lookAheadTime = templateAndRequestSettings.getAsTime(
+                IndexSettings.LOOK_AHEAD_TIME.getKey(),
+                IndexSettings.LOOK_AHEAD_TIME.getDefault(templateAndRequestSettings)
+            );
+            if (indexMode == IndexMode.TIME_SERIES) {
+                Instant start = Instant.ofEpochMilli(resolvedAt).minusMillis(lookAheadTime.getMillis());
+                Instant end = Instant.ofEpochMilli(resolvedAt).plusMillis(lookAheadTime.getMillis());
+
+                Settings.Builder builder = Settings.builder();
+                builder.put(IndexSettings.TIME_SERIES_START_TIME.getKey(), FORMATTER.format(start));
+                builder.put(IndexSettings.TIME_SERIES_END_TIME.getKey(), FORMATTER.format(end));
+                return builder.build();
+            }
+        }
+
+        return Settings.EMPTY;
+    }
+
+}

x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/DataStreamsPlugin.java

@@ -14,6 +14,7 @@ import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.index.shard.IndexSettingProvider;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
@@ -41,6 +42,7 @@ import org.elasticsearch.xpack.datastreams.rest.RestGetDataStreamsAction;
 import org.elasticsearch.xpack.datastreams.rest.RestMigrateToDataStreamAction;
 import org.elasticsearch.xpack.datastreams.rest.RestPromoteDataStreamAction;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.function.Supplier;
 
@@ -86,4 +88,9 @@ public class DataStreamsPlugin extends Plugin implements ActionPlugin {
         var promoteAction = new RestPromoteDataStreamAction();
         return List.of(createDsAction, deleteDsAction, getDsAction, dsStatsAction, migrateAction, promoteAction);
     }
+
+    @Override
+    public Collection<IndexSettingProvider> getAdditionalIndexSettingProviders() {
+        return List.of(new DataStreamIndexSettingsProvider());
+    }
 }

x-pack/plugin/data-streams/src/test/java/org/elasticsearch/xpack/datastreams/DataStreamIndexSettingsProviderTests.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.datastreams;
+
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.test.ESTestCase;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+
+import static org.elasticsearch.common.settings.Settings.builder;
+import static org.hamcrest.Matchers.equalTo;
+
+public class DataStreamIndexSettingsProviderTests extends ESTestCase {
+
+    public void testGetAdditionalIndexSettings() {
+        String dataStreamName = "logs-app1";
+
+        Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+        TimeValue lookAheadTime = TimeValue.timeValueHours(2); // default
+        Settings settings = builder().put("index.mode", "time_series").build();
+        var provider = new DataStreamIndexSettingsProvider();
+        Settings result = provider.getAdditionalIndexSettings(
+            DataStream.getDefaultBackingIndexName(dataStreamName, 1),
+            dataStreamName,
+            true,
+            now.toEpochMilli(),
+            settings
+        );
+        assertThat(IndexSettings.TIME_SERIES_START_TIME.get(result), equalTo(now.minusMillis(lookAheadTime.getMillis())));
+        assertThat(IndexSettings.TIME_SERIES_END_TIME.get(result), equalTo(now.plusMillis(lookAheadTime.getMillis())));
+    }
+
+    public void testGetAdditionalIndexSettingsLookAheadTime() {
+        String dataStreamName = "logs-app1";
+
+        Instant now = Instant.now().truncatedTo(ChronoUnit.MILLIS);
+        TimeValue lookAheadTime = TimeValue.timeValueMinutes(30);
+        Settings settings = builder().put("index.mode", "time_series").put("index.look_ahead_time", lookAheadTime.getStringRep()).build();
+        var provider = new DataStreamIndexSettingsProvider();
+        Settings result = provider.getAdditionalIndexSettings(
+            DataStream.getDefaultBackingIndexName(dataStreamName, 1),
+            dataStreamName,
+            true,
+            now.toEpochMilli(),
+            settings
+        );
+        assertThat(result.size(), equalTo(2));
+        assertThat(IndexSettings.TIME_SERIES_START_TIME.get(result), equalTo(now.minusMillis(lookAheadTime.getMillis())));
+        assertThat(IndexSettings.TIME_SERIES_END_TIME.get(result), equalTo(now.plusMillis(lookAheadTime.getMillis())));
+    }
+
+    public void testGetAdditionalIndexSettingsNoTimeSeries() {
+        String dataStreamName = "logs-app1";
+
+        long now = Instant.now().toEpochMilli();
+        Settings settings = randomBoolean() ? Settings.EMPTY : builder().put("index.mode", "standard").build();
+        var provider = new DataStreamIndexSettingsProvider();
+        Settings result = provider.getAdditionalIndexSettings(
+            DataStream.getDefaultBackingIndexName(dataStreamName, 1),
+            dataStreamName,
+            true,
+            now,
+            settings
+        );
+        assertThat(result, equalTo(Settings.EMPTY));
+    }
+
+    public void testGetAdditionalIndexSettingsDataStreamAlreadyCreated() {
+        String dataStreamName = "logs-app1";
+
+        long now = Instant.now().toEpochMilli();
+        Settings settings = builder().put("index.mode", "time_series").build();
+        var provider = new DataStreamIndexSettingsProvider();
+        Settings result = provider.getAdditionalIndexSettings(
+            DataStream.getDefaultBackingIndexName(dataStreamName, 1),
+            dataStreamName,
+            false,
+            now,
+            settings
+        );
+        assertThat(result, equalTo(Settings.EMPTY));
+    }
+
+}

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/data_stream/150_tsdb.yml

@@ -88,6 +88,16 @@ created the data stream:
   - match: { data_streams.0.template: 'my-template1' }
   - match: { data_streams.0.hidden: false }
   - match: { data_streams.0.system: false }
+  - set:  { data_streams.0.indices.0.index_name: backing_index }
+
+  - do:
+      indices.get:
+        index: $backing_index
+        human: true
+  - match: { $body.$backing_index.data_stream: 'k8s' }
+  - match: { $body.$backing_index.settings.index.mode: 'time_series' }
+  - match: { $body.$backing_index.settings.index.time_series.start_time: '2021-04-28T00:00:00Z' }
+  - match: { $body.$backing_index.settings.index.time_series.end_time: '2021-04-29T00:00:00Z' }
 
 ---
 fetch the tsid: