Add shard write-load to cluster info #131496
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -41,7 +41,7 @@ | |
|
|
||
| /** | ||
| * ClusterInfo is an object representing a map of nodes to {@link DiskUsage} | ||
| * and a map of shard ids to shard sizes and shard write-loads, see | ||
| * <code>InternalClusterInfoService.shardIdentifierFromRouting(String)</code> | ||
| * for the key used in the shardSizes map | ||
| */ | ||
|
|
@@ -59,9 +59,10 @@ public class ClusterInfo implements ChunkedToXContent, Writeable { | |
| final Map<NodeAndPath, ReservedSpace> reservedSpace; | ||
|
There was a problem hiding this comment. The class comment further up ^ could use an update at this point. There was a problem hiding this comment. Fixed in cfe5759 |
||
| final Map<String, EstimatedHeapUsage> estimatedHeapUsages; | ||
| final Map<String, NodeUsageStatsForThreadPools> nodeUsageStatsForThreadPools; | ||
| final Map<ShardId, Double> shardWriteLoads; | ||
|
|
||
| protected ClusterInfo() { | ||
| this(Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of()); | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -85,7 +86,8 @@ public ClusterInfo( | |
| Map<NodeAndShard, String> dataPath, | ||
| Map<NodeAndPath, ReservedSpace> reservedSpace, | ||
| Map<String, EstimatedHeapUsage> estimatedHeapUsages, | ||
| Map<String, NodeUsageStatsForThreadPools> nodeUsageStatsForThreadPools, | ||
| Map<ShardId, Double> shardWriteLoads | ||
| ) { | ||
| this.leastAvailableSpaceUsage = Map.copyOf(leastAvailableSpaceUsage); | ||
| this.mostAvailableSpaceUsage = Map.copyOf(mostAvailableSpaceUsage); | ||
|
|
@@ -95,6 +97,7 @@ public ClusterInfo( | |
| this.reservedSpace = Map.copyOf(reservedSpace); | ||
| this.estimatedHeapUsages = Map.copyOf(estimatedHeapUsages); | ||
| this.nodeUsageStatsForThreadPools = Map.copyOf(nodeUsageStatsForThreadPools); | ||
| this.shardWriteLoads = Map.copyOf(shardWriteLoads); | ||
| } | ||
|
|
||
| public ClusterInfo(StreamInput in) throws IOException { | ||
|
|
@@ -116,6 +119,11 @@ public ClusterInfo(StreamInput in) throws IOException { | |
| } else { | ||
| this.nodeUsageStatsForThreadPools = Map.of(); | ||
| } | ||
| if (in.getTransportVersion().onOrAfter(TransportVersions.SHARD_WRITE_LOAD_IN_CLUSTER_INFO)) { | ||
| this.shardWriteLoads = in.readImmutableMap(ShardId::new, StreamInput::readDouble); | ||
| } else { | ||
| this.shardWriteLoads = Map.of(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -136,6 +144,9 @@ public void writeTo(StreamOutput out) throws IOException { | |
| if (out.getTransportVersion().onOrAfter(TransportVersions.NODE_USAGE_STATS_FOR_THREAD_POOLS_IN_CLUSTER_INFO)) { | ||
| out.writeMap(this.nodeUsageStatsForThreadPools, StreamOutput::writeWriteable); | ||
| } | ||
| if (out.getTransportVersion().onOrAfter(TransportVersions.SHARD_WRITE_LOAD_IN_CLUSTER_INFO)) { | ||
| out.writeMap(this.shardWriteLoads, StreamOutput::writeWriteable, StreamOutput::writeDouble); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -216,7 +227,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params | |
| return builder.endObject(); // NodeAndPath | ||
| }), | ||
| endArray() // end "reserved_sizes" | ||
| // NOTE: We don't serialize estimatedHeapUsages/nodeUsageStatsForThreadPools/shardWriteLoads at this stage, to avoid | ||
| // committing to API payloads until the features are settled | ||
| ); | ||
| } | ||
|
|
@@ -255,6 +266,16 @@ public Map<String, DiskUsage> getNodeMostAvailableDiskUsages() { | |
| return this.mostAvailableSpaceUsage; | ||
| } | ||
|
|
||
| /** | ||
| * Returns a map of shard IDs to the write-loads for use in balancing. The write-loads can be interpreted | ||
| * as the average number of threads that ingestion to the shard will consume. | ||
| * This information may be partial or missing altogether under some circumstances. The absence of a shard | ||
| * write load from the map should be interpreted as "unknown". | ||
| */ | ||
| public Map<ShardId, Double> getShardWriteLoads() { | ||
| return shardWriteLoads; | ||
| } | ||
|
|
||
|
There was a problem hiding this comment. update the There was a problem hiding this comment. I don't like that we add equals and hashCode just for testing, but I updated them in cd3169f |
||
| /** | ||
| * Returns the shard size for the given shardId or <code>null</code> if that metric is not available. | ||
| */ | ||
|
|
@@ -331,7 +352,9 @@ public boolean equals(Object o) { | |
| && shardDataSetSizes.equals(that.shardDataSetSizes) | ||
| && dataPath.equals(that.dataPath) | ||
| && reservedSpace.equals(that.reservedSpace) | ||
| && estimatedHeapUsages.equals(that.estimatedHeapUsages) | ||
| && nodeUsageStatsForThreadPools.equals(that.nodeUsageStatsForThreadPools) | ||
| && shardWriteLoads.equals(that.shardWriteLoads); | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -343,7 +366,9 @@ public int hashCode() { | |
| shardDataSetSizes, | ||
| dataPath, | ||
| reservedSpace, | ||
| estimatedHeapUsages, | ||
| nodeUsageStatsForThreadPools, | ||
| shardWriteLoads | ||
| ); | ||
| } | ||
|
|
||
|
|
@@ -466,6 +491,7 @@ public static class Builder { | |
| private Map<NodeAndPath, ReservedSpace> reservedSpace = Map.of(); | ||
| private Map<String, EstimatedHeapUsage> estimatedHeapUsages = Map.of(); | ||
| private Map<String, NodeUsageStatsForThreadPools> nodeUsageStatsForThreadPools = Map.of(); | ||
| private Map<ShardId, Double> shardWriteLoads = Map.of(); | ||
|
|
||
| public ClusterInfo build() { | ||
| return new ClusterInfo( | ||
|
|
@@ -476,7 +502,8 @@ public ClusterInfo build() { | |
| dataPath, | ||
| reservedSpace, | ||
| estimatedHeapUsages, | ||
| nodeUsageStatsForThreadPools, | ||
| shardWriteLoads | ||
| ); | ||
| } | ||
|
|
||
|
|
@@ -519,5 +546,10 @@ public Builder nodeUsageStatsForThreadPools(Map<String, NodeUsageStatsForThreadP | |
| this.nodeUsageStatsForThreadPools = nodeUsageStatsForThreadPools; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder shardWriteLoads(Map<ShardId, Double> shardWriteLoads) { | ||
| this.shardWriteLoads = shardWriteLoads; | ||
| return this; | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
shouldn't this assert be per index? Since all the indices received writes.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No we insert between 1 and 500 documents in indices with between 1 and 3 shards, so it's possible some shards will not be written to. This is just a sanity check anyhow.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think I follow. Some shards won't receive writes, I agree. But all indices will have writes (to at least one shard), and thus maximumLoadRecorded would have a value per index.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh sorry, you're right. For some reason I read that as per shard. Indeed it could be tightened up to be per index. I'll put up a small PR to do that.