Add initial connection log integration with Jetty

7 files changed, 231 insertions, 5 deletions

diff --git a/jdisc_http_service/pom.xml b/jdisc_http_service/pom.xml
index 662cef983dd..1563a1a3d17 100644
--- a/jdisc_http_service/pom.xml
+++ b/jdisc_http_service/pom.xml
@@ -90,6 +90,12 @@
       <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>com.yahoo.vespa</groupId>
+      <artifactId>vespajlib</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
 
 
     <!-- TEST SCOPE -->
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/AccessLogRequestLog.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/AccessLogRequestLog.java
index 5a7d18caad4..32790534f86 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/AccessLogRequestLog.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/AccessLogRequestLog.java
@@ -17,6 +17,7 @@ import java.security.cert.X509Certificate;
 import java.util.List;
 import java.util.Optional;
 import java.util.OptionalInt;
+import java.util.UUID;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -120,6 +121,11 @@ class AccessLogRequestLog extends AbstractLifeCycle implements RequestLog {
                 }
             });
 
+            UUID connectionId = (UUID) request.getAttribute(JettyConnectionLogger.CONNECTION_ID_REQUEST_ATTRIBUTE);
+            if (connectionId != null) {
+                accessLogEntry.setConnectionId(connectionId.toString());
+            }
+
             accessLog.log(accessLogEntry);
         } catch (Exception e) {
             // Catching any exceptions here as it is unclear how Jetty handles exceptions from a RequestLog.
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java
index d93e03fde54..d7ad12a5c64 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java
@@ -64,9 +64,9 @@ public class ConnectorFactory {
         return connectorConfig;
     }
 
-    public ServerConnector createConnector(final Metric metric, final Server server) {
+    public ServerConnector createConnector(final Metric metric, final Server server, JettyConnectionLogger connectionLogger) {
         ServerConnector connector = new JDiscServerConnector(
-                connectorConfig, metric, server, createConnectionFactories(metric).toArray(ConnectionFactory[]::new));
+                connectorConfig, metric, server, connectionLogger, createConnectionFactories(metric).toArray(ConnectionFactory[]::new));
         connector.setPort(connectorConfig.listenPort());
         connector.setName(connectorConfig.name());
         connector.setAcceptQueueSize(connectorConfig.acceptQueueSize());
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscServerConnector.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscServerConnector.java
index 86841458b13..99d0c5c8d8c 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscServerConnector.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JDiscServerConnector.java
@@ -31,7 +31,7 @@ class JDiscServerConnector extends ServerConnector {
     private final String connectorName;
     private final int listenPort;
 
-    JDiscServerConnector(ConnectorConfig config, Metric metric, Server server, ConnectionFactory... factories) {
+    JDiscServerConnector(ConnectorConfig config, Metric metric, Server server, JettyConnectionLogger connectionLogger, ConnectionFactory... factories) {
         super(server, factories);
         this.config = config;
         this.tcpKeepAlive = config.tcpKeepAliveEnabled();
@@ -47,6 +47,7 @@ class JDiscServerConnector extends ServerConnector {
         if (throttlingConfig.enabled()) {
             new ConnectionThrottler(this, throttlingConfig).registerWithConnector();
         }
+        addBean(connectionLogger);
     }
 
     @Override
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java
new file mode 100644
index 00000000000..26c54ff6742
--- /dev/null
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/JettyConnectionLogger.java
@@ -0,0 +1,208 @@
+// 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.yahoo.container.logging.ConnectionLog;
+import com.yahoo.container.logging.ConnectionLogEntry;
+import com.yahoo.jdisc.http.ServerConfig;
+import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.ssl.SslConnection;
+import org.eclipse.jetty.server.HttpChannel;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.util.component.AbstractLifeCycle;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSession;
+import java.net.InetSocketAddress;
+import java.security.cert.X509Certificate;
+import java.time.Instant;
+import java.util.Date;
+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;
+
+/**
+ * Jetty integration for jdisc connection log (TODO link interface)
+ *
+ * @author bjorncs
+ */
+class JettyConnectionLogger extends AbstractLifeCycle implements Connection.Listener, HttpChannel.Listener {
+
+    static final String CONNECTION_ID_REQUEST_ATTRIBUTE = "jdisc.request.connection.id";
+
+    private static final Logger log = Logger.getLogger(JettyConnectionLogger.class.getName());
+
+    private final ConcurrentMap<Connection, AggregatedConnectionInfo> connectionInfo = new ConcurrentHashMap<>();
+    private final boolean enabled;
+    private final ConnectionLog connectionLog;
+
+    JettyConnectionLogger(ServerConfig.ConnectionLog config, ConnectionLog connectionLog) {
+        this.enabled = config.enabled();
+        this.connectionLog = connectionLog;
+        log.log(Level.FINE, () -> "Jetty connection logger is " + (config.enabled() ? "enabled" : "disabled"));
+    }
+
+    //
+    // AbstractLifeCycle methods start
+    //
+    @Override
+    protected void doStop() {
+        if (!enabled) return;
+        log.log(Level.FINE, () -> "Jetty connection logger is stopped");
+    }
+
+    @Override
+    protected void doStart() {
+        if (!enabled) return;
+        log.log(Level.FINE, () -> "Jetty connection logger is started");
+    }
+    //
+    // AbstractLifeCycle methods stop
+    //
+
+    //
+    // Connection.Listener methods start
+    //
+    @Override
+    public void onOpened(Connection connection) {
+        if (!enabled) return;
+        onListenerInvoked("Connection.Listener", "onOpened", "%h", connection);
+        AggregatedConnectionInfo info = new AggregatedConnectionInfo(UUID.randomUUID());
+        synchronized (info.lock()) {
+            EndPoint endpoint = connection.getEndPoint();
+            info.setCreatedAt(endpoint.getCreatedTimeStamp())
+                    .setLocalAddress(endpoint.getLocalAddress())
+                    .setPeerAddress(endpoint.getRemoteAddress());
+            if (connection instanceof SslConnection) {
+                SslConnection sslConnection = (SslConnection) connection;
+                SSLEngine sslEngine = sslConnection.getSSLEngine();
+                SSLSession sslSession = sslEngine.getSession();
+                info.setSslSessionDetails(sslSession);
+            }
+        }
+        connectionInfo.put(connection, info);
+    }
+
+    @Override
+    public void onClosed(Connection connection) {
+        if (!enabled) return;
+        onListenerInvoked("Connection.Listener", "onClosed", "%h", connection);
+        // TODO Decide on handling of connection upgrade where old connection object is closed and replaced by a new (e.g for proxy-protocol auto detection)
+        AggregatedConnectionInfo builder = Objects.requireNonNull(connectionInfo.remove(connection));
+        ConnectionLogEntry logEntry;
+        synchronized (builder.lock()) {
+            logEntry = builder.setBytesReceived(connection.getBytesIn())
+                    .setBytesSent(connection.getBytesOut())
+                    .toLogEntry();
+        }
+        connectionLog.log(logEntry);
+    }
+    //
+    // Connection.Listener methods end
+    //
+
+    //
+    // HttpChannel.Listener methods start
+    //
+    @Override
+    public void onRequestBegin(Request request) {
+        if (!enabled) return;
+        onListenerInvoked("HttpChannel.Listener", "onRequestBegin", "%h", request);
+        Connection connection = request.getHttpChannel().getConnection();
+        AggregatedConnectionInfo info = Objects.requireNonNull(connectionInfo.get(connection));
+        UUID uuid;
+        synchronized (info.lock()) {
+            info.incrementRequests();
+            uuid = info.uuid();
+        }
+        request.setAttribute(CONNECTION_ID_REQUEST_ATTRIBUTE, uuid);
+    }
+
+    @Override
+    public void onResponseBegin(Request request) {
+        if (!enabled) return;
+        onListenerInvoked("HttpChannel.Listener", "onResponseBegin", "%h", request);
+        Connection connection = request.getHttpChannel().getConnection();
+        AggregatedConnectionInfo info = Objects.requireNonNull(connectionInfo.get(connection));
+        synchronized (info.lock()) {
+            info.incrementResponses();
+        }
+    }
+    //
+    // HttpChannel.Listener methods end
+    //
+
+    private void onListenerInvoked(String listenerType, String methodName, String methodArgumentsFormat, Object... methodArguments) {
+        log.log(Level.FINE, () -> String.format(listenerType + "." + methodName + "(" + methodArgumentsFormat + ")", methodArguments));
+    }
+
+    private static class AggregatedConnectionInfo {
+        private final Object monitor = new Object();
+
+        private final UUID uuid;
+
+        private long createdAt;
+        private InetSocketAddress localAddress;
+        private InetSocketAddress peerAddress;
+        private long bytesReceived;
+        private long bytesSent;
+        private long requests;
+        private long responses;
+        private byte[] sslSessionId;
+        private String sslProtocol;
+        private String sslCipherSuite;
+        private String sslPeerSubject;
+        private Date sslPeerNotBefore;
+        private Date sslPeerNotAfter;
+
+        AggregatedConnectionInfo(UUID uuid) {
+            this.uuid = uuid;
+        }
+
+        Object lock() { return monitor; }
+
+        UUID uuid() { return uuid; }
+
+        AggregatedConnectionInfo setCreatedAt(long createdAt) { this.createdAt = createdAt; return this; }
+
+        AggregatedConnectionInfo setLocalAddress(InetSocketAddress localAddress) { this.localAddress = localAddress; return this; }
+
+        AggregatedConnectionInfo setPeerAddress(InetSocketAddress peerAddress) { this.peerAddress = peerAddress; return this; }
+
+        AggregatedConnectionInfo setBytesReceived(long bytesReceived) { this.bytesReceived = bytesReceived; return this; }
+
+        AggregatedConnectionInfo setBytesSent(long bytesSent) { this.bytesSent = bytesSent; return this; }
+
+        AggregatedConnectionInfo incrementRequests() { ++this.requests; return this; }
+
+        AggregatedConnectionInfo incrementResponses() { ++this.responses; return this; }
+
+        AggregatedConnectionInfo setSslSessionDetails(SSLSession session) {
+            this.sslCipherSuite = session.getCipherSuite();
+            this.sslProtocol = session.getProtocol();
+            this.sslSessionId = session.getId();
+            try {
+                this.sslPeerSubject = session.getPeerPrincipal().getName();
+                X509Certificate peerCertificate = (X509Certificate) session.getPeerCertificates()[0];
+                this.sslPeerNotBefore = peerCertificate.getNotAfter();
+                this.sslPeerNotAfter = peerCertificate.getNotAfter();
+            } catch (SSLPeerUnverifiedException e) {
+                // Throw if peer is not authenticated (e.g when client auth is disabled)
+                // JSSE provides no means of checking for client authentication without catching this exception
+            }
+            return this;
+        }
+
+        ConnectionLogEntry toLogEntry() {
+            return ConnectionLogEntry.builder(uuid, Instant.ofEpochMilli(createdAt))
+                    .withPeerAddress(peerAddress.getHostString())
+                    .withPeerPort(peerAddress.getPort())
+                    .build();
+        }
+
+    }
+}
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 bab9694e98b..1e077c32ea1 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
@@ -86,10 +86,11 @@ public class JettyHttpServer extends AbstractServerProvider {
         server.setRequestLog(new AccessLogRequestLog(accessLog, serverConfig.accessLog()));
         setupJmx(server, serverConfig);
         configureJettyThreadpool(server, serverConfig);
+        JettyConnectionLogger connectionLogger = new JettyConnectionLogger(serverConfig.connectionLog(), connectionLog);
 
         for (ConnectorFactory connectorFactory : connectorFactories.allComponents()) {
             ConnectorConfig connectorConfig = connectorFactory.getConnectorConfig();
-            server.addConnector(connectorFactory.createConnector(metric, server));
+            server.addConnector(connectorFactory.createConnector(metric, server, connectionLogger));
             listenedPorts.add(connectorConfig.listenPort());
         }
 
diff --git a/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactoryTest.java b/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactoryTest.java
index 07bffffdbbd..df794c7ecb8 100644
--- a/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactoryTest.java
+++ b/jdisc_http_service/src/test/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactoryTest.java
@@ -3,6 +3,7 @@ package com.yahoo.jdisc.http.server.jetty;
 
 import com.yahoo.jdisc.Metric;
 import com.yahoo.jdisc.http.ConnectorConfig;
+import com.yahoo.jdisc.http.ServerConfig;
 import com.yahoo.jdisc.http.ssl.impl.ConfiguredSslContextFactoryProvider;
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Server;
@@ -27,8 +28,11 @@ public class ConnectorFactoryTest {
         try {
             ConnectorConfig config = new ConnectorConfig(new ConnectorConfig.Builder());
             ConnectorFactory factory = createConnectorFactory(config);
+            JettyConnectionLogger connectionLogger = new JettyConnectionLogger(
+                    new ServerConfig.ConnectionLog.Builder().enabled(false).build(),
+                    new VoidConnectionLog());
             JDiscServerConnector connector =
-                    (JDiscServerConnector)factory.createConnector(new DummyMetric(), server);
+                    (JDiscServerConnector)factory.createConnector(new DummyMetric(), server, connectionLogger);
             server.addConnector(connector);
             server.setHandler(new HelloWorldHandler());
             server.start();