Allow balancing weights to be set per tier

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

@@ -33,6 +33,7 @@
 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.BalancedAllocatorSettings;
 import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.NodeAllocationStatsAndWeightsCalculator;
 import org.elasticsearch.cluster.routing.allocation.WriteLoadForecaster;
@@ -145,13 +146,15 @@ public ClusterModule(
         this.clusterPlugins = clusterPlugins;
         this.deciderList = createAllocationDeciders(settings, clusterService.getClusterSettings(), clusterPlugins);
         this.allocationDeciders = new AllocationDeciders(deciderList);
+        BalancedAllocatorSettings balancedAllocatorSettings = new BalancedAllocatorSettings(clusterService.getClusterSettings());
         var nodeAllocationStatsAndWeightsCalculator = new NodeAllocationStatsAndWeightsCalculator(
             writeLoadForecaster,
-            clusterService.getClusterSettings()
+            balancedAllocatorSettings
         );
         this.shardsAllocator = createShardsAllocator(
             settings,
             clusterService.getClusterSettings(),
+            balancedAllocatorSettings,
             threadPool,
             clusterPlugins,
             clusterService,
@@ -440,6 +443,7 @@ private static void addAllocationDecider(Map<Class<?>, AllocationDecider> decide
     private static ShardsAllocator createShardsAllocator(
         Settings settings,
         ClusterSettings clusterSettings,
+        BalancedAllocatorSettings balancedAllocatorSettings,
         ThreadPool threadPool,
         List<ClusterPlugin> clusterPlugins,
         ClusterService clusterService,
@@ -449,12 +453,12 @@ private static ShardsAllocator createShardsAllocator(
         NodeAllocationStatsAndWeightsCalculator nodeAllocationStatsAndWeightsCalculator
     ) {
         Map<String, Supplier<ShardsAllocator>> allocators = new HashMap<>();
-        allocators.put(BALANCED_ALLOCATOR, () -> new BalancedShardsAllocator(clusterSettings, writeLoadForecaster));
+        allocators.put(BALANCED_ALLOCATOR, () -> new BalancedShardsAllocator(balancedAllocatorSettings, writeLoadForecaster));
         allocators.put(
             DESIRED_BALANCE_ALLOCATOR,
             () -> new DesiredBalanceShardsAllocator(
                 clusterSettings,
-                new BalancedShardsAllocator(clusterSettings, writeLoadForecaster),
+                new BalancedShardsAllocator(balancedAllocatorSettings, writeLoadForecaster),
                 threadPool,
                 clusterService,
                 reconciler,

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

@@ -0,0 +1,113 @@
+/*
+ * 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;
+
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Settings;
+
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.DISK_USAGE_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.INDEXING_TIER_SHARD_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.INDEXING_TIER_WRITE_LOAD_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.SEARCH_TIER_SHARD_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.SEARCH_TIER_WRITE_LOAD_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.THRESHOLD_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator.WRITE_LOAD_BALANCE_FACTOR_SETTING;
+
+public class BalancedAllocatorSettings {
+
+    public static BalancedAllocatorSettings DEFAULT = new BalancedAllocatorSettings(ClusterSettings.createBuiltInClusterSettings());
+    private volatile float indexBalanceFactor;
+    private volatile float shardBalanceFactor;
+    private volatile float indexingTierShardBalanceFactor;
+    private volatile float searchTierShardBalanceFactor;
+    private volatile float writeLoadBalanceFactor;
+    private volatile float indexingTierWriteLoadBalanceFactor;
+    private volatile float searchTierWriteLoadBalanceFactor;
+    private volatile float diskUsageBalanceFactor;
+    private volatile float threshold;
+
+    public BalancedAllocatorSettings(Settings settings) {
+        this(ClusterSettings.createBuiltInClusterSettings(settings));
+    }
+
+    public BalancedAllocatorSettings(ClusterSettings clusterSettings) {
+        clusterSettings.initializeAndWatch(SHARD_BALANCE_FACTOR_SETTING, value -> this.shardBalanceFactor = value);
+        clusterSettings.initializeAndWatch(
+            INDEXING_TIER_SHARD_BALANCE_FACTOR_SETTING,
+            value -> this.indexingTierShardBalanceFactor = value
+        );
+        clusterSettings.initializeAndWatch(SEARCH_TIER_SHARD_BALANCE_FACTOR_SETTING, value -> this.searchTierShardBalanceFactor = value);
+        clusterSettings.initializeAndWatch(INDEX_BALANCE_FACTOR_SETTING, value -> this.indexBalanceFactor = value);
+        clusterSettings.initializeAndWatch(WRITE_LOAD_BALANCE_FACTOR_SETTING, value -> this.writeLoadBalanceFactor = value);
+        clusterSettings.initializeAndWatch(
+            INDEXING_TIER_WRITE_LOAD_BALANCE_FACTOR_SETTING,
+            value -> this.indexingTierWriteLoadBalanceFactor = value
+        );
+        clusterSettings.initializeAndWatch(
+            SEARCH_TIER_WRITE_LOAD_BALANCE_FACTOR_SETTING,
+            value -> this.searchTierWriteLoadBalanceFactor = value
+        );
+        clusterSettings.initializeAndWatch(DISK_USAGE_BALANCE_FACTOR_SETTING, value -> this.diskUsageBalanceFactor = value);
+        clusterSettings.initializeAndWatch(THRESHOLD_SETTING, value -> this.threshold = value);
+    }
+
+    /**
+     * Returns the index related weight factor.
+     */
+    public float getIndexBalanceFactor() {
+        return indexBalanceFactor;
+    }
+
+    /**
+     * Returns the shard related weight factor.
+     */
+    public float getShardBalanceFactor() {
+        return shardBalanceFactor;
+    }
+
+    /**
+     * Returns the shard related weight factor.
+     */
+    public float getIndexingTierShardBalanceFactor() {
+        return indexingTierShardBalanceFactor;
+    }
+
+    /**
+     * Returns the shard related weight factor.
+     */
+    public float getSearchTierShardBalanceFactor() {
+        return searchTierShardBalanceFactor;
+    }
+
+    public float getWriteLoadBalanceFactor() {
+        return writeLoadBalanceFactor;
+    }
+
+    public float getIndexingTierWriteLoadBalanceFactor() {
+        return indexingTierWriteLoadBalanceFactor;
+    }
+
+    public float getSearchTierWriteLoadBalanceFactor() {
+        return searchTierWriteLoadBalanceFactor;
+    }
+
+    public float getDiskUsageBalanceFactor() {
+        return diskUsageBalanceFactor;
+    }
+
+    /**
+     * Returns the currently configured delta threshold
+     */
+    public float getThreshold() {
+        return threshold;
+    }
+}

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

@@ -15,10 +15,9 @@
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
-import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalance;
+import org.elasticsearch.cluster.routing.allocation.allocator.SpecialisedWeightFunction;
 import org.elasticsearch.cluster.routing.allocation.allocator.WeightFunction;
-import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.core.Nullable;
 
@@ -29,11 +28,7 @@
  */
 public class NodeAllocationStatsAndWeightsCalculator {
     private final WriteLoadForecaster writeLoadForecaster;
-
-    private volatile float indexBalanceFactor;
-    private volatile float shardBalanceFactor;
-    private volatile float writeLoadBalanceFactor;
-    private volatile float diskUsageBalanceFactor;
+    private final BalancedAllocatorSettings settings;
 
     /**
      * Node shard allocation stats and the total node weight.
@@ -47,18 +42,9 @@ public record NodeAllocationStatsAndWeight(
         float currentNodeWeight
     ) {}
 
-    public NodeAllocationStatsAndWeightsCalculator(WriteLoadForecaster writeLoadForecaster, ClusterSettings clusterSettings) {
+    public NodeAllocationStatsAndWeightsCalculator(WriteLoadForecaster writeLoadForecaster, BalancedAllocatorSettings settings) {
         this.writeLoadForecaster = writeLoadForecaster;
-        clusterSettings.initializeAndWatch(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING, value -> this.shardBalanceFactor = value);
-        clusterSettings.initializeAndWatch(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING, value -> this.indexBalanceFactor = value);
-        clusterSettings.initializeAndWatch(
-            BalancedShardsAllocator.WRITE_LOAD_BALANCE_FACTOR_SETTING,
-            value -> this.writeLoadBalanceFactor = value
-        );
-        clusterSettings.initializeAndWatch(
-            BalancedShardsAllocator.DISK_USAGE_BALANCE_FACTOR_SETTING,
-            value -> this.diskUsageBalanceFactor = value
-        );
+        this.settings = settings;
     }
 
     /**
@@ -74,7 +60,7 @@ public Map<String, NodeAllocationStatsAndWeight> nodesAllocationStatsAndWeights(
             // must not use licensed features when just starting up
             writeLoadForecaster.refreshLicense();
         }
-        var weightFunction = new WeightFunction(shardBalanceFactor, indexBalanceFactor, writeLoadBalanceFactor, diskUsageBalanceFactor);
+        var weightFunction = new SpecialisedWeightFunction(settings);
         var avgShardsPerNode = WeightFunction.avgShardPerNode(metadata, routingNodes);
         var avgWriteLoadPerNode = WeightFunction.avgWriteLoadPerNode(writeLoadForecaster, metadata, routingNodes);
         var avgDiskUsageInBytesPerNode = WeightFunction.avgDiskUsageInBytesPerNode(clusterInfo, metadata, routingNodes);
@@ -102,6 +88,7 @@ public Map<String, NodeAllocationStatsAndWeight> nodesAllocationStatsAndWeights(
                 currentDiskUsage += shardSize;
             }
             float currentNodeWeight = weightFunction.calculateNodeWeight(
+                node,
                 shards,
                 avgShardsPerNode,
                 forecastedWriteLoad,