Report HTTP method as a dimension in request/response metrics

3 files changed, 371 insertions, 28 deletions

diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpResponseStatisticsCollector.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpResponseStatisticsCollector.java
new file mode 100644
index 00000000000..1a42fd2b5bd
--- /dev/null
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpResponseStatisticsCollector.java
@@ -0,0 +1,199 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.jdisc.http.server.jetty;
+
+import com.yahoo.jdisc.http.server.jetty.JettyHttpServer.Metrics;
+import org.eclipse.jetty.http.HttpStatus;
+import org.eclipse.jetty.server.AsyncContextEvent;
+import org.eclipse.jetty.server.Handler;
+import org.eclipse.jetty.server.HttpChannelState;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.handler.HandlerWrapper;
+import org.eclipse.jetty.util.FutureCallback;
+import org.eclipse.jetty.util.component.Graceful;
+
+import javax.servlet.AsyncEvent;
+import javax.servlet.AsyncListener;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * HttpResponseStatisticsCollector collects statistics about HTTP response types aggregated by category (1xx, 2xx, etc). It is similar to
+ * {@link org.eclipse.jetty.server.handler.StatisticsHandler} with the distinction that this class collects response type statistics grouped
+ * by HTTP method and only collects the numbers that are reported as metrics from Vespa.
+ *
+ * @author ollivir
+ */
+public class HttpResponseStatisticsCollector extends HandlerWrapper implements Graceful {
+    private final AtomicReference<FutureCallback> shutdown = new AtomicReference<>();
+
+    public static enum HttpMethod {
+        GET, PATCH, POST, PUT, REMOVE, UNKNOWN
+    }
+
+    private static final String[] HTTP_RESPONSE_GROUPS = { Metrics.RESPONSES_1XX, Metrics.RESPONSES_2XX, Metrics.RESPONSES_3XX,
+            Metrics.RESPONSES_4XX, Metrics.RESPONSES_5XX };
+
+    private final AtomicLong inFlight = new AtomicLong();
+    private final LongAdder statistics[][];
+
+    public HttpResponseStatisticsCollector() {
+        super();
+        statistics = new LongAdder[HttpMethod.values().length][];
+        for (int method = 0; method < statistics.length; method++) {
+            statistics[method] = new LongAdder[HTTP_RESPONSE_GROUPS.length];
+            for (int group = 0; group < HTTP_RESPONSE_GROUPS.length; group++) {
+                statistics[method][group] = new LongAdder();
+            }
+        }
+    }
+
+    private final AsyncListener completionWatcher = new AsyncListener() {
+        @Override
+        public void onTimeout(AsyncEvent event) throws IOException {
+        }
+
+        @Override
+        public void onStartAsync(AsyncEvent event) throws IOException {
+            event.getAsyncContext().addListener(this);
+        }
+
+        @Override
+        public void onError(AsyncEvent event) throws IOException {
+        }
+
+        @Override
+        public void onComplete(AsyncEvent event) throws IOException {
+            HttpChannelState state = ((AsyncContextEvent) event).getHttpChannelState();
+            Request request = state.getBaseRequest();
+
+            observeEndOfRequest(request, null);
+        }
+    };
+
+    @Override
+    public void handle(String path, Request baseRequest, HttpServletRequest request, HttpServletResponse response)
+            throws IOException, ServletException {
+        inFlight.incrementAndGet();
+
+        try {
+            Handler handler = getHandler();
+            if (handler != null && shutdown.get() == null && isStarted()) {
+                handler.handle(path, baseRequest, request, response);
+            } else if (!baseRequest.isHandled()) {
+                baseRequest.setHandled(true);
+                response.sendError(HttpStatus.SERVICE_UNAVAILABLE_503);
+            }
+        } finally {
+            HttpChannelState state = baseRequest.getHttpChannelState();
+
+            if (state.isSuspended()) {
+                if (state.isInitial()) {
+                    state.addListener(completionWatcher);
+                }
+            } else if (state.isInitial()) {
+                observeEndOfRequest(baseRequest, response);
+            }
+        }
+    }
+
+    private void observeEndOfRequest(Request request, HttpServletResponse flushableResponse) throws IOException {
+        int group = groupIndex(request);
+        if (group >= 0) {
+            HttpMethod method = getMethod(request);
+            statistics[method.ordinal()][group].increment();
+        }
+
+        long live = inFlight.decrementAndGet();
+        FutureCallback shutdownCb = shutdown.get();
+        if (shutdownCb != null) {
+            if (flushableResponse != null) {
+                flushableResponse.flushBuffer();
+            }
+            if (live == 0) {
+                shutdownCb.succeeded();
+            }
+        }
+    }
+
+    private int groupIndex(Request request) {
+        if (request.isHandled()) {
+            int index = (request.getResponse().getStatus() / 100) - 1; // 1xx = 0, 2xx = 1 etc.
+            if (index < 0 || index > statistics.length) {
+                return -1;
+            } else {
+                return index;
+            }
+        } else {
+            return 3; // 4xx
+        }
+    }
+
+    private HttpMethod getMethod(Request request) {
+        switch (request.getMethod()) {
+        case "GET":
+            return HttpMethod.GET;
+        case "PATCH":
+            return HttpMethod.PATCH;
+        case "POST":
+            return HttpMethod.POST;
+        case "PUT":
+            return HttpMethod.PUT;
+        case "DELETE":
+            return HttpMethod.REMOVE;
+        default:
+            return HttpMethod.UNKNOWN;
+        }
+    }
+
+    public Map<String, Map<String, Long>> takeStatisticsByMethod() {
+        Map<String, Map<String, Long>> ret = new HashMap<>();
+
+        for (HttpMethod method : HttpMethod.values()) {
+            int methodIndex = method.ordinal();
+            Map<String, Long> methodStats = new HashMap<>();
+            ret.put(method.toString(), methodStats);
+
+            for (int group = 0; group < HTTP_RESPONSE_GROUPS.length; group++) {
+                long value = statistics[methodIndex][group].sumThenReset();
+                methodStats.put(HTTP_RESPONSE_GROUPS[group], value);
+            }
+        }
+        return ret;
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        shutdown.set(null);
+        super.doStart();
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        super.doStop();
+        FutureCallback shutdownCb = shutdown.get();
+        if (shutdown != null && !shutdownCb.isDone()) {
+            shutdownCb.failed(new TimeoutException());
+        }
+    }
+
+    @Override
+    public Future<Void> shutdown() {
+        FutureCallback shutdownCb = new FutureCallback(false);
+        shutdown.compareAndSet(null, shutdownCb);
+        shutdownCb = shutdown.get();
+        if (inFlight.get() == 0) {
+            shutdownCb.succeeded();
+        }
+        return shutdownCb;
+    }
+}
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java
index 452d1aed874..31521c02d37 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java
@@ -1,4 +1,4 @@
-// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.jdisc.http.server.jetty;
 
 import com.google.common.annotations.Beta;
