-
Notifications
You must be signed in to change notification settings - Fork 25.6k
simulate disk usage during balance calculation #90061
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.
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
Merged
idegtiarenko
merged 22 commits into
elastic:feature/desired-balance-allocator
from
idegtiarenko:simulate_disk_usage
Oct 5, 2022
Merged
Changes from all commits
Commits
Show all changes
22 commits
Select commit
Hold shift + click to select a range
5c634a7
wip cluster info simulation
idegtiarenko 342ba59
use cluster info simulator
idegtiarenko d0fa4c6
null check
idegtiarenko 105fcef
Fix MockDiskUsagesIT
idegtiarenko 7293edd
do not keep reserved info
idegtiarenko 779016b
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko 5c919f9
multiple data path test case
idegtiarenko 103610a
add range check
idegtiarenko 5aaad09
testComputeConsideringShardSizes
idegtiarenko cb7a174
keep node below watermark
idegtiarenko 5faf34e
move class
idegtiarenko a8072ca
fix move
idegtiarenko 2985353
rework test
idegtiarenko 07be67f
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko 480ad7a
Update server/src/main/java/org/elasticsearch/cluster/ClusterInfo.java
idegtiarenko ee9a19f
Update server/src/main/java/org/elasticsearch/cluster/ClusterInfo.java
idegtiarenko 8c92948
make sure test is failing without simulated disk space
idegtiarenko 2f1e6fb
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko 67ee7dd
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko 52fa7d3
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko 93a72dc
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko b64afdd
Merge branch 'feature/desired-balance-allocator' into simulate_disk_u…
idegtiarenko File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
97 changes: 97 additions & 0 deletions
97
server/src/main/java/org/elasticsearch/cluster/ClusterInfoSimulator.java
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 |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License | ||
| * 2.0 and the Server Side Public License, v 1; you may not use this file except | ||
| * in compliance with, at your election, the Elastic License 2.0 or the Server | ||
| * Side Public License, v 1. | ||
| */ | ||
|
|
||
| package org.elasticsearch.cluster; | ||
|
|
||
| import org.elasticsearch.cluster.routing.ShardRouting; | ||
| import org.elasticsearch.index.shard.ShardId; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.Objects; | ||
|
|
||
| public class ClusterInfoSimulator { | ||
|
|
||
| private final Map<String, DiskUsage> leastAvailableSpaceUsage; | ||
| private final Map<String, DiskUsage> mostAvailableSpaceUsage; | ||
| private final Map<String, Long> shardSizes; | ||
| private final Map<ShardId, Long> shardDataSetSizes; | ||
| private final Map<ClusterInfo.NodeAndShard, String> dataPath; | ||
|
|
||
| public ClusterInfoSimulator(ClusterInfo clusterInfo) { | ||
| this.leastAvailableSpaceUsage = new HashMap<>(clusterInfo.getNodeLeastAvailableDiskUsages()); | ||
| this.mostAvailableSpaceUsage = new HashMap<>(clusterInfo.getNodeMostAvailableDiskUsages()); | ||
| this.shardSizes = new HashMap<>(clusterInfo.shardSizes); | ||
| this.shardDataSetSizes = new HashMap<>(clusterInfo.shardDataSetSizes); | ||
| this.dataPath = new HashMap<>(clusterInfo.dataPath); | ||
| } | ||
|
|
||
| public void simulate(ShardRouting shard) { | ||
| assert shard.initializing(); | ||
|
|
||
| var size = getEstimatedShardSize(shard); | ||
| if (size != null && size > 0) { | ||
| if (shard.relocatingNodeId() != null) { | ||
| // relocation | ||
| modifyDiskUsage(shard.relocatingNodeId(), getShardPath(shard.relocatingNodeId(), mostAvailableSpaceUsage), size); | ||
| modifyDiskUsage(shard.currentNodeId(), getShardPath(shard.currentNodeId(), leastAvailableSpaceUsage), -size); | ||
| } else { | ||
| // new shard | ||
| modifyDiskUsage(shard.currentNodeId(), getShardPath(shard.currentNodeId(), leastAvailableSpaceUsage), -size); | ||
| shardSizes.put(ClusterInfo.shardIdentifierFromRouting(shard), size); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private Long getEstimatedShardSize(ShardRouting routing) { | ||
| if (routing.relocatingNodeId() != null) { | ||
| // relocation existing shard, get size of the source shard | ||
| return shardSizes.get(ClusterInfo.shardIdentifierFromRouting(routing)); | ||
| } else if (routing.primary() == false) { | ||
| // initializing new replica, get size of the source primary shard | ||
| return shardSizes.get(ClusterInfo.shardIdentifierFromRouting(routing.shardId(), true)); | ||
| } else { | ||
| // initializing new (empty) primary | ||
| return 0L; | ||
| } | ||
| } | ||
|
|
||
| private String getShardPath(String nodeId, Map<String, DiskUsage> defaultSpaceUsage) { | ||
| var diskUsage = defaultSpaceUsage.get(nodeId); | ||
| return diskUsage != null ? diskUsage.getPath() : null; | ||
| } | ||
|
|
||
| private void modifyDiskUsage(String nodeId, String path, long delta) { | ||
| var leastUsage = leastAvailableSpaceUsage.get(nodeId); | ||
| if (leastUsage != null && Objects.equals(leastUsage.getPath(), path)) { | ||
| // ensure new value is within bounds | ||
| leastAvailableSpaceUsage.put(nodeId, updateWithFreeBytes(leastUsage, delta)); | ||
| } | ||
| var mostUsage = mostAvailableSpaceUsage.get(nodeId); | ||
| if (mostUsage != null && Objects.equals(mostUsage.getPath(), path)) { | ||
| // ensure new value is within bounds | ||
| mostAvailableSpaceUsage.put(nodeId, updateWithFreeBytes(mostUsage, delta)); | ||
| } | ||
| } | ||
|
|
||
| private static DiskUsage updateWithFreeBytes(DiskUsage usage, long delta) { | ||
| // free bytes might go out of range in case when multiple data path are used | ||
| // we might not know exact disk used to allocate a shard and conservatively update | ||
| // most used disk on a target node and least used disk on a source node | ||
| var freeBytes = withinRange(0, usage.getTotalBytes(), usage.freeBytes() + delta); | ||
| return usage.copyWithFreeBytes(freeBytes); | ||
| } | ||
|
|
||
| private static long withinRange(long min, long max, long value) { | ||
| return Math.max(min, Math.min(max, value)); | ||
| } | ||
|
|
||
| public ClusterInfo getClusterInfo() { | ||
| return new ClusterInfo(leastAvailableSpaceUsage, mostAvailableSpaceUsage, shardSizes, shardDataSetSizes, dataPath, Map.of()); | ||
| } | ||
| } |
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
174 changes: 174 additions & 0 deletions
174
...ava/org/elasticsearch/cluster/routing/allocation/allocator/ClusterInfoSimulatorTests.java
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 |
|---|---|---|
| @@ -0,0 +1,174 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License | ||
| * 2.0 and the Server Side Public License, v 1; you may not use this file except | ||
| * in compliance with, at your election, the Elastic License 2.0 or the Server | ||
| * Side Public License, v 1. | ||
| */ | ||
|
|
||
| package org.elasticsearch.cluster.routing.allocation.allocator; | ||
|
|
||
| import org.elasticsearch.cluster.ClusterInfo; | ||
| import org.elasticsearch.cluster.ClusterInfoSimulator; | ||
| import org.elasticsearch.cluster.DiskUsage; | ||
| import org.elasticsearch.cluster.routing.ShardRouting; | ||
| import org.elasticsearch.test.ESTestCase; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING; | ||
| import static org.elasticsearch.cluster.routing.ShardRoutingState.STARTED; | ||
| import static org.elasticsearch.cluster.routing.TestShardRouting.newShardRouting; | ||
| import static org.hamcrest.Matchers.equalTo; | ||
|
|
||
| public class ClusterInfoSimulatorTests extends ESTestCase { | ||
|
|
||
| public void testInitializeNewPrimary() { | ||
|
|
||
| var newPrimary = newShardRouting("index-1", 0, "node-0", true, INITIALIZING); | ||
|
|
||
| var simulator = new ClusterInfoSimulator( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode("node-0", new DiskUsageBuilder(1000, 1000)) | ||
| .withNode("node-1", new DiskUsageBuilder(1000, 1000)) | ||
| .withShard(newPrimary, 0) | ||
| .build() | ||
| ); | ||
| simulator.simulate(newPrimary); | ||
|
|
||
| assertThat( | ||
| simulator.getClusterInfo(), | ||
| equalTo( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode("node-0", new DiskUsageBuilder(1000, 1000)) | ||
| .withNode("node-1", new DiskUsageBuilder(1000, 1000)) | ||
| .withShard(newPrimary, 0) | ||
| .build() | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| public void testInitializeNewReplica() { | ||
|
|
||
| var existingPrimary = newShardRouting("index-1", 0, "node-0", true, STARTED); | ||
| var newReplica = newShardRouting("index-1", 0, "node-1", false, INITIALIZING); | ||
|
|
||
| var simulator = new ClusterInfoSimulator( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode("node-0", new DiskUsageBuilder(1000, 900)) | ||
| .withNode("node-1", new DiskUsageBuilder(1000, 1000)) | ||
| .withShard(existingPrimary, 100) | ||
| .withShard(newReplica, 0) | ||
| .build() | ||
| ); | ||
| simulator.simulate(newReplica); | ||
|
|
||
| assertThat( | ||
| simulator.getClusterInfo(), | ||
| equalTo( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode("node-0", new DiskUsageBuilder(1000, 900)) | ||
| .withNode("node-1", new DiskUsageBuilder(1000, 900)) | ||
| .withShard(existingPrimary, 100) | ||
| .withShard(newReplica, 100) | ||
| .build() | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| public void testRelocateShard() { | ||
|
|
||
| var fromNodeId = "node-0"; | ||
| var toNodeId = "node-1"; | ||
|
|
||
| var shard = newShardRouting("index-1", 0, toNodeId, fromNodeId, true, INITIALIZING); | ||
|
|
||
| var simulator = new ClusterInfoSimulator( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode(fromNodeId, new DiskUsageBuilder(1000, 900)) | ||
| .withNode(toNodeId, new DiskUsageBuilder(1000, 1000)) | ||
| .withShard(shard, 100) | ||
| .build() | ||
| ); | ||
| simulator.simulate(shard); | ||
|
|
||
| assertThat( | ||
| simulator.getClusterInfo(), | ||
| equalTo( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode(fromNodeId, new DiskUsageBuilder(1000, 1000)) | ||
| .withNode(toNodeId, new DiskUsageBuilder(1000, 900)) | ||
| .withShard(shard, 100) | ||
| .build() | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| public void testRelocateShardWithMultipleDataPath1() { | ||
|
|
||
| var fromNodeId = "node-0"; | ||
| var toNodeId = "node-1"; | ||
|
|
||
| var shard = newShardRouting("index-1", 0, toNodeId, fromNodeId, true, INITIALIZING); | ||
|
|
||
| var simulator = new ClusterInfoSimulator( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode(fromNodeId, new DiskUsageBuilder("/data-1", 1000, 500), new DiskUsageBuilder("/data-2", 1000, 750)) | ||
| .withNode(toNodeId, new DiskUsageBuilder("/data-1", 1000, 750), new DiskUsageBuilder("/data-2", 1000, 900)) | ||
| .withShard(shard, 100) | ||
| .build() | ||
| ); | ||
| simulator.simulate(shard); | ||
|
|
||
| assertThat( | ||
| simulator.getClusterInfo(), | ||
| equalTo( | ||
| new ClusterInfoTestBuilder() // | ||
| .withNode(fromNodeId, new DiskUsageBuilder("/data-1", 1000, 500), new DiskUsageBuilder("/data-2", 1000, 850)) | ||
| .withNode(toNodeId, new DiskUsageBuilder("/data-1", 1000, 650), new DiskUsageBuilder("/data-2", 1000, 900)) | ||
| .withShard(shard, 100) | ||
| .build() | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| private static class ClusterInfoTestBuilder { | ||
|
|
||
| private final Map<String, DiskUsage> leastAvailableSpaceUsage = new HashMap<>(); | ||
| private final Map<String, DiskUsage> mostAvailableSpaceUsage = new HashMap<>(); | ||
| private final Map<String, Long> shardSizes = new HashMap<>(); | ||
|
|
||
| public ClusterInfoTestBuilder withNode(String name, DiskUsageBuilder diskUsageBuilderBuilder) { | ||
| leastAvailableSpaceUsage.put(name, diskUsageBuilderBuilder.toDiskUsage(name)); | ||
| mostAvailableSpaceUsage.put(name, diskUsageBuilderBuilder.toDiskUsage(name)); | ||
| return this; | ||
| } | ||
|
|
||
| public ClusterInfoTestBuilder withNode(String name, DiskUsageBuilder leastAvailableSpace, DiskUsageBuilder mostAvailableSpace) { | ||
| leastAvailableSpaceUsage.put(name, leastAvailableSpace.toDiskUsage(name)); | ||
| mostAvailableSpaceUsage.put(name, mostAvailableSpace.toDiskUsage(name)); | ||
| return this; | ||
| } | ||
|
|
||
| public ClusterInfoTestBuilder withShard(ShardRouting shard, long size) { | ||
| shardSizes.put(ClusterInfo.shardIdentifierFromRouting(shard), size); | ||
| return this; | ||
| } | ||
|
|
||
| public ClusterInfo build() { | ||
| return new ClusterInfo(leastAvailableSpaceUsage, mostAvailableSpaceUsage, shardSizes, Map.of(), Map.of(), Map.of()); | ||
| } | ||
| } | ||
|
|
||
| private record DiskUsageBuilder(String path, long total, long free) { | ||
|
|
||
| private DiskUsageBuilder(long total, long free) { | ||
| this("/data", total, free); | ||
| } | ||
|
|
||
| public DiskUsage toDiskUsage(String name) { | ||
| return new DiskUsage(name, name, name + '/' + path, total, free); | ||
| } | ||
| } | ||
| } |
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.
Could we enhance
DesiredBalanceComputerTeststo verify we're accounting for disk usage in all the right places?