Skip to content

Commit d7ad787

Browse files
martijnvgmartijnvg
committed
Add custom metadata support to data steams. (elastic#63991)
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
1 parent c435246 commit d7ad787

File tree

16 files changed

+153
-62
lines changed

16 files changed

+153
-62
lines changed

‎client/rest-high-level/src/main/java/org/elasticsearch/client/indices/DataStream.java

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,16 +41,19 @@ public final class DataStream {
4141
String indexTemplate;
4242
@Nullable
4343
String ilmPolicyName;
44+
@Nullable
45+
private final Map<String, Object> metadata;
4446

4547
public DataStream(String name, String timeStampField, List<String> indices, long generation, ClusterHealthStatus dataStreamStatus,
46-
@Nullable String indexTemplate, @Nullable String ilmPolicyName) {
48+
@Nullable String indexTemplate, @Nullable String ilmPolicyName, @Nullable Map<String, Object> metadata) {
4749
this.name = name;
4850
this.timeStampField = timeStampField;
4951
this.indices = indices;
5052
this.generation = generation;
5153
this.dataStreamStatus = dataStreamStatus;
5254
this.indexTemplate = indexTemplate;
5355
this.ilmPolicyName = ilmPolicyName;
56+
this.metadata = metadata;
5457
}
5558

5659
public String getName() {
@@ -81,13 +84,18 @@ public String getIlmPolicyName() {
8184
return ilmPolicyName;
8285
}
8386

87+
public Map<String, Object> getMetadata() {
88+
return metadata;
89+
}
90+
8491
public static final ParseField NAME_FIELD = new ParseField("name");
8592
public static final ParseField TIMESTAMP_FIELD_FIELD = new ParseField("timestamp_field");
8693
public static final ParseField INDICES_FIELD = new ParseField("indices");
8794
public static final ParseField GENERATION_FIELD = new ParseField("generation");
8895
public static final ParseField STATUS_FIELD = new ParseField("status");
8996
public static final ParseField INDEX_TEMPLATE_FIELD = new ParseField("template");
9097
public static final ParseField ILM_POLICY_FIELD = new ParseField("ilm_policy");
98+
public static final ParseField METADATA_FIELD = new ParseField("_meta");
9199

92100
@SuppressWarnings("unchecked")
93101
private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream",
@@ -101,7 +109,8 @@ public String getIlmPolicyName() {
101109
ClusterHealthStatus status = ClusterHealthStatus.fromString(statusStr);
102110
String indexTemplate = (String) args[5];
103111
String ilmPolicy = (String) args[6];
104-
return new DataStream(dataStreamName, timeStampField, indices, generation, status, indexTemplate, ilmPolicy);
112+
Map<String, Object> metadata = (Map<String, Object>) args[7];
113+
return new DataStream(dataStreamName, timeStampField, indices, generation, status, indexTemplate, ilmPolicy, metadata);
105114
});
106115

107116
static {
@@ -112,6 +121,7 @@ public String getIlmPolicyName() {
112121
PARSER.declareString(ConstructingObjectParser.constructorArg(), STATUS_FIELD);
113122
PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), INDEX_TEMPLATE_FIELD);
114123
PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), ILM_POLICY_FIELD);
124+
PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA_FIELD);
115125
}
116126

117127
public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -129,11 +139,12 @@ public boolean equals(Object o) {
129139
indices.equals(that.indices) &&
130140
dataStreamStatus == that.dataStreamStatus &&
131141
Objects.equals(indexTemplate, that.indexTemplate) &&
132-
Objects.equals(ilmPolicyName, that.ilmPolicyName);
142+
Objects.equals(ilmPolicyName, that.ilmPolicyName) &&
143+
Objects.equals(metadata, that.metadata);
133144
}
134145

135146
@Override
136147
public int hashCode() {
137-
return Objects.hash(name, timeStampField, indices, generation, dataStreamStatus, indexTemplate, ilmPolicyName);
148+
return Objects.hash(name, timeStampField, indices, generation, dataStreamStatus, indexTemplate, ilmPolicyName, metadata);
138149
}
139150
}

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

