Skip to content

Handle negative free disk space in deciders #48392

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

Merged
merged 5 commits into from
Oct 23, 2019
Merged

Handle negative free disk space in deciders #48392

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
eaa7bab
Handle negative free disk space in deciders
DaveCTurner Oct 23, 2019
144e0f4
Merge branch 'master' into 2019-10-23-handle-negative-free-space
DaveCTurner Oct 23, 2019
9ea918f
New class to avoid bogus DiskUsage instances
DaveCTurner Oct 23, 2019
f5cf7b6
Apparently negative bytes are legit, see DiskUsageTests
DaveCTurner Oct 23, 2019
e3c35fd
We can get Long.MAX_VALUE for the free space so must be careful doing…
DaveCTurner Oct 23, 2019
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
103 changes: 91 additions & 12 deletions .../main/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDecider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,12 +139,25 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing

// subtractLeavingShards is passed as false here, because they still use disk space, and therefore should we should be extra careful
// and take the size into account
DiskUsage usage = getDiskUsage(node, allocation, usages, false);
final DiskUsageWithRelocations usage = getDiskUsage(node, allocation, usages, false);
// First, check that the node currently over the low watermark
double freeDiskPercentage = usage.getFreeDiskAsPercentage();
// Cache the used disk percentage for displaying disk percentages consistent with documentation
double usedDiskPercentage = usage.getUsedDiskAsPercentage();
long freeBytes = usage.getFreeBytes();
if (freeBytes < 0L) {
final long sizeOfRelocatingShards = sizeOfRelocatingShards(node, false, usage.getPath(),
allocation.clusterInfo(), allocation.metaData(), allocation.routingTable());
logger.debug("fewer free bytes remaining than the size of all incoming shards: " +
"usage {} on node {} including {} bytes of relocations, preventing allocation",
usage, node.nodeId(), sizeOfRelocatingShards);

return allocation.decision(Decision.NO, NAME,
"the node has fewer free bytes remaining than the total size of all incoming shards: " +
"free space [%sB], relocating shards [%sB]",
freeBytes + sizeOfRelocatingShards, sizeOfRelocatingShards);
}

ByteSizeValue freeBytesValue = new ByteSizeValue(freeBytes);
if (logger.isTraceEnabled()) {
logger.trace("node [{}] has {}% used disk", node.nodeId(), usedDiskPercentage);
Expand Down Expand Up @@ -242,6 +255,7 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
// Secondly, check that allocating the shard to this node doesn't put it above the high watermark
final long shardSize = getExpectedShardSize(shardRouting, 0L,
allocation.clusterInfo(), allocation.metaData(), allocation.routingTable());
assert shardSize >= 0 : shardSize;
double freeSpaceAfterShard = freeDiskPercentageAfterShardAssigned(usage, shardSize);
long freeBytesAfterShard = freeBytes - shardSize;
if (freeBytesAfterShard < diskThresholdSettings.getFreeBytesThresholdHigh().getBytes()) {
Expand All @@ -268,6 +282,7 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
diskThresholdSettings.getHighWatermarkRaw(), usedDiskThresholdHigh, freeSpaceAfterShard);
}

