Draft of driver metrics

Added Basics of Driver metrics and Connection pool metrics.
The driver metrics are enabled with system property `driver.metrics.enabled=true`

TODO:
Histgram and ConnectionMetrics

Author: Zhen

driver/src/main/java/org/neo4j/driver/internal/DriverFactory.java

@@ -38,6 +38,9 @@
 import org.neo4j.driver.internal.cluster.loadbalancing.LoadBalancingStrategy;
 import org.neo4j.driver.internal.cluster.loadbalancing.RoundRobinLoadBalancingStrategy;
 import org.neo4j.driver.internal.logging.NettyLogging;
+import org.neo4j.driver.internal.metrics.DriverMetricsHandler;
+import org.neo4j.driver.internal.metrics.InternalAbstractDriverMetrics;
+import org.neo4j.driver.internal.metrics.InternalDriverMetrics;
 import org.neo4j.driver.internal.retry.ExponentialBackoffRetryLogic;
 import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.retry.RetrySettings;
@@ -56,6 +59,8 @@
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 
 import static java.lang.String.format;
+import static org.neo4j.driver.internal.metrics.InternalAbstractDriverMetrics.DEV_NULL_METRICS;
+import static org.neo4j.driver.internal.metrics.spi.DriverMetrics.isDriverMetricsEnabled;
 import static org.neo4j.driver.internal.security.SecurityPlan.insecure;
 
 public class DriverFactory
@@ -77,18 +82,19 @@ public final Driver newInstance( URI uri, AuthToken authToken, RoutingSettings r
         EventExecutorGroup eventExecutorGroup = bootstrap.config().group();
         RetryLogic retryLogic = createRetryLogic( retrySettings, eventExecutorGroup, config.logging() );
 
-        ConnectionPool connectionPool = createConnectionPool( authToken, securityPlan, bootstrap, config );
+        InternalAbstractDriverMetrics metrics = createDriverMetrics( config );
+        ConnectionPool connectionPool = createConnectionPool( authToken, securityPlan, bootstrap, metrics, config );
 
         InternalDriver driver = createDriver( uri, address, connectionPool, config, newRoutingSettings,
                 eventExecutorGroup, securityPlan, retryLogic );
 
+        driver.driverMetrics( metrics );
         verifyConnectivity( driver, connectionPool, config );
 
         return driver;
     }
 
-    protected ConnectionPool createConnectionPool( AuthToken authToken, SecurityPlan securityPlan,
-            Bootstrap bootstrap, Config config )
+    protected ConnectionPool createConnectionPool( AuthToken authToken, SecurityPlan securityPlan, Bootstrap bootstrap, DriverMetricsHandler metrics, Config config )
     {
         Clock clock = createClock();
         ConnectionSettings settings = new ConnectionSettings( authToken, config.connectionTimeoutMillis() );
@@ -97,7 +103,19 @@ protected ConnectionPool createConnectionPool( AuthToken authToken, SecurityPlan
                 config.connectionAcquisitionTimeoutMillis(), config.maxConnectionLifetimeMillis(),
                 config.idleTimeBeforeConnectionTest()
         );
-        return new ConnectionPoolImpl( connector, bootstrap, poolSettings, config.logging(), clock );
+        return new ConnectionPoolImpl( connector, bootstrap, poolSettings, metrics, config.logging(), clock );
+    }
+
+    protected static InternalAbstractDriverMetrics createDriverMetrics( Config config )
+    {
+        if( isDriverMetricsEnabled() )
+        {
+            return new InternalDriverMetrics( config );
+        }
+        else
+        {
+            return DEV_NULL_METRICS;
+        }
     }
 
     protected ChannelConnector createConnector( ConnectionSettings settings, SecurityPlan securityPlan,

driver/src/main/java/org/neo4j/driver/internal/InternalDriver.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.neo4j.driver.internal.metrics.spi.DriverMetrics;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.v1.AccessMode;
@@ -38,6 +39,7 @@ public class InternalDriver implements Driver
     private final Logger log;
 
     private AtomicBoolean closed = new AtomicBoolean( false );
+    private DriverMetrics driverMetrics;
 
     InternalDriver( SecurityPlan securityPlan, SessionFactory sessionFactory, Logging logging )
     {
@@ -144,6 +146,16 @@ private void assertOpen()
         }
     }
 
