HBASE-29233: Capture scan metrics at region level (apache#7132) (apache#6868)

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

Author: sanjeet006py

.gitignore

@@ -23,3 +23,4 @@ linklint/
 .java-version
 tmp
 **/.flattened-pom.xml
+.vscode/

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

@@ -26,6 +26,7 @@
 @InterfaceAudience.Private
 public abstract class AbstractClientScanner implements ResultScanner {
   protected ScanMetrics scanMetrics;
+  private boolean isScanMetricsByRegionEnabled = false;
 
   /**
    * Check and initialize if application wants to collect scan metrics
@@ -34,6 +35,9 @@ protected void initScanMetrics(Scan scan) {
     // check if application wants to collect scan metrics
     if (scan.isScanMetricsEnabled()) {
       scanMetrics = new ScanMetrics();
+      if (scan.isScanMetricsByRegionEnabled()) {
+        isScanMetricsByRegionEnabled = true;
+      }
     }
   }
 
@@ -46,4 +50,12 @@ protected void initScanMetrics(Scan scan) {
   public ScanMetrics getScanMetrics() {
     return scanMetrics;
   }
+
+  protected void setIsScanMetricsByRegionEnabled(boolean isScanMetricsByRegionEnabled) {
+    this.isScanMetricsByRegionEnabled = isScanMetricsByRegionEnabled;
+  }
+
+  protected boolean isScanMetricsByRegionEnabled() {
+    return isScanMetricsByRegionEnabled;
+  }
 }

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

@@ -95,6 +95,8 @@ class AsyncClientScanner {
 
   private final Map<String, byte[]> requestAttributes;
 
+  private final boolean isScanMetricsByRegionEnabled;
+
   public AsyncClientScanner(Scan scan, AdvancedScanResultConsumer consumer, TableName tableName,
     AsyncConnectionImpl conn, Timer retryTimer, long pauseNs, long pauseNsForServerOverloaded,
     int maxAttempts, long scanTimeoutNs, long rpcTimeoutNs, int startLogErrorsCnt,
@@ -118,12 +120,17 @@ public AsyncClientScanner(Scan scan, AdvancedScanResultConsumer consumer, TableN
     this.startLogErrorsCnt = startLogErrorsCnt;
     this.resultCache = createScanResultCache(scan);
     this.requestAttributes = requestAttributes;
+    boolean isScanMetricsByRegionEnabled = false;
     if (scan.isScanMetricsEnabled()) {
       this.scanMetrics = new ScanMetrics();
       consumer.onScanMetricsCreated(scanMetrics);
+      if (this.scan.isScanMetricsByRegionEnabled()) {
+        isScanMetricsByRegionEnabled = true;
+      }
     } else {
       this.scanMetrics = null;
     }
+    this.isScanMetricsByRegionEnabled = isScanMetricsByRegionEnabled;
 
     /*
      * Assumes that the `start()` method is called immediately after construction. If this is no
@@ -250,6 +257,9 @@ private long getPrimaryTimeoutNs() {
   }
 
   private void openScanner() {
+    if (this.isScanMetricsByRegionEnabled) {
+      scanMetrics.moveToNextRegion();
+    }
     incRegionCountMetrics(scanMetrics);
     openScannerTries.set(1);
     addListener(timelineConsistentRead(conn.getLocator(), tableName, scan, scan.getStartRow(),
@@ -265,6 +275,11 @@ private void openScanner() {
               span.end();
             }
           }
+          if (this.isScanMetricsByRegionEnabled) {
+            HRegionLocation loc = resp.loc;
+            this.scanMetrics.initScanMetricsRegionInfo(loc.getRegion().getEncodedName(),
+              loc.getServerName());
+          }
           startScan(resp);
         }
       });

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

@@ -34,10 +34,12 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.NotServingRegionException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.UnknownScannerException;
 import org.apache.hadoop.hbase.client.ScannerCallable.MoreResults;
+import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
 import org.apache.hadoop.hbase.exceptions.ScannerResetException;
 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
@@ -135,6 +137,11 @@ public ClientScanner(final Configuration conf, final Scan scan, final Scan scanF
     this.useScannerTimeoutForNextCalls = connectionConfiguration.isUseScannerTimeoutForNextCalls();
     this.requestAttributes = requestAttributes;
 
+    if (scan.isScanMetricsByRegionEnabled() && scan.getConsistency() == Consistency.TIMELINE) {
+      scan.setEnableScanMetricsByRegion(false);
+      scanForMetrics.setEnableScanMetricsByRegion(false);
+      LOG.warn("Scan metrics by region is not supported for timeline consistency in HBase 2");
+    }
     // check if application wants to collect scan metrics
     initScanMetrics(scan);
 
@@ -259,6 +266,9 @@ protected boolean moveToNextRegion() {
     // clear the current region, we will set a new value to it after the first call of the new
     // callable.
     this.currentRegion = null;
+    if (isScanMetricsByRegionEnabled()) {
+      scanMetrics.moveToNextRegion();
+    }
     this.callable = new ScannerCallableWithReplicas(getTable(), getConnection(),
       createScannerCallable(), pool, primaryOperationTimeout, scan, getRetries(), readRpcTimeout,
       scannerTimeout, useScannerTimeoutForNextCalls, caching, conf, caller);
@@ -281,6 +291,7 @@ private Result[] call(ScannerCallableWithReplicas callable, RpcRetryingCaller<Re
     Result[] rrs = caller.callWithoutRetries(callable, scannerTimeout);
     if (currentRegion == null && updateCurrentRegion) {
       currentRegion = callable.getHRegionInfo();
+      initScanMetricsRegionInfo();
     }
     return rrs;
   }
@@ -469,7 +480,8 @@ protected void loadCache() throws IOException {
       }
       long currentTime = EnvironmentEdgeManager.currentTime();
       if (this.scanMetrics != null) {
-        this.scanMetrics.sumOfMillisSecBetweenNexts.addAndGet(currentTime - lastNext);
+        this.scanMetrics.addToCounter(ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME,
+          currentTime - lastNext);
       }
       lastNext = currentTime;
       // Groom the array of Results that we received back from the server before adding that
@@ -622,4 +634,12 @@ public Result next() throws IOException {
       return nextWithSyncCache();
     }
   }
+
+  private void initScanMetricsRegionInfo() {
+    if (isScanMetricsByRegionEnabled()) {
+      HRegionLocation location = callable.getLocation();
+      String encodedRegionName = location.getRegion().getEncodedName();
+      scanMetrics.initScanMetricsRegionInfo(encodedRegionName, location.getServerName());
+    }
+  }
 }

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

@@ -367,19 +367,19 @@ static void incRPCCallsMetrics(ScanMetrics scanMetrics, boolean isRegionServerRe
     if (scanMetrics == null) {
       return;
     }
-    scanMetrics.countOfRPCcalls.incrementAndGet();
+    scanMetrics.addToCounter(ScanMetrics.RPC_CALLS_METRIC_NAME, 1);
     if (isRegionServerRemote) {
-      scanMetrics.countOfRemoteRPCcalls.incrementAndGet();
+      scanMetrics.addToCounter(ScanMetrics.REMOTE_RPC_CALLS_METRIC_NAME, 1);
     }
   }
 
   static void incRPCRetriesMetrics(ScanMetrics scanMetrics, boolean isRegionServerRemote) {
     if (scanMetrics == null) {
       return;
     }
-    scanMetrics.countOfRPCRetries.incrementAndGet();
+    scanMetrics.addToCounter(ScanMetrics.RPC_RETRIES_METRIC_NAME, 1);
     if (isRegionServerRemote) {
-      scanMetrics.countOfRemoteRPCRetries.incrementAndGet();
+      scanMetrics.addToCounter(ScanMetrics.REMOTE_RPC_RETRIES_METRIC_NAME, 1);
     }
   }
 
@@ -394,9 +394,9 @@ static void updateResultsMetrics(ScanMetrics scanMetrics, Result[] rrs,
         resultSize += PrivateCellUtil.estimatedSerializedSizeOf(cell);
       }
     }
-    scanMetrics.countOfBytesInResults.addAndGet(resultSize);
+    scanMetrics.addToCounter(ScanMetrics.BYTES_IN_RESULTS_METRIC_NAME, resultSize);
     if (isRegionServerRemote) {
-      scanMetrics.countOfBytesInRemoteResults.addAndGet(resultSize);
+      scanMetrics.addToCounter(ScanMetrics.BYTES_IN_REMOTE_RESULTS_METRIC_NAME, resultSize);
     }
   }
 
@@ -416,7 +416,7 @@ static void incRegionCountMetrics(ScanMetrics scanMetrics) {
     if (scanMetrics == null) {
       return;
     }
-    scanMetrics.countOfRegions.incrementAndGet();
+    scanMetrics.addToCounter(ScanMetrics.REGIONS_SCANNED_METRIC_NAME, 1);
   }
 
   /**

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

@@ -240,6 +240,12 @@ public Scan setScanMetricsEnabled(final boolean enabled) {
       "ImmutableScan does not allow access to setScanMetricsEnabled");
   }
 
+  @Override
+  public Scan setEnableScanMetricsByRegion(final boolean enable) {
+    throw new UnsupportedOperationException(
+      "ImmutableScan does not allow access to setEnableScanMetricsByRegion");
+  }
+
   @Override
   @Deprecated
   public Scan setAsyncPrefetch(boolean asyncPrefetch) {
@@ -420,6 +426,11 @@ public boolean isScanMetricsEnabled() {
     return this.delegateScan.isScanMetricsEnabled();
   }
 
+  @Override
+  public boolean isScanMetricsByRegionEnabled() {
+    return this.delegateScan.isScanMetricsByRegionEnabled();
+  }
+
   @Override
   public Boolean isAsyncPrefetch() {
     return this.delegateScan.isAsyncPrefetch();

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

@@ -130,6 +130,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}
@@ -1102,11 +1104,15 @@ public Scan setPriority(int priority) {
   }
 
   /**
-   * Enable collection of {@link ScanMetrics}. For advanced users.
+   * Enable collection of {@link ScanMetrics}. For advanced users. While disabling scan metrics,
+   * will also disable region level scan metrics.
    * @param enabled Set to true to enable accumulating scan metrics
    */
   public Scan setScanMetricsEnabled(final boolean enabled) {
     setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_ENABLE, Bytes.toBytes(Boolean.valueOf(enabled)));
+    if (!enabled) {
+      setEnableScanMetricsByRegion(false);
+    }
     return this;
   }
 
@@ -1239,4 +1245,22 @@ public boolean isNeedCursorResult() {
   public static Scan createScanFromCursor(Cursor cursor) {
     return new Scan().withStartRow(cursor.getRow());
   }
+
+  /**
+   * Enables region level scan metrics. If scan metrics are disabled then first enables scan metrics
+   * followed by region level scan metrics.
+   * @param enable Set to true to enable region level scan metrics.
+   */
+  public Scan setEnableScanMetricsByRegion(final boolean enable) {
+    if (enable) {
+      setScanMetricsEnabled(true);
+    }
+    setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_BY_REGION_ENABLE, Bytes.toBytes(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/ScannerCallable.java

@@ -228,7 +228,7 @@ private ScanResponse next() throws IOException {
         // when what we need is to open scanner against new location.
         // Attach NSRE to signal client that it needs to re-setup scanner.
         if (this.scanMetrics != null) {
-          this.scanMetrics.countOfNSRE.incrementAndGet();
+          this.scanMetrics.addToCounter(ScanMetrics.NOT_SERVING_REGION_EXCEPTION_METRIC_NAME, 1);
         }
         throw new DoNotRetryIOException("Resetting the scanner -- see exception cause", ioe);
       } else if (ioe instanceof RegionServerStoppedException) {

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

@@ -31,6 +31,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.RegionLocations;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.ScannerCallable.MoreResults;
@@ -498,4 +499,8 @@ public String getExceptionMessageAdditionalDetail() {
   public long sleep(long pause, int tries) {
     return currentScannerCallable.sleep(pause, tries);
   }
+
+  public HRegionLocation getLocation() {
+    return currentScannerCallable.getLocation();
+  }
 }

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

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client.metrics;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * Captures region level scan metrics as a map of metric name ({@link String}) -> Value
+ * ({@link AtomicLong}). <br/>
+ * <br/>
+ * One instance stores scan metrics for a single region only.
+ */
+@InterfaceAudience.Private
+public class RegionScanMetricsData {
+  private final Map<String, AtomicLong> counters = new HashMap<>();
+  private ScanMetricsRegionInfo scanMetricsRegionInfo =
+    ScanMetricsRegionInfo.EMPTY_SCAN_METRICS_REGION_INFO;
+
+  AtomicLong createCounter(String counterName) {
+    return ScanMetricsUtil.createCounter(counters, counterName);
+  }
+
+  void setCounter(String counterName, long value) {
+    ScanMetricsUtil.setCounter(counters, counterName, value);
+  }
+
+  void addToCounter(String counterName, long delta) {
+    ScanMetricsUtil.addToCounter(counters, counterName, delta);
+  }
+
+  Map<String, Long> collectMetrics(boolean reset) {
+    return ScanMetricsUtil.collectMetrics(counters, reset);
+  }
+
+  @Override
+  public String toString() {
+    return getClass().getSimpleName() + "[" + scanMetricsRegionInfo + "," + "Counters=" + counters
+      + "]";
+  }
+
+  /**
+   * Populate encoded region name and server name details if not already populated. If details are
+   * already populated and a re-attempt is done then {@link UnsupportedOperationException} is
+   * thrown.
+   */
+  void initScanMetricsRegionInfo(String encodedRegionName, ServerName serverName) {
+    // Check by reference
+    if (scanMetricsRegionInfo == ScanMetricsRegionInfo.EMPTY_SCAN_METRICS_REGION_INFO) {
+      scanMetricsRegionInfo = new ScanMetricsRegionInfo(encodedRegionName, serverName);
+    } else {
+      throw new UnsupportedOperationException(
+        "ScanMetricsRegionInfo has already been initialized to " + scanMetricsRegionInfo
+          + " and cannot be re-initialized to region: " + encodedRegionName + " and server: "
+          + serverName);
+    }
+  }
+
+  ScanMetricsRegionInfo getScanMetricsRegionInfo() {
+    return scanMetricsRegionInfo;
+  }
+}