Plumb NiftyMetrics into transport layer to allow exporting of netty counters

Summary: SPINiftyMetrics provides low level counters for netty transport layer events such as channel count, bytes written etc. The metrics class was new’ed up low in the transport factory layer but not surfaces to service framework in a way that would allow the counters to be published to fb303. This was was a sev followup task T234357428 to make these available as additonal way to diagnose conneciton issues.

Reviewed By: RayanRal, adolfojunior

Differential Revision: D89578965

fbshipit-source-id: d7b5f99ca1fb133d86953c328cd0adc593d6fe92

Author: j-bahr

third-party/thrift/src/thrift/lib/java/benchmarks/src/main/java/com/facebook/thrift/jmh/ReactiveRpcBenchmarks.java

@@ -25,6 +25,7 @@
 import com.facebook.thrift.example.ping.PingServiceReactiveClient;
 import com.facebook.thrift.example.ping.PingServiceRpcServerHandler;
 import com.facebook.thrift.legacy.server.LegacyServerTransportFactory;
+import com.facebook.thrift.util.SPINiftyMetrics;
 import com.facebook.thrift.util.resources.RpcResources;
 import io.netty.channel.unix.DomainSocketAddress;
 import java.net.InetSocketAddress;
@@ -84,7 +85,7 @@ public void setup(Blackhole bh) {
       LegacyServerTransportFactory factory =
           new LegacyServerTransportFactory(
               new ThriftServerConfig().setSslEnabled(false).setUdsPath("/tmp/jmh.socket"));
-      factory.createServerTransport(socketAddress, serverHandler).block();
+      factory.createServerTransport(socketAddress, serverHandler, new SPINiftyMetrics()).block();
       System.out.println("Ping Service started...");
 
       System.out.println("Connecting Ping Service Client...");

third-party/thrift/src/thrift/lib/java/benchmarks/src/main/java/com/facebook/thrift/runner/MultiUdsReactiveServer.java

@@ -23,6 +23,7 @@
 import com.facebook.thrift.example.ping.PingServiceRpcServerHandler;
 import com.facebook.thrift.legacy.server.LegacyServerTransport;
 import com.facebook.thrift.legacy.server.LegacyServerTransportFactory;
+import com.facebook.thrift.util.SPINiftyMetrics;
 import io.netty.channel.unix.DomainSocketAddress;
 import io.netty.util.ResourceLeakDetector;
 import java.util.ArrayList;
@@ -62,7 +63,9 @@ public static void main(String... args) {
                   .setUdsPath("/tmp/uds_benchmark.socket" + ix));
       DomainSocketAddress socketAddress = new DomainSocketAddress("/tmp/uds_benchmark.socket" + ix);
       LegacyServerTransport transport =
-          transportFactory.createServerTransport(socketAddress, serverHandler).block();
+          transportFactory
+              .createServerTransport(socketAddress, serverHandler, new SPINiftyMetrics())
+              .block();
       System.out.println("creating server listen on uds: " + socketAddress);
       transports.add(transport);
     }

third-party/thrift/src/thrift/lib/java/benchmarks/src/main/java/com/facebook/thrift/runner/UdsReactiveServer.java

@@ -23,6 +23,7 @@
 import com.facebook.thrift.example.ping.PingServiceRpcServerHandler;
 import com.facebook.thrift.legacy.server.LegacyServerTransport;
 import com.facebook.thrift.legacy.server.LegacyServerTransportFactory;
+import com.facebook.thrift.util.SPINiftyMetrics;
 import io.netty.channel.unix.DomainSocketAddress;
 import io.netty.util.ResourceLeakDetector;
 import java.util.Collections;
@@ -48,7 +49,9 @@ public static void main(String... args) {
             new ThriftServerConfig().setSslEnabled(false).setUdsPath("/tmp/uds_benchmark.socket"));
     DomainSocketAddress socketAddress = new DomainSocketAddress("/tmp/uds_benchmark.socket");
     LegacyServerTransport transport =
-        transportFactory.createServerTransport(socketAddress, serverHandler).block();
+        transportFactory
+            .createServerTransport(socketAddress, serverHandler, new SPINiftyMetrics())
+            .block();
 
     LockSupport.park();
   }

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/swift/service/stats/ServerStats.java

