Skip to content

Add write_load_forecast field to DataStreamsStats #92923

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 13 commits into from
Closed

Add write_load_forecast field to DataStreamsStats #92923

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
3912c7c
Add write_load_forecast field to DataStreamsStats
PhilKes Jan 14, 2023
45c1877
Get optional write_load_forecast only for write index of data stream
PhilKes Feb 11, 2023
abefe58
Add missing Nullable annoationt to writeLoadForecast fields
PhilKes Feb 11, 2023
1a70603
Remove unnecessary indexMetadata in DataStreamsStatsTransportAction
PhilKes Feb 11, 2023
35dbab8
Fix DataStreamsStatsTests writeLoadForecast assertions
PhilKes Feb 11, 2023
bd49c4c
Remove WriteLoadForecaster from DataStreamsStatsTransportAction
PhilKes Feb 11, 2023
0ad66b5
Fix minor formatting issues in DataStreamsStatsTransportAction
PhilKes Feb 11, 2023
680e8ad
Add null-check for writeIndex and use writeLoadForecaster
PhilKes Feb 13, 2023
dc987b6
Fix minor formatting issues
PhilKes Feb 13, 2023
6e68185
Add writeLoadForecastPerShard and writeLoadForecastPerIndex
PhilKes Feb 13, 2023
c33d67e
Add new write load fields to data-stream-stats.asciidoc + 120_data_st…
PhilKes Feb 15, 2023
b989b1d
Add write_load_forecast Test to DataStreamsStatsTests + remove unnece…
PhilKes Feb 15, 2023
3b44190
Merge branch 'main' into data-stream-write-load-stats
PhilKes May 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions ...s/src/main/java/org/elasticsearch/datastreams/action/DataStreamsStatsTransportAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardsIterator;
import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.stream.StreamInput;
Expand Down Expand Up @@ -54,15 +55,16 @@ public class DataStreamsStatsTransportAction extends TransportBroadcastByNodeAct
private final ClusterService clusterService;
private final IndicesService indicesService;
private final IndexNameExpressionResolver indexNameExpressionResolver;
private final WriteLoadForecaster writeLoadForecaster;

@Inject
public DataStreamsStatsTransportAction(
ClusterService clusterService,
TransportService transportService,
IndicesService indicesService,
ActionFilters actionFilters,
IndexNameExpressionResolver indexNameExpressionResolver
) {
IndexNameExpressionResolver indexNameExpressionResolver,
WriteLoadForecaster writeLoadForecaster) {
super(
DataStreamsStatsAction.NAME,
clusterService,
Expand All @@ -75,6 +77,7 @@ public DataStreamsStatsTransportAction(
this.clusterService = clusterService;
this.indicesService = indicesService;
this.indexNameExpressionResolver = indexNameExpressionResolver;
this.writeLoadForecaster = writeLoadForecaster;
}

@Override
Expand Down Expand Up @@ -158,7 +161,10 @@ protected void shardOperation(
maxTimestamp = LongPoint.decodeDimension(maxPackedValue, 0);
}
}
return new DataStreamsStatsAction.DataStreamShardStats(indexShard.routingEntry(), storeStats, maxTimestamp);
var indexMetadata = indexService.getMetadata();
assert indexMetadata != null;
double writeLoadForecast = writeLoadForecaster.getForecastedWriteLoad(indexMetadata).orElse(0.0);
Copy link
Author

@PhilKes PhilKes Jan 14, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was not entirely sure if this usage of the WriteLoadForecaster is correct, can somebody verify that?

Copy link
Contributor

@DaveCTurner DaveCTurner Feb 5, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The usage is ok except:

  • we should preserve empty rather than mapping it to 0. The difference between "write load is zero" and "write load is missing" is important.

  • there's no need to do this for each shard. Instead we should extract the write load from the index metadata in getResponseFactory.

  • we only want the write load from the write index of the data stream

Copy link
Contributor

@DaveCTurner DaveCTurner Feb 5, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NB edited^

Copy link
Author

@PhilKes PhilKes Feb 11, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @DaveCTurner for your feedback, I fixed the things you mentioned, writeLoadForecast is now Nullable and extracted only from the write index of the data stream

return new DataStreamsStatsAction.DataStreamShardStats(indexShard.routingEntry(), storeStats, maxTimestamp, writeLoadForecast);
});
}

Expand Down Expand Up @@ -240,6 +246,7 @@ public DataStreamsStatsAction.Response newResponse(
AggregatedStats stats = aggregatedDataStreamsStats.computeIfAbsent(dataStream.getName(), s -> new AggregatedStats());
stats.storageBytes += shardStat.getStoreStats().sizeInBytes();
stats.maxTimestamp = Math.max(stats.maxTimestamp, shardStat.getMaxTimestamp());
stats.writeLoadForecast += shardStat.getWriteLoadForecast();
}

