Record Jetty connection bytes in/out through network listeners (#4514)

before this change we were dumping in / out bytes from the connection upon the connection close which could result in big traffic whenever the connection got closed
after this change we're measuring the bytes through NetworkTrafficListener that hooks in whenever actual in / out bytes are being processed.
This only works on the server-side connection for now. Future versions of Jetty may extend it to client side.

fixes gh-3873

micrometer-core/src/main/java/io/micrometer/core/instrument/binder/jetty/JettyConnectionMetrics.java

@@ -19,7 +19,12 @@
 import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
 import io.micrometer.core.instrument.distribution.TimeWindowMax;
+
+import java.net.Socket;
+import java.nio.ByteBuffer;
+
 import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.NetworkTrafficListener;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.util.component.AbstractLifeCycle;
@@ -47,7 +52,7 @@
  * @author Jon Schneider
  * @since 1.4.0
  */
-public class JettyConnectionMetrics extends AbstractLifeCycle implements Connection.Listener {
+public class JettyConnectionMetrics extends AbstractLifeCycle implements Connection.Listener, NetworkTrafficListener {
 
     private final MeterRegistry registry;
 
@@ -174,12 +179,18 @@ else if (connection.getClass().getName().contains("client")) {
                 .tags(tags)
                 .register(registry));
         }
-
         messagesIn.increment(connection.getMessagesIn());
         messagesOut.increment(connection.getMessagesOut());
+    }
 
-        bytesIn.record(connection.getBytesIn());
-        bytesOut.record(connection.getBytesOut());
+    @Override
+    public void incoming(Socket socket, ByteBuffer bytes) {
+        bytesIn.record(bytes.limit());
+    }
+
+    @Override
+    public void outgoing(Socket socket, ByteBuffer bytes) {
+        bytesOut.record(bytes.limit());
     }
 
     public static void addToAllConnectors(Server server, MeterRegistry registry, Iterable<Tag> tags) {

micrometer-core/src/test/java/io/micrometer/core/instrument/binder/jetty/JettyConnectionMetricsTest.java

@@ -27,8 +27,8 @@
 import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.client.util.StringContentProvider;
 import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.server.NetworkTrafficServerConnector;
 import org.eclipse.jetty.server.Server;
-import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
@@ -44,12 +44,14 @@ class JettyConnectionMetricsTest {
 
     private Server server = new Server(0);
 
-    private ServerConnector connector = new ServerConnector(server);
+    private NetworkTrafficServerConnector connector = new NetworkTrafficServerConnector(server);
 
     private CloseableHttpClient client = HttpClients.createDefault();
 
     void setup() throws Exception {
-        connector.addBean(new JettyConnectionMetrics(registry));
+        JettyConnectionMetrics metrics = new JettyConnectionMetrics(registry);
+        connector.addBean(metrics);
+        connector.addNetworkTrafficListener(metrics);
         server.setConnectors(new Connector[] { connector });
         server.start();
     }
@@ -88,7 +90,7 @@ public void lifeCycleStopped(LifeCycle event) {
         assertThat(latch.await(10, SECONDS)).isTrue();
         assertThat(registry.get("jetty.connections.max").gauge().value()).isEqualTo(2.0);
         assertThat(registry.get("jetty.connections.request").tag("type", "server").timer().count()).isEqualTo(2);
-        assertThat(registry.get("jetty.connections.bytes.in").summary().totalAmount()).isGreaterThan(1);
+        assertThat(registry.get("jetty.connections.bytes.in").summary().totalAmount()).isPositive();
     }
 
     @Test
@@ -116,7 +118,12 @@ public void lifeCycleStopped(LifeCycle event) {
         assertThat(latch.await(10, SECONDS)).isTrue();
         assertThat(registry.get("jetty.connections.max").gauge().value()).isEqualTo(1.0);
         assertThat(registry.get("jetty.connections.request").tag("type", "client").timer().count()).isEqualTo(1);
-        assertThat(registry.get("jetty.connections.bytes.out").summary().totalAmount()).isGreaterThan(1);
+        // assertThat(registry.get("jetty.connections.bytes.out").summary().totalAmount()).isEqualTo(784);
+        // TODO: explain why there is a difference between what we had before and after
+        // the change
+        // assertThat(registry.get("jetty.connections.bytes.out").summary().totalAmount()).isEqualTo(618);
+        // after the changes
+        assertThat(registry.get("jetty.connections.bytes.out").summary().totalAmount()).isPositive();
     }
 
     @Test