HBASE-22975 Add read and write QPS metrics at server level and table level

Signed-off-by Reid Chan <reidchan@apache.org>

Author: ZhaoBQ

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

@@ -0,0 +1,53 @@
+/*
+ * 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.regionserver;
+
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+/**
+ * Query Per Second for each table in a RegionServer.
+ */
+@InterfaceAudience.Private
+public interface MetricsTableQueryMeter {
+
+  /**
+   * Update table read QPS
+   * @param tableName The table the metric is for
+   * @param count Number of occurrences to record
+   */
+  void updateTableReadQueryMeter(TableName tableName, long count);
+
+  /**
+   * Update table read QPS
+   * @param tableName The table the metric is for
+   */
+  void updateTableReadQueryMeter(TableName tableName);
+
+  /**
+   * Update table write QPS
+   * @param tableName The table the metric is for
+   * @param count Number of occurrences to record
+   */
+  void updateTableWriteQueryMeter(TableName tableName, long count);
+
+  /**
+   * Update table write QPS
+   * @param tableName The table the metric is for
+   */
+  void updateTableWriteQueryMeter(TableName tableName);
+}

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

@@ -22,6 +22,8 @@
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
 import org.apache.hadoop.metrics2.MetricHistogram;
+import org.apache.hadoop.metrics2.MetricsCollector;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
 import org.apache.hadoop.metrics2.lib.DynamicMetricsRegistry;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -172,4 +174,15 @@ public void updateScanSize(String tableName, long scanSize) {
   public void updateScanTime(String tableName, long t) {
     getOrCreateTableHistogram(tableName).updateScanTime(t);
   }
+
+  @Override
+  public void getMetrics(MetricsCollector metricsCollector, boolean all) {
+    MetricsRecordBuilder mrb = metricsCollector.addRecord(metricsName);
+    // source is registered in supers constructor, sometimes called before the whole initialization.
+    metricsRegistry.snapshot(mrb, all);
+    if (metricsAdapter != null) {
+      // snapshot MetricRegistry as well
+      metricsAdapter.snapshotAllMetrics(registry, mrb);
+    }
+  }
 }

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

@@ -0,0 +1,107 @@
+/*
+ * 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.regionserver;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.metrics.Meter;
+import org.apache.hadoop.hbase.metrics.MetricRegistry;
+
+/**
+ * Implementation of {@link MetricsTableQueryMeter} to track query per second for each table in
+ * a RegionServer.
+ */
+@InterfaceAudience.Private
+public class MetricsTableQueryMeterImpl implements MetricsTableQueryMeter {
+  private final Map<TableName, TableMeters> metersByTable = new ConcurrentHashMap<>();
+  private final MetricRegistry metricRegistry;
+
+  private final static String TABLE_READ_QUERY_PER_SECOND = "tableReadQueryPerSecond";
+  private final static String TABLE_WRITE_QUERY_PER_SECOND = "tableWriteQueryPerSecond";
+
+  public MetricsTableQueryMeterImpl(MetricRegistry metricRegistry) {
+    this.metricRegistry = metricRegistry;
+  }
+
+  private static class TableMeters {
+    final Meter tableReadQueryMeter;
+    final Meter tableWriteQueryMeter;
+
+    TableMeters(MetricRegistry metricRegistry, TableName tableName) {
+      this.tableReadQueryMeter = metricRegistry.meter(qualifyMetricsName(tableName,
+        TABLE_READ_QUERY_PER_SECOND));
+      this.tableWriteQueryMeter =
+        metricRegistry.meter(qualifyMetricsName(tableName, TABLE_WRITE_QUERY_PER_SECOND));
+    }
+
+    public void updateTableReadQueryMeter(long count) {
+      tableReadQueryMeter.mark(count);
+    }
+
+    public void updateTableReadQueryMeter() {
+      tableReadQueryMeter.mark();
+    }
+
+    public void updateTableWriteQueryMeter(long count) {
+      tableWriteQueryMeter.mark(count);
+    }
+
+    public void updateTableWriteQueryMeter() {
+      tableWriteQueryMeter.mark();
+    }
+  }
+
+  private static String qualifyMetricsName(TableName tableName, String metric) {
+    StringBuilder sb = new StringBuilder();
+    sb.append("Namespace_").append(tableName.getNamespaceAsString());
+    sb.append("_table_").append(tableName.getQualifierAsString());
+    sb.append("_metric_").append(metric);
+    return sb.toString();
+  }
+
+  private TableMeters getOrCreateTableMeter(TableName tableName) {
+    TableMeters meters = metersByTable.get(tableName);
+    if (meters == null) {
+      meters = new TableMeters(metricRegistry, tableName);
+      metersByTable.put(tableName, meters);
+    }
+    return meters;
+  }
+
+  @Override
+  public void updateTableReadQueryMeter(TableName tableName, long count) {
+    getOrCreateTableMeter(tableName).updateTableReadQueryMeter(count);
+  }
+
+  @Override
+  public void updateTableReadQueryMeter(TableName tableName) {
+    getOrCreateTableMeter(tableName).updateTableReadQueryMeter();
+  }
+
+  @Override
+  public void updateTableWriteQueryMeter(TableName tableName, long count) {
+    getOrCreateTableMeter(tableName).updateTableWriteQueryMeter(count);
+  }
+
+  @Override
+  public void updateTableWriteQueryMeter(TableName tableName) {
+    getOrCreateTableMeter(tableName).updateTableWriteQueryMeter();
+  }
+}

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