assert freeBytesAfterShard >= 0 : freeBytesAfterShard;
return allocation.decision(Decision.YES, NAME,
"enough disk for shard on node, free: [%s], shard size: [%s], free after allocating shard: [%s]",
freeBytesValue,
Expand All @@ -289,7 +304,7 @@ public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAl

// subtractLeavingShards is passed as true here, since this is only for shards remaining, we will *eventually* have enough disk
// since shards are moving away. No new shards will be incoming since in canAllocate we pass false for this check.
final DiskUsage usage = getDiskUsage(node, allocation, usages, true);
final DiskUsageWithRelocations usage = getDiskUsage(node, allocation, usages, true);
final String dataPath = clusterInfo.getDataPath(shardRouting);
// If this node is already above the high threshold, the shard cannot remain (get it off!)
final double freeDiskPercentage = usage.getFreeDiskAsPercentage();
Expand All @@ -301,6 +316,17 @@ public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAl
return allocation.decision(Decision.YES, NAME,
"this shard is not allocated on the most utilized disk and can remain");
}
if (freeBytes < 0L) {
final long sizeOfRelocatingShards = sizeOfRelocatingShards(node, false, usage.getPath(),
allocation.clusterInfo(), allocation.metaData(), allocation.routingTable());
logger.debug("fewer free bytes remaining than the size of all incoming shards: " +
"usage {} on node {} including {} bytes of relocations, shard cannot remain",
usage, node.nodeId(), sizeOfRelocatingShards);
return allocation.decision(Decision.NO, NAME,
"the shard cannot remain on this node because the node has fewer free bytes remaining than the total size of all " +
"incoming shards: free space [%s], relocating shards [%s]",
freeBytes + sizeOfRelocatingShards, sizeOfRelocatingShards);
}
if (freeBytes < diskThresholdSettings.getFreeBytesThresholdHigh().getBytes()) {
if (logger.isDebugEnabled()) {
logger.debug("less than the required {} free bytes threshold ({} bytes free) on node {}, shard cannot remain",
Expand Down Expand Up @@ -330,8 +356,8 @@ public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAl
"there is enough disk on this node for the shard to remain, free: [%s]", new ByteSizeValue(freeBytes));
}

private DiskUsage getDiskUsage(RoutingNode node, RoutingAllocation allocation,
ImmutableOpenMap<String, DiskUsage> usages, boolean subtractLeavingShards) {
private DiskUsageWithRelocations getDiskUsage(RoutingNode node, RoutingAllocation allocation,
ImmutableOpenMap<String, DiskUsage> usages, boolean subtractLeavingShards) {
DiskUsage usage = usages.get(node.nodeId());
if (usage == null) {
// If there is no usage, and we have other nodes in the cluster,
Expand All @@ -341,13 +367,11 @@ private DiskUsage getDiskUsage(RoutingNode node, RoutingAllocation allocation,
node.nodeId(), usage.getTotalBytes(), usage.getFreeBytes(), usage.getFreeDiskAsPercentage());
}

final long relocatingShardsSize = sizeOfRelocatingShards(node, subtractLeavingShards, usage.getPath(),
allocation.clusterInfo(), allocation.metaData(), allocation.routingTable());
final DiskUsage usageIncludingRelocations = new DiskUsage(node.nodeId(), node.node().getName(), usage.getPath(),
usage.getTotalBytes(), usage.getFreeBytes() - relocatingShardsSize);
logger.trace("getDiskUsage: usage [{}] with [{}] bytes relocating yields [{}]",
usage, relocatingShardsSize, usageIncludingRelocations);
return usageIncludingRelocations;
final DiskUsageWithRelocations diskUsageWithRelocations = new DiskUsageWithRelocations(usage,
sizeOfRelocatingShards(node, subtractLeavingShards, usage.getPath(),
allocation.clusterInfo(), allocation.metaData(), allocation.routingTable()));
logger.trace("getDiskUsage(subtractLeavingShards={}) returning {}", subtractLeavingShards, diskUsageWithRelocations);
return diskUsageWithRelocations;
}

/**
Expand Down Expand Up @@ -377,7 +401,7 @@ DiskUsage averageUsage(RoutingNode node, ImmutableOpenMap<String, DiskUsage> usa
* @param shardSize Size in bytes of the shard
* @return Percentage of free space after the shard is assigned to the node
*/
double freeDiskPercentageAfterShardAssigned(DiskUsage usage, Long shardSize) {
double freeDiskPercentageAfterShardAssigned(DiskUsageWithRelocations usage, Long shardSize) {
shardSize = (shardSize == null) ? 0 : shardSize;
DiskUsage newUsage = new DiskUsage(usage.getNodeId(), usage.getNodeName(), usage.getPath(),
usage.getTotalBytes(), usage.getFreeBytes() - shardSize);
Expand Down Expand Up @@ -445,4 +469,59 @@ public static long getExpectedShardSize(ShardRouting shard, long defaultValue, C
return clusterInfo.getShardSize(shard, defaultValue);
}
}

static class DiskUsageWithRelocations {

private final DiskUsage diskUsage;
private final long relocatingShardSize;

DiskUsageWithRelocations(DiskUsage diskUsage, long relocatingShardSize) {
this.diskUsage = diskUsage;
this.relocatingShardSize = relocatingShardSize;
}

@Override
public String toString() {
return "DiskUsageWithRelocations{" +
"diskUsage=" + diskUsage +
", relocatingShardSize=" + relocatingShardSize +
'}';
}

double getFreeDiskAsPercentage() {
if (getTotalBytes() == 0L) {
return 100.0;
}
return 100.0 * ((double)getFreeBytes() / getTotalBytes());
}

double getUsedDiskAsPercentage() {
return 100.0 - getFreeDiskAsPercentage();
}

long getFreeBytes() {
try {
return Math.subtractExact(diskUsage.getFreeBytes(), relocatingShardSize);
Copy link
Contributor

Choose a reason for hiding this comment

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

How could this overflow?
Isn't relocatingShardSize >= 0 and 0 <= diskUsage.getFreeBytes() <= Long.MAX_VALUE? Is this to capture the case where diskUsage.getFreeBytes() == 0 && relocatingShardSize == Long.MAX_VALUE?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

No, relocatingShardSize is the net size of all relocations, i.e. shards in minus shards out, so can indeed be negative :(

} catch (ArithmeticException e) {
return Long.MAX_VALUE;
}
}

String getPath() {
return diskUsage.getPath();
}

String getNodeId() {
return diskUsage.getNodeId();
}

String getNodeName() {
return diskUsage.getNodeName();
}

long getTotalBytes() {
return diskUsage.getTotalBytes();
}
}

}
98 changes: 69 additions & 29 deletions .../java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,13 +56,15 @@
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

import static java.util.Collections.emptyMap;
import static java.util.Collections.singleton;
import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.STARTED;
import static org.elasticsearch.cluster.routing.ShardRoutingState.UNASSIGNED;
import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.nullValue;
Expand Down Expand Up @@ -628,7 +630,8 @@ public void testFreeDiskPercentageAfterShardAssigned() {
usages.put("node2", new DiskUsage("node2", "n2", "/dev/null", 100, 50)); // 50% used
usages.put("node3", new DiskUsage("node3", "n3", "/dev/null", 100, 0)); // 100% used

Double after = decider.freeDiskPercentageAfterShardAssigned(new DiskUsage("node2", "n2", "/dev/null", 100, 30), 11L);
Double after = decider.freeDiskPercentageAfterShardAssigned(
new DiskThresholdDecider.DiskUsageWithRelocations(new DiskUsage("node2", "n2", "/dev/null", 100, 30), 0L), 11L);
assertThat(after, equalTo(19.0));
}

Expand All @@ -653,18 +656,19 @@ public void testShardRelocationsTakenIntoAccount() {
final ClusterInfo clusterInfo = new DevNullClusterInfo(usages, usages, shardSizes);

DiskThresholdDecider decider = makeDecider(diskSettings);
final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
AllocationDeciders deciders = new AllocationDeciders(
new HashSet<>(Arrays.asList(new SameShardAllocationDecider(
Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)
), decider)));
new HashSet<>(Arrays.asList(
new SameShardAllocationDecider(Settings.EMPTY, clusterSettings),
new EnableAllocationDecider(
Settings.builder().put(CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING.getKey(), "none").build(), clusterSettings),
decider)));

ClusterInfoService cis = () -> {
logger.info("--> calling fake getClusterInfo");
return clusterInfo;
};
final AtomicReference<ClusterInfo> clusterInfoReference = new AtomicReference<>(clusterInfo);
final ClusterInfoService cis = clusterInfoReference::get;

AllocationService strategy = new AllocationService(deciders, new TestGatewayAllocator(),
new BalancedShardsAllocator(Settings.EMPTY), cis);
new BalancedShardsAllocator(Settings.EMPTY), cis);

MetaData metaData = MetaData.builder()
.put(IndexMetaData.builder("test").settings(settings(Version.CURRENT)).numberOfShards(1).numberOfReplicas(1))
Expand Down Expand Up @@ -702,30 +706,66 @@ Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLU
.add(newNode("node3"))
).build();