+    public DriverMetrics driverMetrics()
+    {
+        return this.driverMetrics;
+    }
+
+    void driverMetrics( DriverMetrics driverMetrics )
+    {
+        this.driverMetrics = driverMetrics;
+    }
+
     private static RuntimeException driverCloseException()
     {
         return new IllegalStateException( "This driver instance has already been closed" );

driver/src/main/java/org/neo4j/driver/internal/async/pool/ConnectionPoolImpl.java

@@ -36,6 +36,9 @@
 import org.neo4j.driver.internal.BoltServerAddress;
 import org.neo4j.driver.internal.async.ChannelConnector;
 import org.neo4j.driver.internal.async.NettyConnection;
+import org.neo4j.driver.internal.metrics.DriverMetricsHandler;
+import org.neo4j.driver.internal.metrics.ListenerEvent;
+import org.neo4j.driver.internal.metrics.SimpleTimerListenerEvent;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.ConnectionPool;
 import org.neo4j.driver.internal.util.Clock;
@@ -48,29 +51,31 @@ public class ConnectionPoolImpl implements ConnectionPool
 {
     private final ChannelConnector connector;
     private final Bootstrap bootstrap;
-    private final ActiveChannelTracker activeChannelTracker;
+    private final NettyChannelTracker nettyChannelTracker;
     private final NettyChannelHealthChecker channelHealthChecker;
     private final PoolSettings settings;
     private final Clock clock;
     private final Logger log;
+    private DriverMetricsHandler driverMetrics;
 
     private final ConcurrentMap<BoltServerAddress,ChannelPool> pools = new ConcurrentHashMap<>();
     private final AtomicBoolean closed = new AtomicBoolean();
 
     public ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, PoolSettings settings,
-            Logging logging, Clock clock )
+            DriverMetricsHandler metricsHandler, Logging logging, Clock clock )
     {
-        this( connector, bootstrap, new ActiveChannelTracker( logging ), settings, logging, clock );
+        this( connector, bootstrap, new NettyChannelTracker( metricsHandler, logging ), settings, metricsHandler, logging, clock );
     }
 
-    ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, ActiveChannelTracker activeChannelTracker,
-            PoolSettings settings, Logging logging, Clock clock )
+    ConnectionPoolImpl( ChannelConnector connector, Bootstrap bootstrap, NettyChannelTracker nettyChannelTracker,
+            PoolSettings settings, DriverMetricsHandler driverMetrics, Logging logging, Clock clock )
     {
         this.connector = connector;
         this.bootstrap = bootstrap;
-        this.activeChannelTracker = activeChannelTracker;
+        this.nettyChannelTracker = nettyChannelTracker;
         this.channelHealthChecker = new NettyChannelHealthChecker( settings, clock, logging );
         this.settings = settings;
+        this.driverMetrics = driverMetrics;
         this.clock = clock;
         this.log = logging.getLog( ConnectionPool.class.getSimpleName() );
     }
@@ -80,15 +85,26 @@ public CompletionStage<Connection> acquire( BoltServerAddress address )
     {
         log.trace( "Acquiring a connection from pool towards %s", address );
 
+        // TODO no need to init if no driver metrics
+        ListenerEvent acquireEvent = new SimpleTimerListenerEvent();
+
         assertNotClosed();
         ChannelPool pool = getOrCreatePool( address );
+        driverMetrics.beforeAcquiring( address, acquireEvent );
         Future<Channel> connectionFuture = pool.acquire();
 
         return Futures.asCompletionStage( connectionFuture ).handle( ( channel, error ) ->
         {
-            processAcquisitionError( error );
-            assertNotClosed( address, channel, pool );
-            return new NettyConnection( channel, pool, clock );
+            try
+            {
+                processAcquisitionError( error );
+                assertNotClosed( address, channel, pool );
+                return new NettyConnection( channel, pool, clock );
+            }
+            finally
+            {
+                driverMetrics.afterAcquired( address, acquireEvent );
+            }
         } );
     }
 
