feat: Implement TCP metrics collection and pipeline MetricRecorder (Address PR comments)

Author: AgraVator

api/src/main/java/io/grpc/InternalTcpMetrics.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * TCP Metrics defined to be shared across transport implementations.
+ */
+@Internal
+public final class InternalTcpMetrics {
+
+  private InternalTcpMetrics() {
+  }
+
+  private static final List<String> OPTIONAL_LABELS = Arrays.asList(
+      "network.local.address",
+      "network.local.port",
+      "network.peer.address",
+      "network.peer.port");
+
+  public static final DoubleHistogramMetricInstrument MIN_RTT_INSTRUMENT = 
+      MetricInstrumentRegistry.getDefaultRegistry()
+          .registerDoubleHistogram(
+          "grpc.tcp.min_rtt",
+          "Minimum round-trip time of a TCP connection",
+          "s",
+          getMinRttBuckets(),
+          Collections.emptyList(),
+          OPTIONAL_LABELS,
+          false);
+
+  public static final LongCounterMetricInstrument CONNECTIONS_CREATED_INSTRUMENT = 
+      MetricInstrumentRegistry
+      .getDefaultRegistry()
+      .registerLongCounter(
+          "grpc.tcp.connections_created",
+          "The total number of TCP connections established.",
+          "{connection}",
+          Collections.emptyList(),
+          OPTIONAL_LABELS,
+          false);
+
+  public static final LongUpDownCounterMetricInstrument CONNECTION_COUNT_INSTRUMENT = 
+      MetricInstrumentRegistry
+      .getDefaultRegistry()
+      .registerLongUpDownCounter(
+          "grpc.tcp.connection_count",
+          "The current number of active TCP connections.",
+          "{connection}",
+          Collections.emptyList(),
+          OPTIONAL_LABELS,
+          false);
+
+  public static final LongCounterMetricInstrument PACKETS_RETRANSMITTED_INSTRUMENT = 
+      MetricInstrumentRegistry
+      .getDefaultRegistry()
+      .registerLongCounter(
+          "grpc.tcp.packets_retransmitted",
+          "The total number of packets retransmitted for all TCP connections.",
+          "{packet}",
+          Collections.emptyList(),
+          OPTIONAL_LABELS,
+          false);
+
+  public static final LongCounterMetricInstrument RECURRING_RETRANSMITS_INSTRUMENT = 
+      MetricInstrumentRegistry
+      .getDefaultRegistry()
+      .registerLongCounter(
+          "grpc.tcp.recurring_retransmits",
+          "The total number of times the retransmit timer popped for all TCP"
+              + " connections.",
+          "{timeout}",
+          Collections.emptyList(),
+          OPTIONAL_LABELS,
+          false);
+
+  private static List<Double> getMinRttBuckets() {
+    List<Double> buckets = new ArrayList<>(100);
+    for (int i = 1; i <= 100; i++) {
+      buckets.add(1e-6 * Math.pow(2.0, i * 0.24));
+    }
+    return Collections.unmodifiableList(buckets);
+  }
+}

api/src/main/java/io/grpc/NameResolver.java

