HBASE-27532 Add block bytes scanned metrics

hbase-client/src/main/java/org/apache/hadoop/hbase/client/metrics/ServerSideScanMetrics.java

@@ -28,6 +28,7 @@
  * Provides server side metrics related to scan operations.
  */
 @InterfaceAudience.Public
+@SuppressWarnings("checkstyle:VisibilityModifier") // See HBASE-27757
 public class ServerSideScanMetrics {
   /**
    * Hash to hold the String -> Atomic Long mappings for each metric
@@ -47,6 +48,8 @@ protected AtomicLong createCounter(String counterName) {
   public static final String COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME = "ROWS_SCANNED";
   public static final String COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME = "ROWS_FILTERED";
 
+  public static final String BLOCK_BYTES_SCANNED_KEY_METRIC_NAME = "BLOCK_BYTES_SCANNED";
+
   /**
    * number of rows filtered during scan RPC
    */
@@ -59,6 +62,9 @@ protected AtomicLong createCounter(String counterName) {
    */
   public final AtomicLong countOfRowsScanned = createCounter(COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME);
 
+  public final AtomicLong countOfBlockBytesScanned =
+    createCounter(BLOCK_BYTES_SCANNED_KEY_METRIC_NAME);
+
   public void setCounter(String counterName, long value) {
     AtomicLong c = this.counters.get(counterName);
     if (c != null) {

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerSource.java

@@ -85,27 +85,32 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
 
   /**
    * Update checkAndMutate histogram
-   * @param t time it took
+   * @param time              time it took
+   * @param blockBytesScanned how many block bytes were scanned for the check portion of the request
    */
-  void updateCheckAndMutate(long t);
+  void updateCheckAndMutate(long time, long blockBytesScanned);
 
   /**
    * Update the Get time histogram .
-   * @param t time it took
+   * @param time              time it took
+   * @param blockBytesScanned how many block bytes were scanned for the request
    */
-  void updateGet(long t);
+  void updateGet(long time, long blockBytesScanned);
 
   /**
    * Update the Increment time histogram.
-   * @param t time it took
+   * @param time              time it took
+   * @param blockBytesScanned how many block bytes were scanned fetching the current value to
+   *                          increment
    */
-  void updateIncrement(long t);
+  void updateIncrement(long time, long blockBytesScanned);
 
   /**
    * Update the Append time histogram.
-   * @param t time it took
+   * @param time              time it took
+   * @param blockBytesScanned how many block bytes were scanned fetching the current value to append
    */
-  void updateAppend(long t);
+  void updateAppend(long time, long blockBytesScanned);
 
   /**
    * Update the Replay time histogram.
@@ -114,15 +119,12 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
   void updateReplay(long t);
 
   /**
-   * Update the scan size.
-   * @param scanSize size of the scan
-   */
-  void updateScanSize(long scanSize);
-
-  /**
-   * Update the scan time.
+   * Update the scan metrics.
+   * @param time              response time of scan
+   * @param responseCellSize  size of the scan resposne
+   * @param blockBytesScanned size of block bytes scanned to retrieve the response
    */
-  void updateScanTime(long t);
+  void updateScan(long time, long responseCellSize, long blockBytesScanned);
 
   /**
    * Increment the number of slow Puts that have happened.
@@ -445,6 +447,13 @@ public interface MetricsRegionServerSource extends BaseSource, JvmPauseMonitorSo
   String SCAN_SIZE_KEY = "scanSize";
   String SCAN_TIME_KEY = "scanTime";
 
+  String BLOCK_BYTES_SCANNED_KEY = "blockBytesScannedCount";
+  String BLOCK_BYTES_SCANNED_DESC = "Count of block bytes scanned by read requests";
+  String GET_BLOCK_BYTES_SCANNED_KEY = "getBlockBytesScanned";
+  String SCAN_BLOCK_BYTES_SCANNED_KEY = "scanBlockBytesScanned";
+  String CHECK_AND_MUTATE_BLOCK_BYTES_SCANNED_KEY = "checkAndMutateBlockBytesScanned";
+  String INCREMENT_BLOCK_BYTES_SCANNED_KEY = "incrementBlockBytesScanned";
+  String APPEND_BLOCK_BYTES_SCANNED_KEY = "appendBlockBytesScanned";
   String SLOW_PUT_KEY = "slowPutCount";
   String SLOW_GET_KEY = "slowGetCount";
   String SLOW_DELETE_KEY = "slowDeleteCount";

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServerSourceImpl.java

@@ -48,6 +48,13 @@ public class MetricsRegionServerSourceImpl extends BaseSourceImpl
   private final MetricHistogram scanSizeHisto;
   private final MetricHistogram scanTimeHisto;
 
+  private final MutableFastCounter blockBytesScannedCount;
+  private final MetricHistogram checkAndMutateBlockBytesScanned;
+  private final MetricHistogram getBlockBytesScanned;
+  private final MetricHistogram incrementBlockBytesScanned;
+  private final MetricHistogram appendBlockBytesScanned;
+  private final MetricHistogram scanBlockBytesScanned;
+
   private final MutableFastCounter slowPut;
   private final MutableFastCounter slowDelete;
   private final MutableFastCounter slowGet;
@@ -125,6 +132,16 @@ public MetricsRegionServerSourceImpl(String metricsName, String metricsDescripti
     scanSizeHisto = getMetricsRegistry().newSizeHistogram(SCAN_SIZE_KEY);
     scanTimeHisto = getMetricsRegistry().newTimeHistogram(SCAN_TIME_KEY);
 
+    blockBytesScannedCount =
+      getMetricsRegistry().newCounter(BLOCK_BYTES_SCANNED_KEY, BLOCK_BYTES_SCANNED_DESC, 0L);
+    checkAndMutateBlockBytesScanned =
+      getMetricsRegistry().newSizeHistogram(CHECK_AND_MUTATE_BLOCK_BYTES_SCANNED_KEY);
+    getBlockBytesScanned = getMetricsRegistry().newSizeHistogram(GET_BLOCK_BYTES_SCANNED_KEY);
+    incrementBlockBytesScanned =
+      getMetricsRegistry().newSizeHistogram(INCREMENT_BLOCK_BYTES_SCANNED_KEY);
+    appendBlockBytesScanned = getMetricsRegistry().newSizeHistogram(APPEND_BLOCK_BYTES_SCANNED_KEY);
+    scanBlockBytesScanned = getMetricsRegistry().newSizeHistogram(SCAN_BLOCK_BYTES_SCANNED_KEY);
+
     flushTimeHisto = getMetricsRegistry().newTimeHistogram(FLUSH_TIME, FLUSH_TIME_DESC);
     flushMemstoreSizeHisto =
       getMetricsRegistry().newSizeHistogram(FLUSH_MEMSTORE_SIZE, FLUSH_MEMSTORE_SIZE_DESC);
@@ -192,18 +209,30 @@ public void updateDelete(long t) {
   }
 
   @Override
-  public void updateGet(long t) {
+  public void updateGet(long t, long blockBytesScanned) {
     getHisto.add(t);
+    if (blockBytesScanned > 0) {
+      blockBytesScannedCount.incr(blockBytesScanned);
+      getBlockBytesScanned.add(blockBytesScanned);
+    }
   }
 
   @Override
-  public void updateIncrement(long t) {
+  public void updateIncrement(long t, long blockBytesScanned) {
     incrementHisto.add(t);
+    if (blockBytesScanned > 0) {
+      blockBytesScannedCount.incr(blockBytesScanned);
+      incrementBlockBytesScanned.add(blockBytesScanned);
+    }
   }
 
   @Override
-  public void updateAppend(long t) {
+  public void updateAppend(long t, long blockBytesScanned) {
     appendHisto.add(t);
+    if (blockBytesScanned > 0) {
+      blockBytesScannedCount.incr(blockBytesScanned);
+      appendBlockBytesScanned.add(blockBytesScanned);
+    }
   }
 
   @Override
@@ -212,13 +241,13 @@ public void updateReplay(long t) {
   }
 
   @Override
-  public void updateScanSize(long scanSize) {
-    scanSizeHisto.add(scanSize);
-  }
-
-  @Override
-  public void updateScanTime(long t) {
-    scanTimeHisto.add(t);
+  public void updateScan(long time, long responseSize, long blockBytesScanned) {
+    scanTimeHisto.add(time);
+    scanSizeHisto.add(responseSize);
+    if (blockBytesScanned > 0) {
+      blockBytesScannedCount.incr(blockBytesScanned);
+      scanBlockBytesScanned.add(blockBytesScanned);
+    }
   }
 
   @Override
@@ -646,8 +675,12 @@ public void updateCheckAndPut(long t) {
   }
 
   @Override
-  public void updateCheckAndMutate(long t) {
-    checkAndMutateHisto.add(t);
+  public void updateCheckAndMutate(long time, long blockBytesScanned) {
+    checkAndMutateHisto.add(time);
+    if (blockBytesScanned > 0) {
+      blockBytesScannedCount.incr(blockBytesScanned);
+      checkAndMutateBlockBytesScanned.add(blockBytesScanned);
+    }
   }
 
   @Override

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsUserSource.java

@@ -51,15 +51,17 @@ interface ClientMetrics {
 
   void updateDelete(long t);
 
-  void updateGet(long t);
+  void updateGet(long time, long blockBytesScanned);
 
-  void updateIncrement(long t);
+  void updateIncrement(long time, long blockBytesScanned);
 
-  void updateAppend(long t);
+  void updateAppend(long time, long blockBytesScanned);
 
   void updateReplay(long t);
 
-  void updateScanTime(long t);
+  void updateScan(long time, long blockBytesScanned);
+
+  void updateCheckAndMutate(long blockBytesScanned);
 
   void getMetrics(MetricsCollector metricsCollector, boolean all);
 

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsUserSourceImpl.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.metrics2.MetricsCollector;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
 import org.apache.hadoop.metrics2.lib.DynamicMetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableFastCounter;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,6 +46,7 @@ public class MetricsUserSourceImpl implements MetricsUserSource {
   private final String userIncrementKey;
   private final String userAppendKey;
   private final String userReplayKey;
+  private final String userBlockBytesScannedKey;
 
   private MetricHistogram getHisto;
   private MetricHistogram scanTimeHisto;
@@ -53,6 +55,7 @@ public class MetricsUserSourceImpl implements MetricsUserSource {
   private MetricHistogram incrementHisto;
   private MetricHistogram appendHisto;
   private MetricHistogram replayHisto;
+  private MutableFastCounter blockBytesScannedCount;
 
   private final int hashCode;
 
@@ -116,7 +119,7 @@ public MetricsUserSourceImpl(String user, MetricsUserAggregateSourceImpl agg) {
     this.user = user;
     this.registry = agg.getMetricsRegistry();
 
-    this.userNamePrefix = "user_" + user + "_metric_";
+    this.userNamePrefix = "User_" + user + "_metric_";
 
     hashCode = userNamePrefix.hashCode();
 
@@ -127,6 +130,7 @@ public MetricsUserSourceImpl(String user, MetricsUserAggregateSourceImpl agg) {
     userIncrementKey = userNamePrefix + MetricsRegionServerSource.INCREMENT_KEY;
     userAppendKey = userNamePrefix + MetricsRegionServerSource.APPEND_KEY;
     userReplayKey = userNamePrefix + MetricsRegionServerSource.REPLAY_KEY;
+    userBlockBytesScannedKey = userNamePrefix + MetricsRegionServerSource.BLOCK_BYTES_SCANNED_KEY;
     clientMetricsMap = new ConcurrentHashMap<>();
     agg.register(this);
   }
@@ -141,6 +145,7 @@ public void register() {
       incrementHisto = registry.newTimeHistogram(userIncrementKey);
       appendHisto = registry.newTimeHistogram(userAppendKey);
       replayHisto = registry.newTimeHistogram(userReplayKey);
+      blockBytesScannedCount = registry.newCounter(userBlockBytesScannedKey, "", 0);
     }
   }
 
@@ -165,6 +170,7 @@ public void deregister() {
       registry.removeMetric(userIncrementKey);
       registry.removeMetric(userAppendKey);
       registry.removeMetric(userReplayKey);
+      registry.removeMetric(userBlockBytesScannedKey);
     }
   }
 
@@ -231,18 +237,21 @@ public void updateDelete(long t) {
   }
 
   @Override
-  public void updateGet(long t) {
-    getHisto.add(t);
+  public void updateGet(long time, long blockBytesScanned) {
+    getHisto.add(time);
+    blockBytesScannedCount.incr(blockBytesScanned);
   }
 
   @Override
-  public void updateIncrement(long t) {
-    incrementHisto.add(t);
+  public void updateIncrement(long time, long blockBytesScanned) {
+    incrementHisto.add(time);
+    blockBytesScannedCount.incr(blockBytesScanned);
   }
 
   @Override
-  public void updateAppend(long t) {
-    appendHisto.add(t);
+  public void updateAppend(long time, long blockBytesScanned) {
+    appendHisto.add(time);
+    blockBytesScannedCount.incr(blockBytesScanned);
   }
 
   @Override
@@ -251,8 +260,14 @@ public void updateReplay(long t) {
   }
 
   @Override
-  public void updateScanTime(long t) {
-    scanTimeHisto.add(t);
+  public void updateScan(long time, long blockBytesScanned) {
+    scanTimeHisto.add(time);
+    blockBytesScannedCount.incr(blockBytesScanned);
+  }
+
+  @Override
+  public void updateCheckAndMutate(long blockBytesScanned) {
+    blockBytesScannedCount.incr(blockBytesScanned);
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcCallContext.java

@@ -87,9 +87,18 @@ default Optional<String> getRequestUserName() {
    */
   void incrementResponseCellSize(long cellSize);
 
-  long getResponseBlockSize();
+  /**
+   * Get the number of block bytes scanned by the current call. In order to serve a response, 1 or
+   * more lower level blocks must be loaded (from disk or cache) and scanned for the requested
+   * cells. This value includes the total block size for each block loaded for the request.
+   */
+  long getBlockBytesScanned();
 
-  void incrementResponseBlockSize(long blockSize);
+  /**
+   * Increment the number of block bytes scanned by the current call. See
+   * {@link #getBlockBytesScanned()} for details.
+   */
+  void incrementBlockBytesScanned(long blockSize);
 
   long getResponseExceptionSize();
 

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java

@@ -428,7 +428,7 @@ public Pair<Message, CellScanner> call(RpcCall call, MonitoredRPCHandler status)
       // Use the raw request call size for now.
       long requestSize = call.getSize();
       long responseSize = result.getSerializedSize();
-      long responseBlockSize = call.getResponseBlockSize();
+      long responseBlockSize = call.getBlockBytesScanned();
       if (call.isClientCellBlockSupported()) {
         // Include the payload size in HBaseRpcController
         responseSize += call.getResponseCellSize();

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/ServerCall.java

@@ -424,12 +424,12 @@ public void incrementResponseCellSize(long cellSize) {
   }
 
   @Override
-  public long getResponseBlockSize() {
+  public long getBlockBytesScanned() {
     return responseBlockSize;
   }
 
   @Override
-  public void incrementResponseBlockSize(long blockSize) {
+  public void incrementBlockBytesScanned(long blockSize) {
     responseBlockSize += blockSize;
   }