Merge pull request #470 from zhenlineo/1.6-metrics

Draft of driver metrics

Author: zhenlineo

driver/pom.xml

@@ -29,6 +29,10 @@
       <groupId>io.netty</groupId>
       <artifactId>netty-handler</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.hdrhistogram</groupId>
+      <artifactId>HdrHistogram</artifactId>
+    </dependency>
 
     <!-- Test dependencies -->
     <dependency>
@@ -208,13 +212,18 @@
               <artifactSet>
                 <includes>
                   <include>io.netty:*</include>
+                  <include>org.hdrhistogram:*</include>
                 </includes>
               </artifactSet>
               <relocations>
                 <relocation>
                   <pattern>io.netty</pattern>
                   <shadedPattern>org.neo4j.driver.internal.shaded.io.netty</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.HdrHistogram</pattern>
+                  <shadedPattern>org.neo4j.driver.internal.shaded.org.HdrHistogram</shadedPattern>
+                </relocation>
               </relocations>
               <shadeTestJar>true</shadeTestJar>
             </configuration>

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

@@ -38,6 +38,10 @@
 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.MetricsListener;
+import org.neo4j.driver.internal.metrics.InternalAbstractMetrics;
+import org.neo4j.driver.internal.metrics.InternalMetrics;
+import org.neo4j.driver.internal.metrics.spi.Metrics;
 import org.neo4j.driver.internal.retry.ExponentialBackoffRetryLogic;
 import org.neo4j.driver.internal.retry.RetryLogic;
 import org.neo4j.driver.internal.retry.RetrySettings;
@@ -56,6 +60,8 @@
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 
 import static java.lang.String.format;
+import static org.neo4j.driver.internal.metrics.InternalAbstractMetrics.DEV_NULL_METRICS;
+import static org.neo4j.driver.internal.metrics.spi.Metrics.isMetricsEnabled;
 import static org.neo4j.driver.internal.security.SecurityPlan.insecure;
 
 public class DriverFactory
@@ -77,18 +83,17 @@ 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 );
+        InternalAbstractMetrics metrics = createDriverMetrics( config );
+        ConnectionPool connectionPool = createConnectionPool( authToken, securityPlan, bootstrap, metrics, config );
 
