Using index setting providers for data stream setting validation (#136214) (#136308)

Author: masseyke

docs/changelog/136214.yaml

@@ -0,0 +1,6 @@
+pr: 136214
+summary: Using index setting providers for data stream setting validation
+area: Data streams
+type: bug
+issues:
+ - 136166

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamSettingsIT.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Template;
+import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
@@ -323,12 +324,57 @@ public void testIndexBlock() throws Exception {
         }
     }
 
+    public void testPutDataStreamSettingsTSDS() throws Exception {
+        String dataStreamName = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
+        putComposableIndexTemplate(
+            "my-template",
+            List.of(dataStreamName),
+            indexSettings(1, 0).put("index.mode", "time_series").build(),
+            CompressedXContent.fromJSON("""
+                {
+                  "properties": {
+                    "host.name": {
+                      "type": "keyword",
+                      "time_series_dimension": true
+                    }
+                  }
+                }
+                """)
+        );
+        final var createDataStreamRequest = new CreateDataStreamAction.Request(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT, dataStreamName);
+        assertAcked(client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).actionGet());
+        final int numberOfShards = randomIntBetween(2, 7);
+        Settings dataStreamSettings = Settings.builder().put("index.number_of_shards", numberOfShards).build();
+        UpdateDataStreamSettingsAction.Response putSettingsResponse = client().execute(
+            UpdateDataStreamSettingsAction.INSTANCE,
+            new UpdateDataStreamSettingsAction.Request(dataStreamSettings, TimeValue.THIRTY_SECONDS, TimeValue.THIRTY_SECONDS).indices(
+                dataStreamName
+            )
+        ).actionGet();
+        List<UpdateDataStreamSettingsAction.DataStreamSettingsResponse> dataStreamSettingsResponses = putSettingsResponse
+            .getDataStreamSettingsResponses();
+        assertThat(dataStreamSettingsResponses.size(), equalTo(1));
+        UpdateDataStreamSettingsAction.DataStreamSettingsResponse dataStreamSettingsResponse = dataStreamSettingsResponses.get(0);
+        assertThat(dataStreamSettingsResponse.dataStreamName(), equalTo(dataStreamName));
+        assertThat(dataStreamSettingsResponse.dataStreamSucceeded(), equalTo(true));
+        assertThat(dataStreamSettingsResponse.settings().get("index.number_of_shards"), equalTo(Integer.toString(numberOfShards)));
+    }
+
     static void putComposableIndexTemplate(String id, List<String> patterns, @Nullable Settings settings) throws IOException {
+        putComposableIndexTemplate(id, patterns, settings, null);
+    }
+
+    static void putComposableIndexTemplate(
+        String id,
+        List<String> patterns,
+        @Nullable Settings settings,
+        @Nullable CompressedXContent mappings
+    ) throws IOException {
         TransportPutComposableIndexTemplateAction.Request request = new TransportPutComposableIndexTemplateAction.Request(id);
         request.indexTemplate(
             ComposableIndexTemplate.builder()
                 .indexPatterns(patterns)
-                .template(Template.builder().settings(settings))
+                .template(Template.builder().settings(settings).mappings(mappings))
                 .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
                 .build()
         );

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

@@ -32,6 +32,9 @@
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.IndexSettingProviders;
+import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.logging.LogManager;
@@ -40,6 +43,7 @@
 import org.elasticsearch.snapshots.SnapshotsServiceUtils;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -62,15 +66,18 @@ public class MetadataDataStreamsService {
     private final MasterServiceTaskQueue<UpdateOptionsTask> updateOptionsTaskQueue;
     private final MasterServiceTaskQueue<UpdateSettingsTask> updateSettingsTaskQueue;
     private final MasterServiceTaskQueue<UpdateMappingsTask> updateMappingsTaskQueue;
+    private final IndexSettingProviders indexSettingProviders;
 
     public MetadataDataStreamsService(
         ClusterService clusterService,
         IndicesService indicesService,
-        DataStreamGlobalRetentionSettings globalRetentionSettings
+        DataStreamGlobalRetentionSettings globalRetentionSettings,
+        IndexSettingProviders indexSettingProviders
     ) {
         this.clusterService = clusterService;
         this.indicesService = indicesService;
         this.globalRetentionSettings = globalRetentionSettings;
+        this.indexSettingProviders = indexSettingProviders;
         ClusterStateTaskExecutor<UpdateLifecycleTask> updateLifecycleExecutor = new SimpleBatchedAckListenerTaskExecutor<>() {
 
             @Override
@@ -497,15 +504,56 @@ private DataStream createDataStreamForUpdatedDataStreamSettings(
         final ComposableIndexTemplate template = lookupTemplateForDataStream(dataStreamName, projectMetadata);
         Settings templateSettings = MetadataIndexTemplateService.resolveSettings(template, projectMetadata.componentTemplates());
         Settings mergedEffectiveSettings = templateSettings.merge(mergedDataStreamSettings);
+        CompressedXContent effectiveMappings = dataStream.getEffectiveMappings(projectMetadata, indicesService);
         MetadataIndexTemplateService.validateTemplate(
-            mergedEffectiveSettings,
-            dataStream.getEffectiveMappings(projectMetadata, indicesService),
+            addSettingsFromIndexSettingProviders(dataStreamName, effectiveMappings, projectMetadata, mergedEffectiveSettings),
+            effectiveMappings,
             indicesService
         );
 
         return dataStream.copy().setSettings(mergedDataStreamSettings).build();
     }
 
+    private Settings addSettingsFromIndexSettingProviders(
+        String dataStreamName,
+        CompressedXContent effectiveMappings,
+        ProjectMetadata projectMetadata,
+        Settings settings
+    ) throws IOException {
+        Settings.Builder additionalSettings = Settings.builder();
+        IndexMode indexMode = projectMetadata.dataStreams().get(dataStreamName).getIndexMode();
+        Set<String> overrulingSettings = new HashSet<>();
+        indexSettingProviders.getIndexSettingProviders().forEach(indexSettingProvider -> {
+            Settings.Builder providerSettingsBuilder = Settings.builder();
+            indexSettingProvider.provideAdditionalSettings(
+                dataStreamName,
+                dataStreamName,
+                indexMode,
+                projectMetadata,
+                Instant.now(),
+                settings,
+                List.of(effectiveMappings),
+                IndexVersion.current(),
+                providerSettingsBuilder
+            );
+            Settings providerSettings = providerSettingsBuilder.build();
+            if (indexSettingProvider.overrulesTemplateAndRequestSettings()) {
+                overrulingSettings.addAll(providerSettings.keySet());
+            }
+            additionalSettings.put(providerSettings);
+        });
+        Settings filteredmergedEffectiveSettings = settings;
+        if (overrulingSettings.isEmpty() == false) {
+            // Filter any conflicting settings from overruling providers, to avoid overwriting their values from templates.
+            final Settings.Builder filtered = Settings.builder().put(settings);
+            for (String setting : overrulingSettings) {
+                filtered.remove(setting);
+            }
+            filteredmergedEffectiveSettings = filtered.build();
+        }
+        return additionalSettings.put(filteredmergedEffectiveSettings).build();
+    }
+
     private DataStream createDataStreamForUpdatedDataStreamMappings(
         ProjectId projectId,
         String dataStreamName,

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -646,7 +646,8 @@ private DataStreamGlobalRetentionSettings createDataStreamServicesAndGlobalReten
         ThreadPool threadPool,
         ClusterService clusterService,
         IndicesService indicesService,
-        MetadataCreateIndexService metadataCreateIndexService
+        MetadataCreateIndexService metadataCreateIndexService,
+        IndexSettingProviders indexSettingProviders
     ) {
         DataStreamGlobalRetentionSettings dataStreamGlobalRetentionSettings = DataStreamGlobalRetentionSettings.create(
             clusterService.getClusterSettings()
@@ -662,7 +663,7 @@ private DataStreamGlobalRetentionSettings createDataStreamServicesAndGlobalReten
         );
         modules.bindToInstance(
             MetadataDataStreamsService.class,
-            new MetadataDataStreamsService(clusterService, indicesService, dataStreamGlobalRetentionSettings)
+            new MetadataDataStreamsService(clusterService, indicesService, dataStreamGlobalRetentionSettings, indexSettingProviders)
         );
         return dataStreamGlobalRetentionSettings;
     }
@@ -990,7 +991,8 @@ public Map<String, String> queryFields() {
             threadPool,
             clusterService,
             indicesService,
-            metadataCreateIndexService
+            metadataCreateIndexService,
+            indexSettingProviders
         );
 
         final MetadataIndexTemplateService metadataIndexTemplateService = new MetadataIndexTemplateService(

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

@@ -141,7 +141,8 @@ public void testUpdatingLifecycleOnADataStream() {
         MetadataDataStreamsService metadataDataStreamsService = new MetadataDataStreamsService(
             mock(ClusterService.class),
             mock(IndicesService.class),
-            DataStreamGlobalRetentionSettings.create(ClusterSettings.createBuiltInClusterSettings(settingsWithDefaultRetention))
+            DataStreamGlobalRetentionSettings.create(ClusterSettings.createBuiltInClusterSettings(settingsWithDefaultRetention)),
+            IndexSettingProviders.EMPTY
         );
 
         ProjectMetadata after = metadataDataStreamsService.updateDataLifecycle(

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

@@ -20,6 +20,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexSettingProviders;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
@@ -472,7 +473,8 @@ public void testUpdateLifecycle() {
         MetadataDataStreamsService service = new MetadataDataStreamsService(
             mock(ClusterService.class),
             mock(IndicesService.class),
-            DataStreamGlobalRetentionSettings.create(ClusterSettings.createBuiltInClusterSettings())
+            DataStreamGlobalRetentionSettings.create(ClusterSettings.createBuiltInClusterSettings()),
+            IndexSettingProviders.EMPTY
         );
         {
             // Remove lifecycle
@@ -503,7 +505,8 @@ public void testUpdateDataStreamOptions() {
         MetadataDataStreamsService service = new MetadataDataStreamsService(
             mock(ClusterService.class),
             mock(IndicesService.class),
-            DataStreamGlobalRetentionSettings.create(ClusterSettings.createBuiltInClusterSettings())
+            DataStreamGlobalRetentionSettings.create(ClusterSettings.createBuiltInClusterSettings()),
+            IndexSettingProviders.EMPTY
         );
 
         // Ensure no data stream options are stored

x-pack/plugin/otel-data/src/yamlRestTest/resources/rest-api-spec/test/20_metrics_tests.yml

@@ -351,3 +351,27 @@ host.name pass-through:
   - length: { hits.hits: 1 }
   - match: { hits.hits.0.fields.resource\.attributes\.host\.name: [ "localhost" ] }
   - match: { hits.hits.0.fields.host\.name: [ "localhost" ] }
+
+---
+data stream settings ilm:
+  - do:
+      bulk:
+        index: metrics-generic.otel-default
+        refresh: true
+        body:
+          - create: {"dynamic_templates":{"metrics.foo.bar":"counter_long"}}
+          - "@timestamp": 2024-07-18T14:00:00Z
+            resource:
+              attributes:
+                host.name: localhost
+            metrics:
+              foo.bar: 42
+  - is_false: errors
+  - do:
+      indices.put_data_stream_settings:
+        name: metrics-generic.otel-default
+        body:
+          index:
+            number_of_shards: 2
+  - match: { data_streams.0.name: metrics-generic.otel-default }
+  - match: { data_streams.0.applied_to_data_stream: true }