HBASE-25818 Move StochasticLoadBalancer to hbase-balancer module

Author: Apache9

hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/ClusterInfoProvider.java

@@ -22,13 +22,15 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
 import org.apache.hadoop.hbase.ServerMetrics;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.master.RegionPlan;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**
@@ -78,4 +80,18 @@ List<ServerName> getOnlineServersListWithPredicator(List<ServerName> servers,
    * Get a snapshot of the current assignment status.
    */
   Map<ServerName, List<RegionInfo>> getSnapShotOfAssignment(Collection<RegionInfo> regions);
+
+  /**
+   * Test whether we are in off peak hour.
+   * <p/>
+   * For peak and off peak hours we may have different cost for the same balancing operation.
+   * @return
+   */
+  boolean isOffPeakHour();
+
+  /**
+   * Record the given region plans.
+   */
+  void recordBalancerDecision(List<RegionPlan> plans, double currentCost, double initCost,
+    String initFunctionTotalCosts, Supplier<String> totalCostsPerFunc, long step);
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/StochasticLoadBalancer.java renamed to hbase-balancer/src/main/java/org/apache/hadoop/hbase/master/balancer/StochasticLoadBalancer.java

@@ -28,6 +28,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ClusterMetrics;
@@ -36,13 +37,9 @@
 import org.apache.hadoop.hbase.ServerMetrics;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.BalancerDecision;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.master.RegionPlan;
 import org.apache.hadoop.hbase.master.balancer.BalancerClusterState.LocalityType;
-import org.apache.hadoop.hbase.namequeues.BalancerDecisionDetails;
-import org.apache.hadoop.hbase.namequeues.NamedQueueRecorder;
-import org.apache.hadoop.hbase.regionserver.compactions.OffPeakHours;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.ReflectionUtils;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -151,11 +148,6 @@ public class StochasticLoadBalancer extends BaseLoadBalancer {
   private RegionReplicaHostCostFunction regionReplicaHostCostFunction;
   private RegionReplicaRackCostFunction regionReplicaRackCostFunction;
 
-  /**
-   * Use to add balancer decision history to ring-buffer
-   */
-  NamedQueueRecorder namedQueueRecorder;
-
   /**
    * The constructor that pass a MetricsStochasticBalancer to BaseLoadBalancer to replace its
    * default MetricsBalancer
@@ -205,7 +197,7 @@ public synchronized void setConf(Configuration conf) {
     costFunctions = new ArrayList<>();
     addCostFunction(new RegionCountSkewCostFunction(conf));
     addCostFunction(new PrimaryRegionCountSkewCostFunction(conf));
-    addCostFunction(new MoveCostFunction(conf));
+    addCostFunction(new MoveCostFunction(conf, () -> provider));
     addCostFunction(localityCost);
     addCostFunction(rackLocalityCost);
     addCostFunction(new TableSkewCostFunction(conf));
@@ -221,13 +213,6 @@ public synchronized void setConf(Configuration conf) {
     curFunctionCosts= new Double[costFunctions.size()];
     tempFunctionCosts= new Double[costFunctions.size()];
 
-    boolean isBalancerDecisionRecording = getConf()
-      .getBoolean(BaseLoadBalancer.BALANCER_DECISION_BUFFER_ENABLED,
-        BaseLoadBalancer.DEFAULT_BALANCER_DECISION_BUFFER_ENABLED);
-    if (this.namedQueueRecorder == null && isBalancerDecisionRecording) {
-      this.namedQueueRecorder = NamedQueueRecorder.getInstance(getConf());
-    }
-
     LOG.info("Loaded config; maxSteps=" + maxSteps + ", stepsPerRegion=" + stepsPerRegion +
             ", maxRunningTime=" + maxRunningTime + ", isByTable=" + isByTable + ", CostFunctions=" +
             Arrays.toString(getCostFunctionNames()) + " etc.");
@@ -483,24 +468,9 @@ public synchronized List<RegionPlan> balanceTable(TableName tableName, Map<Serve
   }
 
   private void sendRegionPlansToRingBuffer(List<RegionPlan> plans, double currentCost,
-      double initCost, String initFunctionTotalCosts, long step) {
-    if (this.namedQueueRecorder != null) {
-      List<String> regionPlans = new ArrayList<>();
-      for (RegionPlan plan : plans) {
-        regionPlans.add(
-          "table: " + plan.getRegionInfo().getTable() + " , region: " + plan.getRegionName()
-            + " , source: " + plan.getSource() + " , destination: " + plan.getDestination());
-      }
-      BalancerDecision balancerDecision =
-        new BalancerDecision.Builder()
-          .setInitTotalCost(initCost)
-          .setInitialFunctionCosts(initFunctionTotalCosts)
-          .setComputedTotalCost(currentCost)
-          .setFinalFunctionCosts(totalCostsPerFunc())
-          .setComputedSteps(step)
-          .setRegionPlans(regionPlans).build();
-      namedQueueRecorder.addRecord(new BalancerDecisionDetails(balancerDecision));
-    }
+    double initCost, String initFunctionTotalCosts, long step) {
+    provider.recordBalancerDecision(plans, currentCost, initCost, initFunctionTotalCosts,
+      this::totalCostsPerFunc, step);
   }
 
   /**
@@ -837,10 +807,15 @@ static class MoveCostFunction extends CostFunction {
 
     private final float maxMovesPercent;
     private final Configuration conf;
+    // we call setConf before setClusterInfoProvider, so when we create MoveCostFunction, the
+    // provider is still null, thus here we need to provide a supplier instead of passing the
+    // provider directly.
+    private final Supplier<ClusterInfoProvider> provider;
 
-    MoveCostFunction(Configuration conf) {
+    MoveCostFunction(Configuration conf, Supplier<ClusterInfoProvider> provider) {
       super(conf);
       this.conf = conf;
+      this.provider = provider;
       // What percent of the number of regions a single run of the balancer can move.
       maxMovesPercent = conf.getFloat(MAX_MOVES_PERCENT_KEY, DEFAULT_MAX_MOVE_PERCENT);
 
@@ -853,7 +828,8 @@ static class MoveCostFunction extends CostFunction {
     protected double cost() {
       // Move cost multiplier should be the same cost or higher than the rest of the costs to ensure
       // that large benefits are need to overcome the cost of a move.
-      if (OffPeakHours.getInstance(conf).isOffPeakHour()) {
+      ClusterInfoProvider p = provider.get(); 
+      if (p != null && p.isOffPeakHour()) {
         this.setMultiplier(conf.getFloat(MOVE_COST_OFFPEAK_KEY, DEFAULT_MOVE_COST_OFFPEAK));
       } else {
         this.setMultiplier(conf.getFloat(MOVE_COST_KEY, DEFAULT_MOVE_COST));

hbase-balancer/src/test/java/org/apache/hadoop/hbase/master/balancer/DummyClusterInfoProvider.java

@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.master.balancer;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HDFSBlocksDistribution;
+import org.apache.hadoop.hbase.ServerMetrics;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.master.RegionPlan;
+
+public class DummyClusterInfoProvider implements ClusterInfoProvider {
+
+  @Override
+  public List<RegionInfo> getAssignedRegions() {
+    return Collections.emptyList();
+  }
+
+  @Override
+  public TableDescriptor getTableDescriptor(TableName tableName) throws IOException {
+    return null;
+  }
+
+  @Override
+  public int getNumberOfTables() throws IOException {
+    return 0;
+  }
+
+  @Override
+  public HDFSBlocksDistribution computeHDFSBlocksDistribution(Configuration conf,
+    TableDescriptor tableDescriptor, RegionInfo regionInfo) throws IOException {
+    return new HDFSBlocksDistribution();
+  }
+
+  @Override
+  public boolean hasRegionReplica(Collection<RegionInfo> regions) throws IOException {
+    return false;
+  }
+
+  @Override
+  public List<ServerName> getOnlineServersListWithPredicator(List<ServerName> servers,
+    Predicate<ServerMetrics> filter) {
+    return Collections.emptyList();
+  }
+
+  @Override
+  public Map<ServerName, List<RegionInfo>> getSnapShotOfAssignment(Collection<RegionInfo> regions) {
+    return Collections.emptyMap();
+  }
+
+  @Override
+  public boolean isOffPeakHour() {
+    return false;
+  }
+
+  @Override
+  public void recordBalancerDecision(List<RegionPlan> plans, double currentCost, double initCost,
+    String initFunctionTotalCosts, Supplier<String> totalCostsPerFunc, long step) {
+  }
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/DummyCostFunction.java renamed to hbase-balancer/src/test/java/org/apache/hadoop/hbase/master/balancer/DummyCostFunction.java

@@ -19,6 +19,7 @@
 
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
 
 import java.util.List;
 import java.util.Map;
@@ -48,6 +49,7 @@ public static void beforeAllTests() throws Exception {
     conf.setFloat("hbase.master.balancer.stochastic.localityCost", 0);
     loadBalancer = new StochasticLoadBalancer();
     loadBalancer.setConf(conf);
+    loadBalancer.setClusterInfoProvider(new DummyClusterInfoProvider());
   }
 
   protected void testWithCluster(int numNodes, int numRegions, int numRegionsPerServer,

hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/StochasticBalancerTestBase.java renamed to hbase-balancer/src/test/java/org/apache/hadoop/hbase/master/balancer/StochasticBalancerTestBase.java

@@ -92,8 +92,7 @@ public static void beforeAllTests() throws Exception {
     conf.setClass("hbase.util.ip.to.rack.determiner", MockMapping.class, DNSToSwitchMapping.class);
     loadBalancer = new MockBalancer();
     loadBalancer.setConf(conf);
-    ClusterInfoProvider provider = mock(ClusterInfoProvider.class);
-    loadBalancer.setClusterInfoProvider(provider);
+    loadBalancer.setClusterInfoProvider(new DummyClusterInfoProvider());
 
     // Set up the rack topologies (5 machines per rack)
     rackManager = mock(RackManager.class);

hbase-server/src/test/java/org/apache/hadoop/hbase/master/balancer/StochasticBalancerTestBase2.java renamed to hbase-balancer/src/test/java/org/apache/hadoop/hbase/master/balancer/StochasticBalancerTestBase2.java

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ClusterMetrics;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
@@ -47,6 +48,7 @@
 import org.apache.hadoop.hbase.client.RegionInfoBuilder;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.master.RegionPlan;
 import org.apache.hadoop.hbase.testclassification.MasterTests;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -97,7 +99,7 @@ public static void setUpBeforeClass() {
   @Before
   public void setUp() {
     finder = new RegionHDFSBlockLocationFinder();
-    finder.setClusterInfoProvider(new ClusterInfoProvider() {
+    finder.setClusterInfoProvider(new DummyClusterInfoProvider() {
 
       @Override
       public TableDescriptor getTableDescriptor(TableName tableName) throws IOException {
@@ -114,28 +116,6 @@ public HDFSBlocksDistribution computeHDFSBlocksDistribution(Configuration conf,
         TableDescriptor tableDescriptor, RegionInfo regionInfo) throws IOException {
         return generate(regionInfo);
       }
-
-      @Override
-      public boolean hasRegionReplica(Collection<RegionInfo> regions) throws IOException {
-        return false;
-      }
-
-      @Override
-      public List<ServerName> getOnlineServersListWithPredicator(List<ServerName> servers,
-        Predicate<ServerMetrics> filter) {
-        return Collections.emptyList();
-      }
-
-      @Override
-      public Map<ServerName, List<RegionInfo>>
-        getSnapShotOfAssignment(Collection<RegionInfo> regions) {
-        return Collections.emptyMap();
-      }
-
-      @Override
-      public int getNumberOfTables() {
-        return 0;
-      }
     });
   }