Merge pull request #17823 from vespa-engine/bjorncs/connection-metrics

Bjorncs/connection metrics [run-systemtest]

9 files changed, 115 insertions, 74 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/http/JettyHttpServer.java b/config-model/src/main/java/com/yahoo/vespa/model/container/http/JettyHttpServer.java
index ce79a124e81..7afebe2ed52 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/http/JettyHttpServer.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/http/JettyHttpServer.java
@@ -5,6 +5,7 @@ import com.yahoo.component.ComponentId;
 import com.yahoo.component.ComponentSpecification;
 import com.yahoo.container.bundle.BundleInstantiationSpecification;
 import com.yahoo.jdisc.http.ServerConfig;
+import com.yahoo.jdisc.http.server.jetty.Janitor;
 import com.yahoo.osgi.provider.model.ComponentModel;
 import com.yahoo.vespa.model.container.ApplicationContainerCluster;
 import com.yahoo.vespa.model.container.ContainerCluster;
@@ -14,8 +15,6 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
-import static com.yahoo.component.ComponentSpecification.fromString;
-
 /**
  * @author Einar M R Rosenvinge
  * @author bjorncs
@@ -33,6 +32,7 @@ public class JettyHttpServer extends SimpleComponent implements ServerConfig.Pro
         final FilterBindingsProviderComponent filterBindingsProviderComponent = new FilterBindingsProviderComponent(componentId);
         addChild(filterBindingsProviderComponent);
         inject(filterBindingsProviderComponent);
+        inject(new SimpleComponent(Janitor.class.getName()));
     }
 
     public void setHostedVespa(boolean isHostedVespa) { this.isHostedVespa = isHostedVespa; }
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscContext.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscContext.java
index b37a7352dc6..48c70095918 100644
--- a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscContext.java
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscContext.java
@@ -5,18 +5,16 @@ import com.yahoo.jdisc.Metric;
 import com.yahoo.jdisc.http.ServerConfig;
 import com.yahoo.jdisc.service.CurrentContainer;
 
-import java.util.concurrent.Executor;
-
 public class JDiscContext {
     final FilterResolver filterResolver;
     final CurrentContainer container;
-    final Executor janitor;
+    final Janitor janitor;
     final Metric metric;
     final ServerConfig serverConfig;
 
     public JDiscContext(FilterBindings filterBindings,
                         CurrentContainer container,
-                        Executor janitor,
+                        Janitor janitor,
                         Metric metric,
                         ServerConfig serverConfig) {
 
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/Janitor.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/Janitor.java
new file mode 100644
index 00000000000..cd2b9ca23c0
--- /dev/null
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/Janitor.java
@@ -0,0 +1,46 @@
+// Copyright Verizon Media. 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.inject.Inject;
+import com.yahoo.component.AbstractComponent;
+import com.yahoo.concurrent.DaemonThreadFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+/**
+ * Separate janitor threadpool for tasks that cannot be executed on the jdisc default threadpool due to risk of deadlock.
+ * Modelled as a separate component as the underlying executor must be available across {@link JettyHttpServer} instances.
+ *
+ * @author bjorncs
+ */
+public class Janitor extends AbstractComponent {
+
+    private static final Logger log = Logger.getLogger(Janitor.class.getName());
+
+    private final ExecutorService executor;
+
+    @Inject
+    public Janitor() {
+        int threadPoolSize = Math.max(2, Runtime.getRuntime().availableProcessors()/4);
+        log.info("Creating janitor executor with " + threadPoolSize + " threads");
+        this.executor = Executors.newFixedThreadPool(threadPoolSize, new DaemonThreadFactory("jdisc-janitor-"));
+    }
+
+    public void scheduleTask(Runnable task) { executor.execute(task); }
+
+    @Override
+    public void deconstruct() {
+        try {
+            executor.shutdown();
+            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
+                log.warning("Failed to shutdown janitor in time");
+            }
+        } catch (InterruptedException e) {
+            log.warning("Interrupted while shutting down janitor");
+            Thread.currentThread().interrupt();
+        }
+    }
+}
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java
index 1923153f970..9a6465cce3b 100644
--- a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java
@@ -35,8 +35,6 @@ import java.util.Date;
 import java.util.List;
 import java.util.Objects;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -51,8 +49,8 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
 
     private static final Logger log = Logger.getLogger(JettyConnectionLogger.class.getName());
 
