Implement WriteLoadConstraintDecider#canAllocate #132041
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
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
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,124 @@ | ||
| /* | ||
| * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public | ||
| * License v3.0 only", or the "Server Side Public License, v 1". | ||
| */ | ||
|
|
||
| package org.elasticsearch.cluster.routing.allocation.decider; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.elasticsearch.cluster.NodeUsageStatsForThreadPools.ThreadPoolUsageStats; | ||
| import org.elasticsearch.cluster.metadata.IndexMetadata; | ||
| import org.elasticsearch.cluster.routing.RoutingNode; | ||
| import org.elasticsearch.cluster.routing.ShardMovementWriteLoadSimulator; | ||
| import org.elasticsearch.cluster.routing.ShardRouting; | ||
| import org.elasticsearch.cluster.routing.allocation.RoutingAllocation; | ||
| import org.elasticsearch.cluster.routing.allocation.WriteLoadConstraintSettings; | ||
| import org.elasticsearch.common.settings.ClusterSettings; | ||
| import org.elasticsearch.core.Strings; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
|
|
||
| /** | ||
| * Decides whether shards can be allocated to cluster nodes, or can remain on cluster nodes, based on the target node's current write thread | ||
| * pool usage stats and any candidate shard's write load estimate. | ||
| */ | ||
| public class WriteLoadConstraintDecider extends AllocationDecider { | ||
| private static final Logger logger = LogManager.getLogger(WriteLoadConstraintDecider.class); | ||
|
|
||
| public static final String NAME = "write_load"; | ||
|
|
||
| private final WriteLoadConstraintSettings writeLoadConstraintSettings; | ||
|
|
||
| public WriteLoadConstraintDecider(ClusterSettings clusterSettings) { | ||
| this.writeLoadConstraintSettings = new WriteLoadConstraintSettings(clusterSettings); | ||
| } | ||
|
|
||
| @Override | ||
| public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) { | ||
| if (writeLoadConstraintSettings.getWriteLoadConstraintEnabled().disabled()) { | ||
| return Decision.single(Decision.Type.YES, NAME, "Decider is disabled"); | ||
| } | ||
DiannaHohensee marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| // Check whether the shard being relocated has any write load estimate. If it does not, then this decider has no opinion. | ||
| var allShardWriteLoads = allocation.clusterInfo().getShardWriteLoads(); | ||
| var shardWriteLoad = allShardWriteLoads.get(shardRouting.shardId()); | ||
nicktindall marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| if (shardWriteLoad == null || shardWriteLoad == 0) { | ||
mhl-b marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| return Decision.single(Decision.Type.YES, NAME, "Shard has no estimated write load. Decider takes no action."); | ||
| } | ||
|
|
||
| var allNodeUsageStats = allocation.clusterInfo().getNodeUsageStatsForThreadPools(); | ||
| var nodeUsageStatsForThreadPools = allNodeUsageStats.get(node.nodeId()); | ||
| if (nodeUsageStatsForThreadPools == null) { | ||
mhl-b marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| // No node-level thread pool usage stats were reported for this node. Let's assume this is OK and that the simulator will handle | ||
| // setting a node-level write load for this node after this shard is assigned. | ||
| return Decision.single(Decision.Type.YES, NAME, "The node has no write load estimate. Decider takes no action."); | ||
| } | ||
|
|
||
| assert nodeUsageStatsForThreadPools.threadPoolUsageStatsMap().isEmpty() == false; | ||
| assert nodeUsageStatsForThreadPools.threadPoolUsageStatsMap().get(ThreadPool.Names.WRITE) != null; | ||
| var nodeWriteThreadPoolStats = nodeUsageStatsForThreadPools.threadPoolUsageStatsMap().get(ThreadPool.Names.WRITE); | ||
| var nodeWriteThreadPoolLoadThreshold = writeLoadConstraintSettings.getWriteThreadPoolHighUtilizationThresholdSetting(); | ||
| if (nodeWriteThreadPoolStats.averageThreadPoolUtilization() >= nodeWriteThreadPoolLoadThreshold) { | ||
mhl-b marked this conversation as resolved.
Show resolved
Hide resolved
Contributor
There was a problem hiding this comment.
Contributor
Author
There was a problem hiding this comment. Yes, the next check would also catch this. It's a matter of explanation message for the NO decision, really, at this point. They could be combined; separately the messages can be clearer for the user to understand, I think.
Contributor
There was a problem hiding this comment.
Contributor
Author
There was a problem hiding this comment. I'm not sure what you're suggesting. Are you proposing to add logic to I like the clarity of the separate explain messages. Do you feel strongly about merging the two if statements? |
||
| // The node's write thread pool usage stats already show high utilization above the threshold for accepting new shards. | ||
| String explain = Strings.format( | ||
| "Node [%s] with write thread pool utilization [%.2f] already exceeds the high utilization threshold of [%f]. Cannot " | ||
| + "allocate shard [%s] to node without risking increased write latencies.", | ||
| node.nodeId(), | ||
| nodeWriteThreadPoolStats.averageThreadPoolUtilization(), | ||
| nodeWriteThreadPoolLoadThreshold, | ||
| shardRouting.shardId() | ||
| ); | ||
| logger.debug(explain); | ||
| return Decision.single(Decision.Type.NO, NAME, explain); | ||
|
Contributor
There was a problem hiding this comment. I think this should respond "not-preferred" instead?
Contributor
Author
There was a problem hiding this comment. There was disagreement on implementing |
||
| } | ||
|
|
||
| if (calculateShardMovementChange(nodeWriteThreadPoolStats, shardWriteLoad) >= nodeWriteThreadPoolLoadThreshold) { | ||
| // The node's write thread pool usage would be raised above the high utilization threshold with assignment of the new shard. | ||
| // This could lead to a hot spot on this node and is undesirable. | ||
| String explain = Strings.format( | ||
| "The high utilization threshold of [%f] would be exceeded on node [%s] with utilization [%.2f] if shard [%s] with " | ||
| + "estimated additional utilisation [%.5f] (write load [%.5f] / threads [%d]) were assigned to it. Cannot allocate " | ||
| + "shard to node without risking increased write latencies.", | ||
| nodeWriteThreadPoolLoadThreshold, | ||
| node.nodeId(), | ||
| nodeWriteThreadPoolStats.averageThreadPoolUtilization(), | ||
nicktindall marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| shardRouting.shardId(), | ||
| shardWriteLoad / nodeWriteThreadPoolStats.totalThreadPoolThreads(), | ||
| shardWriteLoad, | ||
| nodeWriteThreadPoolStats.totalThreadPoolThreads() | ||
| ); | ||
| logger.debug(explain); | ||
| return Decision.single(Decision.Type.NO, NAME, explain); | ||
| } | ||
|
|
||
| return Decision.YES; | ||
| } | ||
|
|
||
| @Override | ||
| public Decision canRemain(IndexMetadata indexMetadata, ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) { | ||
| if (writeLoadConstraintSettings.getWriteLoadConstraintEnabled().notFullyEnabled()) { | ||
| return Decision.YES; | ||
| } | ||
|
|
||
| // TODO: implement | ||
|
|
||
| return Decision.YES; | ||
| } | ||
|
|
||
| /** | ||
| * Calculates the change to the node's write thread pool utilization percentage if the shard is added to the node. | ||
| * Returns the percent thread pool utilization change. | ||
| */ | ||
| private float calculateShardMovementChange(ThreadPoolUsageStats nodeWriteThreadPoolStats, double shardWriteLoad) { | ||
| assert shardWriteLoad > 0; | ||
| return ShardMovementWriteLoadSimulator.updateNodeUtilizationWithShardMovements( | ||
| nodeWriteThreadPoolStats.averageThreadPoolUtilization(), | ||
| (float) shardWriteLoad, | ||
| nodeWriteThreadPoolStats.totalThreadPoolThreads() | ||
| ); | ||
| } | ||
| } | ||
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.
Nit: alternatively we could have a method
WriteLoadDeciderStatus#requiresShardLevelWriteLoads()(andrequiresNodeLevelWriteLoads()) which would returntrueforLOW_THRESHOLD_ONLYandENABLEDbutfalseforDISABLED.It would read nicer if the
writeLoadConstraintEnabledfield was calledwriteLoadDeciderStatusif we went that way.Don't feel strongly about this naming thing though.
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.
requiresShardLevelWriteLoadsandrequiresNodeLevelWriteLoadsdoesn't seem like the right split, as I understand it. I was imagining LOW as the best-effort hot-spot prevention (canAllocate) without hot-spot correction (canRemain), and fully enabled as including hot-spot correction. Both node and shard level stats are needed for prevention, to compare the shard move write load change with the node's overall write load.I'll leave this as is until some follow up.