HBASE-29626: Refactor server side scan metrics for Coproc hooks (#7348)

Signed-off-by: Viraj Jasani <[email protected]>

Author: sanjeet006py

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

@@ -40,7 +40,7 @@
 import org.apache.hadoop.hbase.io.compress.Compression;
 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
 import org.apache.hadoop.hbase.io.hfile.ReaderContext.ReaderType;
-import org.apache.hadoop.hbase.ipc.RpcServer;
+import org.apache.hadoop.hbase.monitoring.ThreadLocalServerSideScanMetrics;
 import org.apache.hadoop.hbase.regionserver.CellSink;
 import org.apache.hadoop.hbase.regionserver.ShipperListener;
 import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
@@ -190,7 +190,7 @@ public static final long getChecksumFailuresCount() {
   }
 
   public static final void updateReadLatency(long latencyMillis, boolean pread, boolean tooSlow) {
-    RpcServer.getCurrentCall().ifPresent(call -> call.updateFsReadTime(latencyMillis));
+    ThreadLocalServerSideScanMetrics.addFsReadTime(latencyMillis);
     if (pread) {
       MetricsIO.getInstance().updateFsPreadTime(latencyMillis);
     } else {

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

@@ -132,8 +132,4 @@ public interface RpcCall extends RpcCallContext {
 
   /** Returns A short string format of this call without possibly lengthy params */
   String toShortString();
-
-  void updateFsReadTime(long latencyMillis);
-
-  long getFsReadTime();
 }

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

@@ -44,6 +44,7 @@
 import org.apache.hadoop.hbase.io.ByteBuffAllocator;
 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler;
 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
+import org.apache.hadoop.hbase.monitoring.ThreadLocalServerSideScanMetrics;
 import org.apache.hadoop.hbase.namequeues.NamedQueueRecorder;
 import org.apache.hadoop.hbase.namequeues.RpcLogDetails;
 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
@@ -447,19 +448,18 @@ public Pair<Message, CellScanner> call(RpcCall call, MonitoredRPCHandler status)
       int processingTime = (int) (endTime - startTime);
       int qTime = (int) (startTime - receiveTime);
       int totalTime = (int) (endTime - receiveTime);
+      long fsReadTime = ThreadLocalServerSideScanMetrics.getFsReadTimeCounter().get();
       if (LOG.isTraceEnabled()) {
         LOG.trace(
           "{}, response: {}, receiveTime: {}, queueTime: {}, processingTime: {}, "
             + "totalTime: {}, fsReadTime: {}",
           CurCall.get().toString(), TextFormat.shortDebugString(result),
-          CurCall.get().getReceiveTime(), qTime, processingTime, totalTime,
-          CurCall.get().getFsReadTime());
+          CurCall.get().getReceiveTime(), qTime, processingTime, totalTime, fsReadTime);
       }
       // Use the raw request call size for now.
       long requestSize = call.getSize();
       long responseSize = result.getSerializedSize();
       long responseBlockSize = call.getBlockBytesScanned();
-      long fsReadTime = call.getFsReadTime();
       if (call.isClientCellBlockSupported()) {
         // Include the payload size in HBaseRpcController
         responseSize += call.getResponseCellSize();

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

@@ -102,7 +102,6 @@ public abstract class ServerCall<T extends ServerRpcConnection> implements RpcCa
 
   private long responseCellSize = 0;
   private long responseBlockSize = 0;
-  private long fsReadTimeMillis = 0;
   // cumulative size of serialized exceptions
   private long exceptionSize = 0;
   private final boolean retryImmediatelySupported;
@@ -610,14 +609,4 @@ public synchronized BufferChain getResponse() {
   public synchronized RpcCallback getCallBack() {
     return this.rpcCallback;
   }
-
-  @Override
-  public void updateFsReadTime(long latencyMillis) {
-    fsReadTimeMillis += latencyMillis;
-  }
-
-  @Override
-  public long getFsReadTime() {
-    return fsReadTimeMillis;
-  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/monitoring/ThreadLocalServerSideScanMetrics.java

@@ -18,10 +18,12 @@
 package org.apache.hadoop.hbase.monitoring;
 
 import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics;
 import org.apache.hadoop.hbase.regionserver.RegionScanner;
 import org.apache.hadoop.hbase.regionserver.ScannerContext;
 import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.yetus.audience.InterfaceStability;
 
 /**
  * Thread-local storage for server-side scan metrics that captures performance data separately for
@@ -61,7 +63,8 @@
  * @see RegionScanner
  * @see org.apache.hadoop.hbase.regionserver.handler.ParallelSeekHandler
  */
-@InterfaceAudience.Private
+@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.PHOENIX)
+@InterfaceStability.Evolving
 public final class ThreadLocalServerSideScanMetrics {
   private ThreadLocalServerSideScanMetrics() {
   }
@@ -81,6 +84,9 @@ private ThreadLocalServerSideScanMetrics() {
   private static final ThreadLocal<AtomicLong> BLOCK_READ_OPS_COUNT =
     ThreadLocal.withInitial(() -> new AtomicLong(0));
 
+  private static final ThreadLocal<AtomicLong> FS_READ_TIME =
+    ThreadLocal.withInitial(() -> new AtomicLong(0));
+
   public static void setScanMetricsEnabled(boolean enable) {
     IS_SCAN_METRICS_ENABLED.set(enable);
   }
@@ -101,6 +107,10 @@ public static long addBlockReadOpsCount(long count) {
     return BLOCK_READ_OPS_COUNT.get().addAndGet(count);
   }
 
+  public static long addFsReadTime(long time) {
+    return FS_READ_TIME.get().addAndGet(time);
+  }
+
   public static boolean isScanMetricsEnabled() {
     return IS_SCAN_METRICS_ENABLED.get();
   }
@@ -121,6 +131,10 @@ public static AtomicLong getBlockReadOpsCountCounter() {
     return BLOCK_READ_OPS_COUNT.get();
   }
 
+  public static AtomicLong getFsReadTimeCounter() {
+    return FS_READ_TIME.get();
+  }
+
   public static long getBytesReadFromFsAndReset() {
     return getBytesReadFromFsCounter().getAndSet(0);
   }
@@ -137,11 +151,16 @@ public static long getBlockReadOpsCountAndReset() {
     return getBlockReadOpsCountCounter().getAndSet(0);
   }
 
+  public static long getFsReadTimeAndReset() {
+    return getFsReadTimeCounter().getAndSet(0);
+  }
+
   public static void reset() {
     getBytesReadFromFsAndReset();
     getBytesReadFromBlockCacheAndReset();
     getBytesReadFromMemstoreAndReset();
     getBlockReadOpsCountAndReset();
+    getFsReadTimeAndReset();
   }
 
   public static void populateServerSideScanMetrics(ServerSideScanMetrics metrics) {
@@ -156,5 +175,7 @@ public static void populateServerSideScanMetrics(ServerSideScanMetrics metrics)
       getBytesReadFromMemstoreCounter().get());
     metrics.addToCounter(ServerSideScanMetrics.BLOCK_READ_OPS_COUNT_METRIC_NAME,
       getBlockReadOpsCountCounter().get());
+    metrics.addToCounter(ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME,
+      getFsReadTimeCounter().get());
   }
 }

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

@@ -112,6 +112,7 @@
 import org.apache.hadoop.hbase.log.HBaseMarkers;
 import org.apache.hadoop.hbase.master.HMaster;
 import org.apache.hadoop.hbase.master.MasterRpcServices;
+import org.apache.hadoop.hbase.monitoring.ThreadLocalServerSideScanMetrics;
 import org.apache.hadoop.hbase.namequeues.NamedQueuePayload;
 import org.apache.hadoop.hbase.namequeues.NamedQueueRecorder;
 import org.apache.hadoop.hbase.namequeues.RpcLogDetails;
@@ -3570,10 +3571,6 @@ private void scan(HBaseRpcController controller, ScanRequest request, RegionScan
           // from block size progress before writing into the response
           scanMetrics.setCounter(ServerSideScanMetrics.BLOCK_BYTES_SCANNED_KEY_METRIC_NAME,
             scannerContext.getBlockSizeProgress());
-          if (rpcCall != null) {
-            scanMetrics.setCounter(ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME,
-              rpcCall.getFsReadTime());
-          }
         }
       }
     } finally {
@@ -3639,6 +3636,11 @@ public ScanResponse scan(final RpcController controller, final ScanRequest reque
       }
       throw new ServiceException(e);
     }
+    boolean trackMetrics = request.hasTrackScanMetrics() && request.getTrackScanMetrics();
+    ThreadLocalServerSideScanMetrics.setScanMetricsEnabled(trackMetrics);
+    if (trackMetrics) {
+      ThreadLocalServerSideScanMetrics.reset();
+    }
     requestCount.increment();
     rpcScanRequestCount.increment();
     RegionScannerContext rsx;
@@ -3709,7 +3711,6 @@ public ScanResponse scan(final RpcController controller, final ScanRequest reque
     boolean scannerClosed = false;
     try {
       List<Result> results = new ArrayList<>(Math.min(rows, 512));
-      boolean trackMetrics = request.hasTrackScanMetrics() && request.getTrackScanMetrics();
       ServerSideScanMetrics scanMetrics = trackMetrics ? new ServerSideScanMetrics() : null;
       if (rows > 0) {
         boolean done = false;
@@ -3791,6 +3792,7 @@ public ScanResponse scan(final RpcController controller, final ScanRequest reque
           scanMetrics.addToCounter(ServerSideScanMetrics.RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME,
             rpcQueueWaitTime);
         }
+        ThreadLocalServerSideScanMetrics.populateServerSideScanMetrics(scanMetrics);
         Map<String, Long> metrics = scanMetrics.getMetricsMap();
         ScanMetrics.Builder metricBuilder = ScanMetrics.newBuilder();
         NameInt64Pair.Builder pairBuilder = NameInt64Pair.newBuilder();

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

@@ -44,7 +44,6 @@
 import org.apache.hadoop.hbase.ipc.RpcCall;
 import org.apache.hadoop.hbase.ipc.RpcCallback;
 import org.apache.hadoop.hbase.ipc.RpcServer;
-import org.apache.hadoop.hbase.monitoring.ThreadLocalServerSideScanMetrics;
 import org.apache.hadoop.hbase.regionserver.Region.Operation;
 import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope;
 import org.apache.hadoop.hbase.regionserver.ScannerContext.NextState;
@@ -95,8 +94,6 @@ public class RegionScannerImpl implements RegionScanner, Shipper, RpcCallback {
 
   private RegionServerServices rsServices;
 
-  private ServerSideScanMetrics scannerInitMetrics = null;
-
   @Override
   public RegionInfo getRegionInfo() {
     return region.getRegionInfo();
@@ -147,16 +144,7 @@ private static boolean hasNonce(HRegion region, long nonce) {
     } finally {
       region.smallestReadPointCalcLock.unlock(ReadPointCalculationLock.LockType.RECORDING_LOCK);
     }
-    boolean isScanMetricsEnabled = scan.isScanMetricsEnabled();
-    ThreadLocalServerSideScanMetrics.setScanMetricsEnabled(isScanMetricsEnabled);
-    if (isScanMetricsEnabled) {
-      this.scannerInitMetrics = new ServerSideScanMetrics();
-      ThreadLocalServerSideScanMetrics.reset();
-    }
     initializeScanners(scan, additionalScanners);
-    if (isScanMetricsEnabled) {
-      ThreadLocalServerSideScanMetrics.populateServerSideScanMetrics(scannerInitMetrics);
-    }
   }
 
   public ScannerContext getContext() {
@@ -289,16 +277,6 @@ public boolean nextRaw(List<Cell> outResults, ScannerContext scannerContext) thr
       throw new UnknownScannerException("Scanner was closed");
     }
     boolean moreValues = false;
-    boolean isScanMetricsEnabled = scannerContext.isTrackingMetrics();
-    ThreadLocalServerSideScanMetrics.setScanMetricsEnabled(isScanMetricsEnabled);
-    if (isScanMetricsEnabled) {
-      ThreadLocalServerSideScanMetrics.reset();
-      ServerSideScanMetrics scanMetrics = scannerContext.getMetrics();
-      if (scannerInitMetrics != null) {
-        scannerInitMetrics.getMetricsMap().forEach(scanMetrics::addToCounter);
-        scannerInitMetrics = null;
-      }
-    }
     if (outResults.isEmpty()) {
       // Usually outResults is empty. This is true when next is called
       // to handle scan or get operation.
@@ -308,10 +286,6 @@ public boolean nextRaw(List<Cell> outResults, ScannerContext scannerContext) thr
       moreValues = nextInternal(tmpList, scannerContext);
       outResults.addAll(tmpList);
     }
-    if (isScanMetricsEnabled) {
-      ServerSideScanMetrics scanMetrics = scannerContext.getMetrics();
-      ThreadLocalServerSideScanMetrics.populateServerSideScanMetrics(scanMetrics);
-    }
     region.addReadRequestsCount(1);
     if (region.getMetrics() != null) {
       region.getMetrics().updateReadRequestCount();

hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestNamedQueueRecorder.java

@@ -919,16 +919,6 @@ public long getResponseExceptionSize() {
       @Override
       public void incrementResponseExceptionSize(long exceptionSize) {
       }
-
-      @Override
-      public void updateFsReadTime(long latencyMillis) {
-
-      }
-
-      @Override
-      public long getFsReadTime() {
-        return 0;
-      }
     };
     return rpcCall;
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestRpcLogDetails.java

@@ -264,16 +264,6 @@ public long getResponseExceptionSize() {
       @Override
       public void incrementResponseExceptionSize(long exceptionSize) {
       }
-
-      @Override
-      public void updateFsReadTime(long latencyMillis) {
-
-      }
-
-      @Override
-      public long getFsReadTime() {
-        return 0;
-      }
     };
     return rpcCall;
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/procedure2/store/region/TestRegionProcedureStore.java

@@ -326,16 +326,6 @@ public long getResponseExceptionSize() {
       @Override
       public void incrementResponseExceptionSize(long exceptionSize) {
       }
-
-      @Override
-      public void updateFsReadTime(long latencyMillis) {
-
-      }
-
-      @Override
-      public long getFsReadTime() {
-        return 0;
-      }
     };
   }
 }