DataStreamsStatsAction.DataStreamStats[] dataStreamStats = aggregatedDataStreamsStats.entrySet()
Expand All @@ -249,8 +256,8 @@ public DataStreamsStatsAction.Response newResponse(
entry.getKey(),
entry.getValue().backingIndices.size(),
ByteSizeValue.ofBytes(entry.getValue().storageBytes),
entry.getValue().maxTimestamp
)
entry.getValue().maxTimestamp,
entry.getValue().writeLoadForecast)
)
.toArray(DataStreamsStatsAction.DataStreamStats[]::new);

Expand All @@ -271,5 +278,6 @@ private static class AggregatedStats {
Set<String> backingIndices = new HashSet<>();
long storageBytes = 0L;
long maxTimestamp = 0L;
double writeLoadForecast = 0.0;
}
}
9 changes: 9 additions & 0 deletions modules/data-streams/src/test/java/org/elasticsearch/datastreams/DataStreamsStatsTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@
import java.util.concurrent.TimeUnit;

import static java.lang.Math.max;
import static org.hamcrest.Matchers.closeTo;
import static org.hamcrest.Matchers.is;

public class DataStreamsStatsTests extends ESSingleNodeTestCase {

Expand Down Expand Up @@ -86,6 +88,7 @@ public void testStatsEmptyDataStream() throws Exception {
assertEquals(0L, stats.getDataStreams()[0].getMaximumTimestamp());
assertNotEquals(0L, stats.getDataStreams()[0].getStoreSize().getBytes());
assertEquals(stats.getTotalStoreSize().getBytes(), stats.getDataStreams()[0].getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));
}

public void testStatsExistingDataStream() throws Exception {
Expand All @@ -104,6 +107,7 @@ public void testStatsExistingDataStream() throws Exception {
assertEquals(timestamp, stats.getDataStreams()[0].getMaximumTimestamp());
assertNotEquals(0L, stats.getDataStreams()[0].getStoreSize().getBytes());
assertEquals(stats.getTotalStoreSize().getBytes(), stats.getDataStreams()[0].getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));
}

public void testStatsExistingHiddenDataStream() throws Exception {
Expand All @@ -122,6 +126,7 @@ public void testStatsExistingHiddenDataStream() throws Exception {
assertEquals(timestamp, stats.getDataStreams()[0].getMaximumTimestamp());
assertNotEquals(0L, stats.getDataStreams()[0].getStoreSize().getBytes());
assertEquals(stats.getTotalStoreSize().getBytes(), stats.getDataStreams()[0].getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));
}

public void testStatsClosedBackingIndexDataStream() throws Exception {
Expand Down Expand Up @@ -153,6 +158,7 @@ public void testStatsClosedBackingIndexDataStream() throws Exception {
assertEquals(0L, stats.getDataStreams()[0].getMaximumTimestamp());
assertNotEquals(0L, stats.getDataStreams()[0].getStoreSize().getBytes());
assertEquals(stats.getTotalStoreSize().getBytes(), stats.getDataStreams()[0].getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));

// Call stats again after writing a new event into the write index
long timestamp = createDocument(dataStreamName);
Expand All @@ -169,6 +175,7 @@ public void testStatsClosedBackingIndexDataStream() throws Exception {
assertEquals(timestamp, stats.getDataStreams()[0].getMaximumTimestamp());
assertNotEquals(0L, stats.getDataStreams()[0].getStoreSize().getBytes());
assertEquals(stats.getTotalStoreSize().getBytes(), stats.getDataStreams()[0].getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));
}

public void testStatsRolledDataStream() throws Exception {
Expand All @@ -189,6 +196,7 @@ public void testStatsRolledDataStream() throws Exception {
assertEquals(timestamp, stats.getDataStreams()[0].getMaximumTimestamp());
assertNotEquals(0L, stats.getDataStreams()[0].getStoreSize().getBytes());
assertEquals(stats.getTotalStoreSize().getBytes(), stats.getDataStreams()[0].getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));
}

public void testStatsMultipleDataStreams() throws Exception {
Expand Down Expand Up @@ -220,6 +228,7 @@ public void testStatsMultipleDataStreams() throws Exception {
assertEquals(1, dataStreamStats.getBackingIndices());
assertEquals(expectedMaxTS.longValue(), dataStreamStats.getMaximumTimestamp());
assertNotEquals(0L, dataStreamStats.getStoreSize().getBytes());
assertThat(0.0, is(closeTo(stats.getDataStreams()[0].getWriteLoadForecast(), 1e-8)));
}
}

