allocation: add balancer round summary as metrics (#136043)

This commit adds the BalancerRoundSummary as a collection of APM/open telemetry
metrics. (These are already logged.) The summary provided every balancing round
is sent synchronously into a counter or histogram. The round counter is
implicitly incremented with each report. The shards moved are recorded with a
counter and histogram. The per-node metrics, shard count, disk usage, write
load, and total weight, are recorded into histograms with node_id and node_name
attributes set. This change also includes a minor refactoring of the
BalancingRoundSummary to move its key from node name to DiscoveryNode. This
preserves the node id and name for use in histogram per-node attributes.

Author: schase-es

docs/changelog/136043.yaml

@@ -0,0 +1,5 @@
+pr: 136043
+summary: "Allocation: add balancer round summary as metrics"
+area: Allocation
+type: enhancement
+issues: []

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

@@ -41,6 +41,7 @@
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
+import org.elasticsearch.cluster.routing.allocation.allocator.AllocationBalancingRoundMetrics;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancerSettings;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancingWeightsFactory;
@@ -141,6 +142,7 @@ public class ClusterModule extends AbstractModule {
     private final AllocationStatsService allocationStatsService;
     private final TelemetryProvider telemetryProvider;
     private final DesiredBalanceMetrics desiredBalanceMetrics;
+    private final AllocationBalancingRoundMetrics balancingRoundMetrics;
 
     public ClusterModule(
         Settings settings,
@@ -168,6 +170,7 @@ public ClusterModule(
             balancingWeightsFactory
         );
         this.desiredBalanceMetrics = new DesiredBalanceMetrics(telemetryProvider.getMeterRegistry());
+        this.balancingRoundMetrics = new AllocationBalancingRoundMetrics(telemetryProvider.getMeterRegistry());
         this.shardsAllocator = createShardsAllocator(
             settings,
             clusterService.getClusterSettings(),
@@ -180,7 +183,8 @@ public ClusterModule(
             writeLoadForecaster,
             nodeAllocationStatsAndWeightsCalculator,
             this::explainShardAllocation,
-            desiredBalanceMetrics
+            desiredBalanceMetrics,
+            balancingRoundMetrics
         );
         this.clusterService = clusterService;
         this.indexNameExpressionResolver = new IndexNameExpressionResolver(threadPool.getThreadContext(), systemIndices, projectResolver);
@@ -521,7 +525,8 @@ private static ShardsAllocator createShardsAllocator(
         WriteLoadForecaster writeLoadForecaster,
         NodeAllocationStatsAndWeightsCalculator nodeAllocationStatsAndWeightsCalculator,
         ShardAllocationExplainer shardAllocationExplainer,
-        DesiredBalanceMetrics desiredBalanceMetrics
+        DesiredBalanceMetrics desiredBalanceMetrics,
+        AllocationBalancingRoundMetrics balancingRoundMetrics
     ) {
         Map<String, Supplier<ShardsAllocator>> allocators = new HashMap<>();
         allocators.put(
@@ -538,7 +543,8 @@ private static ShardsAllocator createShardsAllocator(
                 reconciler,
                 nodeAllocationStatsAndWeightsCalculator,
                 shardAllocationExplainer,
-                desiredBalanceMetrics
+                desiredBalanceMetrics,
+                balancingRoundMetrics
             )
         );
 

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

@@ -0,0 +1,114 @@
+/*
+ * 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.allocator;
+
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.allocator.BalancingRoundSummary.NodesWeightsChanges;
+import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.telemetry.metric.DoubleHistogram;
+import org.elasticsearch.telemetry.metric.LongCounter;
+import org.elasticsearch.telemetry.metric.LongHistogram;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
+
+import java.util.Map;
+
+/**
+ * A telemetry metrics sender for {@link BalancingRoundSummary}
+ */
+public class AllocationBalancingRoundMetrics {
+
+    // counters that measure rounds and moves from the last balancing round
+    public static final String NUMBER_OF_BALANCING_ROUNDS_METRIC_NAME = "es.allocator.balancing_round.balancing_rounds.total";
+    public static final String NUMBER_OF_SHARD_MOVES_METRIC_NAME = "es.allocator.balancing_round.shard_moves.total";
+    public static final String NUMBER_OF_SHARD_MOVES_HISTOGRAM_METRIC_NAME = "es.allocator.balancing_round.shard_moves.histogram";
+
+    // histograms that measure current utilization
+    public static final String NUMBER_OF_SHARDS_METRIC_NAME = "es.allocator.balancing_round.shard_count.histogram";
+    public static final String DISK_USAGE_BYTES_METRIC_NAME = "es.allocator.balancing_round.disk_usage_bytes.histogram";
+    public static final String WRITE_LOAD_METRIC_NAME = "es.allocator.balancing_round.write_load.histogram";
+    public static final String TOTAL_WEIGHT_METRIC_NAME = "es.allocator.balancing_round.total_weight.histogram";
+
+    private final LongCounter balancingRoundCounter;
+    private final LongCounter shardMovesCounter;
+    private final LongHistogram shardMovesHistogram;
+
+    private final LongHistogram shardCountHistogram;
+    private final DoubleHistogram diskUsageHistogram;
+    private final DoubleHistogram writeLoadHistogram;
+    private final DoubleHistogram totalWeightHistogram;
+
+    public static AllocationBalancingRoundMetrics NOOP = new AllocationBalancingRoundMetrics(MeterRegistry.NOOP);
+
+    public AllocationBalancingRoundMetrics(MeterRegistry meterRegistry) {
+        this.balancingRoundCounter = meterRegistry.registerLongCounter(
+            NUMBER_OF_BALANCING_ROUNDS_METRIC_NAME,
+            "Total number of balancing rounds",
+            "unit"
+        );
+        this.shardMovesCounter = meterRegistry.registerLongCounter(
+            NUMBER_OF_SHARD_MOVES_METRIC_NAME,
+            "Total number of shard moves",
+            "unit"
+        );
+
+        this.shardMovesHistogram = meterRegistry.registerLongHistogram(
+            NUMBER_OF_SHARD_MOVES_HISTOGRAM_METRIC_NAME,
+            "Number of shard movements executed in a balancing round",
+            "unit"
+        );
+        this.shardCountHistogram = meterRegistry.registerLongHistogram(
+            NUMBER_OF_SHARDS_METRIC_NAME,
+            "change in node shard count per balancing round",
+            "unit"
+        );
+        this.diskUsageHistogram = meterRegistry.registerDoubleHistogram(
+            DISK_USAGE_BYTES_METRIC_NAME,
+            "change in disk usage in bytes per balancing round",
+            "unit"
+        );
+        this.writeLoadHistogram = meterRegistry.registerDoubleHistogram(
+            WRITE_LOAD_METRIC_NAME,
+            "change in write load per balancing round",
+            "1.0"
+        );
+        this.totalWeightHistogram = meterRegistry.registerDoubleHistogram(
+            TOTAL_WEIGHT_METRIC_NAME,
+            "change in total weight per balancing round",
+            "1.0"
+        );
+    }
+
+    @SuppressForbidden(reason = "ForbiddenAPIs bans Math.abs(long) because of overflow on Long.MIN_VALUE, but this is impossible here")
+    private long longAbsNegativeSafe(long value) {
+        assert value != Long.MIN_VALUE : "value must not be Long.MIN_VALUE";
+        return Math.abs(value);
+    }
+
+    public void addBalancingRoundSummary(BalancingRoundSummary summary) {
+        balancingRoundCounter.increment();
+        shardMovesCounter.incrementBy(summary.numberOfShardsToMove());
+        shardMovesHistogram.record(summary.numberOfShardsToMove());
+
+        for (Map.Entry<DiscoveryNode, NodesWeightsChanges> changesEntry : summary.nodeToWeightChanges().entrySet()) {
+            DiscoveryNode node = changesEntry.getKey();
+            NodesWeightsChanges weightChanges = changesEntry.getValue();
+            BalancingRoundSummary.NodeWeightsDiff weightsDiff = weightChanges.weightsDiff();
+
+            shardCountHistogram.record(longAbsNegativeSafe(weightsDiff.shardCountDiff()), getNodeAttributes(node));
+            diskUsageHistogram.record(Math.abs(weightsDiff.diskUsageInBytesDiff()), getNodeAttributes(node));
+            writeLoadHistogram.record(Math.abs(weightsDiff.writeLoadDiff()), getNodeAttributes(node));
+            totalWeightHistogram.record(Math.abs(weightsDiff.totalWeightDiff()), getNodeAttributes(node));
+        }
+    }
+
+    private Map<String, Object> getNodeAttributes(DiscoveryNode node) {
+        return Map.of("node_name", node.getName(), "node_id", node.getId());
+    }
+}

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

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.core.TimeValue;
@@ -58,6 +59,7 @@ public class AllocationBalancingRoundSummaryService {
     private final ThreadPool threadPool;
     private volatile boolean enableBalancerRoundSummaries;
     private volatile TimeValue summaryReportInterval;
+    private final AllocationBalancingRoundMetrics balancingRoundMetrics;
 
     /**
      * A concurrency-safe list of balancing round summaries. Balancer rounds are run and added here serially, so the queue will naturally
@@ -68,12 +70,17 @@ public class AllocationBalancingRoundSummaryService {
     /** This reference is set when reporting is scheduled. If it is null, then reporting is inactive. */
     private final AtomicReference<Scheduler.Cancellable> scheduledReportFuture = new AtomicReference<>();
 
-    public AllocationBalancingRoundSummaryService(ThreadPool threadPool, ClusterSettings clusterSettings) {
+    public AllocationBalancingRoundSummaryService(
+        ThreadPool threadPool,
+        ClusterSettings clusterSettings,
+        AllocationBalancingRoundMetrics balancingRoundMetrics
+    ) {
         this.threadPool = threadPool;
         // Initialize the local setting values to avoid a null access when ClusterSettings#initializeAndWatch is called on each setting:
         // updating enableBalancerRoundSummaries accesses summaryReportInterval.
         this.enableBalancerRoundSummaries = clusterSettings.get(ENABLE_BALANCER_ROUND_SUMMARIES_SETTING);
         this.summaryReportInterval = clusterSettings.get(BALANCER_ROUND_SUMMARIES_LOG_INTERVAL_SETTING);
+        this.balancingRoundMetrics = balancingRoundMetrics;
 
         clusterSettings.initializeAndWatch(ENABLE_BALANCER_ROUND_SUMMARIES_SETTING, value -> {
             this.enableBalancerRoundSummaries = value;
@@ -99,14 +106,14 @@ public static BalancingRoundSummary createBalancerRoundSummary(DesiredBalance ol
      * Creates a summary of the node weight changes from {@code oldDesiredBalance} to {@code newDesiredBalance}.
      * See {@link BalancingRoundSummary.NodesWeightsChanges} for content details.
      */
-    private static Map<String, BalancingRoundSummary.NodesWeightsChanges> createWeightsSummary(
+    private static Map<DiscoveryNode, BalancingRoundSummary.NodesWeightsChanges> createWeightsSummary(
         DesiredBalance oldDesiredBalance,
         DesiredBalance newDesiredBalance
     ) {
         var oldWeightsPerNode = oldDesiredBalance.weightsPerNode();
         var newWeightsPerNode = newDesiredBalance.weightsPerNode();
 
-        Map<String, BalancingRoundSummary.NodesWeightsChanges> nodeNameToWeightInfo = new HashMap<>(oldWeightsPerNode.size());
+        Map<DiscoveryNode, BalancingRoundSummary.NodesWeightsChanges> nodeNameToWeightInfo = new HashMap<>(oldWeightsPerNode.size());
         for (var nodeAndWeights : oldWeightsPerNode.entrySet()) {
             var discoveryNode = nodeAndWeights.getKey();
             var oldNodeWeightStats = nodeAndWeights.getValue();
@@ -116,7 +123,7 @@ private static Map<String, BalancingRoundSummary.NodesWeightsChanges> createWeig
             var newNodeWeightStats = newWeightsPerNode.getOrDefault(discoveryNode, DesiredBalanceMetrics.NodeWeightStats.ZERO);
 
             nodeNameToWeightInfo.put(
-                discoveryNode.getName(),
+                discoveryNode,
                 new BalancingRoundSummary.NodesWeightsChanges(
                     oldNodeWeightStats,
                     BalancingRoundSummary.NodeWeightsDiff.create(oldNodeWeightStats, newNodeWeightStats)
@@ -128,11 +135,11 @@ private static Map<String, BalancingRoundSummary.NodesWeightsChanges> createWeig
         // the new DesiredBalance to check.
         for (var nodeAndWeights : newWeightsPerNode.entrySet()) {
             var discoveryNode = nodeAndWeights.getKey();
-            if (nodeNameToWeightInfo.containsKey(discoveryNode.getName()) == false) {
+            if (nodeNameToWeightInfo.containsKey(discoveryNode) == false) {
                 // This node is new in the new DesiredBalance, there was no entry added during iteration of the nodes in the old
                 // DesiredBalance. So we'll make a new entry with a base of zero value weights and a weights diff of the new node's weights.
                 nodeNameToWeightInfo.put(
-                    discoveryNode.getName(),
+                    discoveryNode,
                     new BalancingRoundSummary.NodesWeightsChanges(
                         DesiredBalanceMetrics.NodeWeightStats.ZERO,
                         BalancingRoundSummary.NodeWeightsDiff.create(DesiredBalanceMetrics.NodeWeightStats.ZERO, nodeAndWeights.getValue())
@@ -164,6 +171,7 @@ public void addBalancerRoundSummary(BalancingRoundSummary summary) {
         }
 
         summaries.add(summary);
+        balancingRoundMetrics.addBalancingRoundSummary(summary);
     }
 
     /**

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

@@ -9,19 +9,22 @@
 
 package org.elasticsearch.cluster.routing.allocation.allocator;
 
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.core.Strings;
+
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 /**
  * Summarizes the impact to the cluster as a result of a rebalancing round.
  *
- * @param nodeNameToWeightChanges The shard balance weight changes for each node (by name), comparing a previous DesiredBalance shard
+ * @param nodeToWeightChanges The shard balance weight changes for each DiscoveryNode, comparing a previous DesiredBalance shard
  *                                allocation to a new DesiredBalance allocation.
  * @param numberOfShardsToMove The number of shard moves required to move from the previous desired balance to the new one. Does not include
  *                             new (index creation) or removed (index deletion) shard assignements.
  */
-public record BalancingRoundSummary(Map<String, NodesWeightsChanges> nodeNameToWeightChanges, long numberOfShardsToMove) {
+public record BalancingRoundSummary(Map<DiscoveryNode, NodesWeightsChanges> nodeToWeightChanges, long numberOfShardsToMove) {
 
     /**
      * Represents the change in weights for a node going from an old DesiredBalance to a new DesiredBalance
@@ -75,8 +78,8 @@ public NodeWeightsDiff combine(NodeWeightsDiff otherDiff) {
     @Override
     public String toString() {
         return "BalancingRoundSummary{"
-            + "nodeNameToWeightChanges"
-            + nodeNameToWeightChanges
+            + "nodeToWeightChanges"
+            + nodeToWeightChanges
             + ", numberOfShardsToMove="
             + numberOfShardsToMove
             + '}';
@@ -93,17 +96,34 @@ public String toString() {
      * latest desired balance.
      *
      * @param numberOfBalancingRounds How many balancing round summaries are combined in this report.
-     * @param nodeNameToWeightChanges
+     * @param nodeToWeightChanges
      * @param numberOfShardMoves The sum of shard moves for each balancing round being combined into a single summary.
      */
     public record CombinedBalancingRoundSummary(
         int numberOfBalancingRounds,
-        Map<String, NodesWeightsChanges> nodeNameToWeightChanges,
+        Map<DiscoveryNode, NodesWeightsChanges> nodeToWeightChanges,
         long numberOfShardMoves
     ) {
 
         public static final CombinedBalancingRoundSummary EMPTY_RESULTS = new CombinedBalancingRoundSummary(0, new HashMap<>(), 0);
 
+        /**
+         * Serialize the CombinedBalancingRoundSummary to a compact log representation, where {@link DiscoveryNode#getName()} is used
+         * instead of the entire {@link DiscoveryNode#toString()} method.
+         */
+        @Override
+        public String toString() {
+            Map<String, NodesWeightsChanges> nodeNameToWeightChanges = new HashMap<>(nodeToWeightChanges.size());
+            nodeToWeightChanges.forEach((node, nodesWeightChanges) -> nodeNameToWeightChanges.put(node.getName(), nodesWeightChanges));
+
+            return Strings.format(
+                "CombinedBalancingRoundSummary[numberOfBalancingRounds=%d, nodeToWeightChange=%s, numberOfShardMoves=%d]",
+                numberOfBalancingRounds,
+                nodeNameToWeightChanges,
+                numberOfShardMoves
+            );
+        }
+
         /**
          * Merges multiple {@link BalancingRoundSummary} summaries into a single {@link CombinedBalancingRoundSummary}.
          */
@@ -113,7 +133,7 @@ public static CombinedBalancingRoundSummary combine(List<BalancingRoundSummary>
             }
 
             // We will loop through the summaries and sum the weight diffs for each node entry.
-            Map<String, NodesWeightsChanges> combinedNodeNameToWeightChanges = new HashMap<>();
+            Map<DiscoveryNode, NodesWeightsChanges> combinedNodeNameToWeightChanges = new HashMap<>();
 
             // Number of shards moves are simply summed across summaries. Each new balancing round is built upon the last one, so it is
             // possible that a shard is reassigned back to a node before it even moves away, and that will still be counted as 2 moves here.
@@ -128,7 +148,7 @@ public static CombinedBalancingRoundSummary combine(List<BalancingRoundSummary>
 
                 // We'll build the weight changes by keeping the node weight base from the first summary in which a node appears and then
                 // summing the weight diffs in each summary to get total weight diffs across summaries.
-                for (var nodeNameAndWeights : summary.nodeNameToWeightChanges.entrySet()) {
+                for (var nodeNameAndWeights : summary.nodeToWeightChanges.entrySet()) {
                     var combined = combinedNodeNameToWeightChanges.get(nodeNameAndWeights.getKey());
                     if (combined == null) {
                         // Either this is the first summary, and combinedNodeNameToWeightChanges hasn't been initialized yet for this node;