Lines changed: 4 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -19,43 +19,25 @@
1919

2020
package org.elasticsearch.client.indices;
2121

22-
import org.elasticsearch.xpack.core.action.GetDataStreamAction;
23-
import org.elasticsearch.xpack.core.action.GetDataStreamAction.Response.DataStreamInfo;
2422
import org.elasticsearch.client.AbstractResponseTestCase;
23+
import org.elasticsearch.cluster.DataStreamTestHelper;
2524
import org.elasticsearch.cluster.health.ClusterHealthStatus;
2625
import org.elasticsearch.cluster.metadata.DataStream;
27-
import org.elasticsearch.common.UUIDs;
2826
import org.elasticsearch.common.xcontent.XContentParser;
2927
import org.elasticsearch.common.xcontent.XContentType;
3028
import org.elasticsearch.index.Index;
29+
import org.elasticsearch.xpack.core.action.GetDataStreamAction;
30+
import org.elasticsearch.xpack.core.action.GetDataStreamAction.Response.DataStreamInfo;
3131

3232
import java.io.IOException;
3333
import java.util.ArrayList;
3434
import java.util.Iterator;
35-
import java.util.List;
36-
import java.util.Locale;
3735
import java.util.stream.Collectors;
3836

39-
import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
40-
import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
41-
4237
public class GetDataStreamResponseTests extends AbstractResponseTestCase<GetDataStreamAction.Response, GetDataStreamResponse> {
4338

44-
private static List<Index> randomIndexInstances() {
45-
int numIndices = randomIntBetween(0, 128);
46-
List<Index> indices = new ArrayList<>(numIndices);
47-
for (int i = 0; i < numIndices; i++) {
48-
indices.add(new Index(randomAlphaOfLength(10).toLowerCase(Locale.ROOT), UUIDs.randomBase64UUID(random())));
49-
}
50-
return indices;
51-
}
52-
5339
private static DataStreamInfo randomInstance() {
54-
List<Index> indices = randomIndexInstances();
55-
long generation = indices.size() + randomLongBetween(1, 128);
56-
String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
57-
indices.add(new Index(getDefaultBackingIndexName(dataStreamName, generation), UUIDs.randomBase64UUID(random())));
58-
DataStream dataStream = new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation);
40+
DataStream dataStream = DataStreamTestHelper.randomInstance();
5941
return new DataStreamInfo(dataStream, ClusterHealthStatus.YELLOW, randomAlphaOfLengthBetween(2, 10),
6042
randomAlphaOfLengthBetween(2, 10));
6143
}

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

Lines changed: 30 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,10 @@
1818
*/
1919
package org.elasticsearch.cluster.metadata;
2020

21+
import org.elasticsearch.Version;
2122
import org.elasticsearch.cluster.AbstractDiffable;
2223
import org.elasticsearch.cluster.Diff;
24+
import org.elasticsearch.common.Nullable;
2325
import org.elasticsearch.common.ParseField;
2426
import org.elasticsearch.common.io.stream.StreamInput;
2527
import org.elasticsearch.common.io.stream.StreamOutput;
@@ -35,6 +37,7 @@
3537
import java.util.Collections;
3638
import java.util.List;
3739
import java.util.Locale;
40+
import java.util.Map;
3841
import java.util.Objects;
3942

