Add custom metadata support to data steams.

Composable index template may hold custom metadata. This change adds behaviour that
when a data stream gets created the custom metadata of the matching composable index
template is copied to new data stream. The get data stream api can then be used to
view the custom metadata.

Example:

```
PUT /_index_template/my-logs-template
{
  "index_patterns": [ "logs-*" ],
  "data_stream": { },
  "template": {
      "settings": {
          "index.number_of_replicas": 0
      }
  },
  "_meta": {
      "managed": true
  }
}

PUT /_data_stream/logs-myapp

GET /_data_stream
```

The get data stream api then yields the following response:

```
{
    "data_streams": [
        {
            "name": "logs-myapp",
            "timestamp_field": {
                "name": "@timestamp"
            },
            "indices": [
                {
                    "index_name": ".ds-logs-myapp-000001",
                    "index_uuid": "3UaBxM3mQXuHR6qx0IDVCw"
                }
            ],
            "generation": 1,
            "_meta": {
                "managed": true
            },
            "status": "GREEN",
            "template": "my-logs-template"
        }
    ]
}
```

Closes elastic#59195

Author: martijnvg

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

@@ -19,43 +19,25 @@
 
 package org.elasticsearch.client.indices;
 
-import org.elasticsearch.xpack.core.action.GetDataStreamAction;
-import org.elasticsearch.xpack.core.action.GetDataStreamAction.Response.DataStreamInfo;
 import org.elasticsearch.client.AbstractResponseTestCase;
+import org.elasticsearch.cluster.DataStreamTestHelper;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.metadata.DataStream;
-import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.xpack.core.action.GetDataStreamAction;
+import org.elasticsearch.xpack.core.action.GetDataStreamAction.Response.DataStreamInfo;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
-import java.util.List;
-import java.util.Locale;
 import java.util.stream.Collectors;
 
-import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
-import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
-
 public class GetDataStreamResponseTests extends AbstractResponseTestCase<GetDataStreamAction.Response, GetDataStreamResponse> {
 
-    private static List<Index> randomIndexInstances() {
-        int numIndices = randomIntBetween(0, 128);
-        List<Index> indices = new ArrayList<>(numIndices);
-        for (int i = 0; i < numIndices; i++) {
-            indices.add(new Index(randomAlphaOfLength(10).toLowerCase(Locale.ROOT), UUIDs.randomBase64UUID(random())));
-        }
-        return indices;
-    }
-
     private static DataStreamInfo randomInstance() {
-        List<Index> indices = randomIndexInstances();
-        long generation = indices.size() + randomLongBetween(1, 128);
-        String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-        indices.add(new Index(getDefaultBackingIndexName(dataStreamName, generation), UUIDs.randomBase64UUID(random())));
-        DataStream dataStream = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation);
+        DataStream dataStream = DataStreamTestHelper.randomInstance();
         return new DataStreamInfo(dataStream, ClusterHealthStatus.YELLOW, randomAlphaOfLengthBetween(2, 10),
             randomAlphaOfLengthBetween(2, 10));
     }

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