@@ -8,6 +8,7 @@ import com.yahoo.component.ComponentId;
 import com.yahoo.component.provider.ComponentRegistry;
 import com.yahoo.container.logging.AccessLog;
 import com.yahoo.jdisc.Metric;
+import com.yahoo.jdisc.Metric.Context;
 import com.yahoo.jdisc.application.OsgiFramework;
 import com.yahoo.jdisc.http.ServerConfig;
 import com.yahoo.jdisc.http.ServletPathsConfig;
@@ -24,7 +25,6 @@ import org.eclipse.jetty.server.ServerConnectionStatistics;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.server.handler.AbstractHandlerContainer;
 import org.eclipse.jetty.server.handler.HandlerCollection;
-import org.eclipse.jetty.server.handler.StatisticsHandler;
 import org.eclipse.jetty.server.handler.gzip.GzipHandler;
 import org.eclipse.jetty.server.handler.gzip.GzipHttpOutputInterceptor;
 import org.eclipse.jetty.servlet.FilterHolder;
@@ -39,6 +39,7 @@ import org.osgi.framework.ServiceReference;
 
 import javax.management.remote.JMXServiceURL;
 import javax.servlet.DispatcherType;
+
 import java.lang.management.ManagementFactory;
 import java.net.BindException;
 import java.net.MalformedURLException;
@@ -46,7 +47,9 @@ import java.nio.channels.ServerSocketChannel;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -233,11 +236,11 @@ public class JettyHttpServer extends AbstractServerProvider {
         GzipHandler gzipHandler = newGzipHandler(serverConfig);
         gzipHandler.setHandler(servletContextHandler);
 
-        StatisticsHandler statisticsHandler = newStatisticsHandler();
-        statisticsHandler.setHandler(gzipHandler);
+        HttpResponseStatisticsCollector statisticsCollector = new HttpResponseStatisticsCollector();
+        statisticsCollector.setHandler(gzipHandler);
 
         HandlerCollection handlerCollection = new HandlerCollection();
-        handlerCollection.setHandlers(new Handler[]{statisticsHandler});
+        handlerCollection.setHandlers(new Handler[] { statisticsCollector });
         return handlerCollection;
     }
 