@@ -355,7 +355,7 @@ public static final class Args {
     @Nullable private final ChannelLogger channelLogger;
     @Nullable private final Executor executor;
     @Nullable private final String overrideAuthority;
-    @Nullable private final MetricRecorder metricRecorder;
+    private final MetricRecorder metricRecorder;
     @Nullable private final NameResolverRegistry nameResolverRegistry;
     @Nullable private final IdentityHashMap<Key<?>, Object> customArgs;
 
@@ -497,7 +497,6 @@ public String getOverrideAuthority() {
     /**
      * Returns the {@link MetricRecorder} that the channel uses to record metrics.
      */
-    @Nullable
     public MetricRecorder getMetricRecorder() {
       return metricRecorder;
     }

core/src/main/java/io/grpc/internal/ClientTransportFactory.java

@@ -24,6 +24,7 @@
 import io.grpc.ChannelCredentials;
 import io.grpc.ChannelLogger;
 import io.grpc.HttpConnectProxiedSocketAddress;
+import io.grpc.MetricRecorder;
 import java.io.Closeable;
 import java.net.SocketAddress;
 import java.util.Collection;
@@ -91,6 +92,7 @@ final class ClientTransportOptions {
     private Attributes eagAttributes = Attributes.EMPTY;
     @Nullable private String userAgent;
     @Nullable private HttpConnectProxiedSocketAddress connectProxiedSocketAddr;
+    @Nullable private MetricRecorder metricRecorder;
 
     public ChannelLogger getChannelLogger() {
       return channelLogger;
@@ -101,6 +103,16 @@ public ClientTransportOptions setChannelLogger(ChannelLogger channelLogger) {
       return this;
     }
 
+    @Nullable
+    public MetricRecorder getMetricRecorder() {
+      return metricRecorder;
+    }
+
+    public ClientTransportOptions setMetricRecorder(@Nullable MetricRecorder metricRecorder) {
+      this.metricRecorder = metricRecorder;
+      return this;
+    }
+
     public String getAuthority() {
       return authority;
     }

core/src/main/java/io/grpc/internal/InternalServer.java

@@ -58,7 +58,8 @@ public interface InternalServer {
   /**
    * Returns the first listen socket stats of this server. May return {@code null}.
    */
-  @Nullable InternalInstrumented<SocketStats> getListenSocketStats();
+  @Nullable
+  InternalInstrumented<SocketStats> getListenSocketStats();
 
   /**
    * Returns a list of listening socket addresses.  May change after {@link #start(ServerListener)}
@@ -69,6 +70,7 @@ public interface InternalServer {
   /**
    * Returns a list of listen socket stats of this server. May return {@code null}.
    */
-  @Nullable List<InternalInstrumented<SocketStats>> getListenSocketStatsList();
+  @Nullable
+  List<InternalInstrumented<SocketStats>> getListenSocketStatsList();
 
 }

core/src/main/java/io/grpc/internal/InternalSubchannel.java

@@ -80,6 +80,7 @@ final class InternalSubchannel implements InternalInstrumented<ChannelStats>, Tr
   private final InternalChannelz channelz;
   private final CallTracer callsTracer;
   private final ChannelTracer channelTracer;
+  private final MetricRecorder metricRecorder;
   private final ChannelLogger channelLogger;
   private final boolean reconnectDisabled;
 
@@ -191,6 +192,7 @@ protected void handleNotInUse() {
     this.scheduledExecutor = scheduledExecutor;
     this.connectingTimer = stopwatchSupplier.get();
     this.syncContext = syncContext;
+    this.metricRecorder = metricRecorder;
     this.callback = callback;
     this.channelz = channelz;
     this.callsTracer = callsTracer;
@@ -265,6 +267,7 @@ private void startNewTransport() {
           .setAuthority(eagChannelAuthority != null ? eagChannelAuthority : authority)
           .setEagAttributes(currentEagAttributes)
           .setUserAgent(userAgent)
+          .setMetricRecorder(metricRecorder)
           .setHttpConnectProxiedSocketAddress(proxiedAddr);
     TransportLogger transportLogger = new TransportLogger();
     // In case the transport logs in the constructor, use the subchannel logId

core/src/main/java/io/grpc/internal/ServerImpl.java

@@ -143,6 +143,7 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
       InternalServer transportServer,
       Context rootContext) {
     this.executorPool = Preconditions.checkNotNull(builder.executorPool, "executorPool");
+
     this.registry = Preconditions.checkNotNull(builder.registryBuilder.build(), "registryBuilder");
     this.fallbackRegistry =
         Preconditions.checkNotNull(builder.fallbackRegistry, "fallbackRegistry");

core/src/main/java/io/grpc/internal/ServerImplBuilder.java

@@ -31,6 +31,9 @@
 import io.grpc.HandlerRegistry;
 import io.grpc.InternalChannelz;
 import io.grpc.InternalConfiguratorRegistry;
+import io.grpc.MetricInstrumentRegistry;
+import io.grpc.MetricRecorder;
+import io.grpc.MetricSink;
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 import io.grpc.ServerCallExecutorSupplier;
@@ -80,6 +83,7 @@ public static ServerBuilder<?> forPort(int port) {
   final List<ServerTransportFilter> transportFilters = new ArrayList<>();
   final List<ServerInterceptor> interceptors = new ArrayList<>();
   private final List<ServerStreamTracer.Factory> streamTracerFactories = new ArrayList<>();
+  final List<MetricSink> metricSinks = new ArrayList<>();
   private final ClientTransportServersBuilder clientTransportServersBuilder;
   HandlerRegistry fallbackRegistry = DEFAULT_FALLBACK_REGISTRY;
   ObjectPool<? extends Executor> executorPool = DEFAULT_EXECUTOR_POOL;
@@ -104,7 +108,8 @@ public static ServerBuilder<?> forPort(int port) {
    */
   public interface ClientTransportServersBuilder {
     InternalServer buildClientTransportServers(
-        List<? extends ServerStreamTracer.Factory> streamTracerFactories);
+        List<? extends ServerStreamTracer.Factory> streamTracerFactories,
+        MetricRecorder metricRecorder);
   }
 
   /**
@@ -157,6 +162,14 @@ public ServerImplBuilder intercept(ServerInterceptor interceptor) {
     return this;
   }
 
+  /**
+   * Adds a MetricSink to the server.
+   */
+  public ServerImplBuilder addMetricSink(MetricSink metricSink) {
+    metricSinks.add(checkNotNull(metricSink, "metricSink"));
+    return this;
+  }
+
   @Override
   public ServerImplBuilder addStreamTracerFactory(ServerStreamTracer.Factory factory) {
     streamTracerFactories.add(checkNotNull(factory, "factory"));
@@ -241,8 +254,11 @@ public void setDeadlineTicker(Deadline.Ticker ticker) {
 
   @Override
   public Server build() {
+    MetricRecorder metricRecorder = new MetricRecorderImpl(metricSinks,
+        MetricInstrumentRegistry.getDefaultRegistry());
     return new ServerImpl(this,
-        clientTransportServersBuilder.buildClientTransportServers(getTracerFactories()),
+        clientTransportServersBuilder.buildClientTransportServers(
+                getTracerFactories(), metricRecorder),
         Context.ROOT);
   }
 

core/src/test/java/io/grpc/internal/ServerImplBuilderTest.java

@@ -73,7 +73,8 @@ public void setUp() throws Exception {
         new ClientTransportServersBuilder() {
           @Override
           public InternalServer buildClientTransportServers(
-              List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
+              List<? extends ServerStreamTracer.Factory> streamTracerFactories,
+              io.grpc.MetricRecorder metricRecorder) {
             throw new UnsupportedOperationException();
           }
         });
@@ -139,7 +140,7 @@ public static final class StaticTestingClassLoaderCallsGet implements Runnable {
     public void run() {
       ServerImplBuilder builder =
           new ServerImplBuilder(
-              streamTracerFactories -> {
+              (streamTracerFactories, metricRecorder) -> {
                 throw new UnsupportedOperationException();
               });
       assertThat(builder.getTracerFactories()).hasSize(2);
@@ -169,7 +170,7 @@ public void configureServerBuilder(ServerBuilder<?> builder) {
           }));
       ServerImplBuilder builder =
           new ServerImplBuilder(
-              streamTracerFactories -> {
+              (streamTracerFactories, metricRecorder) -> {
                 throw new UnsupportedOperationException();
               });
       assertThat(builder.getTracerFactories()).containsExactly(DUMMY_USER_TRACER);
@@ -192,7 +193,7 @@ public void run() {
       InternalConfiguratorRegistry.setConfigurators(Collections.emptyList());
       ServerImplBuilder builder =
           new ServerImplBuilder(
-              streamTracerFactories -> {
+              (streamTracerFactories, metricRecorder) -> {
                 throw new UnsupportedOperationException();
               });
       assertThat(builder.getTracerFactories()).isEmpty();

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -206,7 +206,8 @@ public void startUp() throws IOException {
         new ClientTransportServersBuilder() {
           @Override
           public InternalServer buildClientTransportServers(
-              List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
+              List<? extends ServerStreamTracer.Factory> streamTracerFactories,
+              io.grpc.MetricRecorder metricRecorder) {
             throw new UnsupportedOperationException();
           }
         });

inprocess/src/main/java/io/grpc/inprocess/InProcessServerBuilder.java

@@ -120,7 +120,8 @@ private InProcessServerBuilder(SocketAddress listenAddress) {
     final class InProcessClientTransportServersBuilder implements ClientTransportServersBuilder {
       @Override
       public InternalServer buildClientTransportServers(
-          List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
+          List<? extends ServerStreamTracer.Factory> streamTracerFactories,
+          io.grpc.MetricRecorder metricRecorder) {
         return buildTransportServers(streamTracerFactories);
       }
     }