Skip to content

Commit 5781848

Browse files
phananh1010phananh1010
committed
Mirror upstream elastic#133958 as single snapshot commit for AI review
BASE=fabf32c3f1dd08a0e51d6b86412bbea7dee50bc7 HEAD=56dfacba4d973bd825df6f7b24a516b307b8fff7 Branch=main
1 parent fabf32c commit 5781848

File tree

9 files changed

+266
-22
lines changed

9 files changed

+266
-22
lines changed

‎server/src/main/java/org/elasticsearch/cluster/ClusterModule.java‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,8 @@
3737
import org.elasticsearch.cluster.routing.allocation.AllocationStatsService;
3838
import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
3939
import org.elasticsearch.cluster.routing.allocation.NodeAllocationStatsAndWeightsCalculator;
40+
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
41+
import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
4042
import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
4143
import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
4244
import org.elasticsearch.cluster.routing.allocation.allocator.BalancerSettings;
@@ -107,6 +109,8 @@
107109
import java.util.Objects;
108110
import java.util.function.Supplier;
109111

112+
import static org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceShardsAllocator.ShardAllocationExplainer;
113+
110114
/**
111115
* Configures classes and services that affect the entire cluster.
112116
*/
@@ -171,7 +175,8 @@ public ClusterModule(
171175
this::reconcile,
172176
writeLoadForecaster,
173177
telemetryProvider,
174-
nodeAllocationStatsAndWeightsCalculator
178+
nodeAllocationStatsAndWeightsCalculator,
179+
this::explainShardAllocation
175180
);
176181
this.clusterService = clusterService;
177182
this.indexNameExpressionResolver = new IndexNameExpressionResolver(threadPool.getThreadContext(), systemIndices, projectResolver);
@@ -237,6 +242,10 @@ private ClusterState reconcile(ClusterState clusterState, RerouteStrategy rerout
237242
return allocationService.executeWithRoutingAllocation(clusterState, "reconcile-desired-balance", rerouteStrategy);
238243
}
239244

