HBASE-22407 add an option to use Hadoop metrics tags for table metrics (and fix some issues in metrics)

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

@@ -523,4 +523,6 @@ public interface MetricsRegionServerWrapper {
   long getTrailerHitCount();
 
   long getTotalRowActionRequestCount();
+
+  String getConfVar(String name, String defaultValue);
 }

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

@@ -44,15 +44,15 @@ public interface MetricsTableLatencies {
    */
   String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME;
 
-  String GET_TIME = "getTime";
+  String GET_TIME = "get";
   String SCAN_TIME = "scanTime";
   String SCAN_SIZE = "scanSize";
-  String PUT_TIME = "putTime";
-  String PUT_BATCH_TIME = "putBatchTime";
-  String DELETE_TIME = "deleteTime";
-  String DELETE_BATCH_TIME = "deleteBatchTime";
-  String INCREMENT_TIME = "incrementTime";
-  String APPEND_TIME = "appendTime";
+  String PUT_TIME = "put";
+  String PUT_BATCH_TIME = "putBatch";
+  String DELETE_TIME = "delete";
+  String DELETE_BATCH_TIME = "deleteBatch";
+  String INCREMENT_TIME = "increment";
+  String APPEND_TIME = "append";
 
   /**
    * Update the Put time histogram
@@ -125,4 +125,9 @@ public interface MetricsTableLatencies {
    * @param t time it took
    */
   void updateScanTime(String tableName, long t);
+
+  /**
+   * @return Whether this metrics writer creates scoped metrics (see DynamicMetricsRegistry).
+   */
+  boolean isScoped();
 }

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

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.hbase.regionserver;
 
+import org.apache.hadoop.conf.Configuration;
+
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**
@@ -107,6 +109,8 @@ public interface MetricsTableWrapperAggregate {
    */
   long getNumReferenceFiles(String table);
 
-
-
+  /**
+   * @return Get HBase configuration; allows reflection-based classes to have access to it.
+   */
+  Configuration getConf();
 }

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsAssignmentManagerSourceImpl.java

@@ -29,7 +29,7 @@
 public class MetricsAssignmentManagerSourceImpl
     extends BaseSourceImpl
     implements MetricsAssignmentManagerSource {
-
+  // TODO: it may make sense to add some table-scoped metrics for these.
   private MutableGaugeLong ritGauge;
   private MutableGaugeLong ritCountOverThresholdGauge;
   private MutableGaugeLong ritOldestAgeGauge;

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterFilesystemSourceImpl.java

@@ -26,7 +26,7 @@
 public class MetricsMasterFilesystemSourceImpl
     extends BaseSourceImpl
     implements MetricsMasterFileSystemSource {
-
+  // TODO: it may make sense to add some table-scoped metrics for these.
   private MetricHistogram splitSizeHisto;
   private MetricHistogram splitTimeHisto;
   private MetricHistogram metaSplitTimeHisto;

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSourceImpl.java

@@ -28,6 +28,7 @@
 @InterfaceAudience.Private
 public class ExceptionTrackingSourceImpl extends BaseSourceImpl
     implements ExceptionTrackingSource {
+  // TODO: it may make sense to add some table-scoped metrics for these.
   protected MutableFastCounter exceptions;
   protected MutableFastCounter exceptionsOOO;
   protected MutableFastCounter exceptionsBusy;

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

@@ -31,7 +31,7 @@
 @InterfaceAudience.Private
 public class MetricsHeapMemoryManagerSourceImpl extends BaseSourceImpl implements
     MetricsHeapMemoryManagerSource {
-
+  // TODO: it may make sense to add some table-scoped metrics for these.
   private final MetricHistogram blockedFlushHistogram;
   private final MetricHistogram unblockedFlushHistogram;
   private final MetricHistogram incMemStoreSizeHistogram;

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

@@ -77,7 +77,11 @@ public MetricsRegionSource createRegion(MetricsRegionWrapper wrapper) {
 
   @Override
   public MetricsTableSource createTable(String table, MetricsTableWrapperAggregate wrapper) {
-    return new MetricsTableSourceImpl(table, getTableAggregate(), wrapper);
+    if (MetricsTableAggregateSourceImpl.areTablesViaTags(wrapper)) {
+      return new MetricsTableSourceImplWithTags(table, getTableAggregate(), wrapper);
+    } else {
+      return new MetricsTableSourceImpl(table, getTableAggregate(), wrapper);
+    }
   }
 
   public MetricsIOSource createIO(MetricsIOWrapper wrapper) {

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

@@ -65,6 +65,10 @@ public class MetricsRegionSourceImpl implements MetricsRegionSource {
 
   private final int hashCode;
 
+  // TODO: add tagged region metrics option, similar to the tables one?
+  //       Note that this would require refactoring them together to avoid metrics duplication.
+  //       See also the ugly way server-level metrics handle duplication now - all of these
+  //       could be merged and handled nicely and consistently.
   public MetricsRegionSourceImpl(MetricsRegionWrapper regionWrapper,
                                  MetricsRegionAggregateSourceImpl aggregate) {
     this.regionWrapper = regionWrapper;

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

@@ -18,8 +18,11 @@
 
 package org.apache.hadoop.hbase.regionserver;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
 import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
 import org.apache.hadoop.hbase.metrics.Interns;
@@ -32,6 +35,8 @@
 @InterfaceAudience.Private
 public class MetricsTableAggregateSourceImpl extends BaseSourceImpl
         implements MetricsTableAggregateSource {
+  public static final String TABLE_METRICS_TAGS = "hbase.metrics.table.via.tags";
+  public static final boolean TABLE_METRICS_TAGS_DEFAULT = false;
 
   private static final Logger LOG = LoggerFactory.getLogger(MetricsTableAggregateSourceImpl.class);
   private ConcurrentHashMap<String, MetricsTableSource> tableSources = new ConcurrentHashMap<>();
@@ -47,8 +52,17 @@ public MetricsTableAggregateSourceImpl(String metricsName,
     super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
   }
 
-  private void register(MetricsTableSource source) {
-    source.registerMetrics();
+  public static boolean areTablesViaTags(MetricsTableWrapperAggregate wrapper) {
+    return areTablesViaTags(wrapper.getConf());
+  }
+
+  public static boolean areTablesViaTags(MetricsRegionServerWrapper wrapper) {
+    return Boolean.parseBoolean(wrapper.getConfVar(
+        TABLE_METRICS_TAGS, Boolean.toString(TABLE_METRICS_TAGS_DEFAULT)));
+  }
+
+  public static boolean areTablesViaTags(Configuration conf) {
+    return conf != null && conf.getBoolean(TABLE_METRICS_TAGS, TABLE_METRICS_TAGS_DEFAULT);
   }
 
   @Override
@@ -92,14 +106,33 @@ public MetricsTableSource getOrCreateTableSource(String table,
   @Override
   public void getMetrics(MetricsCollector collector, boolean all) {
     MetricsRecordBuilder mrb = collector.addRecord(metricsName);
+    Map<String, MetricsRecordBuilder> scopedMrbs = new HashMap<>();
+
+    // Note that there are two metrics models in one being merged here...
+    // 1) MetricsTableSource.snapshot writes a subset of metrics
+    // 2) DynamicMetricsRegistry.snapshot writes a DIFFERENT set of metrics via a different model.
+    // See MetricsTableSourceImpl of registerMetrics and snapshot for the example of two models.
 
     if (tableSources != null) {
       for (MetricsTableSource tableMetricSource : tableSources.values()) {
         if (tableMetricSource instanceof MetricsTableSourceImpl) {
-          ((MetricsTableSourceImpl) tableMetricSource).snapshot(mrb, all);
+          MetricsTableSourceImpl impl = (MetricsTableSourceImpl) tableMetricSource;
+          MetricsRecordBuilder scopedMrb = mrb;
+          String scope = impl.getScope();
+          if (scope != null) {
+            scopedMrb = collector.addRecord(metricsName + "." + scope);
+            if (scopedMrbs.put(scope, scopedMrb) != null) {
+              LOG.error("Found multiple collectors for " + scope);
+            }
+          }
+          impl.snapshot(scopedMrb, all);
         }
       }
       mrb.addGauge(Interns.info(NUM_TABLES, NUMBER_OF_TABLES_DESC), tableSources.size());
+
+      for (Map.Entry<String, MetricsRecordBuilder> e : scopedMrbs.entrySet()) {
+        metricsRegistry.snapshotScoped(e.getKey(), e.getValue(), all);
+      }
       metricsRegistry.snapshot(mrb, all);
     }
   }

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

@@ -16,23 +16,37 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
-import java.util.HashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
 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.MetricsTag;
 import org.apache.hadoop.metrics2.lib.DynamicMetricsRegistry;
+import org.apache.hadoop.metrics2.lib.MutableSizeHistogram;
+import org.apache.hadoop.metrics2.lib.MutableTimeHistogram;
 import org.apache.yetus.audience.InterfaceAudience;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
+import org.apache.hbase.thirdparty.com.google.common.collect.Maps;import org.slf4j.Logger;import org.slf4j.LoggerFactory;
 
 /**
  * Implementation of {@link MetricsTableLatencies} to track latencies for one table in a
  * RegionServer.
  */
 @InterfaceAudience.Private
 public class MetricsTableLatenciesImpl extends BaseSourceImpl implements MetricsTableLatencies {
+  private static final Logger LOG = LoggerFactory.getLogger(MetricsTableLatenciesImpl.class);
 
-  private final HashMap<TableName,TableHistograms> histogramsByTable = new HashMap<>();
+  private final ConcurrentMap<TableName, TableHistograms> histogramsByTable =
+    Maps.newConcurrentMap();
+  private Configuration conf;
+  private boolean useTags;
 
   @VisibleForTesting
   public static class TableHistograms {
@@ -46,18 +60,27 @@ public static class TableHistograms {
     final MetricHistogram scanTimeHisto;
     final MetricHistogram scanSizeHisto;
 
+    // TODO: this doesn't appear to remove metrics like the other impl
     TableHistograms(DynamicMetricsRegistry registry, TableName tn) {
-      getTimeHisto = registry.newTimeHistogram(qualifyMetricsName(tn, GET_TIME));
-      incrementTimeHisto = registry.newTimeHistogram(
-          qualifyMetricsName(tn, INCREMENT_TIME));
-      appendTimeHisto = registry.newTimeHistogram(qualifyMetricsName(tn, APPEND_TIME));
-      putTimeHisto = registry.newTimeHistogram(qualifyMetricsName(tn, PUT_TIME));
-      putBatchTimeHisto = registry.newTimeHistogram(qualifyMetricsName(tn, PUT_BATCH_TIME));
-      deleteTimeHisto = registry.newTimeHistogram(qualifyMetricsName(tn, DELETE_TIME));
-      deleteBatchTimeHisto = registry.newTimeHistogram(
-          qualifyMetricsName(tn, DELETE_BATCH_TIME));
-      scanTimeHisto = registry.newTimeHistogram(qualifyMetricsName(tn, SCAN_TIME));
-      scanSizeHisto = registry.newSizeHistogram(qualifyMetricsName(tn, SCAN_SIZE));
+      getTimeHisto = newTimeHistogram(registry, tn, GET_TIME);
+      incrementTimeHisto = newTimeHistogram(registry, tn, INCREMENT_TIME);
+      appendTimeHisto = newTimeHistogram(registry, tn, APPEND_TIME);
+      putTimeHisto = newTimeHistogram(registry, tn, PUT_TIME);
+      putBatchTimeHisto = newTimeHistogram(registry, tn, PUT_BATCH_TIME);
+      deleteTimeHisto = newTimeHistogram(registry, tn, DELETE_TIME);
+      deleteBatchTimeHisto = newTimeHistogram(registry, tn, DELETE_BATCH_TIME);
+      scanTimeHisto = newTimeHistogram(registry, tn, SCAN_TIME);
+      scanSizeHisto = newSizeHistogram(registry, tn, SCAN_SIZE);
+    }
+
+    protected MutableTimeHistogram newTimeHistogram(
+        DynamicMetricsRegistry registry, TableName tn, String name) {
+      return registry.newTimeHistogram(qualifyMetricsName(tn, name));
+    }
+
+    protected MutableSizeHistogram newSizeHistogram(
+        DynamicMetricsRegistry registry, TableName tn, String name) {
+      return registry.newSizeHistogram(qualifyMetricsName(tn, name));
     }
 
     public void updatePut(long time) {
@@ -97,6 +120,24 @@ public void updateScanTime(long t) {
     }
   }
 
+  private static class TableHistogramsWithTags extends TableHistograms {
+    TableHistogramsWithTags(DynamicMetricsRegistry registry, TableName tn) {
+      super(registry, tn);
+    }
+
+    @Override
+    protected MutableSizeHistogram newSizeHistogram(
+        DynamicMetricsRegistry registry, TableName tn, String name) {
+      return registry.newScopedSizeHistogram(tn.getNameAsString(), name, "");
+    }
+
+    @Override
+    protected MutableTimeHistogram newTimeHistogram(
+        DynamicMetricsRegistry registry, TableName tn, String name) {
+      return registry.newScopedTimeHistogram(tn.getNameAsString(), name, "");
+    }
+  }
+
   @VisibleForTesting
   public static String qualifyMetricsName(TableName tableName, String metric) {
     StringBuilder sb = new StringBuilder();
@@ -112,7 +153,8 @@ public TableHistograms getOrCreateTableHistogram(String tableName) {
     final TableName tn = TableName.valueOf(tableName);
     TableHistograms latency = histogramsByTable.get(tn);
     if (latency == null) {
-      latency = new TableHistograms(getMetricsRegistry(), tn);
+      DynamicMetricsRegistry reg = getMetricsRegistry();
+      latency = useTags ? new TableHistogramsWithTags(reg, tn) : new TableHistograms(reg, tn);
       histogramsByTable.put(tn, latency);
     }
     return latency;
@@ -171,4 +213,34 @@ public void updateScanSize(String tableName, long scanSize) {
   public void updateScanTime(String tableName, long t) {
     getOrCreateTableHistogram(tableName).updateScanTime(t);
   }
+
+  public void setConf(Configuration conf) {
+    if (this.conf != null) {
+      LOG.warn("The object was already initialized with {}", this.useTags);
+      return;
+    }
+    this.conf = conf;
+    this.useTags = MetricsTableAggregateSourceImpl.areTablesViaTags(conf);
+  }
+
+  @Override
+  public void getMetrics(MetricsCollector metricsCollector, boolean all) {
+    if (!useTags) {
+      super.getMetrics(metricsCollector, all);
+    } else {
+      for (TableName tn : histogramsByTable.keySet()) {
+        String scope = tn.getNameAsString();
+        String recName = metricsRegistry.info().name() + "." + scope;
+        MetricsRecordBuilder mrb = metricsCollector.addRecord(recName);
+        mrb.add(new MetricsTag(MetricsTableSourceImplWithTags.TABLE_TAG_INFO, scope));
+        metricsRegistry.snapshotScoped(scope, mrb, all);
+      }
+      // There are no metrics here not scoped to the table, so don't snapshot().
+    }
+  }
+
+  @Override
+  public boolean isScoped() {
+    return useTags;
+  }
 }