Metrics and Tracing are now handled related to the connection and reported by a dedicated reporter around the command scheduler when it can access a connection.

Author: vietj

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

@@ -24,12 +24,13 @@
 import io.vertx.core.net.NetClient;
 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;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.ConnectionFactoryBase;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 
 import java.util.Map;
 import java.util.function.Predicate;
@@ -43,6 +44,7 @@ public DB2ConnectionFactory(VertxInternal vertx, Supplier<DB2ConnectOptions> opt
 
   @Override
   protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventLoopContext context) {
+    DB2ConnectOptions db2Options = DB2ConnectOptions.wrap(options);
     SocketAddress server = options.getSocketAddress();
     boolean cachePreparedStatements = options.getCachePreparedStatements();
     int preparedStatementCacheSize = options.getPreparedStatementCacheMaxSize();
@@ -51,10 +53,12 @@ protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventL
     String password = options.getPassword();
     String database = options.getDatabase();
     Map<String, String> properties = options.getProperties();
-    int pipeliningLimit = ((DB2ConnectOptions) options).getPipeliningLimit();
+    int pipeliningLimit = db2Options.getPipeliningLimit();
     NetClient netClient = netClient(options);
     return netClient.connect(server).flatMap(so -> {
-      DB2SocketConnection conn = new DB2SocketConnection((NetSocketInternal) so, cachePreparedStatements,
+      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();
       return Future.future(p -> conn.sendStartupMessage(username, password, database, properties, p));
@@ -64,11 +68,10 @@ protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventL
   @Override
   public Future<SqlConnection> connect(Context context, SqlConnectOptions options) {
     ContextInternal contextInternal = (ContextInternal) context;
-    QueryTracer tracer = contextInternal.tracer() == null ? null : new QueryTracer(contextInternal.tracer(), options);
     Promise<SqlConnection> promise = contextInternal.promise();
     connect(asEventLoopContext(contextInternal), options)
       .map(conn -> {
-        DB2ConnectionImpl db2Connection = new DB2ConnectionImpl(contextInternal, this, conn, tracer, null);
+        DB2ConnectionImpl db2Connection = new DB2ConnectionImpl(contextInternal, this, conn);
         conn.init(db2Connection);
         return (SqlConnection)db2Connection;
       }).onComplete(promise);

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

@@ -20,14 +20,12 @@
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.ContextInternal;
-import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.db2client.DB2Connection;
 import io.vertx.db2client.impl.command.PingCommand;
 import io.vertx.db2client.spi.DB2Driver;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.SqlConnectionBase;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
 public class DB2ConnectionImpl extends SqlConnectionBase<DB2ConnectionImpl> implements DB2Connection {
@@ -44,8 +42,8 @@ public static Future<DB2Connection> connect(Vertx vertx, DB2ConnectOptions optio
     return (Future) client.connect(ctx);
   }
 
-  public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
-    super(context, factory, conn, DB2Driver.INSTANCE, tracer, metrics);
+  public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {
+    super(context, factory, conn, DB2Driver.INSTANCE);
   }
 
   @Override

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

@@ -22,12 +22,14 @@
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
-import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.EventLoopContext;
 import io.vertx.core.net.impl.NetSocketInternal;
+import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.db2client.impl.codec.DB2Codec;
 import io.vertx.db2client.impl.command.InitialHandshakeCommand;
 import io.vertx.db2client.impl.drda.ConnectionMetaData;
+import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.QueryResultHandler;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
@@ -40,17 +42,21 @@
 
 public class DB2SocketConnection extends SocketConnectionBase {
 
+  private final DB2ConnectOptions connectOptions;
   private DB2Codec codec;
   private Handler<Void> closeHandler;
   public final ConnectionMetaData connMetadata = new ConnectionMetaData();
 
   public DB2SocketConnection(NetSocketInternal socket,
+      ClientMetrics clientMetrics,
+      DB2ConnectOptions connectOptions,
       boolean cachePreparedStatements,
       int preparedStatementCacheSize,
       Predicate<String> preparedStatementCacheSqlFilter,
       int pipeliningLimit,
       EventLoopContext context) {
-    super(socket, cachePreparedStatements, preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
+    super(socket, clientMetrics, cachePreparedStatements, preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
+    this.connectOptions = connectOptions;
   }
 
   // TODO RETURN FUTURE ???
@@ -63,6 +69,11 @@ void sendStartupMessage(String username,
     schedule(context, cmd).onComplete(completionHandler);
   }
 
+  @Override
+  protected SqlConnectOptions connectOptions() {
+    return connectOptions;
+  }
+
   @Override
   public void init() {
     codec = new DB2Codec(this);

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

@@ -20,8 +20,6 @@
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.json.JsonObject;
-import io.vertx.core.spi.metrics.ClientMetrics;
-import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.db2client.impl.*;
 import io.vertx.sqlclient.Pool;
@@ -30,7 +28,6 @@
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolImpl;
 import io.vertx.sqlclient.impl.SqlConnectionInternal;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 import io.vertx.sqlclient.spi.Driver;
 
@@ -56,12 +53,9 @@ public Pool newPool(Vertx vertx, Supplier<? extends SqlConnectOptions> databases
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
     DB2ConnectOptions baseConnectOptions = DB2ConnectOptions.wrap(databases.get());
-    QueryTracer tracer = vertx.tracer() == null ? null : new QueryTracer(vertx.tracer(), baseConnectOptions);
-    VertxMetrics vertxMetrics = vertx.metricsSPI();
-    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(baseConnectOptions.getSocketAddress(), "sql", baseConnectOptions.getMetricsName()) : null;
     boolean pipelinedPool = options instanceof Db2PoolOptions && ((Db2PoolOptions) options).isPipelined();
     int pipeliningLimit = pipelinedPool ? baseConnectOptions.getPipeliningLimit() : 1;
-    PoolImpl pool = new PoolImpl(vertx, this, tracer, metrics, pipeliningLimit, options, null, null, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this, pipeliningLimit, options, null, null, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
     pool.connectionProvider(factory::connect);
     pool.init();
@@ -91,7 +85,7 @@ public ConnectionFactory createConnectionFactory(Vertx vertx, Supplier<? extends
   }
 
   @Override
-  public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
-    return new DB2ConnectionImpl(context, factory, conn, tracer, metrics);
+  public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection conn) {
+    return new DB2ConnectionImpl(context, factory, conn);
   }
 }

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

@@ -22,12 +22,13 @@
 import io.vertx.core.net.NetSocket;
 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.mssqlclient.MSSQLConnectOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.ConnectionFactoryBase;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 
 import java.util.Map;
 import java.util.function.Supplier;
@@ -42,7 +43,7 @@ public MSSQLConnectionFactory(VertxInternal vertx, Supplier<MSSQLConnectOptions>
 
   @Override
   protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventLoopContext context) {
-    return connectOrRedirect((MSSQLConnectOptions) options, context, 0);
+    return connectOrRedirect(MSSQLConnectOptions.wrap(options), context, 0);
   }
 
   private Future<Connection> connectOrRedirect(MSSQLConnectOptions options, EventLoopContext context, int redirections) {
@@ -55,7 +56,7 @@ private Future<Connection> connectOrRedirect(MSSQLConnectOptions options, EventL
     // Always start unencrypted, the connection will be upgraded if client and server agree
     NetClient netClient = netClient(new NetClientOptions(options).setSsl(false));
     return netClient.connect(server)
-      .map(so -> createSocketConnection(so, desiredPacketSize, context))
+      .map(so -> createSocketConnection(so, options, desiredPacketSize, context))
       .compose(conn -> conn.sendPreLoginMessage(clientSslConfig)
         .compose(encryptionLevel -> login(conn, options, encryptionLevel, context))
       )
@@ -71,8 +72,10 @@ private Future<Connection> connectOrRedirect(MSSQLConnectOptions options, EventL
       });
   }
 
-  private MSSQLSocketConnection createSocketConnection(NetSocket so, int desiredPacketSize, EventLoopContext context) {
-    MSSQLSocketConnection conn = new MSSQLSocketConnection((NetSocketInternal) so, desiredPacketSize, false, 0, sql -> true, 1, context);
+  private MSSQLSocketConnection createSocketConnection(NetSocket so, MSSQLConnectOptions options, int desiredPacketSize, EventLoopContext context) {
+    VertxMetrics vertxMetrics = vertx.metricsSPI();
+    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(options.getSocketAddress(), "sql", options.getMetricsName()) : null;
+    MSSQLSocketConnection conn = new MSSQLSocketConnection((NetSocketInternal) so, metrics, options, desiredPacketSize, false, 0, sql -> true, 1, context);
     conn.init();
     return conn;
   }
@@ -102,11 +105,10 @@ private Future<Connection> login(MSSQLSocketConnection conn, MSSQLConnectOptions
   @Override
   public Future<SqlConnection> connect(Context context, SqlConnectOptions options) {
     ContextInternal ctx = (ContextInternal) context;
-    QueryTracer tracer = ctx.tracer() == null ? null : new QueryTracer(ctx.tracer(), options);
     Promise<SqlConnection> promise = ctx.promise();
     connect(asEventLoopContext(ctx), options)
       .map(conn -> {
-        MSSQLConnectionImpl msConn = new MSSQLConnectionImpl(ctx, this, conn, tracer, null);
+        MSSQLConnectionImpl msConn = new MSSQLConnectionImpl(ctx, this, conn);
         conn.init(msConn);
         return (SqlConnection)msConn;
       })

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

@@ -15,23 +15,21 @@
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.ContextInternal;
-import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.mssqlclient.MSSQLConnectOptions;
 import io.vertx.mssqlclient.MSSQLConnection;
 import io.vertx.mssqlclient.MSSQLInfo;
 import io.vertx.mssqlclient.spi.MSSQLDriver;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
 import io.vertx.sqlclient.impl.SqlConnectionBase;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
 public class MSSQLConnectionImpl extends SqlConnectionBase<MSSQLConnectionImpl> implements MSSQLConnection {
 
   private volatile Handler<MSSQLInfo> infoHandler;
 
-  public MSSQLConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
-    super(context, factory, conn, MSSQLDriver.INSTANCE, tracer, metrics);
+  public MSSQLConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {
+    super(context, factory, conn, MSSQLDriver.INSTANCE);
   }
 
   public static Future<MSSQLConnection> connect(Vertx vertx, MSSQLConnectOptions options) {

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

@@ -23,13 +23,15 @@
 import io.vertx.core.net.impl.NetSocketInternal;
 import io.vertx.core.net.impl.SSLHelper;
 import io.vertx.core.net.impl.SslHandshakeCompletionHandler;
+import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.mssqlclient.MSSQLConnectOptions;
 import io.vertx.mssqlclient.MSSQLInfo;
 import io.vertx.mssqlclient.impl.codec.TdsLoginSentCompletionHandler;
 import io.vertx.mssqlclient.impl.codec.TdsMessageCodec;
 import io.vertx.mssqlclient.impl.codec.TdsPacketDecoder;
 import io.vertx.mssqlclient.impl.codec.TdsSslHandshakeCodec;
 import io.vertx.mssqlclient.impl.command.PreLoginCommand;
+import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.QueryResultHandler;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
@@ -43,22 +45,31 @@
 
 public class MSSQLSocketConnection extends SocketConnectionBase {
 
+  private final MSSQLConnectOptions connectOptions;
   private final int packetSize;
 
   private MSSQLDatabaseMetadata databaseMetadata;
   private SocketAddress alternateServer;
 
   MSSQLSocketConnection(NetSocketInternal socket,
+                        ClientMetrics clientMetrics,
+                        MSSQLConnectOptions connectOptions,
                         int packetSize,
                         boolean cachePreparedStatements,
                         int preparedStatementCacheSize,
                         Predicate<String> preparedStatementCacheSqlFilter,
                         int pipeliningLimit,
                         EventLoopContext context) {
-    super(socket, cachePreparedStatements, preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
+    super(socket, clientMetrics, cachePreparedStatements, preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
+    this.connectOptions = connectOptions;
     this.packetSize = packetSize;
   }
 
+  @Override
+  protected SqlConnectOptions connectOptions() {
+    return connectOptions;
+  }
+
   Future<Byte> sendPreLoginMessage(boolean clientConfigSsl) {
     PreLoginCommand cmd = new PreLoginCommand(clientConfigSsl);
     return schedule(context, cmd).map(resp -> {

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

@@ -20,8 +20,6 @@
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.json.JsonObject;
-import io.vertx.core.spi.metrics.ClientMetrics;
-import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.mssqlclient.MSSQLConnectOptions;
 import io.vertx.mssqlclient.impl.MSSQLConnectionFactory;
 import io.vertx.mssqlclient.impl.MSSQLConnectionImpl;
@@ -33,7 +31,6 @@
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolImpl;
 import io.vertx.sqlclient.impl.SqlConnectionInternal;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 import io.vertx.sqlclient.spi.Driver;
 
@@ -59,12 +56,9 @@ public Pool newPool(Vertx vertx, Supplier<? extends SqlConnectOptions> databases
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
     MSSQLConnectOptions baseConnectOptions = MSSQLConnectOptions.wrap(databases.get());
-    QueryTracer tracer = vertx.tracer() == null ? null : new QueryTracer(vertx.tracer(), baseConnectOptions);
-    VertxMetrics vertxMetrics = vertx.metricsSPI();
-    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(baseConnectOptions.getSocketAddress(), "sql", baseConnectOptions.getMetricsName()) : null;
-    PoolImpl pool = new PoolImpl(vertx, this, tracer, metrics, 1, options, null, null, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this, 1, options, null, null, closeFuture);
     ConnectionFactory factory = createConnectionFactory(vertx, databases);
-    pool.connectionProvider(factory::connect);
+    pool.connectionProvider(context -> factory.connect(context, databases.get()));
     pool.init();
     closeFuture.add(factory);
     return pool;
@@ -98,7 +92,7 @@ public int appendQueryPlaceholder(StringBuilder queryBuilder, int index, int cur
   }
 
   @Override
-  public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
-    return new MSSQLConnectionImpl(context, factory, conn, tracer, metrics);
+  public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection conn) {
+    return new MSSQLConnectionImpl(context, factory, conn);
   }
 }