Skip to content

Commit f14da48

Browse files
martijnvgmartijnvg
authored
Add time series related information to get data stream API (#86395)
In case if a data stream is a time series data stream then include time series information. This includes the continuous temporal ranges a time series data stream encapsulates. This is computed based on combing the index.time_series.start_time and index.time_series.end_time ranges of all backing indices of a time series data stream Closes #83518
1 parent fae5099 commit f14da48

File tree

5 files changed

+221
-12
lines changed

5 files changed

+221
-12
lines changed

‎modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/GetDataStreamsTransportAction.java‎

Lines changed: 58 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,18 +18,23 @@
1818
import org.elasticsearch.cluster.block.ClusterBlockLevel;
1919
import org.elasticsearch.cluster.health.ClusterStateHealth;
2020
import org.elasticsearch.cluster.metadata.DataStream;
21+
import org.elasticsearch.cluster.metadata.IndexMetadata;
2122
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
2223
import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
2324
import org.elasticsearch.cluster.service.ClusterService;
2425
import org.elasticsearch.common.inject.Inject;
2526
import org.elasticsearch.common.settings.Settings;
27+
import org.elasticsearch.core.Tuple;
2628
import org.elasticsearch.index.Index;
29+
import org.elasticsearch.index.IndexMode;
30+
import org.elasticsearch.index.IndexSettings;
2731
import org.elasticsearch.indices.SystemDataStreamDescriptor;
2832
import org.elasticsearch.indices.SystemIndices;
2933
import org.elasticsearch.tasks.Task;
3034
import org.elasticsearch.threadpool.ThreadPool;
3135
import org.elasticsearch.transport.TransportService;
3236

37+
import java.time.Instant;
3338
import java.util.ArrayList;
3439
import java.util.Comparator;
3540
import java.util.List;
@@ -72,6 +77,15 @@ protected void masterOperation(
7277
ClusterState state,
7378
ActionListener<GetDataStreamAction.Response> listener
7479
) throws Exception {
80+
listener.onResponse(innerOperation(state, request, indexNameExpressionResolver, systemIndices));
81+
}
82+
83+
static GetDataStreamAction.Response innerOperation(
84+
ClusterState state,
85+
GetDataStreamAction.Request request,
86+
IndexNameExpressionResolver indexNameExpressionResolver,
87+
SystemIndices systemIndices
88+
) {
7589
List<DataStream> dataStreams = getDataStreams(state, indexNameExpressionResolver, request);
7690
List<GetDataStreamAction.Response.DataStreamInfo> dataStreamInfos = new ArrayList<>(dataStreams.size());
7791
for (DataStream dataStream : dataStreams) {
@@ -105,11 +119,53 @@ protected void masterOperation(
105119
state,
106120
dataStream.getIndices().stream().map(Index::getName).toArray(String[]::new)
107121
);
122+
123+
GetDataStreamAction.Response.TimeSeries timeSeries = null;
124+
if (dataStream.getIndexMode() == IndexMode.TIME_SERIES) {
125+
List<Tuple<Instant, Instant>> ranges = new ArrayList<>();
126+
Tuple<Instant, Instant> current = null;
127+
for (Index index : dataStream.getIndices()) {
128+
IndexMetadata metadata = state.getMetadata().index(index);
129+
Instant start = IndexSettings.TIME_SERIES_START_TIME.get(metadata.getSettings());
130+
Instant end = IndexSettings.TIME_SERIES_END_TIME.get(metadata.getSettings());
131+
if (current == null) {
132+
current = new Tuple<>(start, end);
133+
} else if (current.v2().compareTo(start) == 0) {
134+
current = new Tuple<>(current.v1(), end);
135+
} else if (current.v2().compareTo(start) < 0) {
136+
ranges.add(current);
137+
current = new Tuple<>(start, end);
138+
} else {
139+
String message = "previous backing index ["
140+
+ current.v1()
141+
+ "/"
142+
+ current.v2()
143+
+ "] range is colliding with current backing index range ["
144+
+ start
145+
+ "/"
146+
+ end
147+
+ "]";
148+
assert current.v2().compareTo(start) < 0 : message;
149+
LOGGER.warn(message);
150+
}
151+
}
152+
if (current != null) {
153+
ranges.add(current);
154+
}
155+
timeSeries = new GetDataStreamAction.Response.TimeSeries(ranges);
156+
}
157+
108158
dataStreamInfos.add(
109-
new GetDataStreamAction.Response.DataStreamInfo(dataStream, streamHealth.getStatus(), indexTemplate, ilmPolicyName)
159+
new GetDataStreamAction.Response.DataStreamInfo(
160+
dataStream,
161+
streamHealth.getStatus(),
162+
indexTemplate,
163+
ilmPolicyName,
164+
timeSeries
165+
)
110166
);
111167
}
112-
listener.onResponse(new GetDataStreamAction.Response(dataStreamInfos));
168+
return new GetDataStreamAction.Response(dataStreamInfos);
113169
}
114170

115171
static List<DataStream> getDataStreams(

‎modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsResponseTests.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,10 @@
1111
import org.elasticsearch.cluster.health.ClusterHealthStatus;
1212
import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
1313
import org.elasticsearch.common.io.stream.Writeable;
14+
import org.elasticsearch.core.Tuple;
1415
import org.elasticsearch.test.AbstractWireSerializingTestCase;
1516

17+
import java.time.Instant;
1618
import java.util.ArrayList;
1719
import java.util.List;
1820

@@ -28,12 +30,22 @@ protected Response createTestInstance() {
2830
int numDataStreams = randomIntBetween(0, 8);
2931
List<Response.DataStreamInfo> dataStreams = new ArrayList<>();
3032
for (int i = 0; i < numDataStreams; i++) {
33+
List<Tuple<Instant, Instant>> timeSeries = null;
34+
if (randomBoolean()) {
35+
timeSeries = new ArrayList<>();
36+
int numTimeSeries = randomIntBetween(0, 3);
37+
for (int j = 0; j < numTimeSeries; j++) {
38+
timeSeries.add(new Tuple<>(Instant.now(), Instant.now()));
39+
}
40+
}
41+
3142
dataStreams.add(
3243
new Response.DataStreamInfo(
3344
DataStreamTestHelper.randomInstance(),
3445
ClusterHealthStatus.GREEN,
3546
randomAlphaOfLengthBetween(2, 10),
36-
randomAlphaOfLengthBetween(2, 10)
47+
randomAlphaOfLengthBetween(2, 10),
48+
timeSeries != null ? new Response.TimeSeries(timeSeries) : null
3749
)
3850
);
3951
}

‎modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsTransportActionTests.java‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,21 +11,30 @@
1111
import org.elasticsearch.cluster.ClusterName;
1212
import org.elasticsearch.cluster.ClusterState;
1313
import org.elasticsearch.cluster.metadata.DataStream;
14+
import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
1415
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
16+
import org.elasticsearch.cluster.metadata.Metadata;
1517
import org.elasticsearch.core.Tuple;
1618
import org.elasticsearch.index.IndexNotFoundException;
19+
import org.elasticsearch.indices.SystemIndices;
1720
import org.elasticsearch.indices.TestIndexNameExpressionResolver;
1821
import org.elasticsearch.test.ESTestCase;
1922

23+
import java.time.Instant;
24+
import java.time.temporal.ChronoUnit;
2025
import java.util.List;
26+
import java.util.Map;
2127

2228
import static org.elasticsearch.cluster.metadata.DataStreamTestHelper.getClusterStateWithDataStreams;
29+
import static org.hamcrest.Matchers.contains;
2330
import static org.hamcrest.Matchers.containsString;
2431
import static org.hamcrest.Matchers.equalTo;
32+
import static org.hamcrest.Matchers.hasSize;
2533

2634
public class GetDataStreamsTransportActionTests extends ESTestCase {
2735

2836
private final IndexNameExpressionResolver resolver = TestIndexNameExpressionResolver.newInstance();
37+
private final SystemIndices systemIndices = new SystemIndices(Map.of());
2938

3039
public void testGetDataStream() {
3140
final String dataStreamName = "my-data-stream";
@@ -107,4 +116,64 @@ public void testGetNonexistentDataStream() {
107116
assertThat(e.getMessage(), containsString("no such index [" + dataStreamName + "]"));
108117
}
109118

119+
public void testGetTimeSeriesDataStream() {
120+
Instant now = Instant.now().truncatedTo(ChronoUnit.SECONDS);
121+
String dataStream1 = "ds-1";
122+
String dataStream2 = "ds-2";
123+
Instant sixHoursAgo = now.minus(6, ChronoUnit.HOURS);
124+
Instant fourHoursAgo = now.minus(4, ChronoUnit.HOURS);
125+
Instant twoHoursAgo = now.minus(2, ChronoUnit.HOURS);
126+
Instant twoHoursAhead = now.plus(2, ChronoUnit.HOURS);
127+
128+
ClusterState state;
129+
{
130+
var mBuilder = new Metadata.Builder();
131+
DataStreamTestHelper.getClusterStateWithDataStream(
132+
mBuilder,
133+
dataStream1,
134+
List.of(
135+
new Tuple<>(sixHoursAgo, fourHoursAgo),
136+
new Tuple<>(fourHoursAgo, twoHoursAgo),
137+
new Tuple<>(twoHoursAgo, twoHoursAhead)
138+
)
139+
);
140+
DataStreamTestHelper.getClusterStateWithDataStream(
141+
mBuilder,
142+
dataStream2,
143+
List.of(
144+
new Tuple<>(sixHoursAgo, fourHoursAgo),
145+
new Tuple<>(fourHoursAgo, twoHoursAgo),
146+
new Tuple<>(twoHoursAgo, twoHoursAhead)
147+
)
148+
);
149+
state = ClusterState.builder(new ClusterName("_name")).metadata(mBuilder).build();
150+
}
151+
152+
var req = new GetDataStreamAction.Request(new String[] {});
153+
var response = GetDataStreamsTransportAction.innerOperation(state, req, resolver, systemIndices);
154+
assertThat(response.getDataStreams(), hasSize(2));
155+
assertThat(response.getDataStreams().get(0).getDataStream().getName(), equalTo(dataStream1));
156+
assertThat(response.getDataStreams().get(0).getTimeSeries().temporalRanges(), contains(new Tuple<>(sixHoursAgo, twoHoursAhead)));
157+
assertThat(response.getDataStreams().get(1).getDataStream().getName(), equalTo(dataStream2));
158+
assertThat(response.getDataStreams().get(1).getTimeSeries().temporalRanges(), contains(new Tuple<>(sixHoursAgo, twoHoursAhead)));
159+
160+
// Remove the middle backing index first data stream, so that there is time gap in the data stream:
161+
{
162+
Metadata.Builder mBuilder = Metadata.builder(state.getMetadata());
163+
DataStream dataStream = state.getMetadata().dataStreams().get(dataStream1);
164+
mBuilder.put(dataStream.removeBackingIndex(dataStream.getIndices().get(1)));
165+
mBuilder.remove(dataStream.getIndices().get(1).getName());
166+
state = ClusterState.builder(state).metadata(mBuilder).build();
167+
}
168+
response = GetDataStreamsTransportAction.innerOperation(state, req, resolver, systemIndices);
169+
assertThat(response.getDataStreams(), hasSize(2));
170+
assertThat(response.getDataStreams().get(0).getDataStream().getName(), equalTo(dataStream1));
171+
assertThat(
172+
response.getDataStreams().get(0).getTimeSeries().temporalRanges(),
173+
contains(new Tuple<>(sixHoursAgo, fourHoursAgo), new Tuple<>(twoHoursAgo, twoHoursAhead))
174+
);
175+
assertThat(response.getDataStreams().get(1).getDataStream().getName(), equalTo(dataStream2));
176+
assertThat(response.getDataStreams().get(1).getTimeSeries().temporalRanges(), contains(new Tuple<>(sixHoursAgo, twoHoursAhead)));
177+
}
178+
110179
}

‎modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/150_tsdb.yml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,8 @@ created the data stream:
103103
- match: { data_streams.0.template: 'my-template1' }
104104
- match: { data_streams.0.hidden: false }
105105
- match: { data_streams.0.system: false }
106+
- match: { data_streams.0.time_series.temporal_ranges.0.start: 2021-04-28T00:00:00.000Z }
107+
- match: { data_streams.0.time_series.temporal_ranges.0.end: 2021-04-29T00:00:00.000Z }
106108
- set: { data_streams.0.indices.0.index_name: backing_index }
107109

108110
- do:

‎server/src/main/java/org/elasticsearch/action/datastreams/GetDataStreamAction.java‎

Lines changed: 79 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
*/
88
package org.elasticsearch.action.datastreams;
99

10+
import org.elasticsearch.Version;
1011
import org.elasticsearch.action.ActionRequestValidationException;
1112
import org.elasticsearch.action.ActionResponse;
1213
import org.elasticsearch.action.ActionType;
@@ -18,12 +19,16 @@
1819
import org.elasticsearch.cluster.metadata.DataStream;
1920
import org.elasticsearch.common.io.stream.StreamInput;
2021
import org.elasticsearch.common.io.stream.StreamOutput;
22+
import org.elasticsearch.common.io.stream.Writeable;
2123
import org.elasticsearch.core.Nullable;
24+
import org.elasticsearch.core.Tuple;
25+
import org.elasticsearch.index.mapper.DateFieldMapper;
2226
import org.elasticsearch.xcontent.ParseField;
2327
import org.elasticsearch.xcontent.ToXContentObject;
2428
import org.elasticsearch.xcontent.XContentBuilder;
2529

2630
import java.io.IOException;
31+
import java.time.Instant;
2732
import java.util.Arrays;
2833
import java.util.List;
2934
import java.util.Objects;
@@ -122,28 +127,42 @@ public static class DataStreamInfo implements SimpleDiffable<DataStreamInfo>, To
122127
public static final ParseField SYSTEM_FIELD = new ParseField("system");
123128
public static final ParseField ALLOW_CUSTOM_ROUTING = new ParseField("allow_custom_routing");
124129
public static final ParseField REPLICATED = new ParseField("replicated");
130+
public static final ParseField TIME_SERIES = new ParseField("time_series");
131+
public static final ParseField TEMPORAL_RANGES = new ParseField("temporal_ranges");
132+
public static final ParseField TEMPORAL_RANGE_START = new ParseField("start");
133+
public static final ParseField TEMPORAL_RANGE_END = new ParseField("end");
125134

126-
DataStream dataStream;
127-
ClusterHealthStatus dataStreamStatus;
135+
private final DataStream dataStream;
136+
private final ClusterHealthStatus dataStreamStatus;
128137
@Nullable
129-
String indexTemplate;
138+
private final String indexTemplate;
130139
@Nullable
131-
String ilmPolicyName;
140+
private final String ilmPolicyName;
141+
@Nullable
142+
private final TimeSeries timeSeries;
132143

133144
public DataStreamInfo(
134145
DataStream dataStream,
135146
ClusterHealthStatus dataStreamStatus,
136147
@Nullable String indexTemplate,
137-
@Nullable String ilmPolicyName
148+
@Nullable String ilmPolicyName,
149+
@Nullable TimeSeries timeSeries
138150
) {
139151
this.dataStream = dataStream;
140152
this.dataStreamStatus = dataStreamStatus;
141153
this.indexTemplate = indexTemplate;
142154
this.ilmPolicyName = ilmPolicyName;
155+
this.timeSeries = timeSeries;
143156
}
144157

145-
public DataStreamInfo(StreamInput in) throws IOException {
146-
this(new DataStream(in), ClusterHealthStatus.readFrom(in), in.readOptionalString(), in.readOptionalString());
158+
DataStreamInfo(StreamInput in) throws IOException {
159+
this(
160+
new DataStream(in),
161+
ClusterHealthStatus.readFrom(in),
162+
in.readOptionalString(),
163+
in.readOptionalString(),
164+
in.getVersion().onOrAfter(Version.V_8_3_0) ? in.readOptionalWriteable(TimeSeries::new) : null
165+
);
147166
}
148167

149168
public DataStream getDataStream() {
@@ -164,12 +183,20 @@ public String getIlmPolicy() {
164183
return ilmPolicyName;
165184
}
166185

186+
@Nullable
187+
public TimeSeries getTimeSeries() {
188+
return timeSeries;
189+
}
190+
167191
@Override
168192
public void writeTo(StreamOutput out) throws IOException {
169193
dataStream.writeTo(out);
170194
dataStreamStatus.writeTo(out);
171195
out.writeOptionalString(indexTemplate);
172196
out.writeOptionalString(ilmPolicyName);
197+
if (out.getVersion().onOrAfter(Version.V_8_3_0)) {
198+
out.writeOptionalWriteable(timeSeries);
199+
}
173200
}
174201

175202
@Override
@@ -193,6 +220,20 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
193220
builder.field(SYSTEM_FIELD.getPreferredName(), dataStream.isSystem());
194221
builder.field(ALLOW_CUSTOM_ROUTING.getPreferredName(), dataStream.isAllowCustomRouting());
195222
builder.field(REPLICATED.getPreferredName(), dataStream.isReplicated());
223+
if (timeSeries != null) {
224+
builder.startObject(TIME_SERIES.getPreferredName());
225+
builder.startArray(TEMPORAL_RANGES.getPreferredName());
226+
for (var range : timeSeries.temporalRanges()) {
227+
builder.startObject();
228+
Instant start = range.v1();
229+
builder.field(TEMPORAL_RANGE_START.getPreferredName(), DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.format(start));
230+
Instant end = range.v2();
231+
builder.field(TEMPORAL_RANGE_END.getPreferredName(), DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.format(end));
232+
builder.endObject();
233+
}
234+
builder.endArray();
235+
builder.endObject();
236+
}
196237
builder.endObject();
197238
return builder;
198239
}
@@ -205,12 +246,41 @@ public boolean equals(Object o) {
205246
return dataStream.equals(that.dataStream)
206247
&& dataStreamStatus == that.dataStreamStatus
207248
&& Objects.equals(indexTemplate, that.indexTemplate)
208-
&& Objects.equals(ilmPolicyName, that.ilmPolicyName);
249+
&& Objects.equals(ilmPolicyName, that.ilmPolicyName)
250+
&& Objects.equals(timeSeries, that.timeSeries);
251+
}
252+
253+
@Override
254+
public int hashCode() {
255+
return Objects.hash(dataStream, dataStreamStatus, indexTemplate, ilmPolicyName, timeSeries);
256+
}
257+
}
258+
259+
public static record TimeSeries(List<Tuple<Instant, Instant>> temporalRanges) implements Writeable {
260+
261+
TimeSeries(StreamInput in) throws IOException {
262+
this(in.readList(in1 -> new Tuple<>(in1.readInstant(), in1.readInstant())));
263+
}
264+
265+
@Override
266+
public void writeTo(StreamOutput out) throws IOException {
267+
out.writeCollection(temporalRanges, (out1, value) -> {
268+
out1.writeInstant(value.v1());
269+
out1.writeInstant(value.v2());
270+
});
271+
}
272+
273+
@Override
274+
public boolean equals(Object o) {
275+
if (this == o) return true;
276+
if (o == null || getClass() != o.getClass()) return false;
277+
TimeSeries that = (TimeSeries) o;
278+
return temporalRanges.equals(that.temporalRanges);
209279
}
210280

211281
@Override
212282
public int hashCode() {
213-
return Objects.hash(dataStream, dataStreamStatus, indexTemplate, ilmPolicyName);
283+
return Objects.hash(temporalRanges);
214284
}
215285
}
216286

0 commit comments

Comments
 (0)