@@ -18,8 +18,10 @@
  */
 package org.elasticsearch.cluster.metadata;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -35,6 +37,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Objects;
 
 public final class DataStream extends AbstractDiffable<DataStream> implements ToXContentObject {
@@ -45,18 +48,20 @@ public final class DataStream extends AbstractDiffable<DataStream> implements To
     private final TimestampField timeStampField;
     private final List<Index> indices;
     private final long generation;
+    private final Map<String, Object> metadata;
 
-    public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation) {
+    public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation, Map<String, Object> metadata) {
         this.name = name;
         this.timeStampField = timeStampField;
         this.indices = Collections.unmodifiableList(indices);
         this.generation = generation;
+        this.metadata = metadata;
         assert indices.size() > 0;
         assert indices.get(indices.size() - 1).getName().equals(getDefaultBackingIndexName(name, generation));
     }
 
     public DataStream(String name, TimestampField timeStampField, List<Index> indices) {
-        this(name, timeStampField, indices, indices.size());
+        this(name, timeStampField, indices, indices.size(), null);
     }
 
     public String getName() {
@@ -75,6 +80,11 @@ public long getGeneration() {
         return generation;
     }
 
+    @Nullable
+    public Map<String, Object> getMetadata() {
+        return metadata;
+    }
+
     /**
      * Performs a rollover on a {@code DataStream} instance and returns a new instance containing
      * the updated list of backing indices and incremented generation.
@@ -87,7 +97,7 @@ public DataStream rollover(Index newWriteIndex) {
         assert newWriteIndex.getName().equals(getDefaultBackingIndexName(name, generation + 1));
         List<Index> backingIndices = new ArrayList<>(indices);
         backingIndices.add(newWriteIndex);
-        return new DataStream(name, timeStampField, backingIndices, generation + 1);
+        return new DataStream(name, timeStampField, backingIndices, generation + 1, metadata);
     }
 
     /**
@@ -101,7 +111,7 @@ public DataStream removeBackingIndex(Index index) {
         List<Index> backingIndices = new ArrayList<>(indices);
         backingIndices.remove(index);
         assert backingIndices.size() == indices.size() - 1;
-        return new DataStream(name, timeStampField, backingIndices, generation);
+        return new DataStream(name, timeStampField, backingIndices, generation, metadata);
     }
 
     /**
@@ -126,7 +136,7 @@ public DataStream replaceBackingIndex(Index existingBackingIndex, Index newBacki
                 "it is the write index", existingBackingIndex.getName(), name));
         }
         backingIndices.set(backingIndexPosition, newBackingIndex);
-        return new DataStream(name, timeStampField, backingIndices, generation);
+        return new DataStream(name, timeStampField, backingIndices, generation, metadata);
     }
 
     /**
@@ -142,7 +152,8 @@ public static String getDefaultBackingIndexName(String dataStreamName, long gene
     }
 
     public DataStream(StreamInput in) throws IOException {
-        this(in.readString(), new TimestampField(in), in.readList(Index::new), in.readVLong());
+        this(in.readString(), new TimestampField(in), in.readList(Index::new), in.readVLong(),
+            in.getVersion().onOrAfter(Version.V_8_0_0) ? in.readMap(): null);
     }
 
     public static Diff<DataStream> readDiffFrom(StreamInput in) throws IOException {
@@ -155,22 +166,28 @@ public void writeTo(StreamOutput out) throws IOException {
         timeStampField.writeTo(out);
         out.writeList(indices);
         out.writeVLong(generation);
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeMap(metadata);
+        }
     }
 
     public static final ParseField NAME_FIELD = new ParseField("name");
     public static final ParseField TIMESTAMP_FIELD_FIELD = new ParseField("timestamp_field");
     public static final ParseField INDICES_FIELD = new ParseField("indices");
     public static final ParseField GENERATION_FIELD = new ParseField("generation");
+    public static final ParseField METADATA_FIELD = new ParseField("_meta");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream",
-        args -> new DataStream((String) args[0], (TimestampField) args[1], (List<Index>) args[2], (Long) args[3]));
+        args -> new DataStream((String) args[0], (TimestampField) args[1], (List<Index>) args[2], (Long) args[3],
+            (Map<String, Object>) args[4]));
 
     static {
         PARSER.declareString(ConstructingObjectParser.constructorArg(), NAME_FIELD);
         PARSER.declareObject(ConstructingObjectParser.constructorArg(), TimestampField.PARSER, TIMESTAMP_FIELD_FIELD);
         PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), (p, c) -> Index.fromXContent(p), INDICES_FIELD);
         PARSER.declareLong(ConstructingObjectParser.constructorArg(), GENERATION_FIELD);
+        PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA_FIELD);
     }
 
     public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -184,6 +201,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.field(TIMESTAMP_FIELD_FIELD.getPreferredName(), timeStampField);
         builder.field(INDICES_FIELD.getPreferredName(), indices);
         builder.field(GENERATION_FIELD.getPreferredName(), generation);
+        if (metadata != null) {
+            builder.field(METADATA_FIELD.getPreferredName(), metadata);
+        }
         builder.endObject();
         return builder;
     }
@@ -196,12 +216,13 @@ public boolean equals(Object o) {
         return name.equals(that.name) &&
             timeStampField.equals(that.timeStampField) &&
             indices.equals(that.indices) &&
-            generation == that.generation;
+            generation == that.generation &&
+            Objects.equals(metadata, that.metadata);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(name, timeStampField, indices, generation);
+        return Objects.hash(name, timeStampField, indices, generation, metadata);
     }
 
     public static final class TimestampField implements Writeable, ToXContentObject {

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

@@ -158,7 +158,9 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
 
         String fieldName = template.getDataStreamTemplate().getTimestampField();
         DataStream.TimestampField timestampField = new DataStream.TimestampField(fieldName);
-        DataStream newDataStream = new DataStream(request.name, timestampField, List.of(firstBackingIndex.getIndex()));
+        DataStream newDataStream =
+            new DataStream(request.name, timestampField, List.of(firstBackingIndex.getIndex()), 1L,
+                template.metadata() != null ? Map.copyOf(template.metadata()) : null);
         Metadata.Builder builder = Metadata.builder(currentState.metadata()).put(newDataStream);
         logger.info("adding data stream [{}]", request.name);
         return ClusterState.builder(currentState).metadata(builder).build();

server/src/main/java/org/elasticsearch/snapshots/RestoreService.java

@@ -608,7 +608,8 @@ static DataStream updateDataStream(DataStream dataStream, Metadata.Builder metad
         List<Index> updatedIndices = dataStream.getIndices().stream()
             .map(i -> metadata.get(renameIndex(i.getName(), request, true)).getIndex())
             .collect(Collectors.toList());
-        return new DataStream(dataStreamName, dataStream.getTimeStampField(), updatedIndices, dataStream.getGeneration());
+        return new DataStream(dataStreamName, dataStream.getTimeStampField(), updatedIndices, dataStream.getGeneration(),
+            dataStream.getMetadata());
     }
 
     public static RestoreInProgress updateRestoreStateWithDeletedIndices(RestoreInProgress oldRestore, Set<Index> deletedIndices) {

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

@@ -1727,7 +1727,7 @@ public void testIndicesAliasesRequestTargetDataStreams() {
 
         Metadata.Builder mdBuilder = Metadata.builder()
             .put(backingIndex, false)
-            .put(new DataStream(dataStreamName, createTimestampField("@timestamp"), List.of(backingIndex.getIndex()), 1));
+            .put(new DataStream(dataStreamName, createTimestampField("@timestamp"), List.of(backingIndex.getIndex())));
         ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(mdBuilder).build();
 
         {
@@ -1916,7 +1916,7 @@ public void testDataStreams() {
         Metadata.Builder mdBuilder = Metadata.builder()
             .put(index1, false)
             .put(index2, false)
-            .put(new DataStream(dataStreamName, createTimestampField("@timestamp"), List.of(index1.getIndex(), index2.getIndex()), 2));
+            .put(new DataStream(dataStreamName, createTimestampField("@timestamp"), List.of(index1.getIndex(), index2.getIndex())));
         ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(mdBuilder).build();
 
         {

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

@@ -1016,7 +1016,7 @@ public void testBuilderForDataStreamWithRandomlyNumberedBackingIndices() {
             backingIndices.add(im.getIndex());
         }
 
-        b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum));
+        b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum, null));
         Metadata metadata = b.build();
         assertThat(metadata.dataStreams().size(), equalTo(1));
         assertThat(metadata.dataStreams().get(dataStreamName).getName(), equalTo(dataStreamName));
@@ -1034,7 +1034,7 @@ public void testBuildIndicesLookupForDataStreams() {
                 indices.add(idx.getIndex());
                 b.put(idx, true);
             }
-            b.put(new DataStream(name, createTimestampField("@timestamp"), indices, indices.size()));
+            b.put(new DataStream(name, createTimestampField("@timestamp"), indices));
         }
 
         Metadata metadata = b.build();
@@ -1100,8 +1100,7 @@ public void testValidateDataStreamsThrowsExceptionOnConflict() {
         DataStream dataStream = new DataStream(
             dataStreamName,
             createTimestampField("@timestamp"),
-            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
-            backingIndices.size()
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
         );
 
         IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, backingIndices);
@@ -1174,8 +1173,7 @@ public void testValidateDataStreamsAllowsPrefixedBackingIndices() {
         DataStream dataStream = new DataStream(
             dataStreamName,
             createTimestampField("@timestamp"),
-            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
-            backingIndices.size()
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
         );
 
         IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, backingIndices);
@@ -1275,7 +1273,7 @@ private static CreateIndexResult createIndices(int numIndices, int numBackingInd
             b.put(im, false);
             backingIndices.add(im.getIndex());
         }
-        b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum));
+        b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum, null));
         return new CreateIndexResult(indices, backingIndices, b.build());
     }
 

test/framework/src/main/java/org/elasticsearch/cluster/DataStreamTestHelper.java

@@ -32,11 +32,13 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
 import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_INDEX_UUID;
 import static org.elasticsearch.test.ESTestCase.randomAlphaOfLength;
+import static org.elasticsearch.test.ESTestCase.randomBoolean;
 
 public final class DataStreamTestHelper {
 
@@ -103,7 +105,11 @@ public static DataStream randomInstance() {
         long generation = indices.size() + ESTestCase.randomLongBetween(1, 128);
         String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
         indices.add(new Index(getDefaultBackingIndexName(dataStreamName, generation), UUIDs.randomBase64UUID(LuceneTestCase.random())));
-        return new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation);
+        Map<String, Object> metadata = null;
+        if (randomBoolean()) {
+            metadata = Map.of("key", "value");
+        }
+        return new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation, metadata);
     }
 
     /**
@@ -127,7 +133,8 @@ public static ClusterState getClusterStateWithDataStreams(List<Tuple<String, Int
                 dsTuple.v1(),
                 createTimestampField("@timestamp"),
                 backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
-                dsTuple.v2()
+                dsTuple.v2(),
+                null
             );
             builder.put(ds);
         }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/action/GetDataStreamAction.java

@@ -159,6 +159,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 builder.field(DataStream.TIMESTAMP_FIELD_FIELD.getPreferredName(), dataStream.getTimeStampField());
                 builder.field(DataStream.INDICES_FIELD.getPreferredName(), dataStream.getIndices());
                 builder.field(DataStream.GENERATION_FIELD.getPreferredName(), dataStream.getGeneration());
+                if (dataStream.getMetadata() != null) {
+                    builder.field(DataStream.METADATA_FIELD.getPreferredName(), dataStream.getMetadata());
+                }
                 builder.field(STATUS_FIELD.getPreferredName(), dataStreamStatus);
                 if (indexTemplate != null) {
                     builder.field(INDEX_TEMPLATE_FIELD.getPreferredName(), indexTemplate);

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/RolloverStepTests.java

@@ -136,7 +136,7 @@ public void testPerformActionOnDataStream() {
             .metadata(
                 Metadata.builder()
                     .put(new DataStream(dataStreamName, createTimestampField("@timestamp"),
-                        List.of(indexMetadata.getIndex()), 1L))
+                        List.of(indexMetadata.getIndex())))
                     .put(indexMetadata, true)
             )
             .build();

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/UpdateRolloverLifecycleDateStepTests.java

@@ -98,7 +98,7 @@ public void testPerformActionOnDataStream() {
             .metadata(
                 Metadata.builder()
                     .put(new DataStream(dataStreamName, createTimestampField("@timestamp"),
-                        List.of(originalIndexMeta.getIndex(), rolledIndexMeta.getIndex()), 2L))
+                        List.of(originalIndexMeta.getIndex(), rolledIndexMeta.getIndex())))
                     .put(originalIndexMeta, true)
                     .put(rolledIndexMeta, true)
             ).build();

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/WaitForActiveShardsTests.java

@@ -169,7 +169,7 @@ public void testResultEvaluatedOnDataStream() throws IOException {
             .metadata(
                 Metadata.builder()
                     .put(new DataStream(dataStreamName, createTimestampField("@timestamp"),
-                        List.of(originalIndexMeta.getIndex(), rolledIndexMeta.getIndex()), 2L))
+                        List.of(originalIndexMeta.getIndex(), rolledIndexMeta.getIndex())))
                     .put(originalIndexMeta, true)
                     .put(rolledIndexMeta, true)
             )

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/WaitForRolloverReadyStepTests.java

@@ -152,7 +152,7 @@ public void testEvaluateConditionOnDataStreamTarget() {
         SetOnce<Boolean> conditionsMet = new SetOnce<>();
         Metadata metadata = Metadata.builder().put(indexMetadata, true)
             .put(new DataStream(dataStreamName, createTimestampField("@timestamp"),
-                List.of(indexMetadata.getIndex()), 1L))
+                List.of(indexMetadata.getIndex())))
             .build();
         step.evaluateCondition(metadata, indexMetadata.getIndex(), new AsyncWaitStep.Listener() {