HBASE-13965 Stochastic Load Balancer JMX Metrics (Lei Chen)

hbase-client/src/main/java/org/apache/hadoop/hbase/filter/Filter.java

@@ -181,6 +181,10 @@ public enum ReturnCode {
      * Seek to next key which is given as hint by the filter.
      */
     SEEK_NEXT_USING_HINT,
+    /**
+     * Include KeyValue and done with row, seek to next.
+     */
+    INCLUDE_AND_SEEK_NEXT_ROW,
 }
 
   /**

hbase-client/src/main/java/org/apache/hadoop/hbase/filter/FilterWrapper.java

@@ -134,7 +134,8 @@ public void filterRowCells(List<Cell> kvs) throws IOException {
   public enum FilterRowRetCode {
     NOT_CALLED,
     INCLUDE,     // corresponds to filter.filterRow() returning false
-    EXCLUDE      // corresponds to filter.filterRow() returning true
+    EXCLUDE,     // corresponds to filter.filterRow() returning true
+    INCLUDE_THIS_FAMILY  // exclude other families
   }
   public FilterRowRetCode filterRowCellsWithRet(List<Cell> kvs) throws IOException {
     //To fix HBASE-6429,

hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java

@@ -124,6 +124,12 @@ public enum OperationStatusCode {
   /** Config for pluggable load balancers */
   public static final String HBASE_MASTER_LOADBALANCER_CLASS = "hbase.master.loadbalancer.class";
 
