HBASE-27999 Implement cache prefetch aware load balancer (#5527)

this commit is part of the rebase of HBASE-28186

Signed-off-by: Wellington Chevreuil <[email protected]>
Signed-off-by: Tak Lon (Stephen) Wu <[email protected]>
Co-authored-by: Rahul Agarkar <[email protected]>

(cherry picked from commit e799ee0)

Author: wchevreuil

hbase-client/src/main/java/org/apache/hadoop/hbase/RegionLoad.java

@@ -389,6 +389,16 @@ public CompactionState getCompactionState() {
     return metrics.getCompactionState();
   }
 
+  @Override
+  public Size getRegionSizeMB() {
+    return metrics.getRegionSizeMB();
+  }
+
+  @Override
+  public float getCurrentRegionCachedRatio() {
+    return metrics.getCurrentRegionCachedRatio();
+  }
+
   /**
    * @see java.lang.Object#toString()
    */

hbase-client/src/main/java/org/apache/hadoop/hbase/ServerLoad.java

@@ -430,6 +430,11 @@ public List<ServerTask> getTasks() {
     return metrics.getTasks();
   }
 
+  @Override
+  public Map<String, Integer> getRegionCachedInfo() {
+    return metrics.getRegionCachedInfo();
+  }
+
   /**
    * Originally, this method factored in the effect of requests going to the server as well.
    * However, this does not interact very well with the current region rebalancing code, which only

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

@@ -1428,6 +1428,18 @@ public enum OperationStatusCode {
    */
   public static final String BUCKET_CACHE_SIZE_KEY = "hbase.bucketcache.size";
 
+  /**
+   * If the chosen ioengine can persist its state across restarts, the path to the file to persist
+   * to. This file is NOT the data file. It is a file into which we will serialize the map of what
+   * is in the data file. For example, if you pass the following argument as
+   * BUCKET_CACHE_IOENGINE_KEY ("hbase.bucketcache.ioengine"),
+   * <code>file:/tmp/bucketcache.data </code>, then we will write the bucketcache data to the file
+   * <code>/tmp/bucketcache.data</code> but the metadata on where the data is in the supplied file
+   * is an in-memory map that needs to be persisted across restarts. Where to store this in-memory
+   * state is what you supply here: e.g. <code>/tmp/bucketcache.map</code>.
+   */
+  public static final String BUCKET_CACHE_PERSISTENT_PATH_KEY = "hbase.bucketcache.persistent.path";
+
   /**
    * HConstants for fast fail on the client side follow
    */

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

@@ -86,6 +86,8 @@ private ServerMetrics mockServerMetricsWithReadRequests(ServerName server,
       when(rl.getWriteRequestCount()).thenReturn(0L);
       when(rl.getMemStoreSize()).thenReturn(Size.ZERO);
       when(rl.getStoreFileSize()).thenReturn(Size.ZERO);
+      when(rl.getRegionSizeMB()).thenReturn(Size.ZERO);
+      when(rl.getCurrentRegionCachedRatio()).thenReturn(0.0f);
       regionLoadMap.put(info.getRegionName(), rl);
     }
     when(serverMetrics.getRegionMetrics()).thenReturn(regionLoadMap);

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockCacheFactory.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.io.hfile;
 
 import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_IOENGINE_KEY;
+import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_PERSISTENT_PATH_KEY;
 import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_SIZE_KEY;
 
 import java.io.IOException;
@@ -47,18 +48,6 @@ public final class BlockCacheFactory {
   public static final String BLOCKCACHE_POLICY_KEY = "hfile.block.cache.policy";
   public static final String BLOCKCACHE_POLICY_DEFAULT = "LRU";
 
-  /**
-   * If the chosen ioengine can persist its state across restarts, the path to the file to persist
-   * to. This file is NOT the data file. It is a file into which we will serialize the map of what
-   * is in the data file. For example, if you pass the following argument as
-   * BUCKET_CACHE_IOENGINE_KEY ("hbase.bucketcache.ioengine"),
-   * <code>file:/tmp/bucketcache.data </code>, then we will write the bucketcache data to the file
-   * <code>/tmp/bucketcache.data</code> but the metadata on where the data is in the supplied file
-   * is an in-memory map that needs to be persisted across restarts. Where to store this in-memory
-   * state is what you supply here: e.g. <code>/tmp/bucketcache.map</code>.
-   */
-  public static final String BUCKET_CACHE_PERSISTENT_PATH_KEY = "hbase.bucketcache.persistent.path";
-
   public static final String BUCKET_CACHE_WRITER_THREADS_KEY = "hbase.bucketcache.writer.threads";
 
   public static final String BUCKET_CACHE_WRITER_QUEUE_KEY = "hbase.bucketcache.writer.queuelength";

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java

@@ -337,6 +337,7 @@ public BucketCache(String ioEngineName, long capacity, int blockSize, int[] buck
         fullyCachedFiles.clear();
         backingMapValidated.set(true);
         bucketAllocator = new BucketAllocator(capacity, bucketSizes);
+        regionCachedSizeMap.clear();
       }
     } else {
       bucketAllocator = new BucketAllocator(capacity, bucketSizes);
@@ -1503,6 +1504,7 @@ private void disableCache() {
       this.backingMap.clear();
       this.blocksByHFile.clear();
       this.fullyCachedFiles.clear();
+      this.regionCachedSizeMap.clear();
     }
   }
 
@@ -2024,9 +2026,9 @@ public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int d
       final MutableInt count = new MutableInt();
       LOG.debug("iterating over {} entries in the backing map", backingMap.size());
       backingMap.entrySet().stream().forEach(entry -> {
-        if (entry.getKey().getHfileName().equals(fileName)) {
+        if (entry.getKey().getHfileName().equals(fileName.getName())) {
           LOG.debug("found block for file {} in the backing map. Acquiring read lock for offset {}",
-            fileName, entry.getKey().getOffset());
+            fileName.getName(), entry.getKey().getOffset());
           ReentrantReadWriteLock lock = offsetLock.getLock(entry.getKey().getOffset());
           lock.readLock().lock();
           locks.add(lock);
@@ -2037,23 +2039,23 @@ public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int d
       });
       // We may either place only data blocks on the BucketCache or all type of blocks
       if (dataBlockCount == count.getValue() || totalBlockCount == count.getValue()) {
-        LOG.debug("File {} has now been fully cached.", fileName);
+        LOG.debug("File {} has now been fully cached.", fileName.getName());
         fileCacheCompleted(fileName, size);
       } else {
         LOG.debug(
           "Prefetch executor completed for {}, but only {} blocks were cached. "
             + "Total blocks for file: {}. Checking for blocks pending cache in cache writer queue.",
-          fileName, count.getValue(), dataBlockCount);
+          fileName.getName(), count.getValue(), dataBlockCount);
         if (ramCache.hasBlocksForFile(fileName.getName())) {
           LOG.debug("There are still blocks pending caching for file {}. Will sleep 100ms "
-            + "and try the verification again.", fileName);
+            + "and try the verification again.", fileName.getName());
           Thread.sleep(100);
           notifyFileCachingCompleted(fileName, totalBlockCount, dataBlockCount, size);
         } else {
           LOG.info(
             "We found only {} blocks cached from a total of {} for file {}, "
               + "but no blocks pending caching. Maybe cache is full?",
-            count, dataBlockCount, fileName);
+            count, dataBlockCount, fileName.getName());
         }
       }
     } catch (InterruptedException e) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/PersistentIOEngine.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import org.apache.hadoop.hbase.procedure2.util.StringUtils;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.util.Shell;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -91,7 +92,8 @@ protected byte[] calculateChecksum(String algorithm) {
   private static long getFileSize(String filePath) throws IOException {
     DU.setExecCommand(filePath);
     DU.execute();
-    return Long.parseLong(DU.getOutput().split("\t")[0]);
+    String size = DU.getOutput().split("\t")[0];
+    return StringUtils.isEmpty(size.trim()) ? 0 : Long.parseLong(size);
   }
 
   private static class DuFileCommand extends Shell.ShellCommandExecutor {

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

@@ -34,6 +34,7 @@
 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
 import org.apache.hadoop.hbase.master.RackManager;
 import org.apache.hadoop.hbase.net.Address;
+import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -114,6 +115,12 @@ class BalancerClusterState {
   private float[][] rackLocalities;
   // Maps localityType -> region -> [server|rack]Index with highest locality
   private int[][] regionsToMostLocalEntities;
+  // Maps region -> serverIndex -> regionCacheRatio of a region on a server
+  private Map<Pair<Integer, Integer>, Float> regionIndexServerIndexRegionCachedRatio;
+  // Maps regionIndex -> serverIndex with best region cache ratio
+  private int[] regionServerIndexWithBestRegionCachedRatio;
+  // Maps regionName -> oldServerName -> cache ratio of the region on the old server
+  Map<String, Pair<ServerName, Float>> regionCacheRatioOnOldServerMap;
 
   static class DefaultRackManager extends RackManager {
     @Override
@@ -125,13 +132,20 @@ public String getRack(ServerName server) {
   BalancerClusterState(Map<ServerName, List<RegionInfo>> clusterState,
     Map<String, Deque<BalancerRegionLoad>> loads, RegionLocationFinder regionFinder,
     RackManager rackManager) {
-    this(null, clusterState, loads, regionFinder, rackManager);
+    this(null, clusterState, loads, regionFinder, rackManager, null);
+  }
+
+  protected BalancerClusterState(Map<ServerName, List<RegionInfo>> clusterState,
+    Map<String, Deque<BalancerRegionLoad>> loads, RegionLocationFinder regionFinder,
+    RackManager rackManager, Map<String, Pair<ServerName, Float>> oldRegionServerRegionCacheRatio) {
+    this(null, clusterState, loads, regionFinder, rackManager, oldRegionServerRegionCacheRatio);
   }
 
   @SuppressWarnings("unchecked")
   BalancerClusterState(Collection<RegionInfo> unassignedRegions,
     Map<ServerName, List<RegionInfo>> clusterState, Map<String, Deque<BalancerRegionLoad>> loads,
-    RegionLocationFinder regionFinder, RackManager rackManager) {
+    RegionLocationFinder regionFinder, RackManager rackManager,
+    Map<String, Pair<ServerName, Float>> oldRegionServerRegionCacheRatio) {
     if (unassignedRegions == null) {
       unassignedRegions = Collections.emptyList();
     }
@@ -145,6 +159,8 @@ public String getRack(ServerName server) {
     tables = new ArrayList<>();
     this.rackManager = rackManager != null ? rackManager : new DefaultRackManager();
 
+    this.regionCacheRatioOnOldServerMap = oldRegionServerRegionCacheRatio;
+
     numRegions = 0;
 
     List<List<Integer>> serversPerHostList = new ArrayList<>();
@@ -527,6 +543,142 @@ private void computeCachedLocalities() {
 
   }
 
+  /**
+   * Returns the size of hFiles from the most recent RegionLoad for region
+   */
+  public int getTotalRegionHFileSizeMB(int region) {
+    Deque<BalancerRegionLoad> load = regionLoads[region];
+    if (load == null) {
+      // This means, that the region has no actual data on disk
+      return 0;
+    }
+    return regionLoads[region].getLast().getRegionSizeMB();
+  }
+
+  /**
+   * Returns the weighted cache ratio of a region on the given region server
+   */
+  public float getOrComputeWeightedRegionCacheRatio(int region, int server) {
+    return getTotalRegionHFileSizeMB(region) * getOrComputeRegionCacheRatio(region, server);
+  }
+
+  /**
+   * Returns the amount by which a region is cached on a given region server. If the region is not
+   * currently hosted on the given region server, then find out if it was previously hosted there
+   * and return the old cache ratio.
+   */
+  protected float getRegionCacheRatioOnRegionServer(int region, int regionServerIndex) {
+    float regionCacheRatio = 0.0f;
+
+    // Get the current region cache ratio if the region is hosted on the server regionServerIndex
+    for (int regionIndex : regionsPerServer[regionServerIndex]) {
+      if (region != regionIndex) {
+        continue;
+      }
+
+      Deque<BalancerRegionLoad> regionLoadList = regionLoads[regionIndex];
+
+      // The region is currently hosted on this region server. Get the region cache ratio for this
+      // region on this server
+      regionCacheRatio =
+        regionLoadList == null ? 0.0f : regionLoadList.getLast().getCurrentRegionCacheRatio();
+
+      return regionCacheRatio;
+    }
+
+    // Region is not currently hosted on this server. Check if the region was cached on this
+    // server earlier. This can happen when the server was shutdown and the cache was persisted.
+    // Search using the region name and server name and not the index id and server id as these ids
+    // may change when a server is marked as dead or a new server is added.
+    String regionEncodedName = regions[region].getEncodedName();
+    ServerName serverName = servers[regionServerIndex];
+    if (
+      regionCacheRatioOnOldServerMap != null
+        && regionCacheRatioOnOldServerMap.containsKey(regionEncodedName)
+    ) {
+      Pair<ServerName, Float> cacheRatioOfRegionOnServer =
+        regionCacheRatioOnOldServerMap.get(regionEncodedName);
+      if (ServerName.isSameAddress(cacheRatioOfRegionOnServer.getFirst(), serverName)) {
+        regionCacheRatio = cacheRatioOfRegionOnServer.getSecond();
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Old cache ratio found for region {} on server {}: {}", regionEncodedName,
+            serverName, regionCacheRatio);
+        }
+      }
+    }
+    return regionCacheRatio;
+  }
+
+  /**
+   * Populate the maps containing information about how much a region is cached on a region server.
+   */
+  private void computeRegionServerRegionCacheRatio() {
+    regionIndexServerIndexRegionCachedRatio = new HashMap<>();
+    regionServerIndexWithBestRegionCachedRatio = new int[numRegions];
+
+    for (int region = 0; region < numRegions; region++) {
+      float bestRegionCacheRatio = 0.0f;
+      int serverWithBestRegionCacheRatio = 0;
+      for (int server = 0; server < numServers; server++) {
+        float regionCacheRatio = getRegionCacheRatioOnRegionServer(region, server);
+        if (regionCacheRatio > 0.0f || server == regionIndexToServerIndex[region]) {
+          // A region with cache ratio 0 on a server means nothing. Hence, just make a note of
+          // cache ratio only if the cache ratio is greater than 0.
+          Pair<Integer, Integer> regionServerPair = new Pair<>(region, server);
+          regionIndexServerIndexRegionCachedRatio.put(regionServerPair, regionCacheRatio);
+        }
+        if (regionCacheRatio > bestRegionCacheRatio) {
+          serverWithBestRegionCacheRatio = server;
+          // If the server currently hosting the region has equal cache ratio to a historical
+          // server, consider the current server to keep hosting the region
+          bestRegionCacheRatio = regionCacheRatio;
+        } else if (
+          regionCacheRatio == bestRegionCacheRatio && server == regionIndexToServerIndex[region]
+        ) {
+          // If two servers have same region cache ratio, then the server currently hosting the
+          // region
+          // should retain the region
+          serverWithBestRegionCacheRatio = server;
+        }
+      }
+      regionServerIndexWithBestRegionCachedRatio[region] = serverWithBestRegionCacheRatio;
+      Pair<Integer, Integer> regionServerPair =
+        new Pair<>(region, regionIndexToServerIndex[region]);
+      float tempRegionCacheRatio = regionIndexServerIndexRegionCachedRatio.get(regionServerPair);
+      if (tempRegionCacheRatio > bestRegionCacheRatio) {
+        LOG.warn(
+          "INVALID CONDITION: region {} on server {} cache ratio {} is greater than the "
+            + "best region cache ratio {} on server {}",
+          regions[region].getEncodedName(), servers[regionIndexToServerIndex[region]],
+          tempRegionCacheRatio, bestRegionCacheRatio, servers[serverWithBestRegionCacheRatio]);
+      }
+    }
+  }
+
+  protected float getOrComputeRegionCacheRatio(int region, int server) {
+    if (
+      regionServerIndexWithBestRegionCachedRatio == null
+        || regionIndexServerIndexRegionCachedRatio.isEmpty()
+    ) {
+      computeRegionServerRegionCacheRatio();
+    }
+
+    Pair<Integer, Integer> regionServerPair = new Pair<>(region, server);
+    return regionIndexServerIndexRegionCachedRatio.containsKey(regionServerPair)
+      ? regionIndexServerIndexRegionCachedRatio.get(regionServerPair)
+      : 0.0f;
+  }
+
+  public int[] getOrComputeServerWithBestRegionCachedRatio() {
+    if (
+      regionServerIndexWithBestRegionCachedRatio == null
+        || regionIndexServerIndexRegionCachedRatio.isEmpty()
+    ) {
+      computeRegionServerRegionCacheRatio();
+    }
+    return regionServerIndexWithBestRegionCachedRatio;
+  }
+
   /**
    * Maps region index to rack index
    */

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

@@ -33,12 +33,16 @@ class BalancerRegionLoad {
   private final long writeRequestsCount;
   private final int memStoreSizeMB;
   private final int storefileSizeMB;
+  private final int regionSizeMB;
+  private final float currentRegionPrefetchRatio;
 
   BalancerRegionLoad(RegionMetrics regionMetrics) {
     readRequestsCount = regionMetrics.getReadRequestCount();
     writeRequestsCount = regionMetrics.getWriteRequestCount();
     memStoreSizeMB = (int) regionMetrics.getMemStoreSize().get(Size.Unit.MEGABYTE);
     storefileSizeMB = (int) regionMetrics.getStoreFileSize().get(Size.Unit.MEGABYTE);
+    regionSizeMB = (int) regionMetrics.getRegionSizeMB().get(Size.Unit.MEGABYTE);
+    currentRegionPrefetchRatio = regionMetrics.getCurrentRegionCachedRatio();
   }
 
   public long getReadRequestsCount() {
@@ -56,4 +60,12 @@ public int getMemStoreSizeMB() {
   public int getStorefileSizeMB() {
     return storefileSizeMB;
   }
+
+  public int getRegionSizeMB() {
+    return regionSizeMB;
+  }
+
+  public float getCurrentRegionCacheRatio() {
+    return currentRegionPrefetchRatio;
+  }
 }

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

@@ -365,7 +365,8 @@ private BalancerClusterState createCluster(List<ServerName> servers,
         clusterState.put(server, Collections.emptyList());
       }
     }
-    return new BalancerClusterState(regions, clusterState, null, this.regionFinder, rackManager);
+    return new BalancerClusterState(regions, clusterState, null, this.regionFinder, rackManager,
+      null);
   }
 
   private List<ServerName> findIdleServers(List<ServerName> servers) {