@@ -56,6 +56,9 @@ public class ServerStats {
   private static final String ACTIVE_REQUESTS_KEY = "thrift.active_requests.avg";
   private static final String QUEUED_REQUESTS_KEY = "thrift.queued_requests.avg.60";
   // Connection related Counter Keys
+  private static final String CHANNEL_COUNT_KEY = "thrift.channel.count";
+  private static final String BYTES_READ_KEY = "thrift.bytes_read.count";
+  private static final String BYTES_WRITTEN_KEY = "thrift.bytes_written.count";
   private static final String ACCEPTED_CONNS_KEY = "thrift.accepted_connections.count";
   private static final String DROPPED_CONNS_KEY = "thrift.dropped_conns.count";
   private static final String REJECTED_CONNS_KEY = "thrift.rejected_conns.count";
@@ -146,6 +149,11 @@ public Map<String, Long> getCounters() {
       resultCounters.put(QUEUED_REQUESTS_KEY, (long) threadPoolExecutor.getQueue().size());
     }
     if (niftyMetrics != null) {
+      resultCounters.put(CHANNEL_COUNT_KEY, (long) niftyMetrics.getChannelCount());
+
+      resultCounters.put(BYTES_READ_KEY, niftyMetrics.getBytesRead());
+      resultCounters.put(BYTES_WRITTEN_KEY, niftyMetrics.getBytesWritten());
+
       resultCounters.put(ACCEPTED_CONNS_KEY, niftyMetrics.getAcceptedConnections());
       resultCounters.put(
           ACCEPTED_CONNS_KEY + ONE_MINUTE, niftyMetrics.getAcceptedConnectionsOneMin());

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/thrift/legacy/server/LegacyServerTransport.java

@@ -57,15 +57,17 @@ public class LegacyServerTransport implements ServerTransport {
   }
 
   static Mono<LegacyServerTransport> createNewInstance(
-      SocketAddress bindAddress, RpcServerHandler rpcServerHandler, ThriftServerConfig config) {
+      SocketAddress bindAddress,
+      RpcServerHandler rpcServerHandler,
+      ThriftServerConfig config,
+      SPINiftyMetrics serverMetrics) {
 
     return Mono.defer(
         () -> {
           requireNonNull(rpcServerHandler, "methodInvoker is null");
           requireNonNull(config, "config is null");
 
           MonoProcessor<Void> onClose = MonoProcessor.create();
-          SPINiftyMetrics metrics = new SPINiftyMetrics();
 
           Optional<Supplier<SslContext>> sslContext = Optional.empty();
           if (config.isSslEnabled() && bindAddress instanceof InetSocketAddress) {
@@ -80,7 +82,7 @@ static Mono<LegacyServerTransport> createNewInstance(
                   sslContext,
                   config.isAllowPlaintext(),
                   config.isAssumeClientsSupportOutOfOrderResponses(),
-                  metrics,
+                  serverMetrics,
                   config.getConnectionLimit());
 
           EventLoopGroup group = RpcResources.getEventLoopGroup();
@@ -116,7 +118,7 @@ static Mono<LegacyServerTransport> createNewInstance(
           NettyUtil.toMono(channel.closeFuture()).subscribe(onClose);
 
           return NettyUtil.toMono(bind)
-              .thenReturn(new LegacyServerTransport(channel, metrics, onClose));
+              .thenReturn(new LegacyServerTransport(channel, serverMetrics, onClose));
         });
   }
 

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/thrift/legacy/server/LegacyServerTransportFactory.java

@@ -19,6 +19,7 @@
 import com.facebook.swift.service.ThriftServerConfig;
 import com.facebook.thrift.server.RpcServerHandler;
 import com.facebook.thrift.server.ServerTransportFactory;
+import com.facebook.thrift.util.SPINiftyMetrics;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import reactor.core.publisher.Mono;
@@ -32,13 +33,16 @@ public LegacyServerTransportFactory(ThriftServerConfig config) {
 
   @Override
   public Mono<? extends LegacyServerTransport> createServerTransport(
-      SocketAddress bindAddress, RpcServerHandler rpcServerHandler) {
-    return LegacyServerTransport.createNewInstance(bindAddress, rpcServerHandler, config);
+      SocketAddress bindAddress, RpcServerHandler rpcServerHandler, SPINiftyMetrics serverMetrics) {
+    return LegacyServerTransport.createNewInstance(
+        bindAddress, rpcServerHandler, config, serverMetrics);
   }
 
   public Mono<? extends LegacyServerTransport> createServerTransport(
       RpcServerHandler rpcServerHandler) {
     return createServerTransport(
-        new InetSocketAddress("localhost", config.getPort()), rpcServerHandler);
+        new InetSocketAddress("localhost", config.getPort()),
+        rpcServerHandler,
+        new SPINiftyMetrics());
   }
 }

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/thrift/rsocket/server/RSocketServerTransport.java

@@ -44,17 +44,18 @@ public class RSocketServerTransport implements ServerTransport {
   }
 
   static Mono<RSocketServerTransport> createInstance(
-      SocketAddress socketAddress, RpcServerHandler rpcServerHandler, ThriftServerConfig config) {
+      SocketAddress socketAddress,
+      RpcServerHandler rpcServerHandler,
+      ThriftServerConfig config,
+      SPINiftyMetrics serverMetrics) {
     try {
       requireNonNull(rpcServerHandler, "methodInvoker is null");
       requireNonNull(config, "config is null");
 
-      SPINiftyMetrics metrics = new SPINiftyMetrics();
-
       return RSocketServer.create(new ThriftSocketAcceptor(rpcServerHandler))
           .fragment(MAX_FRAME_SIZE)
           .payloadDecoder(PayloadDecoder.ZERO_COPY)
-          .bind(new ReactorServerTransport(socketAddress, config, metrics))
+          .bind(new ReactorServerTransport(socketAddress, config, serverMetrics))
           .map(RSocketServerTransport::new);
     } catch (Exception e) {
       return Mono.error(e);

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/thrift/rsocket/server/RSocketServerTransportFactory.java

@@ -20,6 +20,7 @@
 import com.facebook.thrift.server.RpcServerHandler;
 import com.facebook.thrift.server.ServerTransportFactory;
 import com.facebook.thrift.util.RpcServerUtils;
+import com.facebook.thrift.util.SPINiftyMetrics;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import reactor.core.publisher.Mono;
@@ -34,14 +35,17 @@ public RSocketServerTransportFactory(ThriftServerConfig config) {
 
   @Override
   public Mono<? extends RSocketServerTransport> createServerTransport(
-      SocketAddress bindAddress, RpcServerHandler rpcServerHandler) {
-    return RSocketServerTransport.createInstance(parsePort(bindAddress), rpcServerHandler, config);
+      SocketAddress bindAddress, RpcServerHandler rpcServerHandler, SPINiftyMetrics serverMetrics) {
+    return RSocketServerTransport.createInstance(
+        parsePort(bindAddress), rpcServerHandler, config, serverMetrics);
   }
 
   public Mono<? extends RSocketServerTransport> createServerTransport(
       RpcServerHandler rpcServerHandler) {
     return createServerTransport(
-        new InetSocketAddress("localhost", parsePort(config.getPort())), rpcServerHandler);
+        new InetSocketAddress("localhost", parsePort(config.getPort())),
+        rpcServerHandler,
+        new SPINiftyMetrics());
   }
 
   private int parsePort(int port) {

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/thrift/server/ServerTransportFactory.java

@@ -16,10 +16,11 @@
 
 package com.facebook.thrift.server;
 
+import com.facebook.thrift.util.SPINiftyMetrics;
 import java.net.SocketAddress;
 import reactor.core.publisher.Mono;
 
 public interface ServerTransportFactory<T extends ServerTransport> {
   Mono<? extends T> createServerTransport(
-      SocketAddress bindAddress, RpcServerHandler rpcServerHandler);
+      SocketAddress bindAddress, RpcServerHandler rpcServerHandler, SPINiftyMetrics niftyMetrics);
 }

third-party/thrift/src/thrift/lib/java/runtime/src/main/java/com/facebook/thrift/util/RpcServerUtils.java

@@ -216,15 +216,16 @@ public static Mono<? extends ServerTransport> createServerTransport(
 
     final int port = config.getPort() == 0 ? RpcServerUtils.findFreePort() : config.getPort();
 
+    SPINiftyMetrics serverMetrics = new SPINiftyMetrics();
     if (config.isEnableUDS()) {
       return transportFactory.createServerTransport(
-          new DomainSocketAddress(config.getUdsPath()), rpcServerHandler);
+          new DomainSocketAddress(config.getUdsPath()), rpcServerHandler, serverMetrics);
     } else if (config.isBindAddressEnabled()) {
       return transportFactory.createServerTransport(
-          new InetSocketAddress(config.getBindAddress(), port), rpcServerHandler);
+          new InetSocketAddress(config.getBindAddress(), port), rpcServerHandler, serverMetrics);
     } else {
       return transportFactory.createServerTransport(
-          new InetSocketAddress("localhost", port), rpcServerHandler);
+          new InetSocketAddress("localhost", port), rpcServerHandler, serverMetrics);
     }
   }