publish metrics numJettyConns to see how number of active jetty conne…

…ctions change over time (apache#2839)

this can be compared with numer of active queries to see if requests are waiting in jetty queue

docs/content/operations/metrics.md

@@ -54,10 +54,16 @@ Available Metrics
 |`query/wait/time`|Milliseconds spent waiting for a segment to be scanned.|id, segment.|several hundred milliseconds|
 |`segment/scan/pending`|Number of segments in queue waiting to be scanned.||Close to 0|
 
+### Jetty
+
+|Metric|Description|Normal Value|
+|------|-----------|------------|
+|`jetty/numOpenConnections`|Number of open jetty connections.|Not much higher than number of jetty threads.|
+
 ### Cache
 
-|Metric|Description|Dimensions|Normal Value|
-|------|-----------|----------|------------|
+|Metric|Description|Normal Value|
+|------|-----------|------------|
 |`query/cache/delta/*`|Cache metrics since the last emission.||N/A|
 |`query/cache/total/*`|Total cache metrics.||N/A|
 

server/src/main/java/io/druid/server/initialization/jetty/JettyMonitoringConnectionFactory.java

@@ -0,0 +1,73 @@
+/*
+ *
+ *  Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ *  or more contributor license agreements. See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership. Metamarkets licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License. You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied. See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ * /
+ *
+ */
+
+package io.druid.server.initialization.jetty;
+
+import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.server.ConnectionFactory;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.util.component.ContainerLifeCycle;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class JettyMonitoringConnectionFactory extends ContainerLifeCycle implements ConnectionFactory
+{
+  private final ConnectionFactory connectionFactory;
+  private final AtomicInteger activeConns;
+
+  public JettyMonitoringConnectionFactory(ConnectionFactory connectionFactory, AtomicInteger activeConns)
+  {
+    this.connectionFactory = connectionFactory;
+    addBean(connectionFactory);
+    this.activeConns = activeConns;
+  }
+
+  @Override
+  public String getProtocol()
+  {
+    return connectionFactory.getProtocol();
+  }
+
+  @Override
+  public Connection newConnection(Connector connector, EndPoint endPoint)
+  {
+    final Connection connection = connectionFactory.newConnection(connector, endPoint);
+    connection.addListener(
+        new Connection.Listener()
+        {
+          @Override
+          public void onOpened(Connection connection)
+          {
+            activeConns.incrementAndGet();
+          }
+
+          @Override
+          public void onClosed(Connection connection)
+          {
+            activeConns.decrementAndGet();
+          }
+        }
+    );
+    return connection;
+  }
+}

server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java

@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import com.google.common.primitives.Ints;
 import com.google.inject.Binder;
 import com.google.inject.ConfigurationException;
@@ -34,6 +35,10 @@
 import com.google.inject.multibindings.Multibinder;
 import com.metamx.common.lifecycle.Lifecycle;
 import com.metamx.common.logger.Logger;
+import com.metamx.emitter.service.ServiceEmitter;
+import com.metamx.emitter.service.ServiceMetricEvent;
+import com.metamx.metrics.AbstractMonitor;
+import com.metamx.metrics.MonitorUtils;
 import com.sun.jersey.api.core.DefaultResourceConfig;
 import com.sun.jersey.api.core.ResourceConfig;
 import com.sun.jersey.guice.JerseyServletModule;
@@ -45,25 +50,35 @@
 import io.druid.guice.annotations.JSR311Resource;
 import io.druid.guice.annotations.Json;
 import io.druid.guice.annotations.Self;
+import io.druid.query.DruidMetrics;
 import io.druid.server.DruidNode;
 import io.druid.server.StatusResource;
 import io.druid.server.initialization.ServerConfig;
+import io.druid.server.metrics.MetricsModule;
+import io.druid.server.metrics.MonitorsConfig;
+import org.eclipse.jetty.server.ConnectionFactory;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
 import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
 
 import javax.servlet.ServletException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  */
 public class JettyServerModule extends JerseyServletModule
 {
   private static final Logger log = new Logger(JettyServerModule.class);
 
+  private static final AtomicInteger activeConnections = new AtomicInteger();
+
   @Override
   protected void configureServlets()
   {
@@ -83,6 +98,8 @@ protected void configureServlets()
     //Adding empty binding for ServletFilterHolders so that injector returns
     //an empty set when no external modules provide ServletFilterHolder impls
     Multibinder.newSetBinder(binder, ServletFilterHolder.class);
+
+    MetricsModule.register(binder, JettyMonitor.class);
   }
 
   public static class DruidGuiceContainer extends GuiceContainer
@@ -110,7 +127,9 @@ protected ResourceConfig getDefaultResourceConfig(
 
   @Provides
   @LazySingleton
-  public Server getServer(Injector injector, Lifecycle lifecycle, @Self DruidNode node, ServerConfig config)
+  public Server getServer(
+      final Injector injector, final Lifecycle lifecycle, @Self final DruidNode node, final ServerConfig config
+  )
   {
     final Server server = makeJettyServer(node, config);
     initializeServer(injector, lifecycle, server);
@@ -146,6 +165,12 @@ static Server makeJettyServer(DruidNode node, ServerConfig config)
     // https://bugs.eclipse.org/bugs/show_bug.cgi?id=435322#c66 for jetty half open connection issues during failovers
     connector.setAcceptorPriorityDelta(-1);
 
+    List<ConnectionFactory> monitoredConnFactories = new ArrayList<>();
+    for (ConnectionFactory cf : connector.getConnectionFactories()) {
+      monitoredConnFactories.add(new JettyMonitoringConnectionFactory(cf, activeConnections));
+    }
+    connector.setConnectionFactories(monitoredConnFactories);
+
     server.setConnectors(new Connector[]{connector});
 
     return server;
@@ -184,4 +209,37 @@ public void stop()
     );
   }
 
+  @Provides
+  @Singleton
+  public JettyMonitor getJettyMonitor(Properties props)
+  {
+    return new JettyMonitor(props);
+  }
+
+  public static class JettyMonitor extends AbstractMonitor
+  {
+    private final Map<String, String[]> dimensions;
+
+    public JettyMonitor(Properties props)
+    {
+      this.dimensions = MonitorsConfig.extractDimensions(
+          props,
+          Lists.newArrayList(
+              DruidMetrics.DATASOURCE,
+              DruidMetrics.TASK_ID
+          )
+      );
+    }
+
+    @Override
+    public boolean doMonitor(ServiceEmitter emitter)
+    {
+      final ServiceMetricEvent.Builder builder = new ServiceMetricEvent.Builder();
+      MonitorUtils.addDimensionsToBuilder(
+          builder, dimensions
+      );
+      emitter.emit(builder.build("jetty/numOpenConnections", activeConnections.get()));
+      return true;
+    }
+  }
 }