AllocationCommand relocate1 = new MoveAllocationCommand("test", 0, "node2", "node3");
AllocationCommands cmds = new AllocationCommands(relocate1);
{
AllocationCommand moveAllocationCommand = new MoveAllocationCommand("test", 0, "node2", "node3");
AllocationCommands cmds = new AllocationCommands(moveAllocationCommand);

clusterState = strategy.reroute(clusterState, cmds, false, false).getClusterState();
logShardStates(clusterState);
clusterState = strategy.reroute(clusterState, cmds, false, false).getClusterState();
logShardStates(clusterState);
}

final ImmutableOpenMap.Builder<String, DiskUsage> overfullUsagesBuilder = ImmutableOpenMap.builder();
overfullUsagesBuilder.put("node1", new DiskUsage("node1", "n1", "/dev/null", 100, 40)); // 60% used
overfullUsagesBuilder.put("node2", new DiskUsage("node2", "n2", "/dev/null", 100, 40)); // 60% used
overfullUsagesBuilder.put("node3", new DiskUsage("node3", "n3", "/dev/null", 100, 0)); // 100% used
final ImmutableOpenMap<String, DiskUsage> overfullUsages = overfullUsagesBuilder.build();

final ImmutableOpenMap.Builder<String, Long> largerShardSizesBuilder = ImmutableOpenMap.builder();
largerShardSizesBuilder.put("[test][0][p]", 14L);
largerShardSizesBuilder.put("[test][0][r]", 14L);
largerShardSizesBuilder.put("[test2][0][p]", 2L);
largerShardSizesBuilder.put("[test2][0][r]", 2L);
final ImmutableOpenMap<String, Long> largerShardSizes = largerShardSizesBuilder.build();