Expand Down
5 changes: 3 additions & 2 deletions ...ams/src/test/java/org/elasticsearch/datastreams/action/DataStreamsStatsResponseTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,13 +42,14 @@ public static DataStreamsStatsAction.Response randomStatsResponse() {
long storeSize = randomLongBetween(250, 1000000000);
totalStoreSize += storeSize;
long maximumTimestamp = randomRecentTimestamp();
double writeLoadForecast = randomDouble();
dataStreamStats.add(
new DataStreamsStatsAction.DataStreamStats(
dataStreamName,
backingIndices,
ByteSizeValue.ofBytes(storeSize),
maximumTimestamp
)
maximumTimestamp,
writeLoadForecast)
);
}
int totalShards = randomIntBetween(backingIndicesTotal, backingIndicesTotal * 3);
Expand Down
27 changes: 24 additions & 3 deletions server/src/main/java/org/elasticsearch/action/datastreams/DataStreamsStatsAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,19 +153,22 @@ public static class DataStreamStats implements ToXContentObject, Writeable {
private final int backingIndices;
private final ByteSizeValue storeSize;
private final long maximumTimestamp;
private final double writeLoadForecast;

public DataStreamStats(String dataStream, int backingIndices, ByteSizeValue storeSize, long maximumTimestamp) {
public DataStreamStats(String dataStream, int backingIndices, ByteSizeValue storeSize, long maximumTimestamp, double writeLoadForecast) {
this.dataStream = dataStream;
this.backingIndices = backingIndices;
this.storeSize = storeSize;
this.maximumTimestamp = maximumTimestamp;
this.writeLoadForecast = writeLoadForecast;
}

public DataStreamStats(StreamInput in) throws IOException {
this.dataStream = in.readString();
this.backingIndices = in.readVInt();
this.storeSize = ByteSizeValue.readFrom(in);
this.maximumTimestamp = in.readVLong();
this.writeLoadForecast = in.readDouble();
}

@Override
Expand All @@ -174,6 +177,7 @@ public void writeTo(StreamOutput out) throws IOException {
out.writeVInt(backingIndices);
storeSize.writeTo(out);
out.writeVLong(maximumTimestamp);
out.writeDouble(writeLoadForecast);
}

@Override
Expand All @@ -183,6 +187,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
builder.field("backing_indices", backingIndices);
builder.humanReadableField("store_size_bytes", "store_size", storeSize);
builder.field("maximum_timestamp", maximumTimestamp);
builder.field("write_load_forecast", writeLoadForecast);
builder.endObject();
return builder;
}
Expand All @@ -203,6 +208,10 @@ public long getMaximumTimestamp() {
return maximumTimestamp;
}

public double getWriteLoadForecast() {
return writeLoadForecast;
}

@Override
public boolean equals(Object obj) {
if (this == obj) {
Expand All @@ -214,13 +223,14 @@ public boolean equals(Object obj) {
DataStreamStats that = (DataStreamStats) obj;
return backingIndices == that.backingIndices
&& maximumTimestamp == that.maximumTimestamp
&& writeLoadForecast == that.writeLoadForecast
&& Objects.equals(dataStream, that.dataStream)
&& Objects.equals(storeSize, that.storeSize);
}

@Override
public int hashCode() {
return Objects.hash(dataStream, backingIndices, storeSize, maximumTimestamp);
return Objects.hash(dataStream, backingIndices, storeSize, maximumTimestamp, writeLoadForecast);
}

@Override
Expand All @@ -235,6 +245,8 @@ public String toString() {
+ storeSize
+ ", maximumTimestamp="
+ maximumTimestamp
+ ", writeLoadForecast="
+ writeLoadForecast
+ '}';
}
}
Expand All @@ -243,24 +255,29 @@ public static class DataStreamShardStats implements Writeable {
private final ShardRouting shardRouting;
private final StoreStats storeStats;
private final long maxTimestamp;
private final double writeLoadForecast;


public DataStreamShardStats(ShardRouting shardRouting, StoreStats storeStats, long maxTimestamp) {
public DataStreamShardStats(ShardRouting shardRouting, StoreStats storeStats, long maxTimestamp, double writeLoadForecast) {
this.shardRouting = shardRouting;
this.storeStats = storeStats;
this.maxTimestamp = maxTimestamp;
this.writeLoadForecast = writeLoadForecast;
}

public DataStreamShardStats(StreamInput in) throws IOException {
this.shardRouting = new ShardRouting(in);
this.storeStats = new StoreStats(in);
this.maxTimestamp = in.readVLong();
this.writeLoadForecast = in.readDouble();
}

@Override
public void writeTo(StreamOutput out) throws IOException {
shardRouting.writeTo(out);
storeStats.writeTo(out);
out.writeVLong(maxTimestamp);
out.writeDouble(writeLoadForecast);
}

public ShardRouting getShardRouting() {
Expand All @@ -274,6 +291,10 @@ public StoreStats getStoreStats() {
public long getMaxTimestamp() {
return maxTimestamp;
}

public double getWriteLoadForecast() {
return writeLoadForecast;
}
}

}