@@ -3160,6 +3160,10 @@ OperationStatus[] batchMutate(BatchOperationInProgress<?> batchOp) throws IOExce
         }
       }
     } finally {
+      if (rsServices != null && rsServices.getMetrics() != null) {
+        rsServices.getMetrics().updateWriteQueryMeter(this.htableDescriptor.
+          getTableName(), batchOp.operations.length);
+      }
       closeRegionOperation(op);
     }
     return batchOp.retCodeDetails;
@@ -6258,6 +6262,9 @@ public boolean nextRaw(List<Cell> outResults, ScannerContext scannerContext)
       if (!outResults.isEmpty()) {
         readRequestsCount.increment();
       }
+      if (rsServices != null && rsServices.getMetrics() != null) {
+        rsServices.getMetrics().updateReadQueryMeter(getRegionInfo().getTable());
+      }
 
       // If the size limit was reached it means a partial Result is being returned. Returning a
       // partial Result means that we should not reset the filters; filters should only be reset in
@@ -7644,6 +7651,11 @@ public void processRowsWithLocks(RowProcessor<?,?> processor, long timeout,
           if (txid != 0) {
             syncOrDefer(txid, getEffectiveDurability(processor.useDurability()));
           }
+
+          if (rsServices != null && rsServices.getMetrics() != null) {
+            rsServices.getMetrics().updateWriteQueryMeter(this.htableDescriptor.
+              getTableName(), mutations.size());
+          }
           walSyncSuccessful = true;
           // 12. call postBatchMutate hook
           processor.postBatchMutate(this);
@@ -7949,6 +7961,10 @@ public Result append(Append mutate, long nonceGroup, long nonce) throws IOExcept
       if(txid != 0){
         syncOrDefer(txid, durability);
       }
+      if (rsServices != null && rsServices.getMetrics() != null) {
+        rsServices.getMetrics().updateWriteQueryMeter(this.htableDescriptor.
+          getTableName());
+      }
       doRollBackMemstore = false;
     } finally {
       if (rowLock != null) {
@@ -8078,6 +8094,10 @@ public Result increment(Increment mutation, long nonceGroup, long nonce)
       // will yield indeterminate results.
       return doIncrement(mutation, nonceGroup, nonce);
     } finally {
+      if (rsServices != null && rsServices.getMetrics() != null) {
+        rsServices.getMetrics().updateWriteQueryMeter(this.htableDescriptor.
+          getTableName());
+      }
       if (this.metricsRegion != null) this.metricsRegion.updateIncrement();
       closeRegionOperation(op);
     }

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

@@ -21,6 +21,7 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.hadoop.hbase.metrics.Meter;
 import org.apache.hadoop.hbase.metrics.MetricRegistries;
 import org.apache.hadoop.hbase.metrics.MetricRegistry;
 import org.apache.hadoop.hbase.metrics.Timer;
@@ -49,6 +50,8 @@ public class MetricsRegionServer {
 
   private MetricRegistry metricRegistry;
   private Timer bulkLoadTimer;
+  private Meter serverReadQueryMeter;
+  private Meter serverWriteQueryMeter;
 
   public MetricsRegionServer(MetricsRegionServerWrapper regionServerWrapper, Configuration conf) {
     this(regionServerWrapper,
@@ -62,6 +65,8 @@ public MetricsRegionServer(MetricsRegionServerWrapper regionServerWrapper, Confi
 
     // create and use metrics from the new hbase-metrics based registry.
     bulkLoadTimer = metricRegistry.timer("Bulkload");
+    serverReadQueryMeter = metricRegistry.meter("ServerReadQueryPerSecond");
+    serverWriteQueryMeter = metricRegistry.meter("ServerWriteQueryPerSecond");
   }
 
   MetricsRegionServer(MetricsRegionServerWrapper regionServerWrapper,
@@ -211,4 +216,33 @@ public void updateCompaction(boolean isMajor, long t, int inputFileCount, int ou
   public void updateBulkLoad(long millis) {
     this.bulkLoadTimer.updateMillis(millis);
   }
+
+
+  public void updateReadQueryMeter(TableName tn, long count) {
+    if (tableMetrics != null && tn != null) {
+      tableMetrics.updateTableReadQueryMeter(tn, count);
+    }
+    this.serverReadQueryMeter.mark(count);
+  }
+
+  public void updateReadQueryMeter(TableName tn) {
+    if (tableMetrics != null && tn != null) {
+      tableMetrics.updateTableReadQueryMeter(tn);
+    }
+    this.serverReadQueryMeter.mark();
+  }
+
+  public void updateWriteQueryMeter(TableName tn, long count) {
+    if (tableMetrics != null && tn != null) {
+      tableMetrics.updateTableWriteQueryMeter(tn, count);
+    }
+    this.serverWriteQueryMeter.mark(count);
+  }
+
+  public void updateWriteQueryMeter(TableName tn) {
+    if (tableMetrics != null && tn != null) {
+      tableMetrics.updateTableWriteQueryMeter(tn);
+    }
+    this.serverWriteQueryMeter.mark();
+  }
 }

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

@@ -19,6 +19,7 @@
 import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.metrics.MetricRegistries;
 
 /**
  * Captures operation metrics by table. Separates metrics collection for table metrics away from
@@ -28,9 +29,12 @@
 public class RegionServerTableMetrics {
 
   private final MetricsTableLatencies latencies;
+  private final MetricsTableQueryMeter queryMeter;
 
   public RegionServerTableMetrics() {
     latencies = CompatibilitySingletonFactory.getInstance(MetricsTableLatencies.class);
+    queryMeter = new MetricsTableQueryMeterImpl(MetricRegistries.global().
+      get(((MetricsTableLatenciesImpl) latencies).getMetricRegistryInfo()).get());
   }
 
   public void updatePut(TableName table, long time) {
@@ -68,4 +72,20 @@ public void updateScanTime(TableName table, long time) {
   public void updateScanSize(TableName table, long size) {
     latencies.updateScanSize(table.getNameAsString(), size);
   }
+
+  public void updateTableReadQueryMeter(TableName table, long count) {
+    queryMeter.updateTableReadQueryMeter(table, count);
+  }
+
+  public void updateTableReadQueryMeter(TableName table) {
+    queryMeter.updateTableReadQueryMeter(table);
+  }
+
+  public void updateTableWriteQueryMeter(TableName table, long count) {
+    queryMeter.updateTableWriteQueryMeter(table, count);
+  }
+
+  public void updateTableWriteQueryMeter(TableName table) {
+    queryMeter.updateTableWriteQueryMeter(table);
+  }
 }