HBASE-28468: Integrate the data-tiering logic into cache evictions.

As a part of cache blocks eviction when the cache is full,
the data-tiering logic is integrated into the freeSpace code path
to identify the cold data files and evict the blocks associated with
those files.

The list of files is traversed to identify the cold files based
on the hot-data-age configuration and also the max timestamp associated
with the files.

The blocks associated with those cold files are evicted first and then
the existing logic of LFU blocks is executed to further evict the blocks.

Change-Id: I4b0a9fbffb274fb80dcc7e1303985901e4da22f2

Change-Id: Id09780357ef07549be63b32ed724259729f55563

Author: jhungund

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

@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -76,6 +77,7 @@
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.nio.RefCnt;
 import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
+import org.apache.hadoop.hbase.regionserver.DataTieringManager;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.IdReadWriteLock;
@@ -935,6 +937,7 @@ void freeSpace(final String why) {
     }
     try {
       freeInProgress = true;
+
       long bytesToFreeWithoutExtra = 0;
       // Calculate free byte for each bucketSizeinfo
       StringBuilder msgBuffer = LOG.isDebugEnabled() ? new StringBuilder() : null;
@@ -960,6 +963,7 @@ void freeSpace(final String why) {
       if (bytesToFreeWithoutExtra <= 0) {
         return;
       }
+
       long currentSize = bucketAllocator.getUsedSize();
       long totalSize = bucketAllocator.getTotalSize();
       if (LOG.isDebugEnabled() && msgBuffer != null) {
@@ -972,6 +976,13 @@ void freeSpace(final String why) {
       long bytesToFreeWithExtra =
         (long) Math.floor(bytesToFreeWithoutExtra * (1 + extraFreeFactor));
 
+      // Check the list of files to determine the cold files which can be readily evicted.
+      Map<String, String> coldFiles =
+        DataTieringManager.getInstance().getColdFilesList();
+      //Can we evict the blocks inline during backingMap traversal?
+      // If not, we accumulate the keys and evict them later.
+      // List<BlockCacheKey> coldBlocks = new ArrayList<>();
+
       // Instantiate priority buckets
       BucketEntryGroup bucketSingle =
         new BucketEntryGroup(bytesToFreeWithExtra, blockSize, getPartitionSize(singleFactor));
@@ -980,9 +991,18 @@ void freeSpace(final String why) {
       BucketEntryGroup bucketMemory =
         new BucketEntryGroup(bytesToFreeWithExtra, blockSize, getPartitionSize(memoryFactor));
 
+      long bytesFreed = 0;
+
       // Scan entire map putting bucket entry into appropriate bucket entry
       // group
       for (Map.Entry<BlockCacheKey, BucketEntry> bucketEntryWithKey : backingMap.entrySet()) {
+
+        if (coldFiles.containsKey(bucketEntryWithKey.getKey().getHfileName())) {
+          //coldBlocks.add(bucketEntryWithKey.getKey());
+          bytesFreed += bucketEntryWithKey.getValue().getLength();
+          evictBlock(bucketEntryWithKey.getKey());
+          continue;
+        }
         switch (bucketEntryWithKey.getValue().getPriority()) {
           case SINGLE: {
             bucketSingle.add(bucketEntryWithKey);
@@ -1007,7 +1027,6 @@ void freeSpace(final String why) {
       bucketQueue.add(bucketMemory);
 
       int remainingBuckets = bucketQueue.size();
-      long bytesFreed = 0;
 
       BucketEntryGroup bucketGroup;
       while ((bucketGroup = bucketQueue.poll()) != null) {

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DataTieringManager.java

@@ -17,15 +17,21 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.OptionalLong;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -219,4 +225,43 @@ private long getDataTieringHotDataAge(Configuration conf) {
     return Long.parseLong(
       conf.get(DATATIERING_HOT_DATA_AGE_KEY, String.valueOf(DEFAULT_DATATIERING_HOT_DATA_AGE)));
   }
+
+  /*
+   * This API takes the names of files as input and returns a subset of these file names
+   * that are cold.
+   * @parameter inputFileNames: Input list of file names
+   * @return List of names of files that are cold as per data-tiering logic.
+   */
+  public Map<String, String> getColdFilesList() {
+    Map<String, String> coldFiles = new HashMap<>();
+    for (HRegion r : this.onlineRegions.values()) {
+      for (HStore hStore : r.getStores()) {
+        Configuration conf = hStore.getReadOnlyConfiguration();
+        if (getDataTieringType(conf) != DataTieringType.TIME_RANGE) {
+          // Data-Tiering not enabled for the store. Just skip it.
+          continue;
+        }
+        Long hotDataAge = getDataTieringHotDataAge(conf);
+
+        for (HStoreFile hStoreFile : hStore.getStorefiles()) {
+          String hFileName =
+            hStoreFile.getFileInfo().getHFileInfo().getHFileContext().getHFileName();
+            OptionalLong maxTimestamp = hStoreFile.getMaximumTimestamp();
+            if (!maxTimestamp.isPresent()) {
+              // We could throw from here, But we are already in the critical code-path
+              // of freeing space. Hence, we can ignore that file for now
+              // Or do we want to include it?
+              continue;
+            }
+            long currentTimestamp = EnvironmentEdgeManager.getDelegate().currentTime();
+            long fileAge = currentTimestamp - maxTimestamp.getAsLong();
+            if (fileAge > hotDataAge) {
+              // Values do not matter.
+              coldFiles.put(hFileName, null);
+            }
+          }
+        }
+      }
+    return coldFiles;
+  }
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDataTieringManager.java

@@ -218,6 +218,14 @@ public void testColdDataFiles() {
     }
   }
 
+  @Test
+  public void testPickColdDataFiles() {
+    Map<String, String> coldDataFiles = dataTieringManager.getColdFilesList();
+    assertEquals(1, coldDataFiles.size());
+    // hStoreFiles[3] is the cold file.
+    assert(coldDataFiles.containsKey(hStoreFiles.get(3).getFileInfo().getActiveFileName()));
+  }
+
   private void testDataTieringMethodWithPath(DataTieringMethodCallerWithPath caller, Path path,
     boolean expectedResult, DataTieringException exception) {
     try {