4043
public final class DataStream extends AbstractDiffable<DataStream> implements ToXContentObject {
@@ -45,18 +48,20 @@ public final class DataStream extends AbstractDiffable<DataStream> implements To
4548
private final TimestampField timeStampField;
4649
private final List<Index> indices;
4750
private final long generation;
51+
private final Map<String, Object> metadata;
4852

49-
public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation) {
53+
public DataStream(String name, TimestampField timeStampField, List<Index> indices, long generation, Map<String, Object> metadata) {
5054
this.name = name;
5155
this.timeStampField = timeStampField;
5256
this.indices = Collections.unmodifiableList(indices);
5357
this.generation = generation;
58+
this.metadata = metadata;
5459
assert indices.size() > 0;
5560
assert indices.get(indices.size() - 1).getName().equals(getDefaultBackingIndexName(name, generation));
5661
}
5762

5863
public DataStream(String name, TimestampField timeStampField, List<Index> indices) {
59-
this(name, timeStampField, indices, indices.size());
64+
this(name, timeStampField, indices, indices.size(), null);
6065
}
6166

6267
public String getName() {
@@ -75,6 +80,11 @@ public long getGeneration() {
7580
return generation;
7681
}
7782

83+
@Nullable
84+
public Map<String, Object> getMetadata() {
85+
return metadata;
86+
}
87+
7888
/**
7989
* Performs a rollover on a {@code DataStream} instance and returns a new instance containing
8090
* the updated list of backing indices and incremented generation.
@@ -87,7 +97,7 @@ public DataStream rollover(Index newWriteIndex) {
8797
assert newWriteIndex.getName().equals(getDefaultBackingIndexName(name, generation + 1));
8898
List<Index> backingIndices = new ArrayList<>(indices);
8999
backingIndices.add(newWriteIndex);
90-
return new DataStream(name, timeStampField, backingIndices, generation + 1);
100+
return new DataStream(name, timeStampField, backingIndices, generation + 1, metadata);
91101
}
92102

93103
/**
@@ -101,7 +111,7 @@ public DataStream removeBackingIndex(Index index) {
101111
List<Index> backingIndices = new ArrayList<>(indices);
102112
backingIndices.remove(index);
103113
assert backingIndices.size() == indices.size() - 1;
104-
return new DataStream(name, timeStampField, backingIndices, generation);
114+
return new DataStream(name, timeStampField, backingIndices, generation, metadata);
105115
}
106116

107117
/**
@@ -126,7 +136,7 @@ public DataStream replaceBackingIndex(Index existingBackingIndex, Index newBacki
126136
"it is the write index", existingBackingIndex.getName(), name));
127137
}
128138
backingIndices.set(backingIndexPosition, newBackingIndex);
129-
return new DataStream(name, timeStampField, backingIndices, generation);
139+
return new DataStream(name, timeStampField, backingIndices, generation, metadata);
130140
}
131141

132142
/**
@@ -142,7 +152,8 @@ public static String getDefaultBackingIndexName(String dataStreamName, long gene
142152
}
143153

144154
public DataStream(StreamInput in) throws IOException {
145-
this(in.readString(), new TimestampField(in), in.readList(Index::new), in.readVLong());
155+
this(in.readString(), new TimestampField(in), in.readList(Index::new), in.readVLong(),
156+
in.getVersion().onOrAfter(Version.V_8_0_0) ? in.readMap(): null);
146157
}
147158

148159
public static Diff<DataStream> readDiffFrom(StreamInput in) throws IOException {
@@ -155,22 +166,28 @@ public void writeTo(StreamOutput out) throws IOException {
155166
timeStampField.writeTo(out);
156167
out.writeList(indices);
157168
out.writeVLong(generation);
169+
if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
170+
out.writeMap(metadata);
171+
}
158172
}
159173

160174
public static final ParseField NAME_FIELD = new ParseField("name");
161175
public static final ParseField TIMESTAMP_FIELD_FIELD = new ParseField("timestamp_field");
162176
public static final ParseField INDICES_FIELD = new ParseField("indices");
163177
public static final ParseField GENERATION_FIELD = new ParseField("generation");
178+
public static final ParseField METADATA_FIELD = new ParseField("_meta");
164179

165180
@SuppressWarnings("unchecked")
166181
private static final ConstructingObjectParser<DataStream, Void> PARSER = new ConstructingObjectParser<>("data_stream",
167-
args -> new DataStream((String) args[0], (TimestampField) args[1], (List<Index>) args[2], (Long) args[3]));
182+
args -> new DataStream((String) args[0], (TimestampField) args[1], (List<Index>) args[2], (Long) args[3],
183+
(Map<String, Object>) args[4]));
168184

169185
static {
170186
PARSER.declareString(ConstructingObjectParser.constructorArg(), NAME_FIELD);
171187
PARSER.declareObject(ConstructingObjectParser.constructorArg(), TimestampField.PARSER, TIMESTAMP_FIELD_FIELD);
172188
PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), (p, c) -> Index.fromXContent(p), INDICES_FIELD);
173189
PARSER.declareLong(ConstructingObjectParser.constructorArg(), GENERATION_FIELD);
190+
PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA_FIELD);
174191
}
175192

176193
public static DataStream fromXContent(XContentParser parser) throws IOException {
@@ -184,6 +201,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
184201
builder.field(TIMESTAMP_FIELD_FIELD.getPreferredName(), timeStampField);
185202
builder.field(INDICES_FIELD.getPreferredName(), indices);
186203
builder.field(GENERATION_FIELD.getPreferredName(), generation);
204+
if (metadata != null) {
205+
builder.field(METADATA_FIELD.getPreferredName(), metadata);
206+
}
187207
builder.endObject();
188208
return builder;
189209
}
@@ -196,12 +216,13 @@ public boolean equals(Object o) {
196216
return name.equals(that.name) &&
197217
timeStampField.equals(that.timeStampField) &&
198218
indices.equals(that.indices) &&
199-
generation == that.generation;
219+
generation == that.generation &&
220+
Objects.equals(metadata, that.metadata);
200221
}
201222

202223
@Override
203224
public int hashCode() {
204-
return Objects.hash(name, timeStampField, indices, generation);
225+
return Objects.hash(name, timeStampField, indices, generation, metadata);
205226
}
206227

207228
public static final class TimestampField implements Writeable, ToXContentObject {

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -162,8 +162,10 @@ static ClusterState createDataStream(MetadataCreateIndexService metadataCreateIn
162162

163163
String fieldName = template.getDataStreamTemplate().getTimestampField();
164164
DataStream.TimestampField timestampField = new DataStream.TimestampField(fieldName);
165-
DataStream newDataStream = new DataStream(request.name, timestampField,
166-
Collections.singletonList(firstBackingIndex.getIndex()));
165+
DataStream newDataStream =
166+
new DataStream(request.name, timestampField,
167+
Collections.singletonList(firstBackingIndex.getIndex()), 1L,
168+
template.metadata() != null ? Map.copyOf(template.metadata()) : null);
167169
Metadata.Builder builder = Metadata.builder(currentState.metadata()).put(newDataStream);
168170
logger.info("adding data stream [{}]", request.name);
169171
return ClusterState.builder(currentState).metadata(builder).build();

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -617,7 +617,8 @@ static DataStream updateDataStream(DataStream dataStream, Metadata.Builder metad
617617
List<Index> updatedIndices = dataStream.getIndices().stream()
618618
.map(i -> metadata.get(renameIndex(i.getName(), request, true)).getIndex())
619619
.collect(Collectors.toList());
620-
return new DataStream(dataStreamName, dataStream.getTimeStampField(), updatedIndices, dataStream.getGeneration());
620+
return new DataStream(dataStreamName, dataStream.getTimeStampField(), updatedIndices, dataStream.getGeneration(),
621+
dataStream.getMetadata());
621622
}
622623

623624
public static RestoreInProgress updateRestoreStateWithDeletedIndices(RestoreInProgress oldRestore, Set<Index> deletedIndices) {

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1728,7 +1728,7 @@ public void testIndicesAliasesRequestTargetDataStreams() {
17281728
Metadata.Builder mdBuilder = Metadata.builder()
17291729
.put(backingIndex, false)
17301730
.put(new DataStream(dataStreamName, createTimestampField("@timestamp"),
1731-
org.elasticsearch.common.collect.List.of(backingIndex.getIndex()), 1));
1731+
org.elasticsearch.common.collect.List.of(backingIndex.getIndex())));
17321732
ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(mdBuilder).build();
17331733

17341734
{
@@ -1918,7 +1918,7 @@ public void testDataStreams() {
19181918
.put(index1, false)
19191919
.put(index2, false)
19201920
.put(new DataStream(dataStreamName, createTimestampField("@timestamp"),
1921-
org.elasticsearch.common.collect.List.of(index1.getIndex(), index2.getIndex()), 2));
1921+
org.elasticsearch.common.collect.List.of(index1.getIndex(), index2.getIndex())));
19221922
ClusterState state = ClusterState.builder(new ClusterName("_name")).metadata(mdBuilder).build();
19231923

19241924
{

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

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1054,7 +1054,7 @@ public void testBuilderForDataStreamWithRandomlyNumberedBackingIndices() {
10541054
backingIndices.add(im.getIndex());
10551055
}
10561056

1057-
b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum));
1057+
b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum, null));
10581058
Metadata metadata = b.build();
10591059
assertThat(metadata.dataStreams().size(), equalTo(1));
10601060
assertThat(metadata.dataStreams().get(dataStreamName).getName(), equalTo(dataStreamName));
@@ -1072,7 +1072,7 @@ public void testBuildIndicesLookupForDataStreams() {
10721072
indices.add(idx.getIndex());
10731073
b.put(idx, true);
10741074
}
1075-
b.put(new DataStream(name, createTimestampField("@timestamp"), indices, indices.size()));
1075+
b.put(new DataStream(name, createTimestampField("@timestamp"), indices));
10761076
}
10771077

10781078
Metadata metadata = b.build();
@@ -1138,8 +1138,7 @@ public void testValidateDataStreamsThrowsExceptionOnConflict() {
11381138
DataStream dataStream = new DataStream(
11391139
dataStreamName,
11401140
createTimestampField("@timestamp"),
1141-
backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
1142-
backingIndices.size()
1141+
backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
11431142
);
11441143

11451144
IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, backingIndices);
@@ -1212,8 +1211,7 @@ public void testValidateDataStreamsAllowsPrefixedBackingIndices() {
12121211
DataStream dataStream = new DataStream(
12131212
dataStreamName,
12141213
createTimestampField("@timestamp"),
1215-
backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
1216-
backingIndices.size()
1214+
backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList())
12171215
);
12181216

12191217
IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, backingIndices);
@@ -1313,7 +1311,7 @@ private static CreateIndexResult createIndices(int numIndices, int numBackingInd
13131311
b.put(im, false);
13141312
backingIndices.add(im.getIndex());
13151313
}
1316-
b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum));
1314+
b.put(new DataStream(dataStreamName, createTimestampField("@timestamp"), backingIndices, lastBackingIndexNum, null));
13171315
return new CreateIndexResult(indices, backingIndices, b.build());
13181316
}
13191317

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

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,13 @@
3232
import java.util.ArrayList;
3333
import java.util.List;
3434
import java.util.Locale;
35+
import java.util.Map;
3536
import java.util.stream.Collectors;
3637

3738
import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
3839
import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_INDEX_UUID;
3940
import static org.elasticsearch.test.ESTestCase.randomAlphaOfLength;
41+
import static org.elasticsearch.test.ESTestCase.randomBoolean;
4042

4143
public final class DataStreamTestHelper {
4244

@@ -103,7 +105,11 @@ public static DataStream randomInstance() {
103105
long generation = indices.size() + ESTestCase.randomLongBetween(1, 128);
104106
String dataStreamName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
105107
indices.add(new Index(getDefaultBackingIndexName(dataStreamName, generation), UUIDs.randomBase64UUID(LuceneTestCase.random())));
106-
return new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation);
108+
Map<String, Object> metadata = null;
109+
if (randomBoolean()) {
110+
metadata = Map.of("key", "value");
111+
}
112+
return new DataStream(dataStreamName, createTimestampField("@timestamp"), indices, generation, metadata);
107113
}
108114

109115
/**
@@ -127,7 +133,8 @@ public static ClusterState getClusterStateWithDataStreams(List<Tuple<String, Int
127133
dsTuple.v1(),
128134
createTimestampField("@timestamp"),
129135
backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
130-
dsTuple.v2()
136+
dsTuple.v2(),
137+
null
131138
);
132139
builder.put(ds);
133140
}

0 commit comments

Comments
 (0)