@@ -293,7 +296,7 @@ public class JettyHttpServer extends AbstractServerProvider {
         try {
             server.start();
         } catch (final BindException e) {
-            throw new RuntimeException("Failed to start server due to BindExecption. ListenPorts = " + listenedPorts.toString() , e);
+            throw new RuntimeException("Failed to start server due to BindExecption. ListenPorts = " + listenedPorts.toString(), e);
         } catch (final Exception e) {
             throw new RuntimeException("Failed to start server.", e);
         }
@@ -318,12 +321,12 @@ public class JettyHttpServer extends AbstractServerProvider {
     private class MetricTask implements Runnable {
         @Override
         public void run() {
-            StatisticsHandler statisticsHandler = ((AbstractHandlerContainer)server.getHandler())
-                    .getChildHandlerByClass(StatisticsHandler.class);
-            if (statisticsHandler == null)
+            HttpResponseStatisticsCollector statisticsCollector = ((AbstractHandlerContainer) server.getHandler())
+                    .getChildHandlerByClass(HttpResponseStatisticsCollector.class);
+            if (statisticsCollector == null)
                 return;
 
-            setServerMetrics(statisticsHandler);
+            setServerMetrics(statisticsCollector);
 
             for (Connector connector : server.getConnectors()) {
                 setConnectorMetrics((JDiscServerConnector)connector);
@@ -333,20 +336,22 @@ public class JettyHttpServer extends AbstractServerProvider {
     }
 
     @SuppressWarnings("deprecation")
-    private void setServerMetrics(StatisticsHandler statistics) {
+    private void setServerMetrics(HttpResponseStatisticsCollector statisticsCollector) {
         long timeSinceStarted = System.currentTimeMillis() - timeStarted;
         metric.set(Metrics.STARTED_MILLIS, timeSinceStarted, null);
         metric.set(Metrics.MANHATTAN_STARTED_MILLIS, timeSinceStarted, null);
 
-        metric.add(Metrics.RESPONSES_1XX, statistics.getResponses1xx(), null);
-        metric.add(Metrics.RESPONSES_2XX, statistics.getResponses2xx(), null);
-        metric.add(Metrics.RESPONSES_3XX, statistics.getResponses3xx(), null);
-        metric.add(Metrics.RESPONSES_4XX, statistics.getResponses4xx(), null);
-        metric.add(Metrics.RESPONSES_5XX, statistics.getResponses5xx(), null);
+        addResponseMetrics(statisticsCollector);
+    }
 
-        // Reset to only add the diff for count metrics.
-        // (The alternative to reset would be to preserve the previous value, and only add the diff.)
-        statistics.statsReset();
+    private void addResponseMetrics(HttpResponseStatisticsCollector statisticsCollector) {
+        Map<String, Map<String, Long>> statistics = statisticsCollector.takeStatisticsByMethod();
+        statistics.forEach((httpMethod, statsByResponseType) -> {
+            Map<String, Object> dimensions = new HashMap<>();
+            dimensions.put(Metrics.METHOD_DIMENSION, httpMethod);
+            Context ctx = metric.createContext(dimensions);
+            statsByResponseType.forEach((group, value) -> metric.add(group, value, ctx));
+        });
     }
 
     private void setConnectorMetrics(JDiscServerConnector connector) {
@@ -359,12 +364,6 @@ public class JettyHttpServer extends AbstractServerProvider {
         metric.set(Metrics.CONNECTION_DURATION_STD_DEV, statistics.getConnectionDurationStdDev(), connector.getConnectorMetricContext());
     }
 
-    private StatisticsHandler newStatisticsHandler() {
-        StatisticsHandler statisticsHandler = new StatisticsHandler();
-        statisticsHandler.statsReset();
-        return statisticsHandler;
-    }
-
     private GzipHandler newGzipHandler(ServerConfig serverConfig) {
         GzipHandler gzipHandler = new GzipHandlerWithVaryHeaderFixed();
         gzipHandler.setCompressionLevel(serverConfig.responseCompressionLevel());
@@ -378,9 +377,9 @@ public class JettyHttpServer extends AbstractServerProvider {
 
         @Override
         public HttpField getVaryField() {
-            return GzipHttpOutputInterceptor.VARY_ACCEPT_ENCODING;            
+            return GzipHttpOutputInterceptor.VARY_ACCEPT_ENCODING;
         }
-        
+
     }
-    
+
 }
diff --git a/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/HttpResponseStatisticsCollectorTest.java b/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/HttpResponseStatisticsCollectorTest.java
new file mode 100644
index 00000000000..e3d70fb5bd6
--- /dev/null
+++ b/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/HttpResponseStatisticsCollectorTest.java
@@ -0,0 +1,145 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.jdisc.http.server.jetty;
+
+import com.yahoo.jdisc.http.server.jetty.JettyHttpServer.Metrics;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpURI;
+import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.http.MetaData;
+import org.eclipse.jetty.http.MetaData.Response;
+import org.eclipse.jetty.server.AbstractConnector;
+import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.HttpChannel;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.HttpTransport;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import org.eclipse.jetty.util.Callback;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+
+/**
+ * @author ollivir
+ */
+public class HttpResponseStatisticsCollectorTest {
+    private Connector connector;
+    private HttpResponseStatisticsCollector collector = new HttpResponseStatisticsCollector();
+    private int httpResponseCode = 500;
+
+    @Test
+    public void statistics_are_aggregated_by_category() throws Exception {
+        testRequest(300, "GET");
+        testRequest(301, "GET");
+        testRequest(200, "GET");
+
+        Map<String, Map<String, Long>> stats = collector.takeStatisticsByMethod();
+        assertThat(stats.get("GET").get(Metrics.RESPONSES_2XX), equalTo(1L));
+        assertThat(stats.get("GET").get(Metrics.RESPONSES_3XX), equalTo(2L));
+    }
+
+    @Test
+    public void statistics_are_grouped_by_http_method() throws Exception {
+        testRequest(200, "GET");
+        testRequest(200, "PUT");
+        testRequest(200, "POST");
+        testRequest(200, "POST");
+        testRequest(404, "GET");
+
+        Map<String, Map<String, Long>> stats = collector.takeStatisticsByMethod();
+        assertThat(stats.get("GET").get(Metrics.RESPONSES_2XX), equalTo(1L));
+        assertThat(stats.get("GET").get(Metrics.RESPONSES_4XX), equalTo(1L));
+        assertThat(stats.get("PUT").get(Metrics.RESPONSES_2XX), equalTo(1L));
+        assertThat(stats.get("POST").get(Metrics.RESPONSES_2XX), equalTo(2L));
+    }
+
+    @Test
+    public void retrieving_statistics_resets_the_counters() throws Exception {
+        testRequest(200, "GET");
+        testRequest(200, "GET");
+
+        Map<String, Map<String, Long>> stats = collector.takeStatisticsByMethod();
+        assertThat(stats.get("GET").get(Metrics.RESPONSES_2XX), equalTo(2L));
+
+        testRequest(200, "GET");
+
+        stats = collector.takeStatisticsByMethod();
+        assertThat(stats.get("GET").get(Metrics.RESPONSES_2XX), equalTo(1L));
+    }
+
+    @BeforeTest
+    public void initializeCollector() throws Exception {
+        Server server = new Server();
+        connector = new AbstractConnector(server, null, null, null, 0) {
+            @Override
+            protected void accept(int acceptorID) throws IOException, InterruptedException {
+            }
+
+            @Override
+            public Object getTransport() {
+                return null;
+            }
+        };
+        collector.setHandler(new AbstractHandler() {
+            @Override
+            public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response)
+                    throws IOException, ServletException {
+                baseRequest.setHandled(true);
+                baseRequest.getResponse().setStatus(httpResponseCode);
+            }
+        });
+        server.setHandler(collector);
+        server.start();
+    }
+
+    private Request testRequest(int responseCode, String httpMethod) throws Exception {
+        HttpChannel channel = new HttpChannel(connector, new HttpConfiguration(), null, new DummyTransport());
+        MetaData.Request metaData = new MetaData.Request(httpMethod, new HttpURI("http://foo/bar"), HttpVersion.HTTP_1_1, new HttpFields());
+        Request req = channel.getRequest();
+        req.setMetaData(metaData);
+
+        this.httpResponseCode = responseCode;
+        channel.handle();
+        return req;
+    }
+
+    private final class DummyTransport implements HttpTransport {
+        @Override
+        public void send(Response info, boolean head, ByteBuffer content, boolean lastContent, Callback callback) {
+            callback.succeeded();
+        }
+
+        @Override
+        public boolean isPushSupported() {
+            return false;
+        }
+
+        @Override
+        public boolean isOptimizedForDirectBuffers() {
+            return false;
+        }
+
+        @Override
+        public void push(MetaData.Request request) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+
+        @Override
+        public void abort(Throwable failure) {
+        }
+    }
+}