Skip to content

Always fetch node usage stats for write load decider #136212

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 7 commits into from
Oct 9, 2025
+36 −64
Merged

Always fetch node usage stats for write load decider #136212

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a983080
Always fetch node usage stats for write load decider
DiannaHohensee Oct 8, 2025
9948453
Merge branch 'main' into 2025/10/08/ES-12631-enable-stats
DiannaHohensee Oct 8, 2025
ea022ff
surround all write load enable settings with clearing afterwards in a…
DiannaHohensee Oct 8, 2025
0e9a689
[CI] Auto commit changes from spotless
Oct 8, 2025
68bcc2f
Merge branch 'main' into 2025/10/08/ES-12631-enable-stats
DiannaHohensee Oct 9, 2025
f6ecade
fix test: remove optionality of results based on enable/disable
DiannaHohensee Oct 9, 2025
8e1fa98
[CI] Auto commit changes from spotless
Oct 9, 2025
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
55 changes: 18 additions & 37 deletions server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -305,46 +305,27 @@ public void testHeapUsageEstimateIsPresent() {
}

public void testNodeWriteLoadsArePresent() {
// Disable write load decider to begin with
setWriteLoadDeciderEnablement(WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED);

InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);

// Force a ClusterInfo refresh to run collection of the node thread pool usage stats.
ClusterInfoServiceUtils.refresh(clusterInfoService);
Map<String, NodeUsageStatsForThreadPools> nodeThreadPoolStats = clusterInfoService.getClusterInfo()
.getNodeUsageStatsForThreadPools();
assertNotNull(nodeThreadPoolStats);
/** Not collecting stats yet because allocation write load stats collection is disabled by default.
* see {@link WriteLoadConstraintSettings.WRITE_LOAD_DECIDER_ENABLED_SETTING} */
assertTrue(nodeThreadPoolStats.isEmpty());

// Enable collection for node write loads.
setWriteLoadDeciderEnablement(
randomBoolean()
? WriteLoadConstraintSettings.WriteLoadDeciderStatus.ENABLED
: WriteLoadConstraintSettings.WriteLoadDeciderStatus.LOW_THRESHOLD_ONLY
);
try {
// Force a ClusterInfo refresh to run collection of the node thread pool usage stats.
ClusterInfoServiceUtils.refresh(clusterInfoService);
nodeThreadPoolStats = clusterInfoService.getClusterInfo().getNodeUsageStatsForThreadPools();

/** Verify that each node has usage stats reported. */
ClusterState state = getInstanceFromNode(ClusterService.class).state();
assertEquals(state.nodes().size(), nodeThreadPoolStats.size());
for (DiscoveryNode node : state.nodes()) {
assertTrue(nodeThreadPoolStats.containsKey(node.getId()));
NodeUsageStatsForThreadPools nodeUsageStatsForThreadPools = nodeThreadPoolStats.get(node.getId());
assertThat(nodeUsageStatsForThreadPools.nodeId(), equalTo(node.getId()));
NodeUsageStatsForThreadPools.ThreadPoolUsageStats writeThreadPoolStats = nodeUsageStatsForThreadPools
.threadPoolUsageStatsMap()
.get(ThreadPool.Names.WRITE);
assertNotNull(writeThreadPoolStats);
assertThat(writeThreadPoolStats.totalThreadPoolThreads(), greaterThanOrEqualTo(0));
assertThat(writeThreadPoolStats.averageThreadPoolUtilization(), greaterThanOrEqualTo(0.0f));
assertThat(writeThreadPoolStats.maxThreadPoolQueueLatencyMillis(), greaterThanOrEqualTo(0L));
}
} finally {
clearWriteLoadDeciderEnablementSetting();
/** Verify that each node has usage stats reported. */
ClusterState state = getInstanceFromNode(ClusterService.class).state();
assertEquals(state.nodes().size(), nodeThreadPoolStats.size());
for (DiscoveryNode node : state.nodes()) {
assertTrue(nodeThreadPoolStats.containsKey(node.getId()));
NodeUsageStatsForThreadPools nodeUsageStatsForThreadPools = nodeThreadPoolStats.get(node.getId());
assertThat(nodeUsageStatsForThreadPools.nodeId(), equalTo(node.getId()));
NodeUsageStatsForThreadPools.ThreadPoolUsageStats writeThreadPoolStats = nodeUsageStatsForThreadPools.threadPoolUsageStatsMap()
.get(ThreadPool.Names.WRITE);
assertNotNull(writeThreadPoolStats);
assertThat(writeThreadPoolStats.totalThreadPoolThreads(), greaterThanOrEqualTo(0));
assertThat(writeThreadPoolStats.averageThreadPoolUtilization(), greaterThanOrEqualTo(0.0f));
assertThat(writeThreadPoolStats.maxThreadPoolQueueLatencyMillis(), greaterThanOrEqualTo(0L));
}
}