+  /** Config for balancing the cluster by table */
+  public static final String HBASE_MASTER_LOADBALANCE_BYTABLE = "hbase.master.loadbalance.bytable";
+
+  /** The name of the ensemble table */
+  public static final String ENSEMBLE_TABLE_NAME = "hbase:ensemble";
+
   /** Config for pluggable region normalizer */
   public static final String HBASE_MASTER_NORMALIZER_CLASS =
     "hbase.master.normalizer.class";

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -34,6 +34,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -1256,12 +1257,14 @@ public boolean balance() throws IOException {
         this.assignmentManager.getRegionStates().getAssignmentsByTable();
 
       List<RegionPlan> plans = new ArrayList<RegionPlan>();
+
       //Give the balancer the current cluster state.
       this.balancer.setClusterStatus(getClusterStatus());
-      for (Map<ServerName, List<HRegionInfo>> assignments : assignmentsByTable.values()) {
-        List<RegionPlan> partialPlans = this.balancer.balanceCluster(assignments);
+      for (Entry<TableName, Map<ServerName, List<HRegionInfo>>> e : assignmentsByTable.entrySet()) {
+        List<RegionPlan> partialPlans = this.balancer.balanceCluster(e.getKey(), e.getValue());
         if (partialPlans != null) plans.addAll(partialPlans);
       }
+
       long cutoffTime = System.currentTimeMillis() + maximumBalanceTime;
       int rpCount = 0;  // number of RegionPlans balanced so far
       long totalRegPlanExecTime = 0;

hbase-server/src/main/java/org/apache/hadoop/hbase/master/LoadBalancer.java

@@ -31,6 +31,7 @@
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.TableName;
 
 /**
  * Makes decisions about the placement and movement of Regions across
@@ -64,6 +65,15 @@ public interface LoadBalancer extends Configurable, Stoppable, ConfigurationObse
    */
   void setMasterServices(MasterServices masterServices);
 
+  /**
+   * Perform the major balance operation
+   * @param tableName
+   * @param clusterState
+   * @return List of plans
+   */
+  List<RegionPlan> balanceCluster(TableName tableName, Map<ServerName,
+      List<HRegionInfo>> clusterState) throws HBaseIOException;
+
   /**
    * Perform the major balance operation
    * @param clusterState

hbase-server/src/main/java/org/apache/hadoop/hbase/master/RegionStates.java

@@ -953,13 +953,14 @@ protected synchronized double getAverageLoad() {
     Map<TableName, Map<ServerName, List<HRegionInfo>>> result =
       new HashMap<TableName, Map<ServerName,List<HRegionInfo>>>();
     synchronized (this) {
-      if (!server.getConfiguration().getBoolean("hbase.master.loadbalance.bytable", false)) {
+      if (!server.getConfiguration().getBoolean(
+            HConstants.HBASE_MASTER_LOADBALANCE_BYTABLE, false)) {
         Map<ServerName, List<HRegionInfo>> svrToRegions =
           new HashMap<ServerName, List<HRegionInfo>>(serverHoldings.size());
         for (Map.Entry<ServerName, Set<HRegionInfo>> e: serverHoldings.entrySet()) {
           svrToRegions.put(e.getKey(), new ArrayList<HRegionInfo>(e.getValue()));
         }
-        result.put(TableName.valueOf("ensemble"), svrToRegions);
+        result.put(TableName.valueOf(HConstants.ENSEMBLE_TABLE_NAME), svrToRegions);
       } else {
         for (Map.Entry<ServerName, Set<HRegionInfo>> e: serverHoldings.entrySet()) {
           for (HRegionInfo hri: e.getValue()) {

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

@@ -45,7 +45,6 @@
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
-import org.apache.hadoop.hbase.conf.ConfigurationObserver;
 import org.apache.hadoop.hbase.master.LoadBalancer;
 import org.apache.hadoop.hbase.master.MasterServices;
 import org.apache.hadoop.hbase.master.RackManager;
@@ -80,6 +79,21 @@ public String getRack(ServerName server) {
       return UNKNOWN_RACK;
     }
   }
+
+  /**
+   * The constructor that uses the basic MetricsBalancer
+   */
+  protected BaseLoadBalancer() {
+    metricsBalancer = new MetricsBalancer();
+  }
+
+  /**
+   * This Constructor accepts an instance of MetricsBalancer, 
+   * which will be used instead of creating a new one
+   */
+  protected BaseLoadBalancer(MetricsBalancer metricsBalancer) {
+    this.metricsBalancer = (metricsBalancer != null) ? metricsBalancer : new MetricsBalancer();
+  }
 
   /**
    * An efficient array based implementation similar to ClusterState for keeping
@@ -803,7 +817,7 @@ public String toString() {
     "hbase.balancer.tablesOnMaster";
 
   protected final Set<String> tablesOnMaster = new HashSet<String>();
-  protected final MetricsBalancer metricsBalancer = new MetricsBalancer();
+  protected MetricsBalancer metricsBalancer = null;
   protected ClusterStatus clusterStatus = null;
   protected ServerName masterServerName;
   protected MasterServices services;

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

@@ -28,11 +28,13 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseIOException;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.ServerLoad;
 import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.master.RackManager;
 import org.apache.hadoop.hbase.master.RegionPlan;
 import org.apache.hadoop.hbase.master.ServerManager;
@@ -345,4 +347,10 @@ private void assignSecondaryAndTertiaryNodesForRegion(
       globalFavoredNodesAssignmentPlan.updateFavoredNodesMap(region, favoredNodesForRegion);
     }
   }
+
+  @Override
+  public List<RegionPlan> balanceCluster(TableName tableName,
+      Map<ServerName, List<HRegionInfo>> clusterState) throws HBaseIOException {
+    return balanceCluster(clusterState);
+  }
 }

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

@@ -25,9 +25,17 @@
  */
 public class MetricsBalancer {
 
-  private final MetricsBalancerSource source;
+  private MetricsBalancerSource source = null;
 
   public MetricsBalancer() {
+    initSource();
+  }
+
+  /**
+   * A function to instantiate the metrics source. This function can be overridden in its 
+   * subclasses to provide extended sources
+   */
+  protected void initSource() {
     source = CompatibilitySingletonFactory.getInstance(MetricsBalancerSource.class);
   }
 

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

@@ -30,9 +30,11 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.HBaseIOException;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.master.RegionPlan;
 
 import com.google.common.collect.MinMaxPriorityQueue;
@@ -433,4 +435,10 @@ private void addRegionPlan(final MinMaxPriorityQueue<RegionPlan> regionsToMove,
     rp.setDestination(sn);
     regionsToReturn.add(rp);
   }
+
+  @Override
+  public List<RegionPlan> balanceCluster(TableName tableName,
+      Map<ServerName, List<HRegionInfo>> clusterState) throws HBaseIOException {
+    return balanceCluster(clusterState);
+  }
 }