HBASE-29233: Capture scan metrics at region level (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/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/metrics/RegionScanMetricsData.java

@@ -0,0 +1,77 @@
+/*
+ * 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");
+    }
+  }
+
+  ScanMetricsRegionInfo getScanMetricsRegionInfo() {
+    return scanMetricsRegionInfo;
+  }
+}

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

@@ -96,8 +96,22 @@ public class ScanMetrics extends ServerSideScanMetrics {
   public final AtomicLong countOfRemoteRPCRetries = createCounter(REMOTE_RPC_RETRIES_METRIC_NAME);
 
   /**
-   * constructor
+   * Constructor
    */
   public ScanMetrics() {
   }
+
+  @Override
+  public void moveToNextRegion() {
+    super.moveToNextRegion();
+    currentRegionScanMetricsData.createCounter(RPC_CALLS_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(REMOTE_RPC_CALLS_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(MILLIS_BETWEEN_NEXTS_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(NOT_SERVING_REGION_EXCEPTION_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(BYTES_IN_RESULTS_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(BYTES_IN_REMOTE_RESULTS_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(REGIONS_SCANNED_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(RPC_RETRIES_METRIC_NAME);
+    currentRegionScanMetricsData.createCounter(REMOTE_RPC_RETRIES_METRIC_NAME);
+  }
 }

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

@@ -0,0 +1,81 @@
+/*
+ * 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 org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.yetus.audience.InterfaceStability;
+
+/**
+ * POJO for capturing region level details when region level scan metrics are enabled. <br>
+ * <br>
+ * Currently, encoded region name and server name (host name, ports and startcode) are captured as
+ * region details. <br>
+ * <br>
+ * Instance of this class serves as key in the Map returned by
+ * {@link ServerSideScanMetrics#collectMetricsByRegion()} or
+ * {@link ServerSideScanMetrics#collectMetricsByRegion(boolean)}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class ScanMetricsRegionInfo {
+  /**
+   * Users should only compare against this constant by reference and should not make any
+   * assumptions regarding content of the constant.
+   */
+  public static final ScanMetricsRegionInfo EMPTY_SCAN_METRICS_REGION_INFO =
+    new ScanMetricsRegionInfo(null, null);
+
+  private final String encodedRegionName;
+  private final ServerName serverName;
+
+  ScanMetricsRegionInfo(String encodedRegionName, ServerName serverName) {
+    this.encodedRegionName = encodedRegionName;
+    this.serverName = serverName;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (!(obj instanceof ScanMetricsRegionInfo other)) {
+      return false;
+    }
+    return new EqualsBuilder().append(encodedRegionName, other.encodedRegionName)
+      .append(serverName, other.serverName).isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37).append(encodedRegionName).append(serverName).toHashCode();
+  }
+
+  @Override
+  public String toString() {
+    return getClass().getSimpleName() + "[encodedRegionName=" + encodedRegionName + ",serverName="
+      + serverName + "]";
+  }
+
+  public String getEncodedRegionName() {
+    return encodedRegionName;
+  }
+
+  public ServerName getServerName() {
+    return serverName;
+  }
+}