Add allocation stats (#105894)

This change attempts to add allocation section to the node stats in
order to simplify unbalanced clusters debugging. It is required for
https://github.com/elastic/elasticsearch/pull/97561

docs/changelog/105894.yaml

@@ -0,0 +1,5 @@
+pr: 105894
+summary: Add allocation stats
+area: Allocation
+type: enhancement
+issues: []

docs/reference/cluster/nodes-stats.asciidoc

@@ -50,6 +50,9 @@ using metrics.
   `adaptive_selection`::
       Statistics about <<search-adaptive-replica,adaptive replica selection>>.
 
+  `allocations`::
+      Statistics about allocated shards
+
   `breaker`::
       Statistics about the field data circuit breaker.
 
@@ -2802,6 +2805,44 @@ search requests on the keyed node.
 The rank of this node; used for shard selection when routing search
 requests.
 ======
+
+[[cluster-nodes-stats-api-response-body-allocations]]
+`allocations`::
+(object)
+Contains allocations statistics for the node.
++
+.Properties of `allocations`
+[%collapsible%open]
+======
+`shards`::
+(integer)
+The number of shards currently allocated to this node
+
+`undesired_shards`::
+(integer)
+The amount of shards that are scheduled to be moved elsewhere in the cluster
+or -1 other than desired balance allocator is used
+
+`forecasted_ingest_load`::
+(double)
+Total forecasted ingest load of all shards assigned to this node
+
+`forecasted_disk_usage`::
+(<<byte-units,byte value>>)
+Forecasted size of all shards assigned to the node
+
+`forecasted_disk_usage_bytes`::
+(integer)
+Forecasted size, in bytes, of all shards assigned to the node
+
+`current_disk_usage`::
+(<<byte-units,byte value>>)
+Current size of all shards assigned to the node
+
+`current_disk_usage_bytes`::
+(integer)
+Current size, in bytes, of all shards assigned to the node
+======
 =====
 ====
 

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/nodes.stats/80_allocation_stats.yml

@@ -0,0 +1,22 @@
+---
+"Allocation stats":
+  - skip:
+      version: " - 8.13.99"
+      reason: "allocation stats was added in 8.14.0"
+      features: [arbitrary_key]
+
+  - do:
+      nodes.info: {}
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      nodes.stats:
+        metric: [ allocations ]
+
+  - exists: nodes.$node_id.allocations
+  - exists: nodes.$node_id.allocations.shards
+  - exists: nodes.$node_id.allocations.undesired_shards
+  - exists: nodes.$node_id.allocations.forecasted_ingest_load
+  - exists: nodes.$node_id.allocations.forecasted_disk_usage_in_bytes
+  - exists: nodes.$node_id.allocations.current_disk_usage_in_bytes

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -141,6 +141,7 @@ public class TransportVersions {
     public static final TransportVersion ESQL_SERIALIZE_ARRAY_VECTOR = def(8_601_00_0);
     public static final TransportVersion ESQL_SERIALIZE_ARRAY_BLOCK = def(8_602_00_0);
     public static final TransportVersion ADD_DATA_STREAM_GLOBAL_RETENTION = def(8_603_00_0);
+    public static final TransportVersion ALLOCATION_STATS = def(8_604_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -12,6 +12,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.admin.cluster.allocation.TransportClusterAllocationExplainAction;
 import org.elasticsearch.action.admin.cluster.allocation.TransportDeleteDesiredBalanceAction;
+import org.elasticsearch.action.admin.cluster.allocation.TransportGetAllocationStatsAction;
 import org.elasticsearch.action.admin.cluster.allocation.TransportGetDesiredBalanceAction;
 import org.elasticsearch.action.admin.cluster.configuration.TransportAddVotingConfigExclusionsAction;
 import org.elasticsearch.action.admin.cluster.configuration.TransportClearVotingConfigExclusionsAction;
@@ -645,6 +646,7 @@ public class ActionModule extends AbstractModule {
         actions.register(TransportAddVotingConfigExclusionsAction.TYPE, TransportAddVotingConfigExclusionsAction.class);
         actions.register(TransportClearVotingConfigExclusionsAction.TYPE, TransportClearVotingConfigExclusionsAction.class);
         actions.register(TransportClusterAllocationExplainAction.TYPE, TransportClusterAllocationExplainAction.class);
+        actions.register(TransportGetAllocationStatsAction.TYPE, TransportGetAllocationStatsAction.class);
         actions.register(TransportGetDesiredBalanceAction.TYPE, TransportGetDesiredBalanceAction.class);
         actions.register(TransportDeleteDesiredBalanceAction.TYPE, TransportDeleteDesiredBalanceAction.class);
         actions.register(ClusterStatsAction.INSTANCE, TransportClusterStatsAction.class);

server/src/main/java/org/elasticsearch/action/admin/cluster/allocation/TransportGetAllocationStatsAction.java

@@ -0,0 +1,132 @@
+/*
+ * 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.action.admin.cluster.allocation;
+
+import org.elasticsearch.TransportVersions;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.master.MasterNodeReadRequest;
+import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.routing.allocation.AllocationStatsService;
+import org.elasticsearch.cluster.routing.allocation.NodeAllocationStats;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class TransportGetAllocationStatsAction extends TransportMasterNodeReadAction<
+    TransportGetAllocationStatsAction.Request,
+    TransportGetAllocationStatsAction.Response> {
+
+    public static final ActionType<TransportGetAllocationStatsAction.Response> TYPE = new ActionType<>("cluster:monitor/allocation/stats");
+
+    private final AllocationStatsService allocationStatsService;
+
+    @Inject
+    public TransportGetAllocationStatsAction(
+        TransportService transportService,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ActionFilters actionFilters,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        AllocationStatsService allocationStatsService
+    ) {
+        super(
+            TYPE.name(),
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            TransportGetAllocationStatsAction.Request::new,
+            indexNameExpressionResolver,
+            TransportGetAllocationStatsAction.Response::new,
+            threadPool.executor(ThreadPool.Names.MANAGEMENT)
+        );
+        this.allocationStatsService = allocationStatsService;
+    }
+
+    @Override
+    protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
+        if (clusterService.state().getMinTransportVersion().before(TransportVersions.ALLOCATION_STATS)) {
+            // The action is not available before ALLOCATION_STATS
+            listener.onResponse(new Response(Map.of()));
+            return;
+        }
+        super.doExecute(task, request, listener);
+    }
+
+    @Override
+    protected void masterOperation(Task task, Request request, ClusterState state, ActionListener<Response> listener) throws Exception {
+        listener.onResponse(new Response(allocationStatsService.stats()));
+    }
+
+    @Override
+    protected ClusterBlockException checkBlock(Request request, ClusterState state) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+    }
+
+    public static class Request extends MasterNodeReadRequest<Request> {
+
+        public Request(TaskId parentTaskId) {
+            setParentTask(parentTaskId);
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            assert out.getTransportVersion().onOrAfter(TransportVersions.ALLOCATION_STATS);
+            super.writeTo(out);
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+    }
+
+    public static class Response extends ActionResponse {
+
+        private final Map<String, NodeAllocationStats> nodeAllocationStats;
+
+        public Response(Map<String, NodeAllocationStats> nodeAllocationStats) {
+            this.nodeAllocationStats = nodeAllocationStats;
+        }
+
+        public Response(StreamInput in) throws IOException {
+            super(in);
+            this.nodeAllocationStats = in.readImmutableMap(StreamInput::readString, NodeAllocationStats::new);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeMap(nodeAllocationStats, StreamOutput::writeString, StreamOutput::writeWriteable);
+        }
+
+        public Map<String, NodeAllocationStats> getNodeAllocationStats() {
+            return nodeAllocationStats;
+        }
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStats.java

@@ -12,6 +12,7 @@ import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.support.nodes.BaseNodeResponse;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.cluster.routing.allocation.NodeAllocationStats;
 import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -97,6 +98,9 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
     @Nullable
     private final RepositoriesStats repositoriesStats;
 
+    @Nullable
+    private final NodeAllocationStats nodeAllocationStats;
+
     public NodeStats(StreamInput in) throws IOException {
         super(in);
         timestamp = in.readVLong();
@@ -117,11 +121,12 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
         ingestStats = in.readOptionalWriteable(IngestStats::read);
         adaptiveSelectionStats = in.readOptionalWriteable(AdaptiveSelectionStats::new);
         indexingPressureStats = in.readOptionalWriteable(IndexingPressureStats::new);
-        if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X)) {
-            repositoriesStats = in.readOptionalWriteable(RepositoriesStats::new);
-        } else {
-            repositoriesStats = null;
-        }
+        repositoriesStats = in.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X)
+            ? in.readOptionalWriteable(RepositoriesStats::new)
+            : null;
+        nodeAllocationStats = in.getTransportVersion().onOrAfter(TransportVersions.ALLOCATION_STATS)
+            ? in.readOptionalWriteable(NodeAllocationStats::new)
+            : null;
     }
 
     public NodeStats(
@@ -142,7 +147,8 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
         @Nullable AdaptiveSelectionStats adaptiveSelectionStats,
         @Nullable ScriptCacheStats scriptCacheStats,
         @Nullable IndexingPressureStats indexingPressureStats,
-        @Nullable RepositoriesStats repositoriesStats
+        @Nullable RepositoriesStats repositoriesStats,
+        @Nullable NodeAllocationStats nodeAllocationStats
     ) {
         super(node);
         this.timestamp = timestamp;
@@ -162,6 +168,31 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
         this.scriptCacheStats = scriptCacheStats;
         this.indexingPressureStats = indexingPressureStats;
         this.repositoriesStats = repositoriesStats;
+        this.nodeAllocationStats = nodeAllocationStats;
+    }
+
+    public NodeStats withNodeAllocationStats(@Nullable NodeAllocationStats nodeAllocationStats) {
+        return new NodeStats(
+            getNode(),
+            timestamp,
+            indices,
+            os,
+            process,
+            jvm,
+            threadPool,
+            fs,
+            transport,
+            http,
+            breaker,
+            scriptStats,
+            discoveryStats,
+            ingestStats,
+            adaptiveSelectionStats,
+            scriptCacheStats,
+            indexingPressureStats,
+            repositoriesStats,
+            nodeAllocationStats
+        );
     }
 
     public long getTimestamp() {
@@ -271,6 +302,11 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
         return repositoriesStats;
     }
 
+    @Nullable
+    public NodeAllocationStats getNodeAllocationStats() {
+        return nodeAllocationStats;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
@@ -297,6 +333,9 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
         if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_9_X)) {
             out.writeOptionalWriteable(repositoriesStats);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ALLOCATION_STATS)) {
+            out.writeOptionalWriteable(nodeAllocationStats);
+        }
     }
 
     @Override
@@ -343,7 +382,11 @@ public class NodeStats extends BaseNodeResponse implements ChunkedToXContent {
             ifPresent(getIngestStats()).toXContentChunked(outerParams),
             singleChunk(ifPresent(getAdaptiveSelectionStats())),
             ifPresent(getScriptCacheStats()).toXContentChunked(outerParams),
-            singleChunk((builder, p) -> builder.value(ifPresent(getIndexingPressureStats()), p).value(ifPresent(getRepositoriesStats()), p))
+            singleChunk(
+                (builder, p) -> builder.value(ifPresent(getIndexingPressureStats()), p)
+                    .value(ifPresent(getRepositoriesStats()), p)
+                    .value(ifPresent(getNodeAllocationStats()), p)
+            )
         );
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsRequestBuilder.java

@@ -158,6 +158,11 @@ public class NodesStatsRequestBuilder extends NodesOperationRequestBuilder<
         return this;
     }
 
+    public NodesStatsRequestBuilder setAllocationStats(boolean allocationStats) {
+        addOrRemoveMetric(allocationStats, NodesStatsRequestParameters.Metric.ALLOCATIONS);
+        return this;
+    }
+
     /**
      * Helper method for adding metrics to a request
      */

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsRequestParameters.java

@@ -89,7 +89,8 @@ public class NodesStatsRequestParameters implements Writeable {
         ADAPTIVE_SELECTION("adaptive_selection"),
         SCRIPT_CACHE("script_cache"),
         INDEXING_PRESSURE("indexing_pressure"),
-        REPOSITORIES("repositories");
+        REPOSITORIES("repositories"),
+        ALLOCATIONS("allocations");
 
         private String metricName;
 

server/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -8,11 +8,15 @@
 
 package org.elasticsearch.action.admin.cluster.node.stats;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.admin.cluster.allocation.TransportGetAllocationStatsAction;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.NodeAllocationStats;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
@@ -42,7 +46,9 @@ public class TransportNodesStatsAction extends TransportNodesAction<
     NodeStats> {
 
     public static final ActionType<NodesStatsResponse> TYPE = new ActionType<>("cluster:monitor/nodes/stats");
+
     private final NodeService nodeService;
+    private final NodeClient client;
 
     @Inject
     public TransportNodesStatsAction(
@@ -50,7 +56,8 @@ public class TransportNodesStatsAction extends TransportNodesAction<
         ClusterService clusterService,
         TransportService transportService,
         NodeService nodeService,
-        ActionFilters actionFilters
+        ActionFilters actionFilters,
+        NodeClient client
     ) {
         super(
             TYPE.name(),
@@ -61,6 +68,7 @@ public class TransportNodesStatsAction extends TransportNodesAction<
             threadPool.executor(ThreadPool.Names.MANAGEMENT)
         );
         this.nodeService = nodeService;
+        this.client = client;
     }
 
     @Override
@@ -68,6 +76,34 @@ public class TransportNodesStatsAction extends TransportNodesAction<
         return new NodesStatsResponse(clusterService.getClusterName(), responses, failures);
     }
 
+    @Override
+    protected void newResponseAsync(
+        Task task,
+        NodesStatsRequest request,
+        List<NodeStats> responses,
+        List<FailedNodeException> failures,
+        ActionListener<NodesStatsResponse> listener
+    ) {
+        Set<String> metrics = request.getNodesStatsRequestParameters().requestedMetrics();
+        if (NodesStatsRequestParameters.Metric.ALLOCATIONS.containedIn(metrics)) {
+            client.execute(
+                TransportGetAllocationStatsAction.TYPE,
+                new TransportGetAllocationStatsAction.Request(new TaskId(clusterService.localNode().getId(), task.getId())),
+                listener.delegateFailure((l, r) -> {
+                    ActionListener.respondAndRelease(l, newResponse(request, merge(responses, r.getNodeAllocationStats()), failures));
+                })
+            );
+        } else {
+            ActionListener.run(listener, l -> ActionListener.respondAndRelease(l, newResponse(request, responses, failures)));
+        }
+    }
+
+    private static List<NodeStats> merge(List<NodeStats> responses, Map<String, NodeAllocationStats> allocationStats) {
+        return responses.stream()
+            .map(response -> response.withNodeAllocationStats(allocationStats.get(response.getNode().getId())))
+            .toList();
+    }
+
     @Override
     protected NodeStatsRequest newNodeRequest(NodesStatsRequest request) {
         return new NodeStatsRequest(request);
@@ -80,10 +116,10 @@ public class TransportNodesStatsAction extends TransportNodesAction<
     }
 
     @Override
-    protected NodeStats nodeOperation(NodeStatsRequest nodeStatsRequest, Task task) {
+    protected NodeStats nodeOperation(NodeStatsRequest request, Task task) {
         assert task instanceof CancellableTask;
 
-        final NodesStatsRequestParameters nodesStatsRequestParameters = nodeStatsRequest.getNodesStatsRequestParameters();
+        final NodesStatsRequestParameters nodesStatsRequestParameters = request.getNodesStatsRequestParameters();
         Set<String> metrics = nodesStatsRequestParameters.requestedMetrics();
         return nodeService.stats(
             nodesStatsRequestParameters.indices(),

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

@@ -29,6 +29,7 @@ import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingRoleStrategy;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.AllocationService.RerouteStrategy;
+import org.elasticsearch.cluster.routing.allocation.AllocationStatsService;
 import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
@@ -118,6 +119,7 @@ public class ClusterModule extends AbstractModule {
     final Collection<AllocationDecider> deciderList;
     final ShardsAllocator shardsAllocator;
     private final ShardRoutingRoleStrategy shardRoutingRoleStrategy;
+    private final AllocationStatsService allocationStatsService;
 
     public ClusterModule(
         Settings settings,
@@ -154,6 +156,7 @@ public class ClusterModule extends AbstractModule {
             shardRoutingRoleStrategy
         );
         this.metadataDeleteIndexService = new MetadataDeleteIndexService(settings, clusterService, allocationService);
+        this.allocationStatsService = new AllocationStatsService(clusterService, clusterInfoService, shardsAllocator, writeLoadForecaster);
     }
 
     static ShardRoutingRoleStrategy getShardRoutingRoleStrategy(List<ClusterPlugin> clusterPlugins) {
@@ -440,6 +443,7 @@ public class ClusterModule extends AbstractModule {
         bind(AllocationDeciders.class).toInstance(allocationDeciders);
         bind(ShardsAllocator.class).toInstance(shardsAllocator);
         bind(ShardRoutingRoleStrategy.class).toInstance(shardRoutingRoleStrategy);
+        bind(AllocationStatsService.class).toInstance(allocationStatsService);
     }
 
     public void setExistingShardsAllocators(GatewayAllocator gatewayAllocator) {

server/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationStatsService.java

@@ -0,0 +1,94 @@
+/*
+ * 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;
+
+import org.elasticsearch.cluster.ClusterInfoService;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.routing.RoutingNode;
+import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalance;
+import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceShardsAllocator;
+import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.util.Maps;
+
+import java.util.Map;
+
+public class AllocationStatsService {
+
+    private final ClusterService clusterService;
+    private final ClusterInfoService clusterInfoService;
+    private final DesiredBalanceShardsAllocator desiredBalanceShardsAllocator;
+    private final WriteLoadForecaster writeLoadForecaster;
+
+    public AllocationStatsService(
+        ClusterService clusterService,
+        ClusterInfoService clusterInfoService,
+        ShardsAllocator shardsAllocator,
+        WriteLoadForecaster writeLoadForecaster
+    ) {
+        this.clusterService = clusterService;
+        this.clusterInfoService = clusterInfoService;
+        this.desiredBalanceShardsAllocator = shardsAllocator instanceof DesiredBalanceShardsAllocator allocator ? allocator : null;
+        this.writeLoadForecaster = writeLoadForecaster;
+    }
+
+    public Map<String, NodeAllocationStats> stats() {
+        var state = clusterService.state();
+        var info = clusterInfoService.getClusterInfo();
+        var desiredBalance = desiredBalanceShardsAllocator != null ? desiredBalanceShardsAllocator.getDesiredBalance() : null;
+
+        var stats = Maps.<String, NodeAllocationStats>newMapWithExpectedSize(state.getRoutingNodes().size());
+        for (RoutingNode node : state.getRoutingNodes()) {
+            int shards = 0;
+            int undesiredShards = 0;
+            double forecastedWriteLoad = 0.0;
+            long forecastedDiskUsage = 0;
+            long currentDiskUsage = 0;
+            for (ShardRouting shardRouting : node) {
+                if (shardRouting.relocating()) {
+                    continue;
+                }
+                shards++;
+                IndexMetadata indexMetadata = state.metadata().getIndexSafe(shardRouting.index());
+                if (isDesiredAllocation(desiredBalance, shardRouting) == false) {
+                    undesiredShards++;
+                }
+                long shardSize = info.getShardSize(shardRouting.shardId(), shardRouting.primary(), 0);
+                forecastedWriteLoad += writeLoadForecaster.getForecastedWriteLoad(indexMetadata).orElse(0.0);
+                forecastedDiskUsage += Math.max(indexMetadata.getForecastedShardSizeInBytes().orElse(0), shardSize);
+                currentDiskUsage += shardSize;
+
+            }
+            stats.put(
+                node.nodeId(),
+                new NodeAllocationStats(
+                    shards,
+                    desiredBalanceShardsAllocator != null ? undesiredShards : -1,
+                    forecastedWriteLoad,
+                    forecastedDiskUsage,
+                    currentDiskUsage
+                )
+            );
+        }
+
+        return stats;
+    }
+
+    private static boolean isDesiredAllocation(DesiredBalance desiredBalance, ShardRouting shardRouting) {
+        if (desiredBalance == null) {
+            return true;
+        }
+        var assignment = desiredBalance.getAssignment(shardRouting.shardId());
+        if (assignment == null) {
+            return false;
+        }
+        return assignment.nodeIds().contains(shardRouting.currentNodeId());
+    }
+}

server/src/main/java/org/elasticsearch/cluster/routing/allocation/NodeAllocationStats.java

@@ -0,0 +1,51 @@
+/*
+ * 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;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.xcontent.ToXContentFragment;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+public record NodeAllocationStats(
+    int shards,
+    int undesiredShards,
+    double forecastedIngestLoad,
+    long forecastedDiskUsage,
+    long currentDiskUsage
+) implements Writeable, ToXContentFragment {
+
+    public NodeAllocationStats(StreamInput in) throws IOException {
+        this(in.readVInt(), in.readVInt(), in.readDouble(), in.readVLong(), in.readVLong());
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeVInt(shards);
+        out.writeVInt(undesiredShards);
+        out.writeDouble(forecastedIngestLoad);
+        out.writeVLong(forecastedDiskUsage);
+        out.writeVLong(currentDiskUsage);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return builder.startObject("allocations")
+            .field("shards", shards)
+            .field("undesired_shards", undesiredShards)
+            .field("forecasted_ingest_load", forecastedIngestLoad)
+            .humanReadableField("forecasted_disk_usage_in_bytes", "forecasted_disk_usage", ByteSizeValue.ofBytes(forecastedDiskUsage))
+            .humanReadableField("current_disk_usage_in_bytes", "current_disk_usage", ByteSizeValue.ofBytes(currentDiskUsage))
+            .endObject();
+    }
+}

server/src/main/java/org/elasticsearch/node/NodeService.java

@@ -195,7 +195,8 @@ public class NodeService implements Closeable {
             adaptiveSelection ? responseCollectorService.getAdaptiveStats(searchTransportService.getPendingSearchRequests()) : null,
             scriptCache ? scriptService.cacheStats() : null,
             indexingPressure ? this.indexingPressure.stats() : null,
-            repositoriesStats ? this.repositoriesService.getRepositoriesThrottlingStats() : null
+            repositoriesStats ? this.repositoriesService.getRepositoriesThrottlingStats() : null,
+            null
         );
     }
 

server/src/test/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStatsTests.java

@@ -21,6 +21,7 @@ import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
 import org.elasticsearch.cluster.routing.RecoverySource;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
+import org.elasticsearch.cluster.routing.allocation.NodeAllocationStats;
 import org.elasticsearch.cluster.service.ClusterApplierRecordingService;
 import org.elasticsearch.cluster.service.ClusterApplierRecordingService.Stats.Recording;
 import org.elasticsearch.cluster.service.ClusterStateUpdateStats;
@@ -1043,6 +1044,13 @@ public class NodeStatsTests extends ESTestCase {
         RepositoriesStats repositoriesStats = new RepositoriesStats(
             Map.of("test-repository", new RepositoriesStats.ThrottlingStats(100, 200))
         );
+        NodeAllocationStats nodeAllocationStats = new NodeAllocationStats(
+            randomIntBetween(0, 10000),
+            randomIntBetween(0, 1000),
+            randomDoubleBetween(0, 8, true),
+            randomNonNegativeLong(),
+            randomNonNegativeLong()
+        );
 
         return new NodeStats(
             node,
@@ -1062,7 +1070,8 @@ public class NodeStatsTests extends ESTestCase {
             adaptiveSelectionStats,
             scriptCacheStats,
             indexingPressureStats,
-            repositoriesStats
+            repositoriesStats,
+            nodeAllocationStats
         );
     }
 

server/src/test/java/org/elasticsearch/cluster/DiskUsageTests.java

@@ -183,6 +183,7 @@ public class DiskUsageTests extends ESTestCase {
                 null,
                 null,
                 null,
+                null,
                 null
             );
             DiskUsage leastNode = DiskUsage.findLeastAvailablePath(nodeStats);
@@ -211,6 +212,7 @@ public class DiskUsageTests extends ESTestCase {
                 null,
                 null,
                 null,
+                null,
                 null
             );
             DiskUsage leastNode = DiskUsage.findLeastAvailablePath(nodeStats);
@@ -241,6 +243,7 @@ public class DiskUsageTests extends ESTestCase {
                 null,
                 null,
                 null,
+                null,
                 null
             );
             DiskUsage leastNode = DiskUsage.findLeastAvailablePath(nodeStats);
@@ -275,6 +278,7 @@ public class DiskUsageTests extends ESTestCase {
                 null,
                 null,
                 null,
+                null,
                 null
             );
             DiskUsage leastNode = DiskUsage.findLeastAvailablePath(nodeStats);
@@ -304,6 +308,7 @@ public class DiskUsageTests extends ESTestCase {
                 null,
                 null,
                 null,
+                null,
                 null
             );
             DiskUsage leastNode = DiskUsage.findLeastAvailablePath(nodeStats);
@@ -334,6 +339,7 @@ public class DiskUsageTests extends ESTestCase {
                 null,
                 null,
                 null,
+                null,
                 null
             );
 

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

@@ -0,0 +1,205 @@
+/*
+ * 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;
+
+import org.elasticsearch.cluster.ClusterInfo;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ESAllocationTestCase;
+import org.elasticsearch.cluster.EmptyClusterInfoService;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.routing.IndexRoutingTable;
+import org.elasticsearch.cluster.routing.RoutingTable;
+import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.ShardRoutingState;
+import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalance;
+import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceShardsAllocator;
+import org.elasticsearch.cluster.routing.allocation.allocator.ShardAssignment;
+import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
+import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.telemetry.TelemetryProvider;
+import org.elasticsearch.test.ClusterServiceUtils;
+
+import java.util.Map;
+import java.util.Set;
+
+import static org.elasticsearch.cluster.routing.TestShardRouting.newShardRouting;
+import static org.elasticsearch.cluster.routing.TestShardRouting.shardRoutingBuilder;
+import static org.hamcrest.Matchers.aMapWithSize;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.hasEntry;
+
+public class AllocationStatsServiceTests extends ESAllocationTestCase {
+
+    public void testShardStats() {
+
+        var ingestLoadForecast = randomDoubleBetween(0, 10, true);
+        var shardSizeForecast = randomNonNegativeLong();
+        var currentShardSize = randomNonNegativeLong();
+
+        var indexMetadata = IndexMetadata.builder("my-index")
+            .settings(indexSettings(IndexVersion.current(), 1, 0))
+            .indexWriteLoadForecast(ingestLoadForecast)
+            .shardSizeInBytesForecast(shardSizeForecast)
+            .build();
+        var shardId = new ShardId(indexMetadata.getIndex(), 0);
+
+        var state = ClusterState.builder(ClusterName.DEFAULT)
+            .nodes(DiscoveryNodes.builder().add(newNode("node-1")))
+            .metadata(Metadata.builder().put(indexMetadata, false))
+            .routingTable(
+                RoutingTable.builder()
+                    .add(
+                        IndexRoutingTable.builder(indexMetadata.getIndex())
+                            .addShard(newShardRouting(shardId, "node-1", true, ShardRoutingState.STARTED))
+                            .build()
+                    )
+            )
+            .build();
+
+        var clusterInfo = new ClusterInfo(
+            Map.of(),
+            Map.of(),
+            Map.of(ClusterInfo.shardIdentifierFromRouting(shardId, true), currentShardSize),
+            Map.of(),
+            Map.of(),
+            Map.of()
+        );
+
+        var queue = new DeterministicTaskQueue();
+        try (var clusterService = ClusterServiceUtils.createClusterService(state, queue.getThreadPool())) {
+            var service = new AllocationStatsService(clusterService, () -> clusterInfo, createShardAllocator(), TEST_WRITE_LOAD_FORECASTER);
+            assertThat(
+                service.stats(),
+                allOf(
+                    aMapWithSize(1),
+                    hasEntry(
+                        "node-1",
+                        new NodeAllocationStats(1, -1, ingestLoadForecast, Math.max(shardSizeForecast, currentShardSize), currentShardSize)
+                    )
+                )
+            );
+        }
+    }
+
+    public void testRelocatingShardIsOnlyCountedOnceOnTargetNode() {
+
+        var indexMetadata = IndexMetadata.builder("my-index").settings(indexSettings(IndexVersion.current(), 1, 0)).build();
+        var state = ClusterState.builder(ClusterName.DEFAULT)
+            .nodes(DiscoveryNodes.builder().add(newNode("node-1")).add(newNode("node-2")))
+            .metadata(Metadata.builder().put(indexMetadata, false))
+            .routingTable(
+                RoutingTable.builder()
+                    .add(
+                        IndexRoutingTable.builder(indexMetadata.getIndex())
+                            .addShard(
+                                shardRoutingBuilder(new ShardId(indexMetadata.getIndex(), 0), "node-1", true, ShardRoutingState.RELOCATING)
+                                    .withRelocatingNodeId("node-2")
+                                    .build()
+                            )
+                            .build()
+                    )
+            )
+            .build();
+
+        var queue = new DeterministicTaskQueue();
+        try (var clusterService = ClusterServiceUtils.createClusterService(state, queue.getThreadPool())) {
+            var service = new AllocationStatsService(
+                clusterService,
+                EmptyClusterInfoService.INSTANCE,
+                createShardAllocator(),
+                TEST_WRITE_LOAD_FORECASTER
+            );
+            assertThat(
+                service.stats(),
+                allOf(
+                    aMapWithSize(2),
+                    hasEntry("node-1", new NodeAllocationStats(0, -1, 0, 0, 0)),
+                    hasEntry("node-2", new NodeAllocationStats(1, -1, 0, 0, 0))
+                )
+            );
+        }
+    }
+
+    public void testUndesiredShardCount() {
+
+        var indexMetadata = IndexMetadata.builder("my-index").settings(indexSettings(IndexVersion.current(), 2, 0)).build();
+
+        var state = ClusterState.builder(ClusterName.DEFAULT)
+            .nodes(DiscoveryNodes.builder().add(newNode("node-1")).add(newNode("node-2")).add(newNode("node-3")))
+            .metadata(Metadata.builder().put(indexMetadata, false))
+            .routingTable(
+                RoutingTable.builder()
+                    .add(
+                        IndexRoutingTable.builder(indexMetadata.getIndex())
+                            .addShard(newShardRouting(new ShardId(indexMetadata.getIndex(), 0), "node-1", true, ShardRoutingState.STARTED))
+                            .addShard(newShardRouting(new ShardId(indexMetadata.getIndex(), 1), "node-3", true, ShardRoutingState.STARTED))
+                            .build()
+                    )
+            )
+            .build();
+
+        var queue = new DeterministicTaskQueue();
+        var threadPool = queue.getThreadPool();
+        try (var clusterService = ClusterServiceUtils.createClusterService(state, threadPool)) {
+            var service = new AllocationStatsService(
+                clusterService,
+                EmptyClusterInfoService.INSTANCE,
+                new DesiredBalanceShardsAllocator(
+                    ClusterSettings.createBuiltInClusterSettings(),
+                    createShardAllocator(),
+                    threadPool,
+                    clusterService,
+                    (innerState, strategy) -> innerState,
+                    TelemetryProvider.NOOP
+                ) {
+                    @Override
+                    public DesiredBalance getDesiredBalance() {
+                        return new DesiredBalance(
+                            1,
+                            Map.ofEntries(
+                                Map.entry(new ShardId(indexMetadata.getIndex(), 0), new ShardAssignment(Set.of("node-1"), 1, 0, 0)),
+                                Map.entry(new ShardId(indexMetadata.getIndex(), 1), new ShardAssignment(Set.of("node-2"), 1, 0, 0))
+                            )
+                        );
+                    }
+                },
+                TEST_WRITE_LOAD_FORECASTER
+            );
+            assertThat(
+                service.stats(),
+                allOf(
+                    aMapWithSize(3),
+                    hasEntry("node-1", new NodeAllocationStats(1, 0, 0, 0, 0)),
+                    hasEntry("node-2", new NodeAllocationStats(0, 0, 0, 0, 0)),
+                    hasEntry("node-3", new NodeAllocationStats(1, 1, 0, 0, 0)) // [my-index][1] should be allocated to [node-2]
+                )
+            );
+        }
+    }
+
+    private ShardsAllocator createShardAllocator() {
+        return new ShardsAllocator() {
+            @Override
+            public void allocate(RoutingAllocation allocation) {
+
+            }
+
+            @Override
+            public ShardAllocationDecision decideShardAllocation(ShardRouting shard, RoutingAllocation allocation) {
+                return null;
+            }
+        };
+    }
+}

server/src/test/java/org/elasticsearch/cluster/routing/allocation/NodeAllocationStatsTests.java

@@ -0,0 +1,76 @@
+/*
+ * 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;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.test.ESTestCase;
+
+import java.io.IOException;
+
+public class NodeAllocationStatsTests extends AbstractWireSerializingTestCase<NodeAllocationStats> {
+
+    @Override
+    protected Writeable.Reader<NodeAllocationStats> instanceReader() {
+        return NodeAllocationStats::new;
+    }
+
+    @Override
+    protected NodeAllocationStats createTestInstance() {
+        return new NodeAllocationStats(
+            randomIntBetween(0, 10000),
+            randomIntBetween(0, 1000),
+            randomDoubleBetween(0, 8, true),
+            randomNonNegativeLong(),
+            randomNonNegativeLong()
+        );
+    }
+
+    @Override
+    protected NodeAllocationStats mutateInstance(NodeAllocationStats instance) throws IOException {
+        return switch (randomInt(4)) {
+            case 0 -> new NodeAllocationStats(
+                randomValueOtherThan(instance.shards(), () -> randomIntBetween(0, 10000)),
+                instance.undesiredShards(),
+                instance.forecastedIngestLoad(),
+                instance.forecastedDiskUsage(),
+                instance.currentDiskUsage()
+            );
+            case 1 -> new NodeAllocationStats(
+                instance.shards(),
+                randomValueOtherThan(instance.undesiredShards(), () -> randomIntBetween(0, 1000)),
+                instance.forecastedIngestLoad(),
+                instance.forecastedDiskUsage(),
+                instance.currentDiskUsage()
+            );
+            case 2 -> new NodeAllocationStats(
+                instance.shards(),
+                instance.undesiredShards(),
+                randomValueOtherThan(instance.forecastedIngestLoad(), () -> randomDoubleBetween(0, 8, true)),
+                instance.forecastedDiskUsage(),
+                instance.currentDiskUsage()
+            );
+            case 3 -> new NodeAllocationStats(
+                instance.shards(),
+                instance.undesiredShards(),
+                instance.forecastedIngestLoad(),
+                randomValueOtherThan(instance.forecastedDiskUsage(), ESTestCase::randomNonNegativeLong),
+                instance.currentDiskUsage()
+            );
+            case 4 -> new NodeAllocationStats(
+                instance.shards(),
+                instance.undesiredShards(),
+                instance.forecastedIngestLoad(),
+                instance.currentDiskUsage(),
+                randomValueOtherThan(instance.forecastedDiskUsage(), ESTestCase::randomNonNegativeLong)
+            );
+            default -> throw new RuntimeException("unreachable");
+        };
+    }
+}

server/src/test/java/org/elasticsearch/health/node/tracker/DiskHealthTrackerTests.java

@@ -325,6 +325,7 @@ public class DiskHealthTrackerTests extends ESTestCase {
             null,
             null,
             null,
+            null,
             null
         );
     }

server/src/test/java/org/elasticsearch/rest/action/info/RestClusterInfoActionTests.java

@@ -121,6 +121,7 @@ public class RestClusterInfoActionTests extends ESTestCase {
             null,
             null,
             null,
+            null,
             null
         );
     }

test/framework/src/main/java/org/elasticsearch/cluster/MockInternalClusterInfoService.java

@@ -89,7 +89,8 @@ public class MockInternalClusterInfoService extends InternalClusterInfoService {
                 nodeStats.getAdaptiveSelectionStats(),
                 nodeStats.getScriptCacheStats(),
                 nodeStats.getIndexingPressureStats(),
-                nodeStats.getRepositoriesStats()
+                nodeStats.getRepositoriesStats(),
+                nodeStats.getNodeAllocationStats()
             );
         }).collect(Collectors.toList());
     }

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/capacity/nodeinfo/AutoscalingNodesInfoServiceTests.java

@@ -443,6 +443,7 @@ public class AutoscalingNodesInfoServiceTests extends AutoscalingTestCase {
             null,
             null,
             null,
+            null,
             null
         );
     }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/action/TransportGetTrainedModelsStatsActionTests.java

@@ -274,6 +274,7 @@ public class TransportGetTrainedModelsStatsActionTests extends ESTestCase {
             null,
             null,
             null,
+            null,
             null
         );
 

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/node/NodeStatsMonitoringDocTests.java

@@ -460,6 +460,7 @@ public class NodeStatsMonitoringDocTests extends BaseFilteredMonitoringDocTestCa
             null,
             null,
             null,
+            null,
             null
         );
     }

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/Constants.java

@@ -340,6 +340,7 @@ public class Constants {
         "cluster:monitor/nodes/info",
         "cluster:monitor/nodes/stats",
         "cluster:monitor/nodes/usage",
+        "cluster:monitor/allocation/stats",
         "cluster:monitor/profiling/status/get",
         "cluster:monitor/remote/info",
         "cluster:monitor/settings",