245+
private ShardAllocationDecision explainShardAllocation(ShardRouting shardRouting, RoutingAllocation allocation) {
246+
return allocationService.explainShardAllocation(shardRouting, allocation);
247+
}
248+
240249
public static List<Entry> getNamedWriteables() {
241250
List<Entry> entries = new ArrayList<>();
242251
// Cluster State
@@ -489,7 +498,8 @@ private static ShardsAllocator createShardsAllocator(
489498
DesiredBalanceReconcilerAction reconciler,
490499
WriteLoadForecaster writeLoadForecaster,
491500
TelemetryProvider telemetryProvider,
492-
NodeAllocationStatsAndWeightsCalculator nodeAllocationStatsAndWeightsCalculator
501+
NodeAllocationStatsAndWeightsCalculator nodeAllocationStatsAndWeightsCalculator,
502+
ShardAllocationExplainer shardAllocationExplainer
493503
) {
494504
Map<String, Supplier<ShardsAllocator>> allocators = new HashMap<>();
495505
allocators.put(
@@ -505,7 +515,8 @@ private static ShardsAllocator createShardsAllocator(
505515
clusterService,
506516
reconciler,
507517
telemetryProvider,
508-
nodeAllocationStatsAndWeightsCalculator
518+
nodeAllocationStatsAndWeightsCalculator,
519+
shardAllocationExplainer
509520
)
510521
);
511522

‎server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/DesiredBalanceComputer.java‎

Lines changed: 65 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,16 @@
1717
import org.elasticsearch.cluster.routing.ShardRouting;
1818
import org.elasticsearch.cluster.routing.UnassignedInfo;
1919
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
20+
import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
21+
import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceShardsAllocator.ShardAllocationExplainer;
2022
import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
2123
import org.elasticsearch.cluster.routing.allocation.decider.Decision;
2224
import org.elasticsearch.common.metrics.MeanMetric;
2325
import org.elasticsearch.common.settings.ClusterSettings;
2426
import org.elasticsearch.common.settings.Setting;
2527
import org.elasticsearch.common.time.TimeProvider;
2628
import org.elasticsearch.common.util.Maps;
29+
import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
2730
import org.elasticsearch.core.Strings;
2831
import org.elasticsearch.core.TimeValue;
2932
import org.elasticsearch.index.shard.ShardId;
@@ -39,6 +42,7 @@
3942
import java.util.TreeMap;
4043
import java.util.TreeSet;
4144
import java.util.function.Predicate;
45+
import java.util.stream.Stream;
4246

4347
import static java.util.stream.Collectors.toUnmodifiableSet;
4448

@@ -48,9 +52,13 @@
4852
public class DesiredBalanceComputer {
4953

5054
private static final Logger logger = LogManager.getLogger(DesiredBalanceComputer.class);
55+
private static final Logger allocationExplainLogger = LogManager.getLogger(
56+
DesiredBalanceComputer.class.getCanonicalName() + ".allocation_explain"
57+
);
5158

5259
private final ShardsAllocator delegateAllocator;
5360
private final TimeProvider timeProvider;
61+
private final ShardAllocationExplainer shardAllocationExplainer;
5462

5563
// stats
5664
protected final MeanMetric iterations = new MeanMetric();
@@ -77,10 +85,17 @@ public class DesiredBalanceComputer {
7785
private long lastConvergedTimeMillis;
7886
private long lastNotConvergedLogMessageTimeMillis;
7987
private Level convergenceLogMsgLevel;
88+
private ShardRouting lastTrackedUnassignedShard;
8089

81-
public DesiredBalanceComputer(ClusterSettings clusterSettings, TimeProvider timeProvider, ShardsAllocator delegateAllocator) {
90+
public DesiredBalanceComputer(
91+
ClusterSettings clusterSettings,
92+
TimeProvider timeProvider,
93+
ShardsAllocator delegateAllocator,
94+
ShardAllocationExplainer shardAllocationExplainer
95+
) {
8296
this.delegateAllocator = delegateAllocator;
8397
this.timeProvider = timeProvider;
98+
this.shardAllocationExplainer = shardAllocationExplainer;
8499
this.numComputeCallsSinceLastConverged = 0;
85100
this.numIterationsSinceLastConverged = 0;
86101
this.lastConvergedTimeMillis = timeProvider.relativeTimeInMillis();
@@ -462,10 +477,59 @@ public DesiredBalance compute(
462477
);
463478
}
464479

480+
maybeLogAllocationExplainForUnassigned(finishReason, routingNodes, routingAllocation);
481+
465482
long lastConvergedIndex = hasChanges ? previousDesiredBalance.lastConvergedIndex() : desiredBalanceInput.index();
466483
return new DesiredBalance(lastConvergedIndex, assignments, routingNodes.getBalanceWeightStatsPerNode(), finishReason);
467484
}
468485

486+
private void maybeLogAllocationExplainForUnassigned(
487+
DesiredBalance.ComputationFinishReason finishReason,
488+
RoutingNodes routingNodes,
489+
RoutingAllocation routingAllocation
490+
) {
491+
if (allocationExplainLogger.isDebugEnabled()) {
492+
if (lastTrackedUnassignedShard != null) {
493+
if (Stream.concat(routingNodes.unassigned().stream(), routingNodes.unassigned().ignored().stream())
494+
.noneMatch(shardRouting -> shardRouting.equals(lastTrackedUnassignedShard))) {
495+
allocationExplainLogger.debug("previously tracked unassigned shard [{}] is now assigned", lastTrackedUnassignedShard);
496+
lastTrackedUnassignedShard = null;
497+
} else {
498+
return; // The last tracked unassigned shard is still unassigned, keep tracking it
499+
}
500+
}
501+
502+
assert lastTrackedUnassignedShard == null : "unexpected non-null lastTrackedUnassignedShard " + lastTrackedUnassignedShard;
503+
if (routingNodes.hasUnassignedShards() && finishReason == DesiredBalance.ComputationFinishReason.CONVERGED) {
504+
final Predicate<ShardRouting> predicate = routingNodes.hasUnassignedPrimaries() ? ShardRouting::primary : shard -> true;
505+
lastTrackedUnassignedShard = Stream.concat(routingNodes.unassigned().stream(), routingNodes.unassigned().ignored().stream())
506+
.filter(predicate)
507+
.findFirst()
508+
.orElseThrow();
509+
510+
final var originalDebugMode = routingAllocation.getDebugMode();
511+
routingAllocation.setDebugMode(RoutingAllocation.DebugMode.EXCLUDE_YES_DECISIONS);
512+
final ShardAllocationDecision shardAllocationDecision;
513+
try {
514+
shardAllocationDecision = shardAllocationExplainer.explain(lastTrackedUnassignedShard, routingAllocation);
515+
} finally {
516+
routingAllocation.setDebugMode(originalDebugMode);
517+
}
518+
allocationExplainLogger.debug(
519+
"unassigned shard [{}] with allocation decision {}",
520+
lastTrackedUnassignedShard,
521+
org.elasticsearch.common.Strings.toString(
522+
p -> ChunkedToXContentHelper.object("node_allocation_decision", shardAllocationDecision.toXContentChunked(p))
523+
)
524+
);
525+
}
526+
} else {
527+
if (lastTrackedUnassignedShard != null) {
528+
lastTrackedUnassignedShard = null;
529+
}
530+
}
531+
}
532+
469533
// visible for testing
470534
boolean hasEnoughIterations(int currentIteration) {
471535
return true;

‎server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/DesiredBalanceShardsAllocator.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -109,20 +109,26 @@ public interface DesiredBalanceReconcilerAction {
109109
ClusterState apply(ClusterState clusterState, RerouteStrategy rerouteStrategy);
110110
}
111111

112+
@FunctionalInterface
113+
public interface ShardAllocationExplainer {
114+
ShardAllocationDecision explain(ShardRouting shard, RoutingAllocation allocation);
115+
}
116+
112117
public DesiredBalanceShardsAllocator(
113118
ClusterSettings clusterSettings,
114119
ShardsAllocator delegateAllocator,
115120
ThreadPool threadPool,
116121
ClusterService clusterService,
117122
DesiredBalanceReconcilerAction reconciler,
118123
TelemetryProvider telemetryProvider,
119-
NodeAllocationStatsAndWeightsCalculator nodeAllocationStatsAndWeightsCalculator
124+
NodeAllocationStatsAndWeightsCalculator nodeAllocationStatsAndWeightsCalculator,
125+
ShardAllocationExplainer shardAllocationExplainer
120126
) {
121127
this(
122128
delegateAllocator,
123129
threadPool,
124130
clusterService,
125-
new DesiredBalanceComputer(clusterSettings, threadPool, delegateAllocator),
131+
new DesiredBalanceComputer(clusterSettings, threadPool, delegateAllocator, shardAllocationExplainer),
126132
reconciler,
127133
telemetryProvider,
128134
nodeAllocationStatsAndWeightsCalculator

‎server/src/test/java/org/elasticsearch/action/admin/cluster/allocation/TransportDeleteDesiredBalanceActionTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ public void testDeleteDesiredBalance() throws Exception {
9999
var clusterSettings = ClusterSettings.createBuiltInClusterSettings(settings);
100100

101101
var delegate = new BalancedShardsAllocator();
102-
var computer = new DesiredBalanceComputer(clusterSettings, threadPool, delegate) {
102+
var computer = new DesiredBalanceComputer(clusterSettings, threadPool, delegate, TEST_ONLY_EXPLAINER) {
103103

104104
final AtomicReference<DesiredBalance> lastComputationInput = new AtomicReference<>();
105105

‎server/src/test/java/org/elasticsearch/cluster/routing/allocation/AllocationStatsServiceTests.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -176,7 +176,8 @@ public void testUndesiredShardCount() {
176176
clusterService,
177177
(innerState, strategy) -> innerState,
178178
TelemetryProvider.NOOP,
179-
EMPTY_NODE_ALLOCATION_STATS
179+
EMPTY_NODE_ALLOCATION_STATS,
180+
TEST_ONLY_EXPLAINER
180181
) {
181182
@Override
182183
public DesiredBalance getDesiredBalance() {

‎server/src/test/java/org/elasticsearch/cluster/routing/allocation/allocator/ClusterAllocationSimulationTests.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -490,7 +490,8 @@ private Map.Entry<MockAllocationService, ShardsAllocator> createNewAllocationSer
490490
(clusterState, routingAllocationAction) -> strategyRef.get()
491491
.executeWithRoutingAllocation(clusterState, "reconcile-desired-balance", routingAllocationAction),
492492
TelemetryProvider.NOOP,
493-
EMPTY_NODE_ALLOCATION_STATS
493+
EMPTY_NODE_ALLOCATION_STATS,
494+
TEST_ONLY_EXPLAINER
494495
) {
495496
@Override
496497
public void allocate(RoutingAllocation allocation, ActionListener<Void> listener) {

0 commit comments

Comments
 (0)