-    private final ConcurrentMap<IdentityKey<SocketChannelEndPoint>, ConnectionInfo> connectionInfo = new ConcurrentHashMap<>();
-    private final ConcurrentMap<IdentityKey<SSLEngine>, ConnectionInfo> sslToConnectionInfo = new ConcurrentHashMap<>();
+    private final SimpleConcurrentIdentityHashMap<SocketChannelEndPoint, ConnectionInfo> connectionInfo = new SimpleConcurrentIdentityHashMap<>();
+    private final SimpleConcurrentIdentityHashMap<SSLEngine, ConnectionInfo> sslToConnectionInfo = new SimpleConcurrentIdentityHashMap<>();
 
     private final boolean enabled;
     private final ConnectionLog connectionLog;
@@ -90,16 +88,15 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
     public void onOpened(Connection connection) {
         handleListenerInvocation("Connection.Listener", "onOpened", "%h", List.of(connection), () -> {
             SocketChannelEndPoint endpoint = findUnderlyingSocketEndpoint(connection.getEndPoint());
-            var endpointKey = IdentityKey.of(endpoint);
-            ConnectionInfo info = connectionInfo.get(endpointKey);
+            ConnectionInfo info = connectionInfo.get(endpoint);
             if (info == null) {
                 info = ConnectionInfo.from(endpoint);
-                connectionInfo.put(IdentityKey.of(endpoint), info);
+                connectionInfo.put(endpoint, info);
             }
             String connectionClassName = connection.getClass().getSimpleName(); // For hidden implementations of Connection
             if (connection instanceof SslConnection) {
                 SSLEngine sslEngine = ((SslConnection) connection).getSSLEngine();
-                sslToConnectionInfo.put(IdentityKey.of(sslEngine), info);
+                sslToConnectionInfo.put(sslEngine, info);
             } else if (connection instanceof HttpConnection) {
                 info.setHttpProtocol("HTTP/1.1");
             } else if (connection instanceof HTTP2ServerConnection) {
@@ -120,24 +117,23 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
     public void onClosed(Connection connection) {
         handleListenerInvocation("Connection.Listener", "onClosed", "%h", List.of(connection), () -> {
             SocketChannelEndPoint endpoint = findUnderlyingSocketEndpoint(connection.getEndPoint());
-            var endpointKey = IdentityKey.of(endpoint);
-            ConnectionInfo info = connectionInfo.get(endpointKey);
+            ConnectionInfo info = connectionInfo.get(endpoint);
             if (info == null) return; // Closed connection already handled
             if (connection instanceof HttpConnection) {
                 info.setHttpBytes(connection.getBytesIn(), connection.getBytesOut());
             } else if (connection instanceof SslConnection) {
                 SSLEngine sslEngine = ((SslConnection) connection).getSSLEngine();
-                sslToConnectionInfo.remove(IdentityKey.of(sslEngine));
+                sslToConnectionInfo.remove(sslEngine);
             } else if (connection instanceof ALPNServerConnection) {
                 SSLEngine sslEngine = ((ALPNServerConnection) connection).getSSLEngine();
                 if (sslEngine != null) {
-                    sslToConnectionInfo.remove(IdentityKey.of(sslEngine));
+                    sslToConnectionInfo.remove(sslEngine);
                 }
             }
             if (!endpoint.isOpen()) {
                 info.setClosedAt(System.currentTimeMillis());
                 connectionLog.log(info.toLogEntry());
-                connectionInfo.remove(endpointKey);
+                connectionInfo.remove(endpoint);
             }
         });
     }
@@ -152,7 +148,7 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
     public void onRequestBegin(Request request) {
         handleListenerInvocation("HttpChannel.Listener", "onRequestBegin", "%h", List.of(request), () -> {
             SocketChannelEndPoint endpoint = findUnderlyingSocketEndpoint(request.getHttpChannel().getEndPoint());
-            ConnectionInfo info = Objects.requireNonNull(connectionInfo.get(IdentityKey.of(endpoint)));
+            ConnectionInfo info = Objects.requireNonNull(connectionInfo.get(endpoint));
             info.incrementRequests();
             request.setAttribute(CONNECTION_ID_REQUEST_ATTRIBUTE, info.uuid());
         });
@@ -162,7 +158,7 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
     public void onResponseBegin(Request request) {
         handleListenerInvocation("HttpChannel.Listener", "onResponseBegin", "%h", List.of(request), () -> {
             SocketChannelEndPoint endpoint = findUnderlyingSocketEndpoint(request.getHttpChannel().getEndPoint());
-            ConnectionInfo info = connectionInfo.get(IdentityKey.of(endpoint));
+            ConnectionInfo info = connectionInfo.get(endpoint);
             if (info == null) return; // Connection closed before response started - observed during Jetty server shutdown
             info.incrementResponses();
         });
@@ -178,7 +174,7 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
     public void handshakeSucceeded(Event event) {
         SSLEngine sslEngine = event.getSSLEngine();
         handleListenerInvocation("SslHandshakeListener", "handshakeSucceeded", "sslEngine=%h", List.of(sslEngine), () -> {
-            ConnectionInfo info = sslToConnectionInfo.remove(IdentityKey.of(sslEngine));
+            ConnectionInfo info = sslToConnectionInfo.remove(sslEngine);
             if (info == null) return;
             info.setSslSessionDetails(sslEngine.getSession());
         });
@@ -189,7 +185,7 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
         SSLEngine sslEngine = event.getSSLEngine();
         handleListenerInvocation("SslHandshakeListener", "handshakeFailed", "sslEngine=%h,failure=%s", List.of(sslEngine, failure), () -> {
             log.log(Level.FINE, failure, failure::toString);
-            ConnectionInfo info = sslToConnectionInfo.remove(IdentityKey.of(sslEngine));
+            ConnectionInfo info = sslToConnectionInfo.remove(sslEngine);
             if (info == null) return;
             info.setSslHandshakeFailure((SSLHandshakeException)failure);
         });
@@ -387,21 +383,4 @@ class JettyConnectionLogger extends AbstractLifeCycle implements Connection.List
 
     }
 
-    private static class IdentityKey<T> {
-        final T instance;
-
-        IdentityKey(T instance) { this.instance = instance; }
-
-        static <T> IdentityKey<T> of(T instance) { return new IdentityKey<>(instance); }
-
-        @Override public int hashCode() { return System.identityHashCode(instance); }
-
-        @Override
-        public boolean equals(Object obj) {
-            if (this == obj) return true;
-            if (!(obj instanceof IdentityKey<?>)) return false;
-            IdentityKey<?> other = (IdentityKey<?>) obj;
-            return this.instance == other.instance;
-        }
-    }
 }
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java
index 510c561c10f..cf9945cc65b 100644
--- a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyHttpServer.java
@@ -4,8 +4,6 @@ package com.yahoo.jdisc.http.server.jetty;
 import com.google.inject.Inject;
 import com.yahoo.component.ComponentId;
 import com.yahoo.component.provider.ComponentRegistry;
-import com.yahoo.concurrent.DaemonThreadFactory;
-import com.yahoo.container.logging.AccessLog;
 import com.yahoo.container.logging.ConnectionLog;
 import com.yahoo.container.logging.RequestLog;
 import com.yahoo.jdisc.Metric;
@@ -43,8 +41,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
@@ -59,8 +55,6 @@ public class JettyHttpServer extends AbstractServerProvider {
 
     private final static Logger log = Logger.getLogger(JettyHttpServer.class.getName());
 
-    private final ExecutorService janitor;
-
     private final Server server;
     private final List<Integer> listenedPorts = new ArrayList<>();
     private final ServerMetricReporter metricsReporter;
@@ -71,6 +65,7 @@ public class JettyHttpServer extends AbstractServerProvider {
                            ServerConfig serverConfig,
                            ServletPathsConfig servletPathsConfig,
                            FilterBindings filterBindings,
+                           Janitor janitor,
                            ComponentRegistry<ConnectorFactory> connectorFactories,
                            ComponentRegistry<ServletHolder> servletHolders,
                            FilterInvoker filterInvoker,
@@ -95,8 +90,6 @@ public class JettyHttpServer extends AbstractServerProvider {
             listenedPorts.add(connectorConfig.listenPort());
         }
 
-        janitor = newJanitor();
-
         JDiscContext jDiscContext = new JDiscContext(filterBindings,
                                                      container,
                                                      janitor,
@@ -208,15 +201,6 @@ public class JettyHttpServer extends AbstractServerProvider {
         return ports.stream().map(Object::toString).collect(Collectors.joining(":"));
     }
 
-    // Separate threadpool for tasks that cannot be executed on the jdisc default threadpool due to risk of deadlock
-    private static ExecutorService newJanitor() {
-        int threadPoolSize = Math.max(1, Runtime.getRuntime().availableProcessors()/8);
-        log.info("Creating janitor executor with " + threadPoolSize + " threads");
-        return Executors.newFixedThreadPool(
-                threadPoolSize,
-                new DaemonThreadFactory(JettyHttpServer.class.getName() + "-Janitor-"));
-    }
-
     @Override
     public void start() {
         try {
@@ -258,7 +242,6 @@ public class JettyHttpServer extends AbstractServerProvider {
         }
 
         metricsReporter.shutdown();
-        janitor.shutdown();
     }
 
     private boolean isGracefulShutdownEnabled() {
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java
index b4d03385c3b..696fd2d51ad 100644
--- a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java
@@ -12,7 +12,6 @@ import java.util.ArrayList;
 import java.util.Deque;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
 import java.util.function.Consumer;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -54,7 +53,7 @@ public class ServletOutputStreamWriter {
 
     // GuardedBy("state")
     private final ServletOutputStream outputStream;
-    private final Executor executor;
+    private final Janitor janitor;
 
     // GuardedBy("monitor")
     private final Deque<ResponseContentPart> responseContentQueue = new ArrayDeque<>();
@@ -70,9 +69,9 @@ public class ServletOutputStreamWriter {
     final CompletableFuture<Void> finishedFuture = new CompletableFuture<>();
 
 
-    public ServletOutputStreamWriter(ServletOutputStream outputStream, Executor executor, RequestMetricReporter metricReporter) {
+    public ServletOutputStreamWriter(ServletOutputStream outputStream, Janitor janitor, RequestMetricReporter metricReporter) {
         this.outputStream = outputStream;
-        this.executor = executor;
+        this.janitor = janitor;
         this.metricReporter = metricReporter;
     }
 
@@ -96,7 +95,7 @@ public class ServletOutputStreamWriter {
 
         synchronized (monitor) {
             if (state == State.FINISHED_OR_ERROR) {
-                executor.execute(() ->  handler.failed(new IllegalStateException("ContentChannel already closed.")));
+                janitor.scheduleTask(() ->  handler.failed(new IllegalStateException("ContentChannel already closed.")));
                 return;
             }
             responseContentQueue.addLast(new ResponseContentPart(buf, handler));
@@ -207,8 +206,7 @@ public class ServletOutputStreamWriter {
                 runCompletionHandler_logOnExceptions(
                         () -> responseContentPart.handler.failed(failReason));
 
-        executor.execute(
-                () -> failedParts.forEach(failCompletionHandler));
+        janitor.scheduleTask(() -> failedParts.forEach(failCompletionHandler));
     }
 
     private void writeBufferToOutputStream(ResponseContentPart contentPart) throws Throwable {
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java
index 1882448757a..26d74bdccb3 100644
--- a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java
@@ -10,7 +10,6 @@ import javax.servlet.ServletInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -42,7 +41,7 @@ class ServletRequestReader implements ReadListener {
     private final ServletInputStream servletInputStream;
     private final ContentChannel requestContentChannel;
 
-    private final Executor executor;
+    private final Janitor janitor;
     private final RequestMetricReporter metricReporter;
 
     private int bytesRead;
@@ -93,17 +92,17 @@ class ServletRequestReader implements ReadListener {
     public ServletRequestReader(
             ServletInputStream servletInputStream,
             ContentChannel requestContentChannel,
-            Executor executor,
+            Janitor janitor,
             RequestMetricReporter metricReporter) {
 
         Preconditions.checkNotNull(servletInputStream);
         Preconditions.checkNotNull(requestContentChannel);
-        Preconditions.checkNotNull(executor);
+        Preconditions.checkNotNull(janitor);
         Preconditions.checkNotNull(metricReporter);
 
         this.servletInputStream = servletInputStream;
         this.requestContentChannel = requestContentChannel;
-        this.executor = executor;
+        this.janitor = janitor;
         this.metricReporter = metricReporter;
     }
 
@@ -163,7 +162,7 @@ class ServletRequestReader implements ReadListener {
         }
 
         if (shouldCloseRequestContentChannel) {
-            executor.execute(this::closeCompletionHandler_noThrow);
+            janitor.scheduleTask(this::closeCompletionHandler_noThrow);
         }
     }
 
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletResponseController.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletResponseController.java
index 60b7878156f..31fa9e9ebaa 100644
--- a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletResponseController.java
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletResponseController.java
@@ -20,7 +20,6 @@ import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -57,7 +56,7 @@ public class ServletResponseController {
     public ServletResponseController(
             HttpServletRequest servletRequest,
             HttpServletResponse servletResponse,
-            Executor executor,
+            Janitor janitor,
             RequestMetricReporter metricReporter,
             boolean developerMode) throws IOException {
 
@@ -65,7 +64,7 @@ public class ServletResponseController {
         this.servletResponse = servletResponse;
         this.developerMode = developerMode;
         this.servletOutputStreamWriter =
-                new ServletOutputStreamWriter(servletResponse.getOutputStream(), executor, metricReporter);
+                new ServletOutputStreamWriter(servletResponse.getOutputStream(), janitor, metricReporter);
     }
 
 
diff --git a/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/SimpleConcurrentIdentityHashMap.java b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/SimpleConcurrentIdentityHashMap.java
new file mode 100644
index 00000000000..52142e534ba
--- /dev/null
+++ b/container-core/src/main/java/com/yahoo/jdisc/http/server/jetty/SimpleConcurrentIdentityHashMap.java
@@ -0,0 +1,39 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.jdisc.http.server.jetty;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * A simplified {@link ConcurrentMap} using reference-equality to compare keys (similarly to {@link java.util.IdentityHashMap})
+ *
+ * @author bjorncs
+ */
+class SimpleConcurrentIdentityHashMap<K, V> {
+
+    private final ConcurrentMap<IdentityKey<K>, V> wrappedMap = new ConcurrentHashMap<>();
+
+    V get(K key) { return wrappedMap.get(IdentityKey.of(key)); }
+
+    V remove(K key) { return wrappedMap.remove(IdentityKey.of(key)); }
+
+    void put(K key, V value) { wrappedMap.put(IdentityKey.of(key), value); }
+
+    private static class IdentityKey<K> {
+        final K instance;
+
+        IdentityKey(K instance) { this.instance = instance; }
+
+        static <K> IdentityKey<K> of(K instance) { return new IdentityKey<>(instance); }
+
+        @Override public int hashCode() { return System.identityHashCode(instance); }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (this == obj) return true;
+            if (!(obj instanceof IdentityKey<?>)) return false;
+            IdentityKey<?> other = (IdentityKey<?>) obj;
+            return this.instance == other.instance;
+        }
+    }
+}