@@ -99,7 +115,7 @@ public void retainAll( Set<BoltServerAddress> addressesToRetain )
         {
             if ( !addressesToRetain.contains( address ) )
             {
-                int activeChannels = activeChannelTracker.activeChannelCount( address );
+                int activeChannels = nettyChannelTracker.inUseChannelCount( address );
                 if ( activeChannels == 0 )
                 {
                     // address is not present in updated routing table and has no active connections
@@ -118,9 +134,15 @@ public void retainAll( Set<BoltServerAddress> addressesToRetain )
     }
 
     @Override
-    public int activeConnections( BoltServerAddress address )
+    public int inUseConnections( BoltServerAddress address )
+    {
+        return nettyChannelTracker.inUseChannelCount( address );
+    }
+
+    @Override
+    public int idleConnections( BoltServerAddress address )
     {
-        return activeChannelTracker.activeChannelCount( address );
+        return nettyChannelTracker.idleChannelCount( address );
     }
 
     @Override
@@ -150,6 +172,12 @@ public CompletionStage<Void> close()
                 .thenApply( ignore -> null );
     }
 
+    @Override
+    public boolean isOpen()
+    {
+        return !closed.get();
+    }
+
     private ChannelPool getOrCreatePool( BoltServerAddress address )
     {
         ChannelPool pool = pools.get( address );
@@ -163,13 +191,19 @@ private ChannelPool getOrCreatePool( BoltServerAddress address )
                 pool.close();
                 return getOrCreatePool( address );
             }
+            else
+            {
+                // We added a new pool as a result we add a new metrics for the pool too
+                driverMetrics.addPoolMetrics( address, this );
+            }
         }
+
         return pool;
     }
 
     ChannelPool newPool( BoltServerAddress address )
     {
-        return new NettyChannelPool( address, connector, bootstrap, activeChannelTracker, channelHealthChecker,
+        return new NettyChannelPool( address, connector, bootstrap, nettyChannelTracker, channelHealthChecker,
                 settings.connectionAcquisitionTimeout(), settings.maxConnectionPoolSize() );
     }
 
@@ -217,4 +251,10 @@ private void assertNotClosed( BoltServerAddress address, Channel channel, Channe
             assertNotClosed();
         }
     }
+
+    @Override
+    public String toString()
+    {
+        return "ConnectionPoolImpl{" + "pools=" + pools + '}';
+    }
 }

driver/src/main/java/org/neo4j/driver/internal/async/pool/NettyChannelPool.java

@@ -19,9 +19,9 @@
 package org.neo4j.driver.internal.async.pool;
 
 import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.pool.ChannelHealthChecker;
-import io.netty.channel.pool.ChannelPoolHandler;
 import io.netty.channel.pool.FixedChannelPool;
 
 import org.neo4j.driver.internal.BoltServerAddress;
@@ -42,28 +42,37 @@ public class NettyChannelPool extends FixedChannelPool
 
     private final BoltServerAddress address;
     private final ChannelConnector connector;
+    private final NettyChannelTracker handler;
 
     public NettyChannelPool( BoltServerAddress address, ChannelConnector connector, Bootstrap bootstrap,
-            ChannelPoolHandler handler, ChannelHealthChecker healthCheck, long acquireTimeoutMillis,
+            NettyChannelTracker handler, ChannelHealthChecker healthCheck, long acquireTimeoutMillis,
             int maxConnections )
     {
         super( bootstrap, handler, healthCheck, AcquireTimeoutAction.FAIL, acquireTimeoutMillis, maxConnections,
                 MAX_PENDING_ACQUIRES, RELEASE_HEALTH_CHECK );
 
         this.address = requireNonNull( address );
         this.connector = requireNonNull( connector );
+        this.handler = requireNonNull( handler );
     }
 
     @Override
     protected ChannelFuture connectChannel( Bootstrap bootstrap )
     {
+        handler.channelCreating( address );
         ChannelFuture channelFuture = connector.connect( address, bootstrap );
         channelFuture.addListener( future ->
         {
             if ( future.isSuccess() )
             {
                 // notify pool handler about a successful connection
-                handler().channelCreated( channelFuture.channel() );
+                Channel channel = channelFuture.channel();
+                handler.channelCreated( channel );
+                channel.closeFuture().addListener( closeFuture -> handler.channelClosed( channel ) );
+            }
+            else
+            {
+                handler.channelFailedToCreate( address );
             }
         } );
         return channelFuture;