HBASE-27466: Making metrics instance containing one or more connections.

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

@@ -34,6 +34,16 @@
 @InterfaceAudience.Public
 public interface AsyncConnection extends Closeable {
 
+  /**
+   * Returns the clusterId of this connection.
+   */
+  String getClusterId2();
+
+  /**
+   * Returns this connection identity.
+   */
+  String getIdentity();
+
   /**
    * Returns the {@link org.apache.hadoop.conf.Configuration} object used by this instance.
    * <p>

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

@@ -118,6 +118,8 @@ public class AsyncConnectionImpl implements AsyncConnection {
 
   private final AtomicBoolean closed = new AtomicBoolean(false);
 
+  private final String clusterId;
+  private final String identity;
   private final Optional<MetricsConnection> metrics;
 
   private final ClusterStatusListener clusterStatusListener;
@@ -126,6 +128,13 @@ public class AsyncConnectionImpl implements AsyncConnection {
 
   public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, String clusterId,
     SocketAddress localAddress, User user) {
+    this(conf, registry, clusterId, localAddress, user, null);
+  }
+
+  public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, String clusterId,
+    SocketAddress localAddress, User user, String identity) {
+    this.clusterId = clusterId;
+    this.identity = identity;
     this.conf = conf;
     this.user = user;
 
@@ -135,8 +144,8 @@ public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, Stri
     this.connConf = new AsyncConnectionConfiguration(conf);
     this.registry = registry;
     if (conf.getBoolean(CLIENT_SIDE_METRICS_ENABLED_KEY, false)) {
-      String scope = MetricsConnection.getScope(conf, clusterId, this);
-      this.metrics = Optional.of(new MetricsConnection(scope, () -> null, () -> null));
+      this.metrics =
+        Optional.of(MetricsConnection.getMetricsConnection(this, () -> null, () -> null));
     } else {
       this.metrics = Optional.empty();
     }
@@ -205,6 +214,16 @@ public ConnectionRegistry getConnectionRegistry() {
     return registry;
   }
 
+  @Override
+  public String getClusterId2() {
+    return clusterId;
+  }
+
+  @Override
+  public String getIdentity() {
+    return identity;
+  }
+
   @Override
   public Configuration getConfiguration() {
     return conf;
@@ -235,7 +254,9 @@ public void close() {
           choreService = null;
         }
       }
-      metrics.ifPresent(MetricsConnection::shutdown);
+      if (metrics.isPresent()) {
+        MetricsConnection.deleteMetricsConnection(this);
+      }
       ConnectionOverAsyncConnection c = this.conn;
       if (c != null) {
         c.closePool();

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

@@ -216,6 +216,37 @@ public static Connection createConnection(Configuration conf, User user) throws
    */
   public static Connection createConnection(Configuration conf, ExecutorService pool,
     final User user) throws IOException {
+    return createConnection(conf, pool, user, null);
+  }
+
+  /**
+   * Create a new Connection instance using the passed <code>conf</code> instance. Connection
+   * encapsulates all housekeeping for a connection to the cluster. All tables and interfaces
+   * created from returned connection share zookeeper connection, meta cache, and connections to
+   * region servers and masters. <br>
+   * The caller is responsible for calling {@link Connection#close()} on the returned connection
+   * instance. Typical usage:
+   *
+   * <pre>
+   * Connection connection = ConnectionFactory.createConnection(conf);
+   * Table table = connection.getTable(TableName.valueOf("table1"));
+   * try {
+   *   table.get(...);
+   *   ...
+   * } finally {
+   *   table.close();
+   *   connection.close();
+   * }
+   * </pre>
+   *
+   * @param conf     configuration
+   * @param user     the user the connection is for
+   * @param pool     the thread pool to use for batch operations
+   * @param identity the identity of the metrics for this connection.
+   * @return Connection object for <code>conf</code>
+   */
+  public static Connection createConnection(Configuration conf, ExecutorService pool,
+    final User user, String identity) throws IOException {
     Class<?> clazz = conf.getClass(ConnectionUtils.HBASE_CLIENT_CONNECTION_IMPL,
       ConnectionOverAsyncConnection.class, Connection.class);
     if (clazz != ConnectionOverAsyncConnection.class) {
@@ -225,7 +256,7 @@ public static Connection createConnection(Configuration conf, ExecutorService po
           clazz.getDeclaredConstructor(Configuration.class, ExecutorService.class, User.class);
         constructor.setAccessible(true);
         return user.runAs((PrivilegedExceptionAction<
-          Connection>) () -> (Connection) constructor.newInstance(conf, pool, user));
+          Connection>) () -> (Connection) constructor.newInstance(conf, pool, user, identity));
       } catch (Exception e) {
         throw new IOException(e);
       }

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

@@ -26,6 +26,10 @@
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.RatioGauge;
 import com.codahale.metrics.Timer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
@@ -54,6 +58,38 @@
 @InterfaceAudience.Private
 public class MetricsConnection implements StatisticTrackable {
 
+  static final Map<String, MetricsConnection> METRICS_INSTANCES =
+    new HashMap<String, MetricsConnection>();
+
+  static MetricsConnection getMetricsConnection(final AsyncConnection conn,
+    Supplier<ThreadPoolExecutor> batchPool, Supplier<ThreadPoolExecutor> metaPool) {
+    String scope = getScope(conn);
+    MetricsConnection metrics;
+    synchronized (METRICS_INSTANCES) {
+      metrics = METRICS_INSTANCES.get(scope);
+      if (metrics == null) {
+        metrics = new MetricsConnection(scope, batchPool, metaPool);
+        METRICS_INSTANCES.put(scope, metrics);
+      } else {
+        metrics.addThreadPools(batchPool, metaPool);
+      }
+      metrics.incrConnectionCount();
+    }
+    return metrics;
+  }
+
+  static void deleteMetricsConnection(final AsyncConnection conn) {
+    String scope = getScope(conn);
+    synchronized (METRICS_INSTANCES) {
+      MetricsConnection metrics = METRICS_INSTANCES.get(scope);
+      metrics.decrConnectionCount();
+      if (metrics.getConnectionCount() == 0) {
+        METRICS_INSTANCES.remove(scope);
+        metrics.shutdown();
+      }
+    }
+  }
+
   /** Set this key to {@code true} to enable metrics collection of client requests. */
   public static final String CLIENT_SIDE_METRICS_ENABLED_KEY = "hbase.client.metrics.enable";
 
@@ -78,9 +114,15 @@ public class MetricsConnection implements StatisticTrackable {
    * @param connectionObj either a Connection or AsyncConnectionImpl, the instance creating this
    *                      MetricsConnection.
    */
-  static String getScope(Configuration conf, String clusterId, Object connectionObj) {
-    return conf.get(METRICS_SCOPE_KEY,
-      clusterId + "@" + Integer.toHexString(connectionObj.hashCode()));
+  private static String getScope(final AsyncConnection conn) {
+    String identity = conn.getIdentity();
+    if (identity != null) {
+      return identity;
+    }
+    Configuration conf = conn.getConfiguration();
+    String clusterId = conn.getClusterId2();
+    int connHashCode = conn.hashCode();
+    return conf.get(METRICS_SCOPE_KEY, clusterId + "@" + Integer.toHexString(connHashCode));
   }
 
   private static final String CNT_BASE = "rpcCount_";
@@ -295,8 +337,13 @@ public Counter newMetric(Class<?> clazz, String name, String scope) {
     }
   };
 
+  // List of thread pool per connection of the metrics.
+  private List batchPools = new ArrayList<Supplier>();
+  private List metaPools = new ArrayList<Supplier>();
+
   // static metrics
 
+  protected final Counter connectionCount;
   protected final Counter metaCacheHits;
   protected final Counter metaCacheMisses;
   protected final CallTracker getTracker;
@@ -334,27 +381,48 @@ public Counter newMetric(Class<?> clazz, String name, String scope) {
   MetricsConnection(String scope, Supplier<ThreadPoolExecutor> batchPool,
     Supplier<ThreadPoolExecutor> metaPool) {
     this.scope = scope;
+    this.batchPools.add(batchPool);
+    this.metaPools.add(metaPool);
     this.registry = new MetricRegistry();
     this.registry.register(getExecutorPoolName(), new RatioGauge() {
       @Override
       protected Ratio getRatio() {
-        ThreadPoolExecutor pool = batchPool.get();
-        if (pool == null) {
-          return Ratio.of(0, 0);
+        int numerator = 0;
+        int denominator = 0;
+        for (int i = 0; i < batchPools.size(); i++) {
+          ThreadPoolExecutor pool = (ThreadPoolExecutor) ((Supplier) batchPools.get(i)).get();
+          if (pool != null) {
+            int activeCount = pool.getActiveCount();
+            int maxPoolSize = pool.getMaximumPoolSize();
+            if (numerator == 0 || (numerator * maxPoolSize) < (activeCount * denominator)) {
+              numerator = activeCount;
+              denominator = maxPoolSize;
+            }
+          }
         }
-        return Ratio.of(pool.getActiveCount(), pool.getMaximumPoolSize());
+        return Ratio.of(numerator, denominator);
       }
     });
     this.registry.register(getMetaPoolName(), new RatioGauge() {
       @Override
       protected Ratio getRatio() {
-        ThreadPoolExecutor pool = metaPool.get();
-        if (pool == null) {
-          return Ratio.of(0, 0);
+        int numerator = 0;
+        int denominator = 0;
+        for (int i = 0; i < metaPools.size(); i++) {
+          ThreadPoolExecutor pool = (ThreadPoolExecutor) ((Supplier) metaPools.get(i)).get();
+          if (pool != null) {
+            int activeCount = pool.getActiveCount();
+            int maxPoolSize = pool.getMaximumPoolSize();
+            if (numerator == 0 || (numerator * maxPoolSize) < (activeCount * denominator)) {
+              numerator = activeCount;
+              denominator = maxPoolSize;
+            }
+          }
         }
-        return Ratio.of(pool.getActiveCount(), pool.getMaximumPoolSize());
+        return Ratio.of(numerator, denominator);
       }
     });
+    this.connectionCount = registry.counter(name(this.getClass(), "connectionCount", scope));
     this.metaCacheHits = registry.counter(name(this.getClass(), "metaCacheHits", scope));
     this.metaCacheMisses = registry.counter(name(this.getClass(), "metaCacheMisses", scope));
     this.metaCacheNumClearServer =
@@ -457,6 +525,27 @@ public void incrementServerOverloadedBackoffTime(long time, TimeUnit timeUnit) {
     overloadedBackoffTimer.update(time, timeUnit);
   }
 
+  /** Return the connection count of the metrics within a scope */
+  private long getConnectionCount() {
+    return connectionCount.getCount();
+  }
+
+  /** Increment the connection count of the metrics within a scope */
+  private void incrConnectionCount() {
+    connectionCount.inc();
+  }
+
+  /** Decrement the connection count of the metrics within a scope */
+  private void decrConnectionCount() {
+    connectionCount.dec();
+  }
+
+  /** Add thread pools of additional connections to the metrics */
+  private void addThreadPools(Supplier batchPool, Supplier metaPool) {
+    batchPools.add(batchPool);
+    metaPools.add(metaPool);
+  }
+
   /**
    * Get a metric for {@code key} from {@code map}, or create it with {@code factory}.
    */

hbase-server/src/main/java/org/apache/hadoop/hbase/client/SharedAsyncConnection.java

@@ -52,6 +52,16 @@ public Configuration getConfiguration() {
     return conn.getConfiguration();
   }
 
+  @Override
+  public String getClusterId2() {
+    return conn.getClusterId2();
+  }
+
+  @Override
+  public String getIdentity() {
+    return conn.getIdentity();
+  }
+
   @Override
   public AsyncTableRegionLocator getRegionLocator(TableName tableName) {
     return conn.getRegionLocator(tableName);

hbase-server/src/test/java/org/apache/hadoop/hbase/client/DummyAsyncClusterConnection.java

@@ -44,6 +44,16 @@ public Configuration getConfiguration() {
     return null;
   }
 
+  @Override
+  public String getClusterId2() {
+    return null;
+  }
+
+  @Override
+  public String getIdentity() {
+    return null;
+  }
+
   @Override
   public AsyncTableRegionLocator getRegionLocator(TableName tableName) {
     return null;