Update the pool metrics implementation to use the pool metrics instead of using the inconvenient client metric queue capabilities that does not match correctly the actual pool usage when it becomes dynamic.

Author: vietj

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionFactory.java

@@ -24,6 +24,7 @@
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.SqlConnection;
@@ -54,7 +55,8 @@ protected Future<Connection> doConnectInternal(SqlConnectOptions options, Contex
     int pipeliningLimit = db2Options.getPipeliningLimit();
     NetClient netClient = netClient(options);
     return netClient.connect(server).flatMap(so -> {
-      ClientMetrics metrics = clientMetricsProvider != null ? clientMetricsProvider.metricsFor(options) : null;
+      VertxMetrics vertxMetrics = vertx.metricsSPI();
+      ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(db2Options.getSocketAddress(), "sql", db2Options.getMetricsName()) : null;
       DB2SocketConnection conn = new DB2SocketConnection((NetSocketInternal) so, metrics, db2Options, cachePreparedStatements,
         preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
       conn.init();

vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java

@@ -55,8 +55,8 @@ public Pool newPool(Vertx vertx, Supplier<? extends Future<? extends SqlConnectO
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
     boolean pipelinedPool = options instanceof Db2PoolOptions && ((Db2PoolOptions) options).isPipelined();
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, null, null, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
-    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, factory.metricsProvider(), null, null, closeFuture);
     pool.connectionProvider(factory::connect);
     pool.init();
     closeFuture.add(factory);

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionFactory.java

@@ -19,6 +19,7 @@
 import io.vertx.core.net.*;
 import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.mssqlclient.MSSQLConnectOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.SqlConnection;
@@ -72,7 +73,8 @@ private Future<Connection> connectOrRedirect(MSSQLConnectOptions options, Contex
   }
 
   private MSSQLSocketConnection createSocketConnection(NetSocket so, MSSQLConnectOptions options, ContextInternal context) {
-    ClientMetrics metrics = clientMetricsProvider != null ? clientMetricsProvider.metricsFor(options) : null;
+    VertxMetrics vertxMetrics = vertx.metricsSPI();
+    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(options.getSocketAddress(), "sql", options.getMetricsName()) : null;
     MSSQLSocketConnection conn = new MSSQLSocketConnection((NetSocketInternal) so, metrics, options, false, 0, sql -> true, 1, context);
     conn.init();
     return conn;

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java

@@ -57,8 +57,8 @@ public Pool newPool(Vertx vertx, Supplier<? extends Future<? extends SqlConnectO
   }
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
+    PoolImpl pool = new PoolImpl(vertx, this, false, options, null, null, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
-    PoolImpl pool = new PoolImpl(vertx, this, false, options, factory.metricsProvider(), null, null, closeFuture);
     pool.connectionProvider(context -> factory.connect(context, databases.get()));
     pool.init();
     closeFuture.add(factory);

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionFactory.java

@@ -23,6 +23,7 @@
 import io.vertx.core.net.TrustOptions;
 import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.mysqlclient.MySQLAuthenticationPlugin;
 import io.vertx.mysqlclient.MySQLConnectOptions;
 import io.vertx.mysqlclient.SslMode;
@@ -126,7 +127,8 @@ private Future<Connection> doConnect(MySQLConnectOptions options, SslMode sslMod
     MySQLAuthenticationPlugin authenticationPlugin = options.getAuthenticationPlugin();
     Future<NetSocket> fut = netClient(new NetClientOptions(options).setSsl(false)).connect(server);
     return fut.flatMap(so -> {
-      ClientMetrics metrics = clientMetricsProvider != null ? clientMetricsProvider.metricsFor(options) : null;
+      VertxMetrics vertxMetrics = vertx.metricsSPI();
+      ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(options.getSocketAddress(), "sql", options.getMetricsName()) : null;
       MySQLSocketConnection conn = new MySQLSocketConnection((NetSocketInternal) so, metrics, options, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
       conn.init();
       return Future.future(promise -> conn.sendStartupMessage(username, password, database, collation, serverRsaPublicKey, properties, sslMode, initialCapabilitiesFlags, charsetEncoding, authenticationPlugin, promise));

vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java

@@ -55,8 +55,8 @@ public Pool newPool(Vertx vertx, Supplier<? extends Future<? extends SqlConnectO
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
     boolean pipelinedPool = options instanceof MySQLPoolOptions && ((MySQLPoolOptions) options).isPipelined();
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, null, null, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
-    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, factory.metricsProvider(), null, null, closeFuture);
     pool.connectionProvider(context -> factory.connect(context, databases.get()));
     pool.init();
     closeFuture.add(factory);

vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionFactory.java

@@ -21,10 +21,6 @@
 import io.vertx.oracleclient.OracleConnectOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.SqlConnection;
-import io.vertx.sqlclient.impl.SingletonSupplier;
-import io.vertx.sqlclient.impl.metrics.ClientMetricsProvider;
-import io.vertx.sqlclient.impl.metrics.DynamicClientMetricsProvider;
-import io.vertx.sqlclient.impl.metrics.SingleServerClientMetricsProvider;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 import oracle.jdbc.OracleConnection;
 import oracle.jdbc.datasource.OracleDataSource;
@@ -40,39 +36,15 @@ public class OracleConnectionFactory implements ConnectionFactory {
 
   private final Supplier<? extends Future<? extends SqlConnectOptions>> options;
   private final Map<JsonObject, OracleDataSource> datasources;
-  private final ClientMetricsProvider clientMetricsProvider;
 
   public OracleConnectionFactory(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> options) {
-    VertxMetrics metrics = vertx.metricsSPI();
-    ClientMetricsProvider clientMetricsProvider;
-    if (metrics != null) {
-      if (options instanceof SingletonSupplier) {
-        SqlConnectOptions option = (SqlConnectOptions) ((SingletonSupplier) options).unwrap();
-        ClientMetrics<?, ?, ?, ?> clientMetrics = metrics.createClientMetrics(option.getSocketAddress(), "sql", option.getMetricsName());
-        clientMetricsProvider = new SingleServerClientMetricsProvider(clientMetrics);
-      } else {
-        clientMetricsProvider = new DynamicClientMetricsProvider(metrics);
-      }
-    } else {
-      clientMetricsProvider = null;
-    }
-    this.clientMetricsProvider = clientMetricsProvider;
     this.options = options;
     this.datasources = new HashMap<>();
   }
 
-  @Override
-  public ClientMetricsProvider metricsProvider() {
-    return clientMetricsProvider;
-  }
-
   @Override
   public void close(Promise<Void> promise) {
-    if (clientMetricsProvider != null) {
-      clientMetricsProvider.close(promise);
-    } else {
-      promise.complete();
-    }
+    promise.complete();
   }
 
   @Override
@@ -96,7 +68,8 @@ private OracleDataSource getDatasource(SqlConnectOptions options) {
   @Override
   public Future<SqlConnection> connect(Context context, SqlConnectOptions options) {
     OracleDataSource datasource = getDatasource(options);
-    ClientMetrics metrics = clientMetricsProvider != null ? clientMetricsProvider.metricsFor(options) : null;
+    VertxMetrics vertxMetrics = ((VertxInternal)context.owner()).metricsSPI();
+    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(options.getSocketAddress(), "sql", options.getMetricsName()) : null;
     ContextInternal ctx = (ContextInternal) context;
     return executeBlocking(context, () -> {
       OracleConnection orac = datasource.createConnectionBuilder().build();

vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java

@@ -52,8 +52,8 @@ public Pool newPool(Vertx vertx, Supplier<? extends Future<? extends SqlConnectO
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
     Function<Connection, Future<Void>> afterAcquire = conn -> ((OracleJdbcConnection) conn).afterAcquire();
     Function<Connection, Future<Void>> beforeRecycle = conn -> ((OracleJdbcConnection) conn).beforeRecycle();
+    PoolImpl pool = new PoolImpl(vertx, this,  false, options, afterAcquire, beforeRecycle, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
-    PoolImpl pool = new PoolImpl(vertx, this,  false, options, factory.metricsProvider(), afterAcquire, beforeRecycle, closeFuture);
     pool.connectionProvider(context -> factory.connect(context, databases.get()));
     pool.init();
     closeFuture.add(factory);

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionFactory.java

@@ -179,7 +179,9 @@ private PgSocketConnection newSocketConnection(ContextInternal context, NetSocke
     Predicate<String> preparedStatementCacheSqlFilter = options.getPreparedStatementCacheSqlFilter();
     int pipeliningLimit = options.getPipeliningLimit();
     boolean useLayer7Proxy = options.getUseLayer7Proxy();
-    ClientMetrics metrics = clientMetricsProvider != null ? clientMetricsProvider.metricsFor(options) : null;
-    return new PgSocketConnection(socket, metrics, options, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, useLayer7Proxy, context);
+    VertxMetrics vertxMetrics = vertx.metricsSPI();
+    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(options.getSocketAddress(), "sql", options.getMetricsName()) : null;
+    PgSocketConnection conn = new PgSocketConnection(socket, metrics, options, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, useLayer7Proxy, context);
+    return conn;
   }
 }

vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java

@@ -42,8 +42,8 @@ public Pool newPool(Vertx vertx, Supplier<? extends Future<? extends SqlConnectO
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
     boolean pipelinedPool = options instanceof PgPoolOptions && ((PgPoolOptions) options).isPipelined();
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, null, null, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
-    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, factory.metricsProvider(), null, null, closeFuture);
     pool.connectionProvider(context -> factory.connect(context));  // BEWARE!!!!
     pool.init();
     closeFuture.add(factory);