HBASE-29233: Capture scan metrics at region level

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

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hbase.client;
 
+import java.util.ArrayList;
+import java.util.List;
 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -25,15 +27,33 @@
  */
 @InterfaceAudience.Private
 public abstract class AbstractClientScanner implements ResultScanner {
-  protected ScanMetrics scanMetrics;
+  protected ScanMetrics scanMetrics = null;
+  protected List<ScanMetrics> scanMetricsByRegion;
+
+  /**
+   * Check and initialize list for collecting scan metrics if application wants to collect scan
+   * metrics per region
+   */
+  protected void initScanMetricsByRegion(Scan scan) {
+    // check if application wants to collect scan metrics
+    if (scan.isScanMetricsEnabled() && scan.isScanMetricsByRegionEnabled()) {
+      scanMetricsByRegion = new ArrayList<>();
+    }
+  }
 
   /**
    * Check and initialize if application wants to collect scan metrics
    */
   protected void initScanMetrics(Scan scan) {
     // check if application wants to collect scan metrics
     if (scan.isScanMetricsEnabled()) {
-      scanMetrics = new ScanMetrics();
+      if (scanMetricsByRegion != null) {
+        scanMetrics = new ScanMetrics();
+        scanMetricsByRegion.add(scanMetrics);
+      } else if (scanMetrics == null) {
+        // Only initialize once
+        this.scanMetrics = new ScanMetrics();
+      }
     }
   }
 
@@ -44,6 +64,24 @@ protected void initScanMetrics(Scan scan) {
    */
   @Override
   public ScanMetrics getScanMetrics() {
-    return scanMetrics;
+    if (scanMetricsByRegion != null) {
+      if (scanMetricsByRegion.isEmpty()) {
+        return null;
+      } else if (scanMetricsByRegion.size() == 1) {
+        return scanMetricsByRegion.get(0);
+      }
+      ScanMetrics overallScanMetrics = new ScanMetrics();
+      for (ScanMetrics otherScanMetrics : scanMetricsByRegion) {
+        overallScanMetrics.combineMetrics(otherScanMetrics);
+      }
+      return overallScanMetrics;
+    } else {
+      return scanMetrics;
+    }
+  }
+
+  @Override
+  public List<ScanMetrics> getScanMetricsByRegion() {
+    return scanMetricsByRegion;
   }
 }

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

@@ -32,6 +32,8 @@
 import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.context.Scope;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -67,7 +69,7 @@ class AsyncClientScanner {
   // AsyncScanSingleRegionRpcRetryingCaller will modify this scan object directly.
   private final Scan scan;
 
-  private final ScanMetrics scanMetrics;
+  private ScanMetrics scanMetrics;
 
   private final AdvancedScanResultConsumer consumer;
 
@@ -92,6 +94,7 @@ class AsyncClientScanner {
   private final ScanResultCache resultCache;
 
   private final Span span;
+  private List<ScanMetrics> scanMetricByRegion;
 
   private final Map<String, byte[]> requestAttributes;
 
@@ -119,8 +122,15 @@ public AsyncClientScanner(Scan scan, AdvancedScanResultConsumer consumer, TableN
     this.resultCache = createScanResultCache(scan);
     this.requestAttributes = requestAttributes;
     if (scan.isScanMetricsEnabled()) {
-      this.scanMetrics = new ScanMetrics();
-      consumer.onScanMetricsCreated(scanMetrics);
+      if (scan.isScanMetricsByRegionEnabled()) {
+        // When scan metrics per region is enabled we will create scan metrics while calling
+        // openScanner.
+        this.scanMetricByRegion = new ArrayList<>();
+        consumer.onScanMetricsByRegionEnabled(scanMetricByRegion);
+      } else {
+        this.scanMetrics = new ScanMetrics();
+        consumer.onScanMetricsCreated(scanMetrics);
+      }
     } else {
       this.scanMetrics = null;
     }
@@ -165,6 +175,10 @@ public OpenScannerResponse(HRegionLocation loc, boolean isRegionServerRemote, In
   private CompletableFuture<OpenScannerResponse> callOpenScanner(HBaseRpcController controller,
     HRegionLocation loc, ClientService.Interface stub) {
     try (Scope ignored = span.makeCurrent()) {
+      if (this.scanMetrics != null && scan.isScanMetricsByRegionEnabled()) {
+        this.scanMetrics.setServerName(loc.getServerName());
+        this.scanMetrics.setEncodedRegionName(loc.getRegion().getEncodedName());
+      }
       boolean isRegionServerRemote = isRemote(loc.getHostname());
       incRPCCallsMetrics(scanMetrics, isRegionServerRemote);
       if (openScannerTries.getAndIncrement() > 1) {
@@ -250,6 +264,10 @@ private long getPrimaryTimeoutNs() {
   }
 
   private void openScanner() {
+    if (scan.isScanMetricsEnabled() && scan.isScanMetricsByRegionEnabled()) {
+      this.scanMetrics = new ScanMetrics();
+      this.scanMetricByRegion.add(this.scanMetrics);
+    }
     incRegionCountMetrics(scanMetrics);
     openScannerTries.set(1);
     addListener(timelineConsistentRead(conn.getLocator(), tableName, scan, scan.getStartRow(),

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

@@ -246,7 +246,11 @@ private void scan0(Scan scan, ScanResultConsumer consumer) {
       span = scanner.getSpan();
       try (Scope ignored = span.makeCurrent()) {
         if (scan.isScanMetricsEnabled()) {
-          consumer.onScanMetricsCreated(scanner.getScanMetrics());
+          if (scan.isScanMetricsByRegionEnabled()) {
+            consumer.onScanMetricsByRegionEnabled(scanner.getScanMetricsByRegion());
+          } else {
+            consumer.onScanMetricsCreated(scanner.getScanMetrics());
+          }
         }
         for (Result result; (result = scanner.next()) != null;) {
           if (!consumer.onNext(result)) {

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

@@ -23,6 +23,7 @@
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.ArrayDeque;
+import java.util.List;
 import java.util.Queue;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.AsyncScanSingleRegionRpcRetryingCaller.ScanResumerImpl;
@@ -62,6 +63,7 @@ class AsyncTableResultScanner implements ResultScanner, AdvancedScanResultConsum
 
   // Used to pass the span instance to the `AsyncTableImpl` from its underlying `rawAsyncTable`.
   private Span span = null;
+  private List<ScanMetrics> scanMetricsByRegion;
 
   public AsyncTableResultScanner(TableName tableName, Scan scan, long maxCacheSize) {
     this.tableName = tableName;
@@ -211,7 +213,30 @@ synchronized boolean isSuspended() {
 
   @Override
   public ScanMetrics getScanMetrics() {
-    return scanMetrics;
+    if (scanMetricsByRegion != null) {
+      if (scanMetricsByRegion.isEmpty()) {
+        return null;
+      } else if (scanMetricsByRegion.size() == 1) {
+        return scanMetricsByRegion.get(0);
+      }
+      ScanMetrics overallScanMetrics = new ScanMetrics();
+      for (ScanMetrics otherScanMetrics : scanMetricsByRegion) {
+        overallScanMetrics.combineMetrics(otherScanMetrics);
+      }
+      return overallScanMetrics;
+    } else {
+      return scanMetrics;
+    }
+  }
+
+  @Override
+  public void onScanMetricsByRegionEnabled(List<ScanMetrics> scanMetricsByRegion) {
+    this.scanMetricsByRegion = scanMetricsByRegion;
+  }
+
+  @Override
+  public List<ScanMetrics> getScanMetricsByRegion() {
+    return scanMetricsByRegion;
   }
 
   int getCacheSize() {

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

@@ -115,4 +115,8 @@ default Result[] next(int nbRows) throws IOException {
 
   /** Returns the scan metrics, or {@code null} if we do not enable metrics. */
   ScanMetrics getScanMetrics();
+
+  default List<ScanMetrics> getScanMetricsByRegion() {
+    throw new UnsupportedOperationException("Scan Metrics by region is not supported");
+  }
 }

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

@@ -115,6 +115,8 @@ public class Scan extends Query {
   // define this attribute with the appropriate table name by calling
   // scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName))
   static public final String SCAN_ATTRIBUTES_TABLE_NAME = "scan.attributes.table.name";
+  static private final String SCAN_ATTRIBUTES_METRICS_BY_REGION_ENABLE =
+    "scan.attributes.metrics.byRegion.enable";
 
   /**
    * -1 means no caching specified and the value of {@link HConstants#HBASE_CLIENT_SCANNER_CACHING}
@@ -1030,4 +1032,15 @@ public boolean isNeedCursorResult() {
   public static Scan createScanFromCursor(Cursor cursor) {
     return new Scan().withStartRow(cursor.getRow());
   }
+
+  public Scan setEnableScanMetricsByRegion(final boolean enable) {
+    setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_BY_REGION_ENABLE,
+      Bytes.toBytes(Boolean.valueOf(enable)));
+    return this;
+  }
+
+  public boolean isScanMetricsByRegionEnabled() {
+    byte[] attr = getAttribute(Scan.SCAN_ATTRIBUTES_METRICS_BY_REGION_ENABLE);
+    return attr != null && Bytes.toBoolean(attr);
+  }
 }

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.client;
 
+import java.util.List;
 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -38,11 +39,22 @@ public interface ScanResultConsumerBase {
   void onComplete();
 
   /**
-   * If {@code scan.isScanMetricsEnabled()} returns true, then this method will be called prior to
-   * all other methods in this interface to give you the {@link ScanMetrics} instance for this scan
-   * operation. The {@link ScanMetrics} instance will be updated on-the-fly during the scan, you can
-   * store it somewhere to get the metrics at any time if you want.
+   * If {@code scan.isScanMetricsEnabled()} returns true and
+   * {@code scan.isScanMetricsByRegionEnabled()} returns false, then this method will be called
+   * prior to all other methods in this interface to give you the {@link ScanMetrics} instance for
+   * this scan operation. The {@link ScanMetrics} instance will be updated on-the-fly during the
+   * scan, you can store it somewhere to get the metrics at any time if you want.
    */
   default void onScanMetricsCreated(ScanMetrics scanMetrics) {
   }
+
+  /**
+   * If {@code scan.isScanMetricsEnabled()} and {@code scan.isScanMetricsByRegionEnabled()} both
+   * return true, then this method will be called prior to calling all the methods in this interface
+   * to give you the list for storing scan metric per region for this scan operation. The list will
+   * be get populated with per region scan metrics on-the-fly during the scan, you can store the
+   * provided list somewhere to get scan metrics by region later when you want.
+   */
+  default void onScanMetricsByRegionEnabled(List<ScanMetrics> scanMetricsByRegion) {
+  }
 }

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

@@ -19,7 +19,9 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.hbase.ServerName;
 import org.apache.yetus.audience.InterfaceAudience;
 
 import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
@@ -33,7 +35,9 @@ public class ServerSideScanMetrics {
   /**
    * Hash to hold the String -> Atomic Long mappings for each metric
    */
-  private final Map<String, AtomicLong> counters = new HashMap<>();
+  protected final Map<String, AtomicLong> counters = new HashMap<>();
+  private ServerName serverName;
+  private String encodedRegionName;
 
   /**
    * Create a new counter with the specified name
@@ -117,4 +121,64 @@ public Map<String, Long> getMetricsMap(boolean reset) {
     // Build the immutable map so that people can't mess around with it.
     return builder.build();
   }
+
+  public ServerName getServerName() {
+    return this.serverName;
+  }
+
+  public void setServerName(ServerName serverName) {
+    if (this.serverName == null) {
+      this.serverName = serverName;
+    }
+  }
+
+  public void setEncodedRegionName(String encodedRegionName) {
+    if (this.encodedRegionName == null) {
+      this.encodedRegionName = encodedRegionName;
+    }
+  }
+
+  public String getEncodedRegionName() {
+    return this.encodedRegionName;
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    sb.append("ServerName:");
+    sb.append(this.serverName);
+    sb.append(",EncodedRegion:");
+    sb.append(this.encodedRegionName);
+    sb.append(",[");
+    boolean isFirstMetric = true;
+    for (Map.Entry<String, AtomicLong> e : this.counters.entrySet()) {
+      if (isFirstMetric) {
+        isFirstMetric = false;
+      } else {
+        sb.append(",");
+      }
+      sb.append(e.getKey());
+      sb.append(":");
+      sb.append(e.getValue().get());
+    }
+    sb.append("]");
+    return sb.toString();
+  }
+
+  public void combineMetrics(ScanMetrics other) {
+    for (Map.Entry<String, AtomicLong> entry : other.counters.entrySet()) {
+      String counterName = entry.getKey();
+      AtomicLong counter = entry.getValue();
+      this.addToCounter(counterName, counter.get());
+    }
+    if (
+      this.encodedRegionName != null
+        && !Objects.equals(this.encodedRegionName, other.getEncodedRegionName())
+    ) {
+      this.encodedRegionName = null;
+    }
+    if (this.serverName != null && !Objects.equals(this.serverName, other.getServerName())) {
+      this.serverName = null;
+    }
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/client/ClientSideRegionScanner.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -87,9 +88,17 @@ public ClientSideRegionScanner(Configuration conf, FileSystem fs, Path rootDir,
     values = new ArrayList<>();
 
     if (scanMetrics == null) {
+      initScanMetricsByRegion(scan);
       initScanMetrics(scan);
     } else {
       this.scanMetrics = scanMetrics;
+      if (scan.isScanMetricsByRegionEnabled()) {
+        this.scanMetricsByRegion = Collections.singletonList(this.scanMetrics);
+      }
+    }
+    if (this.scanMetrics != null && scan.isScanMetricsByRegionEnabled()) {
+      this.scanMetrics.setEncodedRegionName(region.getRegionInfo().getEncodedName());
+      // The server name will be null in scan metrics as this is a client side region scanner
     }
     region.startRegionOperation();
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/client/TableSnapshotScanner.java

@@ -134,7 +134,7 @@ public TableSnapshotScanner(Configuration conf, Path rootDir, Path restoreDir,
       openWithRestoringSnapshot();
     }
 
-    initScanMetrics(scan);
+    initScanMetricsByRegion(scan);
   }
 
   private void openWithoutRestoringSnapshot() throws IOException {
@@ -184,6 +184,7 @@ public Result next() throws IOException {
         }
 
         RegionInfo hri = regions.get(currentRegion);
+        initScanMetrics(scan);
         currentRegionScanner =
           new ClientSideRegionScanner(conf, fs, restoreDir, htd, hri, scan, scanMetrics);
         if (this.scanMetrics != null) {