AllocationCommand relocate2 = new MoveAllocationCommand("test2", 0, "node2", "node3");
cmds = new AllocationCommands(relocate2);

try {
// The shard for the "test" index is already being relocated to
// node3, which will put it over the low watermark when it
// completes, with shard relocations taken into account this should
// throw an exception about not being able to complete
strategy.reroute(clusterState, cmds, false, false);
fail("should not have been able to reroute the shard");
} catch (IllegalArgumentException e) {
assertThat("can't be allocated because there isn't enough room: " + e.getMessage(),
e.getMessage(),
containsString("the node is above the low watermark cluster setting " +
"[cluster.routing.allocation.disk.watermark.low=0.7], using more disk space than the maximum " +
"allowed [70.0%], actual free: [26.0%]"));
final ClusterInfo overfullClusterInfo = new DevNullClusterInfo(overfullUsages, overfullUsages, largerShardSizes);

{
AllocationCommand moveAllocationCommand = new MoveAllocationCommand("test2", 0, "node2", "node3");
AllocationCommands cmds = new AllocationCommands(moveAllocationCommand);

final ClusterState clusterStateThatRejectsCommands = clusterState;

assertThat(expectThrows(IllegalArgumentException.class,
() -> strategy.reroute(clusterStateThatRejectsCommands, cmds, false, false)).getMessage(),
containsString("the node is above the low watermark cluster setting " +
"[cluster.routing.allocation.disk.watermark.low=0.7], using more disk space than the maximum " +
"allowed [70.0%], actual free: [26.0%]"));

clusterInfoReference.set(overfullClusterInfo);

assertThat(expectThrows(IllegalArgumentException.class,
() -> strategy.reroute(clusterStateThatRejectsCommands, cmds, false, false)).getMessage(),
containsString("the node has fewer free bytes remaining than the total size of all incoming shards"));

clusterInfoReference.set(clusterInfo);
}

{
AllocationCommand moveAllocationCommand = new MoveAllocationCommand("test2", 0, "node2", "node3");
AllocationCommands cmds = new AllocationCommands(moveAllocationCommand);

logger.info("--> before starting: {}", clusterState);
clusterState = startInitializingShardsAndReroute(strategy, clusterState);
logger.info("--> after starting: {}", clusterState);
clusterState = strategy.reroute(clusterState, cmds, false, false).getClusterState();
logger.info("--> after running another command: {}", clusterState);
logShardStates(clusterState);

clusterInfoReference.set(overfullClusterInfo);

clusterState = strategy.reroute(clusterState, "foo");
logger.info("--> after another reroute: {}", clusterState);
}
}

public void testCanRemainWithShardRelocatingAway() {
Expand Down
Loading