Adding settings to data streams

modules/data-streams/src/test/java/org/elasticsearch/datastreams/UpdateTimeSeriesRangeServiceTests.java

@@ -257,6 +257,7 @@ public void testUpdateTimeSeriesTemporalOneBadDataStream() {
                 ds2Indices,
                 2,
                 ds2.getMetadata(),
+                ds2.getSettings(),
                 ds2.isHidden(),
                 ds2.isReplicated(),
                 ds2.isSystem(),

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

@@ -227,6 +227,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion DOC_FIELDS_AS_LIST = def(9_061_0_00);
     public static final TransportVersion DENSE_VECTOR_OFF_HEAP_STATS = def(9_062_00_0);
     public static final TransportVersion RANDOM_SAMPLER_QUERY_BUILDER = def(9_063_0_00);
+    public static final TransportVersion SETTINGS_IN_DATA_STREAMS = def(9_064_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -18,6 +18,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.mapper.DataStreamTimestampFieldMapper;
 import org.elasticsearch.index.mapper.MapperService;
@@ -309,6 +310,34 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params, @Nulla
         return builder;
     }
 
+    /*
+     * Merges the given settings into the settings in this ComposableIndexTemplate. Any null values in the
+     * given settings are removed from the settings in the returned ComposableIndexTemplate. If this
+     * ComposableIndexTemplate has no settings, the given settings are the only ones in the returned template
+     * (with any null values removed). If this ComposableIndexTemplate has no template, an empty template with
+     * those settings is created. If the given settings are empty, this ComposableIndexTemplate is just
+     * returned unchanged. This method never changes this object.
+     */
+    public ComposableIndexTemplate mergeSettings(Settings settings) {
+        Objects.requireNonNull(settings);
+        if (Settings.EMPTY.equals(settings)) {
+            return this;
+        }
+        ComposableIndexTemplate.Builder mergedIndexTemplateBuilder = this.toBuilder();
+        Template.Builder mergedTemplateBuilder;
+        Settings templateSettings;
+        if (this.template() == null) {
+            mergedTemplateBuilder = Template.builder();
+            templateSettings = null;
+        } else {
+            mergedTemplateBuilder = Template.builder(this.template());
+            templateSettings = this.template().settings();
+        }
+        mergedTemplateBuilder.settings(templateSettings == null ? settings : templateSettings.merge(settings));
+        mergedIndexTemplateBuilder.template(mergedTemplateBuilder);
+        return mergedIndexTemplateBuilder.build();
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(

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

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.DateFormatters;
 import org.elasticsearch.common.xcontent.XContentHelper;
@@ -67,6 +68,7 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.cluster.metadata.MetadataCreateDataStreamService.lookupTemplateForDataStream;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.elasticsearch.index.IndexSettings.LIFECYCLE_ORIGINATION_DATE;
 import static org.elasticsearch.index.IndexSettings.PREFER_ILM_SETTING;
@@ -114,6 +116,7 @@ public final class DataStream implements SimpleDiffable<DataStream>, ToXContentO
     private final long generation;
     @Nullable
     private final Map<String, Object> metadata;
+    private final Settings settings;
     private final boolean hidden;
     private final boolean replicated;
     private final boolean system;
@@ -146,8 +149,46 @@ public DataStream(
     ) {
         this(
             name,
+            indices,
             generation,
             metadata,
+            Settings.EMPTY,
+            hidden,
+            replicated,
+            system,
+            allowCustomRouting,
+            indexMode,
+            lifecycle,
+            dataStreamOptions,
+            failureIndices,
+            rolloverOnWrite,
+            autoShardingEvent
+        );
+    }
+
+    // visible for testing
+    public DataStream(
+        String name,
+        List<Index> indices,
+        long generation,
+        Map<String, Object> metadata,
+        Settings settings,
+        boolean hidden,
+        boolean replicated,
+        boolean system,
+        boolean allowCustomRouting,
+        IndexMode indexMode,
+        DataStreamLifecycle lifecycle,
+        @Nullable DataStreamOptions dataStreamOptions,
+        List<Index> failureIndices,
+        boolean rolloverOnWrite,
+        @Nullable DataStreamAutoShardingEvent autoShardingEvent
+    ) {
+        this(
+            name,
+            generation,
+            metadata,
+            settings,
             hidden,
             replicated,
             system,
@@ -165,6 +206,7 @@ public DataStream(
         String name,
         long generation,
         Map<String, Object> metadata,
+        Settings settings,
         boolean hidden,
         boolean replicated,
         boolean system,
@@ -179,6 +221,7 @@ public DataStream(
         this.name = name;
         this.generation = generation;
         this.metadata = metadata;
+        this.settings = Objects.requireNonNull(settings);
         assert system == false || hidden; // system indices must be hidden
         this.hidden = hidden;
         this.replicated = replicated;
@@ -231,10 +274,17 @@ public static DataStream read(StreamInput in) throws IOException {
             // is still behind a feature flag in previous version we use the default value instead of explicitly disabling it.
             dataStreamOptions = failureStoreEnabled ? DataStreamOptions.FAILURE_STORE_ENABLED : null;
         }
+        final Settings settings;
+        if (in.getTransportVersion().onOrAfter(TransportVersions.SETTINGS_IN_DATA_STREAMS)) {
+            settings = Settings.readSettingsFromStream(in);
+        } else {
+            settings = Settings.EMPTY;
+        }
         return new DataStream(
             name,
             generation,
             metadata,
+            settings,
             hidden,
             replicated,
             system,
@@ -325,6 +375,20 @@ public boolean rolloverOnWrite() {
         return backingIndices.rolloverOnWrite;
     }
 
+    public ComposableIndexTemplate getEffectiveIndexTemplate(ProjectMetadata projectMetadata) {
+        return getMatchingIndexTemplate(projectMetadata).mergeSettings(settings);
+    }
+
+    public Settings getEffectiveSettings(ProjectMetadata projectMetadata) {
+        ComposableIndexTemplate template = getMatchingIndexTemplate(projectMetadata);
+        Settings templateSettings = template.template() == null ? Settings.EMPTY : template.template().settings();
+        return templateSettings.merge(settings);
+    }
+
+    private ComposableIndexTemplate getMatchingIndexTemplate(ProjectMetadata projectMetadata) {
+        return lookupTemplateForDataStream(name, projectMetadata);
+    }
+
     /**
      * We define that a data stream is considered internal either if it is a system index or if
      * its name starts with a dot.
@@ -436,6 +500,10 @@ public Map<String, Object> getMetadata() {
         return metadata;
     }
 
+    public Settings getSettings() {
+        return settings;
+    }
+
     @Override
     public boolean isHidden() {
         return hidden;
@@ -1263,6 +1331,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(DataStream.ADD_DATA_STREAM_OPTIONS_VERSION)) {
             out.writeOptionalWriteable(dataStreamOptions.isEmpty() ? null : dataStreamOptions);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.SETTINGS_IN_DATA_STREAMS)) {
+            settings.writeTo(out);
+        }
     }
 
     public static final ParseField NAME_FIELD = new ParseField("name");
@@ -1284,6 +1355,7 @@ public void writeTo(StreamOutput out) throws IOException {
     public static final ParseField FAILURE_ROLLOVER_ON_WRITE_FIELD = new ParseField("failure_rollover_on_write");
     public static final ParseField FAILURE_AUTO_SHARDING_FIELD = new ParseField("failure_auto_sharding");
     public static final ParseField DATA_STREAM_OPTIONS_FIELD = new ParseField("options");
+    public static final ParseField SETTINGS_FIELD = new ParseField("settings");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>(
@@ -1292,6 +1364,7 @@ public void writeTo(StreamOutput out) throws IOException {
             (String) args[0],
             (Long) args[2],
             (Map<String, Object>) args[3],
+            args[17] == null ? Settings.EMPTY : (Settings) args[17],
             args[4] != null && (boolean) args[4],
             args[5] != null && (boolean) args[5],
             args[6] != null && (boolean) args[6],
@@ -1358,6 +1431,7 @@ public void writeTo(StreamOutput out) throws IOException {
             (p, c) -> DataStreamOptions.fromXContent(p),
             DATA_STREAM_OPTIONS_FIELD
         );
+        PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> Settings.fromXContent(p), SETTINGS_FIELD);
     }
 
     public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -1419,6 +1493,9 @@ public XContentBuilder toXContent(
             backingIndices.autoShardingEvent.toXContent(builder, params);
             builder.endObject();
         }
+        builder.startObject(SETTINGS_FIELD.getPreferredName());
+        this.settings.toXContent(builder, params);
+        builder.endObject();
         builder.endObject();
         return builder;
     }
@@ -1431,6 +1508,7 @@ public boolean equals(Object o) {
         return name.equals(that.name)
             && generation == that.generation
             && Objects.equals(metadata, that.metadata)
+            && Objects.equals(settings, that.settings)
             && hidden == that.hidden
             && system == that.system
             && replicated == that.replicated
@@ -1448,6 +1526,7 @@ public int hashCode() {
             name,
             generation,
             metadata,
+            settings,
             hidden,
             system,
             replicated,
@@ -1760,6 +1839,7 @@ public static class Builder {
         private long generation = 1;
         @Nullable
         private Map<String, Object> metadata = null;
+        private Settings settings = Settings.EMPTY;
         private boolean hidden = false;
         private boolean replicated = false;
         private boolean system = false;
@@ -1787,6 +1867,7 @@ private Builder(DataStream dataStream) {
             name = dataStream.name;
             generation = dataStream.generation;
             metadata = dataStream.metadata;
+            settings = dataStream.settings;
             hidden = dataStream.hidden;
             replicated = dataStream.replicated;
             system = dataStream.system;
@@ -1818,6 +1899,11 @@ public Builder setMetadata(Map<String, Object> metadata) {
             return this;
         }
 
+        public Builder setSettings(Settings settings) {
+            this.settings = settings;
+            return this;
+        }
+
         public Builder setHidden(boolean hidden) {
             this.hidden = hidden;
             return this;
@@ -1878,6 +1964,7 @@ public DataStream build() {
                 name,
                 generation,
                 metadata,
+                settings,
                 hidden,
                 replicated,
                 system,

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

@@ -331,6 +331,7 @@ static ClusterState createDataStream(
             dataStreamName,
             initialGeneration,
             template.metadata() != null ? Map.copyOf(template.metadata()) : null,
+            Settings.EMPTY,
             hidden,
             false,
             isSystem,

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

@@ -876,6 +876,27 @@ public Set<String> keySet() {
         return newKeySet;
     }
 
+    /*
+     * This method merges the given newSettings into this Settings, returning either a new Settings object or this if the newSettings are
+     * empty. If any values are null in newSettings, those keys are removed from the returned object.
+     */
+    public Settings merge(Settings newSettings) {
+        Objects.requireNonNull(newSettings);
+        if (Settings.EMPTY.equals(newSettings)) {
+            return this;
+        }
+        Settings.Builder builder = Settings.builder().put(this);
+        for (String key : newSettings.keySet()) {
+            String rawValue = newSettings.get(key);
+            if (rawValue == null) {
+                builder.remove(key);
+            } else {
+                builder.put(key, rawValue);
+            }
+        }
+        return builder.build();
+    }
+
     /**
      * A builder allowing to put different settings and then {@link #build()} an immutable
      * settings implementation. Use {@link Settings#builder()} in order to