openTelemetry: add tcp metrics (#12652)

Implements [A80](grpc/proposal#519)

Author: AgraVator

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

@@ -201,6 +201,12 @@ public Server build() {
     return delegate().build();
   }
 
+  @Override
+  public T addMetricSink(MetricSink metricSink) {
+    delegate().addMetricSink(metricSink);
+    return thisT();
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();

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

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2026 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.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * TCP Metrics defined to be shared across transport implementations.
+ * These metrics and their definitions are specified in
+ * <a href=
+ * "https://github.com/grpc/proposal/blob/master/A80-tcp-metrics.md">gRFC
+ * A80</a>.
+ */
+@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",
+              Collections.emptyList(),
+              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);
+
+}

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;
 
@@ -369,7 +369,8 @@ private Args(Builder builder) {
       this.channelLogger = builder.channelLogger;
       this.executor = builder.executor;
       this.overrideAuthority = builder.overrideAuthority;
-      this.metricRecorder = builder.metricRecorder;
+      this.metricRecorder = builder.metricRecorder != null ? builder.metricRecorder
+          : new MetricRecorder() {};
       this.nameResolverRegistry = builder.nameResolverRegistry;
       this.customArgs = cloneCustomArgs(builder.customArgs);
     }
@@ -497,7 +498,6 @@ public String getOverrideAuthority() {
     /**
      * Returns the {@link MetricRecorder} that the channel uses to record metrics.
      */
-    @Nullable
     public MetricRecorder getMetricRecorder() {
       return metricRecorder;
     }
@@ -680,7 +680,7 @@ public <T> Builder setArg(Key<T> key, T value) {
        * See {@link Args#getMetricRecorder()}. This is an optional field.
        */
       public Builder setMetricRecorder(MetricRecorder metricRecorder) {
-        this.metricRecorder = metricRecorder;
+        this.metricRecorder = checkNotNull(metricRecorder, "metricRecorder");
         return this;
       }
 

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

@@ -435,6 +435,17 @@ public T setBinaryLog(BinaryLog binaryLog) {
    */
   public abstract Server build();
 
+  /**
+   * Adds a metric sink to the server.
+   *
+   * @param metricSink the metric sink to add.
+   * @return this
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12693")
+  public T addMetricSink(MetricSink metricSink) {
+    return thisT();
+  }
+
   /**
    * Returns the correctly typed version of the builder.
    */

binder/src/main/java/io/grpc/binder/BinderServerBuilder.java

@@ -68,7 +68,7 @@ private BinderServerBuilder(
 
     serverImplBuilder =
         new ServerImplBuilder(
-            streamTracerFactories -> {
+            (streamTracerFactories, metricRecorder) -> {
               internalBuilder.setStreamTracerFactories(streamTracerFactories);
               BinderServer server = internalBuilder.build();
               BinderInternal.setIBinder(binderReceiver, server.getHostBinder());

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,8 @@ final class ClientTransportOptions {
     private Attributes eagAttributes = Attributes.EMPTY;
     @Nullable private String userAgent;
     @Nullable private HttpConnectProxiedSocketAddress connectProxiedSocketAddr;
+    private MetricRecorder metricRecorder = new MetricRecorder() {
+    };
 
     public ChannelLogger getChannelLogger() {
       return channelLogger;
@@ -101,6 +104,15 @@ public ClientTransportOptions setChannelLogger(ChannelLogger channelLogger) {
       return this;
     }
 
+    public MetricRecorder getMetricRecorder() {
+      return metricRecorder;
+    }
+
+    public ClientTransportOptions setMetricRecorder(MetricRecorder metricRecorder) {
+      this.metricRecorder = Preconditions.checkNotNull(metricRecorder, "metricRecorder");
+      return this;
+    }
+
     public String getAuthority() {
       return authority;
     }

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/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,15 @@ public ServerImplBuilder intercept(ServerInterceptor interceptor) {
     return this;
   }
 
+  /**
+   * Adds a MetricSink to the server.
+   */
+  @Override
+  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 +255,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

@@ -22,6 +22,9 @@
 import io.grpc.InternalConfigurator;
 import io.grpc.InternalConfiguratorRegistry;
 import io.grpc.Metadata;
+import io.grpc.MetricRecorder;
+import io.grpc.MetricSink;
+import io.grpc.NoopMetricSink;
 import io.grpc.ServerBuilder;
 import io.grpc.ServerCall;
 import io.grpc.ServerCallHandler;
@@ -73,7 +76,8 @@ public void setUp() throws Exception {
         new ClientTransportServersBuilder() {
           @Override
           public InternalServer buildClientTransportServers(
-              List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
+              List<? extends ServerStreamTracer.Factory> streamTracerFactories,
+              MetricRecorder metricRecorder) {
             throw new UnsupportedOperationException();
           }
         });
@@ -128,6 +132,13 @@ public void getTracerFactories_disableBoth() {
     assertThat(factories).containsExactly(DUMMY_USER_TRACER);
   }
 
+  @Test
+  public void addMetricSink_addsToSinks() {
+    MetricSink noopMetricSink = new NoopMetricSink();
+    builder.addMetricSink(noopMetricSink);
+    assertThat(builder.metricSinks).containsExactly(noopMetricSink);
+  }
+
   @Test
   public void getTracerFactories_callsGet() throws Exception {
     Class<?> runnable = classLoader.loadClass(StaticTestingClassLoaderCallsGet.class.getName());
@@ -139,7 +150,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 +180,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 +203,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

@@ -65,6 +65,7 @@
 import io.grpc.InternalServerInterceptors;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
+import io.grpc.MetricRecorder;
 import io.grpc.ServerCall;
 import io.grpc.ServerCall.Listener;
 import io.grpc.ServerCallExecutorSupplier;
@@ -206,7 +207,8 @@ public void startUp() throws IOException {
         new ClientTransportServersBuilder() {
           @Override
           public InternalServer buildClientTransportServers(
-              List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
+              List<? extends ServerStreamTracer.Factory> streamTracerFactories,
+              MetricRecorder metricRecorder) {
             throw new UnsupportedOperationException();
           }
         });