diff --git a/api/src/main/java/io/grpc/ForwardingServerBuilder.java b/api/src/main/java/io/grpc/ForwardingServerBuilder.java index 9cef7cfa331..d1f183dd824 100644 --- a/api/src/main/java/io/grpc/ForwardingServerBuilder.java +++ b/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(); diff --git a/api/src/main/java/io/grpc/InternalTcpMetrics.java b/api/src/main/java/io/grpc/InternalTcpMetrics.java new file mode 100644 index 00000000000..dd1048ff765 --- /dev/null +++ b/api/src/main/java/io/grpc/InternalTcpMetrics.java @@ -0,0 +1,104 @@ +/* + * 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.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 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 getMinRttBuckets() { + List 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); + } +} diff --git a/api/src/main/java/io/grpc/NameResolver.java b/api/src/main/java/io/grpc/NameResolver.java index 53dbc5d6888..3494eab93d0 100644 --- a/api/src/main/java/io/grpc/NameResolver.java +++ b/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, 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; } diff --git a/api/src/main/java/io/grpc/ServerBuilder.java b/api/src/main/java/io/grpc/ServerBuilder.java index cd1cddbb93f..7dbc7bf6077 100644 --- a/api/src/main/java/io/grpc/ServerBuilder.java +++ b/api/src/main/java/io/grpc/ServerBuilder.java @@ -435,6 +435,16 @@ 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 + */ + public T addMetricSink(MetricSink metricSink) { + throw new UnsupportedOperationException(); + } + /** * Returns the correctly typed version of the builder. */ diff --git a/core/src/main/java/io/grpc/internal/ClientTransportFactory.java b/core/src/main/java/io/grpc/internal/ClientTransportFactory.java index 6c10ced4652..6023fb14aa9 100644 --- a/core/src/main/java/io/grpc/internal/ClientTransportFactory.java +++ b/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; } diff --git a/core/src/main/java/io/grpc/internal/InternalServer.java b/core/src/main/java/io/grpc/internal/InternalServer.java index a6079081233..fdd4f32d0d8 100644 --- a/core/src/main/java/io/grpc/internal/InternalServer.java +++ b/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 getListenSocketStats(); + @Nullable + InternalInstrumented 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> getListenSocketStatsList(); + @Nullable + List> getListenSocketStatsList(); } diff --git a/core/src/main/java/io/grpc/internal/InternalSubchannel.java b/core/src/main/java/io/grpc/internal/InternalSubchannel.java index 7a48bf642fe..ce31921e316 100644 --- a/core/src/main/java/io/grpc/internal/InternalSubchannel.java +++ b/core/src/main/java/io/grpc/internal/InternalSubchannel.java @@ -80,6 +80,7 @@ final class InternalSubchannel implements InternalInstrumented, 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 diff --git a/core/src/main/java/io/grpc/internal/ServerImpl.java b/core/src/main/java/io/grpc/internal/ServerImpl.java index dc0709e1fb8..20e4a06ac38 100644 --- a/core/src/main/java/io/grpc/internal/ServerImpl.java +++ b/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"); diff --git a/core/src/main/java/io/grpc/internal/ServerImplBuilder.java b/core/src/main/java/io/grpc/internal/ServerImplBuilder.java index f6566e067db..62a0e66f314 100644 --- a/core/src/main/java/io/grpc/internal/ServerImplBuilder.java +++ b/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 transportFilters = new ArrayList<>(); final List interceptors = new ArrayList<>(); private final List streamTracerFactories = new ArrayList<>(); + final List metricSinks = new ArrayList<>(); private final ClientTransportServersBuilder clientTransportServersBuilder; HandlerRegistry fallbackRegistry = DEFAULT_FALLBACK_REGISTRY; ObjectPool executorPool = DEFAULT_EXECUTOR_POOL; @@ -104,7 +108,8 @@ public static ServerBuilder forPort(int port) { */ public interface ClientTransportServersBuilder { InternalServer buildClientTransportServers( - List streamTracerFactories); + List 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); } diff --git a/core/src/test/java/io/grpc/internal/ServerImplBuilderTest.java b/core/src/test/java/io/grpc/internal/ServerImplBuilderTest.java index 7ad7f15f358..54c2d6ef8b1 100644 --- a/core/src/test/java/io/grpc/internal/ServerImplBuilderTest.java +++ b/core/src/test/java/io/grpc/internal/ServerImplBuilderTest.java @@ -18,10 +18,13 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; import io.grpc.InternalConfigurator; import io.grpc.InternalConfiguratorRegistry; import io.grpc.Metadata; +import io.grpc.MetricRecorder; +import io.grpc.MetricSink; 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 streamTracerFactories) { + List 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 mockSink = mock(MetricSink.class); + builder.addMetricSink(mockSink); + assertThat(builder.metricSinks).containsExactly(mockSink); + } + @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(); diff --git a/core/src/test/java/io/grpc/internal/ServerImplTest.java b/core/src/test/java/io/grpc/internal/ServerImplTest.java index 0f18efe078c..3405cb9bb0c 100644 --- a/core/src/test/java/io/grpc/internal/ServerImplTest.java +++ b/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 streamTracerFactories) { + List streamTracerFactories, + MetricRecorder metricRecorder) { throw new UnsupportedOperationException(); } }); diff --git a/inprocess/src/main/java/io/grpc/inprocess/InProcessServerBuilder.java b/inprocess/src/main/java/io/grpc/inprocess/InProcessServerBuilder.java index 190f67603c3..b2004426aae 100644 --- a/inprocess/src/main/java/io/grpc/inprocess/InProcessServerBuilder.java +++ b/inprocess/src/main/java/io/grpc/inprocess/InProcessServerBuilder.java @@ -24,6 +24,7 @@ import io.grpc.ExperimentalApi; import io.grpc.ForwardingServerBuilder; import io.grpc.Internal; +import io.grpc.MetricRecorder; import io.grpc.ServerBuilder; import io.grpc.ServerStreamTracer; import io.grpc.internal.FixedObjectPool; @@ -120,7 +121,8 @@ private InProcessServerBuilder(SocketAddress listenAddress) { final class InProcessClientTransportServersBuilder implements ClientTransportServersBuilder { @Override public InternalServer buildClientTransportServers( - List streamTracerFactories) { + List streamTracerFactories, + MetricRecorder metricRecorder) { return buildTransportServers(streamTracerFactories); } } diff --git a/netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java b/netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java index 258aa15b005..e64f1065681 100644 --- a/netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java +++ b/netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java @@ -856,6 +856,7 @@ public void run() { localSocketPicker, channelLogger, useGetForSafeMethods, + options.getMetricRecorder(), Ticker.systemTicker()); return transport; } diff --git a/netty/src/main/java/io/grpc/netty/NettyClientHandler.java b/netty/src/main/java/io/grpc/netty/NettyClientHandler.java index 8ebf89842ad..5615f5ed75a 100644 --- a/netty/src/main/java/io/grpc/netty/NettyClientHandler.java +++ b/netty/src/main/java/io/grpc/netty/NettyClientHandler.java @@ -30,6 +30,7 @@ import io.grpc.InternalChannelz; import io.grpc.InternalStatus; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.Status; import io.grpc.StatusException; import io.grpc.internal.ClientStreamListener.RpcProgress; @@ -123,6 +124,7 @@ class NettyClientHandler extends AbstractNettyHandler { private final Supplier stopwatchFactory; private final TransportTracer transportTracer; private final Attributes eagAttributes; + private final TcpMetrics.Tracker tcpMetrics; private final String authority; private final InUseStateAggregator inUseState = new InUseStateAggregator() { @@ -164,7 +166,8 @@ static NettyClientHandler newHandler( Attributes eagAttributes, String authority, ChannelLogger negotiationLogger, - Ticker ticker) { + Ticker ticker, + MetricRecorder metricRecorder) { Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive"); Http2HeadersDecoder headersDecoder = new GrpcHttp2ClientHeadersDecoder(maxHeaderListSize); Http2FrameReader frameReader = new DefaultHttp2FrameReader(headersDecoder); @@ -194,7 +197,8 @@ static NettyClientHandler newHandler( eagAttributes, authority, negotiationLogger, - ticker); + ticker, + metricRecorder); } @VisibleForTesting @@ -214,7 +218,8 @@ static NettyClientHandler newHandler( Attributes eagAttributes, String authority, ChannelLogger negotiationLogger, - Ticker ticker) { + Ticker ticker, + MetricRecorder metricRecorder) { Preconditions.checkNotNull(connection, "connection"); Preconditions.checkNotNull(frameReader, "frameReader"); Preconditions.checkNotNull(lifecycleManager, "lifecycleManager"); @@ -269,7 +274,8 @@ static NettyClientHandler newHandler( pingCounter, ticker, maxHeaderListSize, - softLimitHeaderListSize); + softLimitHeaderListSize, + metricRecorder); } private NettyClientHandler( @@ -288,7 +294,8 @@ private NettyClientHandler( PingLimiter pingLimiter, Ticker ticker, int maxHeaderListSize, - int softLimitHeaderListSize) { + int softLimitHeaderListSize, + MetricRecorder metricRecorder) { super( /* channelUnused= */ null, decoder, @@ -350,6 +357,7 @@ public void onStreamClosed(Http2Stream stream) { } } }); + this.tcpMetrics = new TcpMetrics.Tracker(metricRecorder); } /** @@ -478,6 +486,7 @@ private void onRstStreamRead(int streamId, long errorCode) { @Override public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { + tcpMetrics.recordTcpInfo(ctx.channel()); logger.fine("Network channel being closed by the application."); if (ctx.channel().isActive()) { // Ignore notification that the socket was closed lifecycleManager.notifyShutdown( @@ -490,10 +499,17 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce /** * Handler for the Channel shutting down. */ + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + tcpMetrics.channelActive(ctx.channel()); + super.channelActive(ctx); + } + @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { try { logger.fine("Network channel is closed"); + tcpMetrics.channelInactive(ctx.channel()); Status status = Status.UNAVAILABLE.withDescription("Network closed for unknown reason"); lifecycleManager.notifyShutdown(status, SimpleDisconnectError.UNKNOWN); final Status streamStatus; diff --git a/netty/src/main/java/io/grpc/netty/NettyClientTransport.java b/netty/src/main/java/io/grpc/netty/NettyClientTransport.java index 53914b3c877..6585df42df3 100644 --- a/netty/src/main/java/io/grpc/netty/NettyClientTransport.java +++ b/netty/src/main/java/io/grpc/netty/NettyClientTransport.java @@ -34,6 +34,7 @@ import io.grpc.InternalLogId; import io.grpc.Metadata; import io.grpc.MethodDescriptor; +import io.grpc.MetricRecorder; import io.grpc.Status; import io.grpc.internal.ClientStream; import io.grpc.internal.ConnectionClientTransport; @@ -108,6 +109,7 @@ class NettyClientTransport implements ConnectionClientTransport, private final ChannelLogger channelLogger; private final boolean useGetForSafeMethods; private final Ticker ticker; + private final MetricRecorder metricRecorder; NettyClientTransport( @@ -132,6 +134,7 @@ class NettyClientTransport implements ConnectionClientTransport, LocalSocketPicker localSocketPicker, ChannelLogger channelLogger, boolean useGetForSafeMethods, + MetricRecorder metricRecorder, Ticker ticker) { this.negotiator = Preconditions.checkNotNull(negotiator, "negotiator"); @@ -159,6 +162,7 @@ class NettyClientTransport implements ConnectionClientTransport, this.logId = InternalLogId.allocate(getClass(), remoteAddress.toString()); this.channelLogger = Preconditions.checkNotNull(channelLogger, "channelLogger"); this.useGetForSafeMethods = useGetForSafeMethods; + this.metricRecorder = metricRecorder; this.ticker = Preconditions.checkNotNull(ticker, "ticker"); } @@ -251,7 +255,8 @@ public Runnable start(Listener transportListener) { eagAttributes, authorityString, channelLogger, - ticker); + ticker, + metricRecorder); ChannelHandler negotiationHandler = negotiator.newHandler(handler); diff --git a/netty/src/main/java/io/grpc/netty/NettyServer.java b/netty/src/main/java/io/grpc/netty/NettyServer.java index 1cf67ea25ca..80c45b490f8 100644 --- a/netty/src/main/java/io/grpc/netty/NettyServer.java +++ b/netty/src/main/java/io/grpc/netty/NettyServer.java @@ -21,6 +21,7 @@ import static io.netty.channel.ChannelOption.ALLOCATOR; import static io.netty.channel.ChannelOption.SO_KEEPALIVE; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.MoreObjects; import com.google.common.base.Preconditions; import com.google.common.util.concurrent.ListenableFuture; @@ -31,6 +32,7 @@ import io.grpc.InternalInstrumented; import io.grpc.InternalLogId; import io.grpc.InternalWithLogId; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.InternalServer; import io.grpc.internal.ObjectPool; @@ -67,6 +69,7 @@ import java.util.concurrent.Callable; import java.util.logging.Level; import java.util.logging.Logger; +import javax.annotation.Nullable; /** * Netty-based server implementation. @@ -93,6 +96,7 @@ class NettyServer implements InternalServer, InternalWithLogId { private final int maxMessageSize; private final int maxHeaderListSize; private final int softLimitHeaderListSize; + private MetricRecorder metricRecorder; private final long keepAliveTimeInNanos; private final long keepAliveTimeoutInNanos; private final long maxConnectionIdleInNanos; @@ -136,8 +140,10 @@ class NettyServer implements InternalServer, InternalWithLogId { long maxConnectionAgeInNanos, long maxConnectionAgeGraceInNanos, boolean permitKeepAliveWithoutCalls, long permitKeepAliveTimeInNanos, int maxRstCount, long maxRstPeriodNanos, - Attributes eagAttributes, InternalChannelz channelz) { + Attributes eagAttributes, InternalChannelz channelz, + MetricRecorder metricRecorder) { this.addresses = checkNotNull(addresses, "addresses"); + this.metricRecorder = metricRecorder; this.channelFactory = checkNotNull(channelFactory, "channelFactory"); checkNotNull(channelOptions, "channelOptions"); this.channelOptions = new HashMap, Object>(channelOptions); @@ -172,6 +178,13 @@ class NettyServer implements InternalServer, InternalWithLogId { this.channelz = Preconditions.checkNotNull(channelz); this.logId = InternalLogId.allocate(getClass(), addresses.isEmpty() ? "No address" : String.valueOf(addresses)); + this.metricRecorder = metricRecorder; + } + + @VisibleForTesting + @Nullable + MetricRecorder getMetricRecorder() { + return metricRecorder; } @Override @@ -272,7 +285,8 @@ public void initChannel(Channel ch) { permitKeepAliveTimeInNanos, maxRstCount, maxRstPeriodNanos, - eagAttributes); + eagAttributes, + metricRecorder); ServerTransportListener transportListener; // This is to order callbacks on the listener, not to guard access to channel. synchronized (NettyServer.this) { diff --git a/netty/src/main/java/io/grpc/netty/NettyServerBuilder.java b/netty/src/main/java/io/grpc/netty/NettyServerBuilder.java index eb3a6d9b538..21e1a063700 100644 --- a/netty/src/main/java/io/grpc/netty/NettyServerBuilder.java +++ b/netty/src/main/java/io/grpc/netty/NettyServerBuilder.java @@ -32,6 +32,7 @@ import io.grpc.ExperimentalApi; import io.grpc.ForwardingServerBuilder; import io.grpc.Internal; +import io.grpc.MetricSink; import io.grpc.ServerBuilder; import io.grpc.ServerCredentials; import io.grpc.ServerStreamTracer; @@ -164,8 +165,9 @@ public static NettyServerBuilder forAddress(SocketAddress address, ServerCredent private final class NettyClientTransportServersBuilder implements ClientTransportServersBuilder { @Override public InternalServer buildClientTransportServers( - List streamTracerFactories) { - return buildTransportServers(streamTracerFactories); + List streamTracerFactories, + io.grpc.MetricRecorder metricRecorder) { + return buildTransportServers(streamTracerFactories, metricRecorder); } } @@ -703,8 +705,10 @@ void eagAttributes(Attributes eagAttributes) { this.eagAttributes = checkNotNull(eagAttributes, "eagAttributes"); } + @VisibleForTesting NettyServer buildTransportServers( - List streamTracerFactories) { + List streamTracerFactories, + io.grpc.MetricRecorder metricRecorder) { assertEventLoopsAndChannelType(); ProtocolNegotiator negotiator = protocolNegotiatorFactory.newNegotiator( @@ -737,7 +741,8 @@ NettyServer buildTransportServers( maxRstCount, maxRstPeriodNanos, eagAttributes, - this.serverImplBuilder.getChannelz()); + this.serverImplBuilder.getChannelz(), + metricRecorder); } @VisibleForTesting @@ -760,6 +765,13 @@ NettyServerBuilder setTransportTracerFactory(TransportTracer.Factory transportTr return this; } + @CanIgnoreReturnValue + @Override + public NettyServerBuilder addMetricSink(MetricSink metricSink) { + serverImplBuilder.addMetricSink(metricSink); + return this; + } + @CanIgnoreReturnValue @Override public NettyServerBuilder useTransportSecurity(File certChain, File privateKey) { diff --git a/netty/src/main/java/io/grpc/netty/NettyServerHandler.java b/netty/src/main/java/io/grpc/netty/NettyServerHandler.java index 036fde55e2c..53b0f3e0dfd 100644 --- a/netty/src/main/java/io/grpc/netty/NettyServerHandler.java +++ b/netty/src/main/java/io/grpc/netty/NettyServerHandler.java @@ -42,6 +42,7 @@ import io.grpc.InternalMetadata; import io.grpc.InternalStatus; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.Status; import io.grpc.internal.GrpcUtil; @@ -127,6 +128,7 @@ class NettyServerHandler extends AbstractNettyHandler { private final Http2Connection.PropertyKey streamKey; private final ServerTransportListener transportListener; private final int maxMessageSize; + private final TcpMetrics.Tracker tcpMetrics; private final long keepAliveTimeInNanos; private final long keepAliveTimeoutInNanos; private final long maxConnectionAgeInNanos; @@ -174,7 +176,8 @@ static NettyServerHandler newHandler( long permitKeepAliveTimeInNanos, int maxRstCount, long maxRstPeriodNanos, - Attributes eagAttributes) { + Attributes eagAttributes, + MetricRecorder metricRecorder) { Preconditions.checkArgument(maxHeaderListSize > 0, "maxHeaderListSize must be positive: %s", maxHeaderListSize); Http2FrameLogger frameLogger = new Http2FrameLogger(LogLevel.DEBUG, NettyServerHandler.class); @@ -208,7 +211,8 @@ static NettyServerHandler newHandler( maxRstCount, maxRstPeriodNanos, eagAttributes, - Ticker.systemTicker()); + Ticker.systemTicker(), + metricRecorder); } static NettyServerHandler newHandler( @@ -234,7 +238,8 @@ static NettyServerHandler newHandler( int maxRstCount, long maxRstPeriodNanos, Attributes eagAttributes, - Ticker ticker) { + Ticker ticker, + MetricRecorder metricRecorder) { Preconditions.checkArgument(maxStreams > 0, "maxStreams must be positive: %s", maxStreams); Preconditions.checkArgument(flowControlWindow > 0, "flowControlWindow must be positive: %s", flowControlWindow); @@ -294,7 +299,8 @@ static NettyServerHandler newHandler( keepAliveEnforcer, autoFlowControl, rstStreamCounter, - eagAttributes, ticker); + eagAttributes, ticker, + metricRecorder); } private NettyServerHandler( @@ -318,7 +324,8 @@ private NettyServerHandler( boolean autoFlowControl, RstStreamCounter rstStreamCounter, Attributes eagAttributes, - Ticker ticker) { + Ticker ticker, + MetricRecorder metricRecorder) { super( channelUnused, decoder, @@ -362,6 +369,7 @@ public void onStreamClosed(Http2Stream stream) { checkArgument(maxMessageSize >= 0, "maxMessageSize must be non-negative: %s", maxMessageSize); this.maxMessageSize = maxMessageSize; + this.tcpMetrics = new TcpMetrics.Tracker(metricRecorder); this.keepAliveTimeInNanos = keepAliveTimeInNanos; this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos; this.maxConnectionIdleManager = maxConnectionIdleManager; @@ -661,8 +669,15 @@ void setKeepAliveManagerForTest(KeepAliveManager keepAliveManager) { /** * Handler for the Channel shutting down. */ + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + tcpMetrics.channelActive(ctx.channel()); + super.channelActive(ctx); + } + @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + tcpMetrics.channelInactive(ctx.channel()); try { if (keepAliveManager != null) { keepAliveManager.onTransportTermination(); diff --git a/netty/src/main/java/io/grpc/netty/NettyServerTransport.java b/netty/src/main/java/io/grpc/netty/NettyServerTransport.java index 758ffeee5b1..c0e52b75876 100644 --- a/netty/src/main/java/io/grpc/netty/NettyServerTransport.java +++ b/netty/src/main/java/io/grpc/netty/NettyServerTransport.java @@ -25,6 +25,7 @@ import io.grpc.Attributes; import io.grpc.InternalChannelz.SocketStats; import io.grpc.InternalLogId; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.Status; import io.grpc.internal.ServerTransport; @@ -81,6 +82,7 @@ class NettyServerTransport implements ServerTransport { private final int maxRstCount; private final long maxRstPeriodNanos; private final Attributes eagAttributes; + private final MetricRecorder metricRecorder; private final List streamTracerFactories; private final TransportTracer transportTracer; @@ -105,7 +107,8 @@ class NettyServerTransport implements ServerTransport { long permitKeepAliveTimeInNanos, int maxRstCount, long maxRstPeriodNanos, - Attributes eagAttributes) { + Attributes eagAttributes, + MetricRecorder metricRecorder) { this.channel = Preconditions.checkNotNull(channel, "channel"); this.channelUnused = channelUnused; this.protocolNegotiator = Preconditions.checkNotNull(protocolNegotiator, "protocolNegotiator"); @@ -128,6 +131,7 @@ class NettyServerTransport implements ServerTransport { this.maxRstCount = maxRstCount; this.maxRstPeriodNanos = maxRstPeriodNanos; this.eagAttributes = Preconditions.checkNotNull(eagAttributes, "eagAttributes"); + this.metricRecorder = metricRecorder; SocketAddress remote = channel.remoteAddress(); this.logId = InternalLogId.allocate(getClass(), remote != null ? remote.toString() : null); } @@ -289,6 +293,7 @@ private NettyServerHandler createHandler( permitKeepAliveTimeInNanos, maxRstCount, maxRstPeriodNanos, - eagAttributes); + eagAttributes, + metricRecorder); } } diff --git a/netty/src/main/java/io/grpc/netty/TcpMetrics.java b/netty/src/main/java/io/grpc/netty/TcpMetrics.java new file mode 100644 index 00000000000..d7f8d080d11 --- /dev/null +++ b/netty/src/main/java/io/grpc/netty/TcpMetrics.java @@ -0,0 +1,271 @@ +/* + * 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.netty; + +import io.grpc.DoubleHistogramMetricInstrument; +import io.grpc.InternalTcpMetrics; +import io.grpc.LongCounterMetricInstrument; +import io.grpc.LongUpDownCounterMetricInstrument; +import io.grpc.MetricRecorder; +import io.netty.channel.Channel; +import io.netty.util.concurrent.ScheduledFuture; +import java.lang.reflect.Method; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Utility for collecting TCP metrics from Netty channels. + */ +final class TcpMetrics { + private static final Logger log = Logger.getLogger(TcpMetrics.class.getName()); + + private static final Metrics DEFAULT_METRICS; + + static { + boolean epollAvailable = false; + try { + Class epollClass = Class.forName("io.netty.channel.epoll.Epoll"); + Method isAvailableMethod = epollClass.getDeclaredMethod("isAvailable"); + epollAvailable = (Boolean) isAvailableMethod.invoke(null); + } catch (ClassNotFoundException e) { + log.log(Level.FINE, "Epoll is not available", e); + } catch (Exception e) { + log.log(Level.FINE, "Failed to determine Epoll availability", e); + } catch (Error e) { + log.log(Level.FINE, "Failed to load native Epoll library", e); + } + DEFAULT_METRICS = new Metrics(epollAvailable); + } + + static Metrics getDefaultMetrics() { + return DEFAULT_METRICS; + } + + static final class Metrics { + final LongCounterMetricInstrument connectionsCreated; + final LongUpDownCounterMetricInstrument connectionCount; + final LongCounterMetricInstrument packetsRetransmitted; + final LongCounterMetricInstrument recurringRetransmits; + final DoubleHistogramMetricInstrument minRtt; + + Metrics(boolean epollAvailable) { + connectionsCreated = InternalTcpMetrics.CONNECTIONS_CREATED_INSTRUMENT; + connectionCount = InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT; + + if (epollAvailable) { + packetsRetransmitted = InternalTcpMetrics.PACKETS_RETRANSMITTED_INSTRUMENT; + recurringRetransmits = InternalTcpMetrics.RECURRING_RETRANSMITS_INSTRUMENT; + minRtt = InternalTcpMetrics.MIN_RTT_INSTRUMENT; + } else { + packetsRetransmitted = null; + recurringRetransmits = null; + minRtt = null; + } + } + } + + static final class Tracker { + private final MetricRecorder metricRecorder; + private final Metrics metrics; + private final Class epollSocketChannelClass; + private final Method tcpInfoMethod; + private final Method totalRetransMethod; + private final Method retransmitsMethod; + private final Method rttMethod; + private final Object tcpInfo; + + private long lastTotalRetrans = 0; + + Tracker(MetricRecorder metricRecorder) { + this(metricRecorder, DEFAULT_METRICS); + } + + Tracker(MetricRecorder metricRecorder, Metrics metrics) { + this( + metricRecorder, metrics, + "io.netty.channel.epoll.EpollSocketChannel", + "io.netty.channel.epoll.EpollTcpInfo"); + } + + Tracker(MetricRecorder metricRecorder, Metrics metrics, + String epollSocketChannelClassName, String epollTcpInfoClassName) { + this.metricRecorder = metricRecorder; + this.metrics = metrics; + + Class epollSocketChannelClass; + Method tcpInfoMethod; + Object tcpInfo; + Method totalRetransMethod; + Method retransMethod; + Method rttMethod; + + try { + epollSocketChannelClass = Class.forName(epollSocketChannelClassName); + Class epollTcpInfoClass = Class.forName(epollTcpInfoClassName); + tcpInfo = epollTcpInfoClass.getDeclaredConstructor().newInstance(); + tcpInfoMethod = epollSocketChannelClass.getMethod("tcpInfo", epollTcpInfoClass); + totalRetransMethod = epollTcpInfoClass.getMethod("totalRetrans"); + retransMethod = epollTcpInfoClass.getMethod("retrans"); + rttMethod = epollTcpInfoClass.getMethod("rtt"); + } catch (Exception | Error t) { + // Epoll not available or error getting tcp_info, features disabled + log.log(Level.FINE, "Failed to initialize Epoll tcp_info reflection", t); + epollSocketChannelClass = null; + tcpInfoMethod = null; + tcpInfo = null; + totalRetransMethod = null; + retransMethod = null; + rttMethod = null; + } + this.epollSocketChannelClass = epollSocketChannelClass; + this.tcpInfoMethod = tcpInfoMethod; + this.tcpInfo = tcpInfo; + this.totalRetransMethod = totalRetransMethod; + this.retransmitsMethod = retransMethod; + this.rttMethod = rttMethod; + } + + private static final long RECORD_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(5); + private ScheduledFuture reportTimer; + + void channelActive(Channel channel) { + if (metricRecorder != null) { + List labelValues = getLabelValues(channel); + metricRecorder.addLongCounter(metrics.connectionsCreated, 1, + Collections.emptyList(), labelValues); + metricRecorder.addLongUpDownCounter(metrics.connectionCount, 1, + Collections.emptyList(), labelValues); + scheduleNextReport(channel, true); + } + } + + private void scheduleNextReport(final Channel channel, boolean isInitial) { + if (!channel.isActive()) { + return; + } + + double jitter = isInitial + ? 0.1 + ThreadLocalRandom.current().nextDouble() // 10% to 110% + : 0.9 + ThreadLocalRandom.current().nextDouble() * 0.2; // 90% to 110% + long rearmingDelay = (long) (RECORD_INTERVAL_MILLIS * jitter); + + try { + reportTimer = channel.eventLoop().schedule(() -> { + if (channel.isActive()) { + Tracker.this.recordTcpInfo(channel, false); + scheduleNextReport(channel, false); // Re-arm + } + }, rearmingDelay, TimeUnit.MILLISECONDS); + } catch (RejectedExecutionException e) { + log.log(Level.FINE, "Failed to schedule next TCP metrics report", e); + // The event loop is likely shutting down. We can safely ignore this. + } + } + + void channelInactive(Channel channel) { + if (reportTimer != null) { + reportTimer.cancel(false); + } + if (metricRecorder != null) { + List labelValues = getLabelValues(channel); + metricRecorder.addLongUpDownCounter(metrics.connectionCount, -1, + Collections.emptyList(), labelValues); + // Final collection on close + recordTcpInfo(channel, true); + } + } + + void recordTcpInfo(Channel channel) { + recordTcpInfo(channel, false); + } + + private void recordTcpInfo(Channel channel, boolean isClose) { + if (metricRecorder == null || epollSocketChannelClass == null + || !epollSocketChannelClass.isInstance(channel)) { + return; + } + List labelValues = getLabelValues(channel); + long totalRetrans; + long retransmits; + long rtt; + try { + tcpInfoMethod.invoke(channel, tcpInfo); + + totalRetrans = (Long) totalRetransMethod.invoke(tcpInfo); + retransmits = (Long) retransmitsMethod.invoke(tcpInfo); + rtt = (Long) rttMethod.invoke(tcpInfo); + } catch (Exception e) { + log.log(Level.FINE, "Error computing TCP metrics", e); + return; + } + + if (metrics.packetsRetransmitted != null) { + long deltaTotal = totalRetrans - lastTotalRetrans; + if (deltaTotal > 0) { + metricRecorder.addLongCounter(metrics.packetsRetransmitted, deltaTotal, + Collections.emptyList(), labelValues); + lastTotalRetrans = totalRetrans; + } + } + if (metrics.recurringRetransmits != null && isClose) { + if (retransmits > 0) { + metricRecorder.addLongCounter(metrics.recurringRetransmits, retransmits, + Collections.emptyList(), labelValues); + } + } + if (metrics.minRtt != null) { + metricRecorder.recordDoubleHistogram(metrics.minRtt, + rtt / 1000000.0, // Convert microseconds to seconds + Collections.emptyList(), labelValues); + } + } + } + + private static List getLabelValues(Channel channel) { + String localAddress = ""; + String localPort = ""; + String peerAddress = ""; + String peerPort = ""; + + SocketAddress local = channel.localAddress(); + if (local instanceof InetSocketAddress) { + InetSocketAddress inetLocal = (InetSocketAddress) local; + localAddress = inetLocal.getAddress().getHostAddress(); + localPort = String.valueOf(inetLocal.getPort()); + } + + SocketAddress remote = channel.remoteAddress(); + if (remote instanceof InetSocketAddress) { + InetSocketAddress inetRemote = (InetSocketAddress) remote; + peerAddress = inetRemote.getAddress().getHostAddress(); + peerPort = String.valueOf(inetRemote.getPort()); + } + + return Arrays.asList(localAddress, localPort, peerAddress, peerPort); + } + + + private TcpMetrics() {} +} diff --git a/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java b/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java index 53598727efd..9f6be9a2f3e 100644 --- a/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyClientHandlerTest.java @@ -57,6 +57,7 @@ import io.grpc.Attributes; import io.grpc.CallOptions; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.Status; import io.grpc.internal.AbstractStream; import io.grpc.internal.ClientStreamListener; @@ -1165,7 +1166,8 @@ public Stopwatch get() { Attributes.EMPTY, "someauthority", null, - fakeClock().getTicker()); + fakeClock().getTicker(), + new MetricRecorder() {}); } @Override diff --git a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java index db44c8f50fd..08ec1e83b84 100644 --- a/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyClientTransportTest.java @@ -37,6 +37,7 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -56,6 +57,7 @@ import io.grpc.Metadata; import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor.Marshaller; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.Status; import io.grpc.Status.Code; @@ -251,6 +253,8 @@ public void setSoLingerChannelOption() throws IOException, GeneralSecurityExcept new SocketPicker(), new FakeChannelLogger(), false, + new MetricRecorder() { + }, Ticker.systemTicker()); transports.add(transport); callMeMaybe(transport.start(clientTransportListener)); @@ -526,6 +530,8 @@ public void failingToConstructChannelShouldFailGracefully() throws Exception { new SocketPicker(), new FakeChannelLogger(), false, + new MetricRecorder() { + }, Ticker.systemTicker()); transports.add(transport); @@ -1148,6 +1154,8 @@ private NettyClientTransport newTransport(ProtocolNegotiator negotiator, int max new SocketPicker(), new FakeChannelLogger(), false, + new MetricRecorder() { + }, Ticker.systemTicker()); transports.add(transport); return transport; @@ -1195,7 +1203,8 @@ private void startServer(int maxStreamsPerConnection, int maxHeaderListSize, MAX_RST_COUNT_DISABLED, 0, Attributes.EMPTY, - channelz); + channelz, + mock(MetricRecorder.class)); server.start(serverListener); address = TestUtils.testServerAddress((InetSocketAddress) server.getListenSocketAddress()); authority = GrpcUtil.authorityFromHostAndPort(address.getHostString(), address.getPort()); diff --git a/netty/src/test/java/io/grpc/netty/NettyServerBuilderTest.java b/netty/src/test/java/io/grpc/netty/NettyServerBuilderTest.java index 797cfa95c0e..2fef700970e 100644 --- a/netty/src/test/java/io/grpc/netty/NettyServerBuilderTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyServerBuilderTest.java @@ -22,7 +22,7 @@ import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableList; -import io.grpc.ServerStreamTracer; +import io.grpc.MetricRecorder; import io.netty.channel.EventLoopGroup; import io.netty.channel.local.LocalServerChannel; import io.netty.handler.ssl.SslContext; @@ -43,8 +43,9 @@ public class NettyServerBuilderTest { @Test public void addMultipleListenAddresses() { builder.addListenAddress(new InetSocketAddress(8081)); - NettyServer server = - builder.buildTransportServers(ImmutableList.of()); + NettyServer server = builder.buildTransportServers( + ImmutableList.of(), + new MetricRecorder() {}); assertThat(server.getListenSocketAddresses()).hasSize(2); } @@ -189,4 +190,14 @@ public void useNioTransport_shouldNotThrow() { builder.assertEventLoopsAndChannelType(); } + + @Test + public void metricRecorder_propagatedToServer() { + MetricRecorder recorder = mock(MetricRecorder.class); + + NettyServer server = builder.buildTransportServers( + ImmutableList.of(), recorder); + + assertThat(server.getMetricRecorder()).isSameInstanceAs(recorder); + } } diff --git a/netty/src/test/java/io/grpc/netty/NettyServerHandlerTest.java b/netty/src/test/java/io/grpc/netty/NettyServerHandlerTest.java index 0d5a9bab176..5c010793e08 100644 --- a/netty/src/test/java/io/grpc/netty/NettyServerHandlerTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyServerHandlerTest.java @@ -59,6 +59,7 @@ import io.grpc.Attributes; import io.grpc.InternalStatus; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.Status; import io.grpc.Status.Code; @@ -1416,7 +1417,8 @@ protected NettyServerHandler newHandler() { maxRstCount, maxRstPeriodNanos, Attributes.EMPTY, - fakeClock().getTicker()); + fakeClock().getTicker(), + new MetricRecorder() {}); } @Override diff --git a/netty/src/test/java/io/grpc/netty/NettyServerTest.java b/netty/src/test/java/io/grpc/netty/NettyServerTest.java index f9bda4c5af1..61c3f9e219e 100644 --- a/netty/src/test/java/io/grpc/netty/NettyServerTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyServerTest.java @@ -37,6 +37,7 @@ import io.grpc.InternalChannelz.SocketStats; import io.grpc.InternalInstrumented; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.FixedObjectPool; import io.grpc.internal.ServerListener; @@ -161,7 +162,7 @@ class NoHandlerProtocolNegotiator implements ProtocolNegotiator { 0, 0, // ignore Attributes.EMPTY, - channelz); + channelz, mock(MetricRecorder.class)); final SettableFuture serverShutdownCalled = SettableFuture.create(); ns.start(new ServerListener() { @Override @@ -218,7 +219,7 @@ public void multiPortStartStopGet() throws Exception { 0, 0, // ignore Attributes.EMPTY, - channelz); + channelz, mock(MetricRecorder.class)); final SettableFuture shutdownCompleted = SettableFuture.create(); ns.start(new ServerListener() { @Override @@ -298,7 +299,7 @@ public void multiPortConnections() throws Exception { 0, 0, // ignore Attributes.EMPTY, - channelz); + channelz, mock(MetricRecorder.class)); final SettableFuture shutdownCompleted = SettableFuture.create(); ns.start(new ServerListener() { @Override @@ -366,7 +367,7 @@ public void getPort_notStarted() { 0, 0, // ignore Attributes.EMPTY, - channelz); + channelz, mock(MetricRecorder.class)); assertThat(ns.getListenSocketAddress()).isEqualTo(addr); assertThat(ns.getListenSocketAddresses()).isEqualTo(addresses); @@ -447,7 +448,7 @@ class TestProtocolNegotiator implements ProtocolNegotiator { 0, 0, // ignore eagAttributes, - channelz); + channelz, mock(MetricRecorder.class)); ns.start(new ServerListener() { @Override public ServerTransportListener transportCreated(ServerTransport transport) { @@ -501,7 +502,7 @@ public void channelzListenSocket() throws Exception { 0, 0, // ignore Attributes.EMPTY, - channelz); + channelz, mock(MetricRecorder.class)); final SettableFuture shutdownCompleted = SettableFuture.create(); ns.start(new ServerListener() { @Override @@ -649,7 +650,7 @@ private NettyServer getServer(List addr, EventLoopGroup ev) { 0, 0, // ignore Attributes.EMPTY, - channelz); + channelz, mock(MetricRecorder.class)); } private static class NoopServerTransportListener implements ServerTransportListener { diff --git a/netty/src/test/java/io/grpc/netty/NettyTransportTest.java b/netty/src/test/java/io/grpc/netty/NettyTransportTest.java index b779dfbe980..22758a8b727 100644 --- a/netty/src/test/java/io/grpc/netty/NettyTransportTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyTransportTest.java @@ -22,6 +22,7 @@ import com.google.common.util.concurrent.SettableFuture; import io.grpc.Attributes; import io.grpc.ChannelLogger; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.Status; import io.grpc.internal.AbstractTransportTest; @@ -71,7 +72,7 @@ protected InternalServer newServer( .forAddress(new InetSocketAddress("localhost", 0)) .flowControlWindow(AbstractTransportTest.TEST_FLOW_CONTROL_WINDOW) .setTransportTracerFactory(fakeClockTransportTracer) - .buildTransportServers(streamTracerFactories); + .buildTransportServers(streamTracerFactories, new MetricRecorder() {}); } @Override @@ -81,7 +82,7 @@ protected InternalServer newServer( .forAddress(new InetSocketAddress("localhost", port)) .flowControlWindow(AbstractTransportTest.TEST_FLOW_CONTROL_WINDOW) .setTransportTracerFactory(fakeClockTransportTracer) - .buildTransportServers(streamTracerFactories); + .buildTransportServers(streamTracerFactories, new MetricRecorder() {}); } @Override diff --git a/netty/src/test/java/io/grpc/netty/ProtocolNegotiatorsTest.java b/netty/src/test/java/io/grpc/netty/ProtocolNegotiatorsTest.java index 80438532172..403b1b64329 100644 --- a/netty/src/test/java/io/grpc/netty/ProtocolNegotiatorsTest.java +++ b/netty/src/test/java/io/grpc/netty/ProtocolNegotiatorsTest.java @@ -46,6 +46,7 @@ import io.grpc.InternalChannelz; import io.grpc.InternalChannelz.Security; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.SecurityLevel; import io.grpc.ServerCredentials; import io.grpc.ServerStreamTracer; @@ -389,7 +390,9 @@ private Object expectHandshake( .buildTransportFactory(); InternalServer server = NettyServerBuilder .forPort(0, serverCreds) - .buildTransportServers(Collections.emptyList()); + .buildTransportServers( + Collections.emptyList(), + new MetricRecorder() {}); server.start(serverListener); ManagedClientTransport.Listener clientTransportListener = diff --git a/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java b/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java new file mode 100644 index 00000000000..ac59f31c747 --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java @@ -0,0 +1,557 @@ +/* + * 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.netty; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +import io.grpc.MetricRecorder; +import io.netty.channel.Channel; +import io.netty.channel.EventLoop; +import io.netty.util.concurrent.ScheduledFuture; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +@RunWith(JUnit4.class) +public class TcpMetricsTest { + + @Rule public final MockitoRule mocks = MockitoJUnit.rule(); + + @Mock + private MetricRecorder metricRecorder; + @Mock + private Channel channel; + @Mock + private EventLoop eventLoop; + @Mock + private ScheduledFuture scheduledFuture; + + private TcpMetrics.Tracker metrics; + + @Before + public void setUp() { + when(channel.eventLoop()).thenReturn(eventLoop); + when(eventLoop.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) + .thenAnswer(invocation -> scheduledFuture); + metrics = new TcpMetrics.Tracker(metricRecorder); + } + + @Test + public void metricsInitialization_epollUnavailable() { + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(false); + + org.junit.Assert.assertNotNull(metrics.connectionsCreated); + org.junit.Assert.assertNotNull(metrics.connectionCount); + org.junit.Assert.assertNull(metrics.packetsRetransmitted); + org.junit.Assert.assertNull(metrics.recurringRetransmits); + org.junit.Assert.assertNull(metrics.minRtt); + } + + @Test + public void metricsInitialization_epollAvailable() { + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + org.junit.Assert.assertNotNull(metrics.connectionsCreated); + org.junit.Assert.assertNotNull(metrics.connectionCount); + org.junit.Assert.assertNotNull(metrics.packetsRetransmitted); + org.junit.Assert.assertNotNull(metrics.recurringRetransmits); + org.junit.Assert.assertNotNull(metrics.minRtt); + } + + public static class FakeEpollTcpInfo { + long totalRetrans; + long retransmits; + long rtt; + + public void setValues(long totalRetrans, long retransmits, long rtt) { + this.totalRetrans = totalRetrans; + this.retransmits = retransmits; + this.rtt = rtt; + } + + @SuppressWarnings("unused") + public long totalRetrans() { + return totalRetrans; + } + + @SuppressWarnings("unused") + public long retrans() { + return retransmits; + } + + @SuppressWarnings("unused") + public long rtt() { + return rtt; + } + } + + @Test + public void tracker_recordTcpInfo_reflectionSuccess() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + ConfigurableFakeWithTcpInfo.class.getName(), + FakeEpollTcpInfo.class.getName()); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + channel.writeInbound("dummy"); + + tracker.channelInactive(channel); + + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(123L), any(), any()); + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.recurringRetransmits)), + eq(4L), any(), any()); + verify(recorder).recordDoubleHistogram(eq(Objects.requireNonNull(metrics.minRtt)), + eq(0.005), any(), any()); + } + + @Test + public void tracker_periodicRecord_doesNotRecordRecurringRetransmits() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + ConfigurableFakeWithTcpInfo.class.getName(), + FakeEpollTcpInfo.class.getName()); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + ConfigurableFakeWithTcpInfo channel = + org.mockito.Mockito.spy(new ConfigurableFakeWithTcpInfo(infoSource)); + when(channel.eventLoop()).thenReturn(eventLoop); + when(channel.isActive()).thenReturn(true); + + tracker.channelActive(channel); + + ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(eventLoop).schedule(runnableCaptor.capture(), anyLong(), eq(TimeUnit.MILLISECONDS)); + Runnable periodicTask = runnableCaptor.getValue(); + + org.mockito.Mockito.clearInvocations(recorder); + periodicTask.run(); + + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(123L), any(), any()); + verify(recorder).recordDoubleHistogram(eq(Objects.requireNonNull(metrics.minRtt)), + eq(0.005), any(), any()); + // Should NOT record recurring retransmits during periodic polling + verify(recorder, org.mockito.Mockito.never()) + .addLongCounter(eq(Objects.requireNonNull(metrics.recurringRetransmits)), + anyLong(), any(), any()); + } + + @Test + public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + ConfigurableFakeWithTcpInfo.class.getName(), + FakeEpollTcpInfo.class.getName()); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + ConfigurableFakeWithTcpInfo channel = + org.mockito.Mockito.spy(new ConfigurableFakeWithTcpInfo(infoSource)); + when(channel.eventLoop()).thenReturn(eventLoop); + when(channel.isActive()).thenReturn(true); + + // Mimic the periodic schedule invocation + tracker.channelActive(channel); + ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(eventLoop).schedule(runnableCaptor.capture(), anyLong(), eq(TimeUnit.MILLISECONDS)); + + // Fire periodic task once. TotalRetrans=123, retransmits=4. + runnableCaptor.getValue().run(); + + org.mockito.Mockito.clearInvocations(recorder); + + // Let's just create a new channel instance where tcpInfo sets retrans=5. + FakeEpollTcpInfo infoSource2 = new FakeEpollTcpInfo(); + infoSource2.setValues(130, 5, 5000); + ConfigurableFakeWithTcpInfo channel2 = + org.mockito.Mockito.spy(new ConfigurableFakeWithTcpInfo(infoSource2)); + when(channel2.eventLoop()).thenReturn(eventLoop); + + tracker.channelInactive(channel2); + + // It should record delta for totalRetrans (130 - 123 = 7) + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(7L), any(), any()); + // But for recurringRetransmits it MUST record the raw value 5, not the delta! + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.recurringRetransmits)), + eq(5L), any(), any()); + } + + @Test + public void tracker_periodicRecord_reportsDeltaForTotalRetrans() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + ConfigurableFakeWithTcpInfo.class.getName(), + FakeEpollTcpInfo.class.getName()); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + ConfigurableFakeWithTcpInfo channel = + org.mockito.Mockito.spy(new ConfigurableFakeWithTcpInfo(infoSource)); + when(channel.eventLoop()).thenReturn(eventLoop); + when(channel.isActive()).thenReturn(true); + + // Initial Active Trigger + tracker.channelActive(channel); + ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(eventLoop).schedule(runnableCaptor.capture(), anyLong(), eq(TimeUnit.MILLISECONDS)); + Runnable periodicTask = runnableCaptor.getValue(); + + // First periodic record + org.mockito.Mockito.clearInvocations(recorder); + periodicTask.run(); + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(123L), any(), any()); + + // Change tcpInfo for second periodic record + org.mockito.Mockito.doAnswer(invocation -> { + FakeEpollTcpInfo info = invocation.getArgument(0); + info.totalRetrans = 150; + info.retransmits = 2; // Should not be recorded + info.rtt = 6000; + return null; + }).when(channel).tcpInfo(any(FakeEpollTcpInfo.class)); + + org.mockito.Mockito.clearInvocations(recorder); + periodicTask.run(); + + // Only the delta (150 - 123 = 27) should be recorded + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(27L), any(), any()); + verify(recorder).recordDoubleHistogram(eq(Objects.requireNonNull(metrics.minRtt)), + eq(0.006), any(), any()); + verify(recorder, org.mockito.Mockito.never()) + .addLongCounter(eq(Objects.requireNonNull(metrics.recurringRetransmits)), + anyLong(), any(), any()); + } + + @Test + public void tracker_periodicRecord_doesNotReportZeroDeltaForTotalRetrans() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + ConfigurableFakeWithTcpInfo.class.getName(), + FakeEpollTcpInfo.class.getName()); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + ConfigurableFakeWithTcpInfo channel = + org.mockito.Mockito.spy(new ConfigurableFakeWithTcpInfo(infoSource)); + when(channel.eventLoop()).thenReturn(eventLoop); + when(channel.isActive()).thenReturn(true); + + // Initial Active Trigger + tracker.channelActive(channel); + ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class); + verify(eventLoop).schedule(runnableCaptor.capture(), anyLong(), eq(TimeUnit.MILLISECONDS)); + Runnable periodicTask = runnableCaptor.getValue(); + + // First periodic record + periodicTask.run(); + org.mockito.Mockito.clearInvocations(recorder); + + // Keep tcpInfo the same for second periodic record + periodicTask.run(); + + // NO delta (123 - 123 = 0), so it should not be recorded + verify(recorder, org.mockito.Mockito.never()) + .addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + anyLong(), any(), any()); + verify(recorder).recordDoubleHistogram(eq(Objects.requireNonNull(metrics.minRtt)), + eq(0.005), any(), any()); + } + + public static class ConfigurableFakeWithTcpInfo extends + io.netty.channel.embedded.EmbeddedChannel { + private final FakeEpollTcpInfo infoToCopy; + + public ConfigurableFakeWithTcpInfo(FakeEpollTcpInfo infoToCopy) { + this.infoToCopy = infoToCopy; + } + + public void tcpInfo(FakeEpollTcpInfo info) { + info.totalRetrans = infoToCopy.totalRetrans; + info.retransmits = infoToCopy.retransmits; + info.rtt = infoToCopy.rtt; + } + } + + @Test + public void tracker_reportsDeltas_correctly() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + String fakeChannelName = ConfigurableFakeWithTcpInfo.class.getName(); + String fakeInfoName = FakeEpollTcpInfo.class.getName(); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + fakeChannelName, fakeInfoName); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + + // 10 retransmits total + infoSource.setValues(10, 2, 1000); + tracker.recordTcpInfo(channel); + + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(10L), any(), any()); + + // 15 retransmits total (delta 5) + infoSource.setValues(15, 0, 1000); + tracker.recordTcpInfo(channel); + + verify(recorder).addLongCounter(eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(5L), any(), any()); + + // 15 retransmits total (delta 0) - should NOT report + // also set retransmits to 1 + infoSource.setValues(15, 1, 1000); + tracker.recordTcpInfo(channel); + // Verify no new interactions with this specific metric and value + // We can't easily verify "no interaction" for specific value without capturing. + verify(recorder, org.mockito.Mockito.times(1)).addLongCounter( + eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(10L), any(), any()); + verify(recorder, org.mockito.Mockito.times(1)).addLongCounter( + eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + eq(5L), any(), any()); + // Total interactions for packetsRetransmitted should be 2 + verify(recorder, org.mockito.Mockito.times(2)).addLongCounter( + eq(Objects.requireNonNull(metrics.packetsRetransmitted)), + anyLong(), any(), any()); + + // recurringRetransmits should NOT have been reported yet (periodic calls) + verify(recorder, org.mockito.Mockito.times(0)).addLongCounter( + eq(Objects.requireNonNull(metrics.recurringRetransmits)), + anyLong(), any(), any()); + + // Close channel - should report recurringRetransmits + tracker.channelInactive(channel); + verify(recorder, org.mockito.Mockito.times(1)).addLongCounter( + eq(Objects.requireNonNull(metrics.recurringRetransmits)), + eq(1L), // From last infoSource setValues(15, 1, 1000) + any(), any()); + } + + @Test + public void tracker_recordTcpInfo_reflectionFailure() { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.Metrics metrics = new TcpMetrics.Metrics(true); + + TcpMetrics.Tracker tracker = new TcpMetrics.Tracker(recorder, metrics, + "non.existent.Class", "non.existent.Info"); + + Channel channel = org.mockito.Mockito.mock(Channel.class); + when(channel.isActive()).thenReturn(true); + + // Should catch exception and ignore + tracker.channelInactive(channel); + } + + @Test + public void registeredMetrics_haveCorrectOptionalLabels() { + List expectedOptionalLabels = Arrays.asList( + "network.local.address", + "network.local.port", + "network.peer.address", + "network.peer.port" + ); + + org.junit.Assert.assertEquals( + expectedOptionalLabels, + TcpMetrics.getDefaultMetrics().connectionsCreated.getOptionalLabelKeys()); + org.junit.Assert.assertEquals( + expectedOptionalLabels, + TcpMetrics.getDefaultMetrics().connectionCount.getOptionalLabelKeys()); + + if (TcpMetrics.getDefaultMetrics().packetsRetransmitted != null) { + org.junit.Assert.assertEquals( + expectedOptionalLabels, + Objects.requireNonNull(TcpMetrics.getDefaultMetrics().packetsRetransmitted) + .getOptionalLabelKeys()); + org.junit.Assert.assertEquals( + expectedOptionalLabels, + Objects.requireNonNull(TcpMetrics.getDefaultMetrics().recurringRetransmits) + .getOptionalLabelKeys()); + org.junit.Assert.assertEquals( + expectedOptionalLabels, + Objects.requireNonNull(TcpMetrics.getDefaultMetrics().minRtt).getOptionalLabelKeys()); + } + } + + @Test + public void channelActive_extractsLabels_ipv4() throws Exception { + + InetAddress localInet = InetAddress.getByAddress(new byte[] { 127, 0, 0, 1 }); + InetAddress remoteInet = InetAddress.getByAddress(new byte[] { 10, 0, 0, 1 }); + when(channel.localAddress()).thenReturn(new InetSocketAddress(localInet, 8080)); + when(channel.remoteAddress()).thenReturn(new InetSocketAddress(remoteInet, 443)); + + metrics.channelActive(channel); + + verify(metricRecorder).addLongCounter( + eq(TcpMetrics.getDefaultMetrics().connectionsCreated), eq(1L), + eq(Collections.emptyList()), + eq(Arrays.asList( + localInet.getHostAddress(), "8080", remoteInet.getHostAddress(), "443"))); + verify(metricRecorder).addLongUpDownCounter( + eq(TcpMetrics.getDefaultMetrics().connectionCount), eq(1L), + eq(Collections.emptyList()), + eq(Arrays.asList( + localInet.getHostAddress(), "8080", remoteInet.getHostAddress(), "443"))); + verifyNoMoreInteractions(metricRecorder); + } + + @Test + public void channelInactive_extractsLabels_ipv6() throws Exception { + + InetAddress localInet = InetAddress.getByAddress( + new byte[] { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 }); + InetAddress remoteInet = InetAddress.getByAddress( + new byte[] { 32, 1, 13, -72, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 }); + when(channel.localAddress()).thenReturn(new InetSocketAddress(localInet, 8080)); + when(channel.remoteAddress()).thenReturn(new InetSocketAddress(remoteInet, 443)); + + metrics.channelInactive(channel); + + verify(metricRecorder).addLongUpDownCounter( + eq(TcpMetrics.getDefaultMetrics().connectionCount), eq(-1L), + eq(Collections.emptyList()), + eq(Arrays.asList( + localInet.getHostAddress(), "8080", remoteInet.getHostAddress(), "443"))); + verifyNoMoreInteractions(metricRecorder); + } + + @Test + public void channelActive_extractsLabels_nonInetAddress() { + SocketAddress dummyAddress = new SocketAddress() {}; + when(channel.localAddress()).thenReturn(dummyAddress); + when(channel.remoteAddress()).thenReturn(dummyAddress); + + metrics.channelActive(channel); + + verify(metricRecorder).addLongCounter( + eq(TcpMetrics.getDefaultMetrics().connectionsCreated), eq(1L), + eq(Collections.emptyList()), + eq(Arrays.asList("", "", "", ""))); + verify(metricRecorder).addLongUpDownCounter( + eq(TcpMetrics.getDefaultMetrics().connectionCount), eq(1L), + eq(Collections.emptyList()), + eq(Arrays.asList("", "", "", ""))); + verifyNoMoreInteractions(metricRecorder); + } + + @Test + public void channelActive_incrementsCounts() { + metrics.channelActive(channel); + verify(metricRecorder).addLongCounter( + eq(TcpMetrics.getDefaultMetrics().connectionsCreated), eq(1L), + eq(Collections.emptyList()), + eq(Arrays.asList("", "", "", ""))); + verify(metricRecorder).addLongUpDownCounter( + eq(TcpMetrics.getDefaultMetrics().connectionCount), eq(1L), + eq(Collections.emptyList()), + eq(Arrays.asList("", "", "", ""))); + verifyNoMoreInteractions(metricRecorder); + } + + @Test + public void channelInactive_decrementsCount_noEpoll_noError() { + metrics.channelInactive(channel); + verify(metricRecorder).addLongUpDownCounter( + eq(TcpMetrics.getDefaultMetrics().connectionCount), eq(-1L), + eq(Collections.emptyList()), + eq(Arrays.asList("", "", "", ""))); + verifyNoMoreInteractions(metricRecorder); + } + + @Test + public void channelActive_schedulesReportTimer() { + when(channel.isActive()).thenReturn(true); + metrics.channelActive(channel); + + ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class); + ArgumentCaptor delayCaptor = ArgumentCaptor.forClass(Long.class); + verify(eventLoop).schedule( + runnableCaptor.capture(), delayCaptor.capture(), eq(TimeUnit.MILLISECONDS)); + + Runnable task = runnableCaptor.getValue(); + long delay = delayCaptor.getValue(); + + // Default RECORD_INTERVAL_MILLIS is 5 minutes (300,000 ms) + // Initial jitter is 10% to 110%, so 30,000 ms to 330,000 ms + org.junit.Assert.assertTrue("Delay should be >= 30000 but was " + delay, delay >= 30_000); + org.junit.Assert.assertTrue("Delay should be <= 330000 but was " + delay, delay <= 330_000); + + // Run the task to verify rescheduling + task.run(); + + verify(eventLoop, org.mockito.Mockito.times(2)) + .schedule(any(Runnable.class), delayCaptor.capture(), eq(TimeUnit.MILLISECONDS)); + + // Re-arming jitter is 90% to 110%, so 270,000 ms to 330,000 ms + long rearmDelay = delayCaptor.getValue(); + org.junit.Assert.assertTrue( + "Delay should be >= 270000 but was " + rearmDelay, rearmDelay >= 270_000); + org.junit.Assert.assertTrue( + "Delay should be <= 330000 but was " + rearmDelay, rearmDelay <= 330_000); + } + + @Test + public void channelInactive_cancelsReportTimer() { + when(channel.isActive()).thenReturn(true); + metrics.channelActive(channel); + + metrics.channelInactive(channel); + + verify(scheduledFuture).cancel(false); + } +} diff --git a/okhttp/src/main/java/io/grpc/okhttp/InternalOkHttpServerBuilder.java b/okhttp/src/main/java/io/grpc/okhttp/InternalOkHttpServerBuilder.java index 78a409a3f85..df9756333b5 100644 --- a/okhttp/src/main/java/io/grpc/okhttp/InternalOkHttpServerBuilder.java +++ b/okhttp/src/main/java/io/grpc/okhttp/InternalOkHttpServerBuilder.java @@ -17,6 +17,7 @@ package io.grpc.okhttp; import io.grpc.Internal; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.InternalServer; import io.grpc.internal.TransportTracer; @@ -29,8 +30,9 @@ @Internal public final class InternalOkHttpServerBuilder { public static InternalServer buildTransportServers(OkHttpServerBuilder builder, - List streamTracerFactories) { - return builder.buildTransportServers(streamTracerFactories); + List streamTracerFactories, + MetricRecorder metricRecorder) { + return builder.buildTransportServers(streamTracerFactories, metricRecorder); } public static void setTransportTracerFactory(OkHttpServerBuilder builder, diff --git a/okhttp/src/main/java/io/grpc/okhttp/OkHttpServerBuilder.java b/okhttp/src/main/java/io/grpc/okhttp/OkHttpServerBuilder.java index 8daeed42a8c..3ace69052b7 100644 --- a/okhttp/src/main/java/io/grpc/okhttp/OkHttpServerBuilder.java +++ b/okhttp/src/main/java/io/grpc/okhttp/OkHttpServerBuilder.java @@ -27,6 +27,7 @@ import io.grpc.ForwardingServerBuilder; import io.grpc.InsecureServerCredentials; import io.grpc.Internal; +import io.grpc.MetricRecorder; import io.grpc.ServerBuilder; import io.grpc.ServerCredentials; import io.grpc.ServerStreamTracer; @@ -387,7 +388,8 @@ void setStatsEnabled(boolean value) { } InternalServer buildTransportServers( - List streamTracerFactories) { + List streamTracerFactories, + MetricRecorder metricRecorder) { return new OkHttpServer(this, streamTracerFactories, serverImplBuilder.getChannelz()); } diff --git a/okhttp/src/test/java/io/grpc/okhttp/OkHttpTransportTest.java b/okhttp/src/test/java/io/grpc/okhttp/OkHttpTransportTest.java index 076eea3349a..9317ca96639 100644 --- a/okhttp/src/test/java/io/grpc/okhttp/OkHttpTransportTest.java +++ b/okhttp/src/test/java/io/grpc/okhttp/OkHttpTransportTest.java @@ -17,6 +17,7 @@ package io.grpc.okhttp; import io.grpc.InsecureServerCredentials; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.AbstractTransportTest; import io.grpc.internal.ClientTransportFactory; @@ -58,11 +59,12 @@ protected InternalServer newServer( @Override protected InternalServer newServer( int port, List streamTracerFactories) { - return OkHttpServerBuilder + OkHttpServerBuilder builder = OkHttpServerBuilder .forPort(port, InsecureServerCredentials.create()) .flowControlWindow(AbstractTransportTest.TEST_FLOW_CONTROL_WINDOW) - .setTransportTracerFactory(fakeClockTransportTracer) - .buildTransportServers(streamTracerFactories); + .setTransportTracerFactory(fakeClockTransportTracer); + return InternalOkHttpServerBuilder + .buildTransportServers(builder, streamTracerFactories, new MetricRecorder() {}); } @Override diff --git a/opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java b/opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java index 6904340ac74..5b4172e6052 100644 --- a/opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java +++ b/opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java @@ -136,7 +136,7 @@ List getOptionalLabels() { return optionalLabels; } - MetricSink getSink() { + public MetricSink getSink() { return sink; } diff --git a/servlet/src/jettyTest/java/io/grpc/servlet/JettyTransportTest.java b/servlet/src/jettyTest/java/io/grpc/servlet/JettyTransportTest.java index 58143a8516c..240394b1e9e 100644 --- a/servlet/src/jettyTest/java/io/grpc/servlet/JettyTransportTest.java +++ b/servlet/src/jettyTest/java/io/grpc/servlet/JettyTransportTest.java @@ -18,6 +18,7 @@ import io.grpc.InternalChannelz; import io.grpc.InternalInstrumented; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.AbstractTransportTest; import io.grpc.internal.ClientTransportFactory; @@ -59,7 +60,9 @@ public class JettyTransportTest extends AbstractTransportTest { protected InternalServer newServer(List streamTracerFactories) { return new InternalServer() { final InternalServer delegate = - new ServletServerBuilder().buildTransportServers(streamTracerFactories); + new ServletServerBuilder().buildTransportServers( + streamTracerFactories, new MetricRecorder() { + }); @Override public void start(ServerListener listener) throws IOException { diff --git a/servlet/src/main/java/io/grpc/servlet/ServletServerBuilder.java b/servlet/src/main/java/io/grpc/servlet/ServletServerBuilder.java index aee25de01ad..17e32f3007c 100644 --- a/servlet/src/main/java/io/grpc/servlet/ServletServerBuilder.java +++ b/servlet/src/main/java/io/grpc/servlet/ServletServerBuilder.java @@ -31,6 +31,7 @@ import io.grpc.InternalInstrumented; import io.grpc.InternalLogId; import io.grpc.Metadata; +import io.grpc.MetricRecorder; import io.grpc.Server; import io.grpc.ServerBuilder; import io.grpc.ServerStreamTracer; @@ -159,7 +160,8 @@ public void transportTerminated() { @VisibleForTesting InternalServer buildTransportServers( - List streamTracerFactories) { + List streamTracerFactories, + MetricRecorder metricRecorder) { checkNotNull(streamTracerFactories, "streamTracerFactories"); this.streamTracerFactories = streamTracerFactories; internalServer = new InternalServerImpl(); diff --git a/servlet/src/tomcatTest/java/io/grpc/servlet/TomcatTransportTest.java b/servlet/src/tomcatTest/java/io/grpc/servlet/TomcatTransportTest.java index cd73b096ccb..d869f25ca4f 100644 --- a/servlet/src/tomcatTest/java/io/grpc/servlet/TomcatTransportTest.java +++ b/servlet/src/tomcatTest/java/io/grpc/servlet/TomcatTransportTest.java @@ -18,6 +18,7 @@ import io.grpc.InternalChannelz.SocketStats; import io.grpc.InternalInstrumented; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.AbstractTransportTest; import io.grpc.internal.ClientTransportFactory; @@ -71,8 +72,11 @@ public void tearDown() throws InterruptedException { @Override protected InternalServer newServer(List streamTracerFactories) { return new InternalServer() { + final ServletServerBuilder builder = new ServletServerBuilder(); final InternalServer delegate = - new ServletServerBuilder().buildTransportServers(streamTracerFactories); + builder.buildTransportServers( + streamTracerFactories, new MetricRecorder() { + }); @Override public void start(ServerListener listener) throws IOException { diff --git a/servlet/src/undertowTest/java/io/grpc/servlet/UndertowTransportTest.java b/servlet/src/undertowTest/java/io/grpc/servlet/UndertowTransportTest.java index ef897c87d70..c072b465c77 100644 --- a/servlet/src/undertowTest/java/io/grpc/servlet/UndertowTransportTest.java +++ b/servlet/src/undertowTest/java/io/grpc/servlet/UndertowTransportTest.java @@ -22,6 +22,7 @@ import io.grpc.InternalChannelz.SocketStats; import io.grpc.InternalInstrumented; +import io.grpc.MetricRecorder; import io.grpc.ServerStreamTracer; import io.grpc.internal.AbstractTransportTest; import io.grpc.internal.ClientTransportFactory; @@ -91,7 +92,9 @@ protected InternalServer newServer(List streamTracerFactories) { return new InternalServer() { final InternalServer delegate = - new ServletServerBuilder().buildTransportServers(streamTracerFactories); + new ServletServerBuilder().buildTransportServers( + streamTracerFactories, new MetricRecorder() { + }); @Override public void start(ServerListener listener) throws IOException {