HBASE-15519 Add per-user metrics with lossy counting

Introducing property hbase.regionserver.user.metrics.enabled(Default:true)
to disable user metrics in case it accounts for any performance issues

Close #661

Signed-off-by: Josh Elser <elserj@apache.org>

Author: ankitsinghal

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

@@ -44,6 +44,18 @@ public interface MetricsRegionServerSourceFactory {
    */
   MetricsRegionSource createRegion(MetricsRegionWrapper wrapper);
 
+  /**
+   * Create a MetricsUserSource from a user
+   * @return A metrics user source
+   */
+  MetricsUserSource createUser(String shortUserName);
+
+  /**
+   * Return the singleton instance for MetricsUserAggregateSource
+   * @return A metrics user aggregate source
+   */
+  MetricsUserAggregateSource getUserAggregate();
+
   /**
    * Create a MetricsTableSource from a MetricsTableWrapper.
    *

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

@@ -0,0 +1,62 @@
+/**
+ * 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.metrics.BaseSource;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+* This interface will be implemented by a MetricsSource that will export metrics from
+* multiple users into the hadoop metrics system.
+*/
+@InterfaceAudience.Private
+public interface MetricsUserAggregateSource extends BaseSource {
+
+  /**
+   * The name of the metrics
+   */
+  static final String METRICS_NAME = "Users";
+
+  /**
+   * The name of the metrics context that metrics will be under.
+   */
+  static final String METRICS_CONTEXT = "regionserver";
+
+  /**
+   * Description
+   */
+  static final String METRICS_DESCRIPTION = "Metrics about users connected to the regionserver";
+
+  /**
+   * The name of the metrics context that metrics will be under in jmx
+   */
+  static final String METRICS_JMX_CONTEXT = "RegionServer,sub=" + METRICS_NAME;
+
+  static final String NUM_USERS = "numUsers";
+  static final String NUMBER_OF_USERS_DESC = "Number of users in the metrics system";
+
+  /**
+   * Returns a MetricsUserSource if already exists, or creates and registers one for this user
+   * @param user the user name
+   * @return a metrics user source
+   */
+  MetricsUserSource getOrCreateMetricsUser(String user);
+
+  void deregister(MetricsUserSource toRemove);
+}

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

@@ -0,0 +1,45 @@
+/**
+ * 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.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public interface MetricsUserSource extends Comparable<MetricsUserSource> {
+
+  String getUser();
+
+  void register();
+
+  void deregister();
+
+  void updatePut(long t);
+
+  void updateDelete(long t);
+
+  void updateGet(long t);
+
+  void updateIncrement(long t);
+
+  void updateAppend(long t);
+
+  void updateReplay(long t);
+
+  void updateScanTime(long t);
+}

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

@@ -30,17 +30,27 @@ public class MetricsRegionServerSourceFactoryImpl implements MetricsRegionServer
   public static enum FactoryStorage {
     INSTANCE;
     private Object aggLock = new Object();
-    private MetricsRegionAggregateSourceImpl aggImpl;
+    private MetricsRegionAggregateSourceImpl regionAggImpl;
+    private MetricsUserAggregateSourceImpl userAggImpl;
     private MetricsTableAggregateSourceImpl tblAggImpl;
     private MetricsHeapMemoryManagerSourceImpl heapMemMngImpl;
   }
 
-  private synchronized MetricsRegionAggregateSourceImpl getAggregate() {
+  private synchronized MetricsRegionAggregateSourceImpl getRegionAggregate() {
     synchronized (FactoryStorage.INSTANCE.aggLock) {
-      if (FactoryStorage.INSTANCE.aggImpl == null) {
-        FactoryStorage.INSTANCE.aggImpl = new MetricsRegionAggregateSourceImpl();
+      if (FactoryStorage.INSTANCE.regionAggImpl == null) {
+        FactoryStorage.INSTANCE.regionAggImpl = new MetricsRegionAggregateSourceImpl();
       }
-      return FactoryStorage.INSTANCE.aggImpl;
+      return FactoryStorage.INSTANCE.regionAggImpl;
+    }
+  }
+
+  public synchronized MetricsUserAggregateSourceImpl getUserAggregate() {
+    synchronized (FactoryStorage.INSTANCE.aggLock) {
+      if (FactoryStorage.INSTANCE.userAggImpl == null) {
+        FactoryStorage.INSTANCE.userAggImpl = new MetricsUserAggregateSourceImpl();
+      }
+      return FactoryStorage.INSTANCE.userAggImpl;
     }
   }
 
@@ -72,7 +82,7 @@ public synchronized MetricsRegionServerSource createServer(
 
   @Override
   public MetricsRegionSource createRegion(MetricsRegionWrapper wrapper) {
-    return new MetricsRegionSourceImpl(wrapper, getAggregate());
+    return new MetricsRegionSourceImpl(wrapper, getRegionAggregate());
   }
 
   @Override
@@ -83,4 +93,10 @@ public MetricsTableSource createTable(String table, MetricsTableWrapperAggregate
   public MetricsIOSource createIO(MetricsIOWrapper wrapper) {
     return new MetricsIOSourceImpl(wrapper);
   }
+
+  @Override
+  public org.apache.hadoop.hbase.regionserver.MetricsUserSource createUser(String shortUserName) {
+    return new org.apache.hadoop.hbase.regionserver.MetricsUserSourceImpl(shortUserName,
+        getUserAggregate());
+  }
 }

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

@@ -0,0 +1,112 @@
+/**
+ * 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.concurrent.ConcurrentHashMap;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.MetricsCollector;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.metrics2.lib.Interns;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
+
+@InterfaceAudience.Private
+public class MetricsUserAggregateSourceImpl extends BaseSourceImpl
+  implements MetricsUserAggregateSource {
+
+  private static final Logger LOG = LoggerFactory.getLogger(MetricsUserAggregateSourceImpl.class);
+
+  private final ConcurrentHashMap<String, MetricsUserSource> userSources =
+      new ConcurrentHashMap<String, MetricsUserSource>();
+
+  public MetricsUserAggregateSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsUserAggregateSourceImpl(String metricsName,
+      String metricsDescription,
+      String metricsContext,
+      String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  @Override
+  public MetricsUserSource getOrCreateMetricsUser(String user) {
+    MetricsUserSource source = userSources.get(user);
+    if (source != null) {
+      return source;
+    }
+    source = new MetricsUserSourceImpl(user, this);
+    MetricsUserSource prev = userSources.putIfAbsent(user, source);
+
+    if (prev != null) {
+      return prev;
+    } else {
+      // register the new metrics now
+      register(source);
+    }
+    return source;
+  }
+
+  public void register(MetricsUserSource source) {
+    synchronized (this) {
+      source.register();
+    }
+  }
+
+  @Override
+  public void deregister(MetricsUserSource toRemove) {
+    try {
+      synchronized (this) {
+        MetricsUserSource source = userSources.remove(toRemove.getUser());
+        if (source != null) {
+          source.deregister();
+        }
+      }
+    } catch (Exception e) {
+      // Ignored. If this errors out it means that someone is double
+      // closing the user source and the user metrics is already nulled out.
+      LOG.info("Error trying to remove " + toRemove + " from " + getClass().getSimpleName(), e);
+    }
+  }
+
+  @VisibleForTesting
+  public ConcurrentHashMap<String, MetricsUserSource> getUserSources() {
+    return userSources;
+  }
+
+  @Override
+  public void getMetrics(MetricsCollector collector, boolean all) {
+    MetricsRecordBuilder mrb = collector.addRecord(metricsName);
+
+    if (userSources != null) {
+      for (MetricsUserSource userMetricSource : userSources.values()) {
+        if (userMetricSource instanceof MetricsUserSourceImpl) {
+          ((MetricsUserSourceImpl) userMetricSource).snapshot(mrb, all);
+        }
+      }
+      mrb.addGauge(Interns.info(NUM_USERS, NUMBER_OF_USERS_DESC), userSources.size());
+      metricsRegistry.snapshot(mrb, all);
+    }
+  }
+}