Expand All @@ -361,10 +342,10 @@ public void testShardWriteLoadsArePresent() {

final InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);

// Explicitly disable write load decider
setWriteLoadDeciderEnablement(WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED);

try {
// Explicitly disable write load decider
setWriteLoadDeciderEnablement(WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED);

// Stats should not be collected when the decider is disabled
{
ClusterInfoServiceUtils.refresh(clusterInfoService);
Expand Down
45 changes: 18 additions & 27 deletions server/src/main/java/org/elasticsearch/cluster/InternalClusterInfoService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,7 +218,7 @@ void execute() {
maybeFetchIndicesStats(diskThresholdEnabled || writeLoadConstraintEnabled.atLeastLowThresholdEnabled());
maybeFetchNodeStats(diskThresholdEnabled || estimatedHeapThresholdEnabled);
maybeFetchNodesEstimatedHeapUsage(estimatedHeapThresholdEnabled);
maybeFetchNodesUsageStatsForThreadPools(writeLoadConstraintEnabled);
fetchNodesUsageStatsForThreadPools();
}
}

Expand Down Expand Up @@ -257,34 +257,25 @@ private void maybeFetchNodesEstimatedHeapUsage(boolean shouldFetch) {
}
}

private void maybeFetchNodesUsageStatsForThreadPools(WriteLoadDeciderStatus writeLoadConstraintEnabled) {
if (writeLoadConstraintEnabled.atLeastLowThresholdEnabled()) {
try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
fetchNodesUsageStatsForThreadPools();
}
} else {
logger.trace("skipping collecting shard/node write load estimates from cluster, feature currently disabled");
nodeThreadPoolUsageStatsPerNode = Map.of();
}
}

private void fetchNodesUsageStatsForThreadPools() {
nodeUsageStatsForThreadPoolsCollector.collectUsageStats(
client,
clusterStateSupplier.get(),
ActionListener.releaseAfter(new ActionListener<>() {
@Override
public void onResponse(Map<String, NodeUsageStatsForThreadPools> threadPoolStats) {
nodeThreadPoolUsageStatsPerNode = threadPoolStats;
}
try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
nodeUsageStatsForThreadPoolsCollector.collectUsageStats(
client,
clusterStateSupplier.get(),
ActionListener.releaseAfter(new ActionListener<>() {
@Override
public void onResponse(Map<String, NodeUsageStatsForThreadPools> threadPoolStats) {
nodeThreadPoolUsageStatsPerNode = threadPoolStats;
}

@Override
public void onFailure(Exception e) {
logger.warn("failed to fetch thread pool usage estimates for nodes", e);
nodeThreadPoolUsageStatsPerNode = Map.of();
}
}, fetchRefs.acquire())
);
@Override
public void onFailure(Exception e) {
logger.warn("failed to fetch thread pool usage estimates for nodes", e);
nodeThreadPoolUsageStatsPerNode = Map.of();
}
}, fetchRefs.acquire())
);
}
}

private void fetchNodesEstimatedHeapUsage() {
Expand Down