-        InternalDriver driver = createDriver( uri, address, connectionPool, config, newRoutingSettings,
-                eventExecutorGroup, securityPlan, retryLogic );
+        InternalDriver driver = createDriver( uri, securityPlan, address, connectionPool, eventExecutorGroup, newRoutingSettings, retryLogic, metrics, config );
 
         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, MetricsListener metrics, Config config )
     {
         Clock clock = createClock();
         ConnectionSettings settings = new ConnectionSettings( authToken, config.connectionTimeoutMillis() );
@@ -97,7 +102,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 InternalAbstractMetrics createDriverMetrics( Config config )
+    {
+        if( isMetricsEnabled() )
+        {
+            return new InternalMetrics( config );
+        }
+        else
+        {
+            return DEV_NULL_METRICS;
+        }
     }
 
     protected ChannelConnector createConnector( ConnectionSettings settings, SecurityPlan securityPlan,
@@ -106,9 +123,8 @@ protected ChannelConnector createConnector( ConnectionSettings settings, Securit
         return new ChannelConnectorImpl( settings, securityPlan, config.logging(), clock );
     }
 
-    private InternalDriver createDriver( URI uri, BoltServerAddress address,
-            ConnectionPool connectionPool, Config config, RoutingSettings routingSettings,
-            EventExecutorGroup eventExecutorGroup, SecurityPlan securityPlan, RetryLogic retryLogic )
+    private InternalDriver createDriver( URI uri, SecurityPlan securityPlan, BoltServerAddress address, ConnectionPool connectionPool,
+            EventExecutorGroup eventExecutorGroup, RoutingSettings routingSettings, RetryLogic retryLogic, Metrics metrics, Config config )
     {
         try
         {
@@ -117,10 +133,9 @@ private InternalDriver createDriver( URI uri, BoltServerAddress address,
             {
             case BOLT_URI_SCHEME:
                 assertNoRoutingContext( uri, routingSettings );
-                return createDirectDriver( address, config, securityPlan, retryLogic, connectionPool );
+                return createDirectDriver( securityPlan, address, connectionPool, retryLogic, metrics, config );
             case BOLT_ROUTING_URI_SCHEME:
-                return createRoutingDriver( address, connectionPool, config, routingSettings, securityPlan, retryLogic,
-                        eventExecutorGroup );
+                return createRoutingDriver( securityPlan, address, connectionPool, eventExecutorGroup, routingSettings, retryLogic, metrics, config );
             default:
                 throw new ClientException( format( "Unsupported URI scheme: %s", scheme ) );
             }
@@ -138,22 +153,21 @@ private InternalDriver createDriver( URI uri, BoltServerAddress address,
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected InternalDriver createDirectDriver( BoltServerAddress address, Config config,
-            SecurityPlan securityPlan, RetryLogic retryLogic, ConnectionPool connectionPool )
+    protected InternalDriver createDirectDriver( SecurityPlan securityPlan, BoltServerAddress address, ConnectionPool connectionPool, RetryLogic retryLogic,
+            Metrics metrics, Config config )
     {
         ConnectionProvider connectionProvider = new DirectConnectionProvider( address, connectionPool );
         SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
-        return createDriver( sessionFactory, securityPlan, config );
+        return createDriver( securityPlan, sessionFactory, metrics, config );
     }
 
     /**
      * Creates new a new driver for "bolt+routing" scheme.
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected InternalDriver createRoutingDriver( BoltServerAddress address, ConnectionPool connectionPool,
-            Config config, RoutingSettings routingSettings, SecurityPlan securityPlan, RetryLogic retryLogic,
-            EventExecutorGroup eventExecutorGroup )
+    protected InternalDriver createRoutingDriver( SecurityPlan securityPlan, BoltServerAddress address, ConnectionPool connectionPool,
+            EventExecutorGroup eventExecutorGroup, RoutingSettings routingSettings, RetryLogic retryLogic, Metrics metrics, Config config )
     {
         if ( !securityPlan.isRoutingCompatible() )
         {
@@ -162,17 +176,17 @@ protected InternalDriver createRoutingDriver( BoltServerAddress address, Connect
         ConnectionProvider connectionProvider = createLoadBalancer( address, connectionPool, eventExecutorGroup,
                 config, routingSettings );
         SessionFactory sessionFactory = createSessionFactory( connectionProvider, retryLogic, config );
-        return createDriver( sessionFactory, securityPlan, config );
+        return createDriver( securityPlan, sessionFactory, metrics, config );
     }
 
     /**
      * Creates new {@link Driver}.
      * <p>
      * <b>This method is protected only for testing</b>
      */
-    protected InternalDriver createDriver( SessionFactory sessionFactory, SecurityPlan securityPlan, Config config )
+    protected InternalDriver createDriver( SecurityPlan securityPlan, SessionFactory sessionFactory, Metrics metrics, Config config )
     {
-        return new InternalDriver( securityPlan, sessionFactory, config.logging() );
+        return new InternalDriver( securityPlan, sessionFactory, metrics, config.logging() );
     }
 
     /**

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.Metrics;
 import org.neo4j.driver.internal.security.SecurityPlan;
 import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.v1.AccessMode;
@@ -38,11 +39,13 @@ public class InternalDriver implements Driver
     private final Logger log;
 
     private AtomicBoolean closed = new AtomicBoolean( false );
+    private final Metrics metrics;
 
-    InternalDriver( SecurityPlan securityPlan, SessionFactory sessionFactory, Logging logging )
+    InternalDriver( SecurityPlan securityPlan, SessionFactory sessionFactory, Metrics metrics, Logging logging )
     {
         this.securityPlan = securityPlan;
         this.sessionFactory = sessionFactory;
+        this.metrics = metrics;
         this.log = logging.getLog( Driver.class.getSimpleName() );
         log.info( "Driver instance %s created", this );
     }
@@ -144,6 +147,11 @@ private void assertOpen()
         }
     }
 
+    public Metrics metrics()
+    {
+        return this.metrics;
+    }
+
     private static RuntimeException driverCloseException()
     {
         return new IllegalStateException( "This driver instance has already been closed" );

driver/src/main/java/org/neo4j/driver/internal/async/NettyConnection.java

@@ -34,6 +34,8 @@
 import org.neo4j.driver.internal.messaging.PullAllMessage;
 import org.neo4j.driver.internal.messaging.ResetMessage;
 import org.neo4j.driver.internal.messaging.RunMessage;
+import org.neo4j.driver.internal.metrics.ListenerEvent;
+import org.neo4j.driver.internal.metrics.MetricsListener;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.ResponseHandler;
 import org.neo4j.driver.internal.util.Clock;
@@ -54,8 +56,10 @@ public class NettyConnection implements Connection
     private final Clock clock;
 
     private final AtomicReference<Status> status = new AtomicReference<>( Status.OPEN );
+    private final MetricsListener metricsListener;
+    private final ListenerEvent inUseEvent;
 
-    public NettyConnection( Channel channel, ChannelPool channelPool, Clock clock )
+    public NettyConnection( Channel channel, ChannelPool channelPool, Clock clock, MetricsListener metricsListener )
     {
         this.channel = channel;
         this.messageDispatcher = ChannelAttributes.messageDispatcher( channel );
@@ -64,6 +68,9 @@ public NettyConnection( Channel channel, ChannelPool channelPool, Clock clock )
         this.channelPool = channelPool;
         this.releaseFuture = new CompletableFuture<>();
         this.clock = clock;
+        this.metricsListener = metricsListener;
+        this.inUseEvent = metricsListener.createListenerEvent();
+        metricsListener.afterAcquiredOrCreated( this.serverAddress, this.inUseEvent );
     }
 
     @Override
@@ -124,6 +131,7 @@ public CompletionStage<Void> release()
     {
         if ( status.compareAndSet( Status.OPEN, Status.RELEASED ) )
         {
+            metricsListener.afterReleased( this.serverAddress, this.inUseEvent );
             ChannelReleasingResetResponseHandler handler = new ChannelReleasingResetResponseHandler( channel,
                     channelPool, messageDispatcher, clock, releaseFuture );
 
@@ -137,6 +145,7 @@ public void terminateAndRelease( String reason )
     {
         if ( status.compareAndSet( Status.OPEN, Status.TERMINATED ) )
         {
+            metricsListener.afterReleased( this.serverAddress, this.inUseEvent );
             setTerminationReason( channel, reason );
             channel.close();
             channelPool.release( channel );