From e20d2a8af29603fc405badcac8533cce168b8381 Mon Sep 17 00:00:00 2001 From: joecqupt Date: Tue, 20 Oct 2026 14:18:08 +0800 Subject: [PATCH] netty: Populate Channelz TcpInfo from netty-epoll The default NettySocketSupport helper always returned null, so Channelz never reported TCP_INFO for open-source users. When netty-epoll is available, read EpollSocketChannel.tcpInfo() via reflection and map all fields into InternalChannelz.TcpInfo. The epoll reflection previously private to TcpMetrics is extracted into EpollTcpInfoAccessor and shared by both callers. Also fix grpc.tcp.recurring_retransmits, which was reading tcpi_retrans (currently unacknowledged retransmitted segments) instead of tcpi_retransmits (number of retransmission timer expirations), which is what the metric is defined to count. --- .../io/grpc/netty/EpollTcpInfoAccessor.java | 142 ++++++++++++++ .../io/grpc/netty/NettySocketSupport.java | 72 ++++++- .../main/java/io/grpc/netty/TcpMetrics.java | 69 +------ .../grpc/netty/EpollTcpInfoAccessorTest.java | 46 +++++ .../java/io/grpc/netty/FakeEpollTcpInfo.java | 178 ++++++++++++++++++ .../io/grpc/netty/NettySocketSupportTest.java | 156 +++++++++++++++ .../java/io/grpc/netty/TcpMetricsTest.java | 148 ++++----------- 7 files changed, 636 insertions(+), 175 deletions(-) create mode 100644 netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java create mode 100644 netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java create mode 100644 netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java create mode 100644 netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java diff --git a/netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java b/netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java new file mode 100644 index 00000000000..00ed86f2d70 --- /dev/null +++ b/netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java @@ -0,0 +1,142 @@ +/* + * 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 com.google.common.annotations.VisibleForTesting; +import io.netty.channel.Channel; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.util.EnumMap; +import java.util.logging.Level; +import java.util.logging.Logger; +import javax.annotation.Nullable; + +/** + * Reflective access to {@code EpollSocketChannel.tcpInfo()}, since grpc-netty does not depend on + * netty-epoll at compile time. + */ +final class EpollTcpInfoAccessor { + private static final Logger log = Logger.getLogger(EpollTcpInfoAccessor.class.getName()); + + @Nullable + static final EpollTcpInfoAccessor INSTANCE = load(); + + enum Field { + STATE("state"), + CA_STATE("caState"), + RETRANSMITS("retransmits"), + PROBES("probes"), + BACKOFF("backoff"), + OPTIONS("options"), + SND_WSCALE("sndWscale"), + RCV_WSCALE("rcvWscale"), + RTO("rto"), + ATO("ato"), + SND_MSS("sndMss"), + RCV_MSS("rcvMss"), + UNACKED("unacked"), + SACKED("sacked"), + LOST("lost"), + RETRANS("retrans"), + FACKETS("fackets"), + LAST_DATA_SENT("lastDataSent"), + LAST_ACK_SENT("lastAckSent"), + LAST_DATA_RECV("lastDataRecv"), + LAST_ACK_RECV("lastAckRecv"), + PMTU("pmtu"), + RCV_SSTHRESH("rcvSsthresh"), + RTT("rtt"), + RTTVAR("rttvar"), + SND_SSTHRESH("sndSsthresh"), + SND_CWND("sndCwnd"), + ADVMSS("advmss"), + REORDERING("reordering"), + TOTAL_RETRANS("totalRetrans"); + + final String methodName; + + Field(String methodName) { + this.methodName = methodName; + } + } + + private final Class channelClass; + private final Constructor infoConstructor; + private final Method tcpInfo; + private final EnumMap getters; + + private EpollTcpInfoAccessor( + Class channelClass, Constructor infoConstructor, Method tcpInfo, + EnumMap getters) { + this.channelClass = channelClass; + this.infoConstructor = infoConstructor; + this.tcpInfo = tcpInfo; + this.getters = getters; + } + + @Nullable + private static EpollTcpInfoAccessor load() { + boolean epollAvailable = false; + try { + Class epollClass = Class.forName("io.netty.channel.epoll.Epoll"); + Method isAvailableMethod = epollClass.getDeclaredMethod("isAvailable"); + epollAvailable = (Boolean) isAvailableMethod.invoke(null); + if (epollAvailable) { + return create( + Class.forName("io.netty.channel.epoll.EpollSocketChannel"), + Class.forName("io.netty.channel.epoll.EpollTcpInfo")); + } + } catch (ReflectiveOperationException e) { + log.log(Level.FINE, "Failed to initialize Epoll tcp_info reflection", e); + } finally { + log.log(Level.FINE, "Epoll available during static init of EpollTcpInfoAccessor:" + + "{0}", epollAvailable); + } + return null; + } + + @VisibleForTesting + static EpollTcpInfoAccessor create(Class channelClass, Class infoClass) + throws ReflectiveOperationException { + EnumMap getters = new EnumMap<>(Field.class); + for (Field field : Field.values()) { + getters.put(field, infoClass.getMethod(field.methodName)); + } + return new EpollTcpInfoAccessor( + channelClass, + infoClass.getDeclaredConstructor(), + channelClass.getMethod("tcpInfo", infoClass), + getters); + } + + boolean isEpollSocketChannel(Channel channel) { + return channelClass.isInstance(channel); + } + + Object newTcpInfo() throws ReflectiveOperationException { + return infoConstructor.newInstance(); + } + + /** Fills {@code info} with the current {@code TCP_INFO} of {@code channel}. */ + void read(Channel channel, Object info) throws ReflectiveOperationException { + tcpInfo.invoke(channel, info); + } + + long get(Object info, Field field) throws ReflectiveOperationException { + return ((Number) getters.get(field).invoke(info)).longValue(); + } +} diff --git a/netty/src/main/java/io/grpc/netty/NettySocketSupport.java b/netty/src/main/java/io/grpc/netty/NettySocketSupport.java index 528a3978ce9..0f12783beec 100644 --- a/netty/src/main/java/io/grpc/netty/NettySocketSupport.java +++ b/netty/src/main/java/io/grpc/netty/NettySocketSupport.java @@ -16,18 +16,25 @@ package io.grpc.netty; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; import io.grpc.InternalChannelz.TcpInfo; +import io.grpc.netty.EpollTcpInfoAccessor.Field; import io.netty.channel.Channel; import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; import javax.annotation.Nullable; /** * An class for getting low level socket info. */ final class NettySocketSupport { - private static volatile Helper instance = new NettySocketHelperImpl(); + private static final Logger log = Logger.getLogger(NettySocketSupport.class.getName()); + + private static volatile Helper instance = + new NettySocketHelperImpl(EpollTcpInfoAccessor.INSTANCE); interface Helper { /** @@ -63,12 +70,67 @@ static void setHelper(Helper helper) { instance = Preconditions.checkNotNull(helper); } - private static final class NettySocketHelperImpl implements Helper { + @VisibleForTesting + static final class NettySocketHelperImpl implements Helper { + @Nullable + private final EpollTcpInfoAccessor epollInfo; + + NettySocketHelperImpl(@Nullable EpollTcpInfoAccessor epollInfo) { + this.epollInfo = epollInfo; + } + @Override + @Nullable public NativeSocketOptions getNativeSocketOptions(Channel ch) { - // TODO(zpencer): if netty-epoll, use reflection to call EpollSocketChannel.tcpInfo() - // And/or if some other low level socket support library is available, call it now. - return null; + if (epollInfo == null || !epollInfo.isEpollSocketChannel(ch)) { + return null; + } + TcpInfo tcpInfo; + try { + tcpInfo = readTcpInfo(epollInfo, ch); + } catch (ReflectiveOperationException e) { + log.log(Level.FINE, "Failed to read Epoll tcp_info", e); + return null; + } + return new NativeSocketOptions(tcpInfo, ImmutableMap.of()); + } + + // Channelz fields are uint32, so narrowing the epoll values to int keeps their bits. + private static TcpInfo readTcpInfo(EpollTcpInfoAccessor epollInfo, Channel ch) + throws ReflectiveOperationException { + Object info = epollInfo.newTcpInfo(); + epollInfo.read(ch, info); + return new TcpInfo.Builder() + .setState((int) epollInfo.get(info, Field.STATE)) + .setCaState((int) epollInfo.get(info, Field.CA_STATE)) + .setRetransmits((int) epollInfo.get(info, Field.RETRANSMITS)) + .setProbes((int) epollInfo.get(info, Field.PROBES)) + .setBackoff((int) epollInfo.get(info, Field.BACKOFF)) + .setOptions((int) epollInfo.get(info, Field.OPTIONS)) + .setSndWscale((int) epollInfo.get(info, Field.SND_WSCALE)) + .setRcvWscale((int) epollInfo.get(info, Field.RCV_WSCALE)) + .setRto((int) epollInfo.get(info, Field.RTO)) + .setAto((int) epollInfo.get(info, Field.ATO)) + .setSndMss((int) epollInfo.get(info, Field.SND_MSS)) + .setRcvMss((int) epollInfo.get(info, Field.RCV_MSS)) + .setUnacked((int) epollInfo.get(info, Field.UNACKED)) + .setSacked((int) epollInfo.get(info, Field.SACKED)) + .setLost((int) epollInfo.get(info, Field.LOST)) + .setRetrans((int) epollInfo.get(info, Field.RETRANS)) + .setFackets((int) epollInfo.get(info, Field.FACKETS)) + .setLastDataSent((int) epollInfo.get(info, Field.LAST_DATA_SENT)) + .setLastAckSent((int) epollInfo.get(info, Field.LAST_ACK_SENT)) + .setLastDataRecv((int) epollInfo.get(info, Field.LAST_DATA_RECV)) + .setLastAckRecv((int) epollInfo.get(info, Field.LAST_ACK_RECV)) + .setPmtu((int) epollInfo.get(info, Field.PMTU)) + .setRcvSsthresh((int) epollInfo.get(info, Field.RCV_SSTHRESH)) + .setRtt((int) epollInfo.get(info, Field.RTT)) + .setRttvar((int) epollInfo.get(info, Field.RTTVAR)) + .setSndSsthresh((int) epollInfo.get(info, Field.SND_SSTHRESH)) + .setSndCwnd((int) epollInfo.get(info, Field.SND_CWND)) + .setAdvmss((int) epollInfo.get(info, Field.ADVMSS)) + .setReordering((int) epollInfo.get(info, Field.REORDERING)) + .build(); } } } diff --git a/netty/src/main/java/io/grpc/netty/TcpMetrics.java b/netty/src/main/java/io/grpc/netty/TcpMetrics.java index 10f3a55014d..34edf1d484a 100644 --- a/netty/src/main/java/io/grpc/netty/TcpMetrics.java +++ b/netty/src/main/java/io/grpc/netty/TcpMetrics.java @@ -21,7 +21,6 @@ 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; @@ -38,57 +37,7 @@ final class TcpMetrics { private static final Logger log = Logger.getLogger(TcpMetrics.class.getName()); - static EpollInfo epollInfo = loadEpollInfo(); - - static final class EpollInfo { - final Class channelClass; - final java.lang.reflect.Constructor infoConstructor; - final Method tcpInfo; - final Method totalRetrans; - final Method retransmits; - final Method rtt; - - EpollInfo( - Class channelClass, - java.lang.reflect.Constructor infoConstructor, - Method tcpInfo, - Method totalRetrans, - Method retransmits, - Method rtt) { - this.channelClass = channelClass; - this.infoConstructor = infoConstructor; - this.tcpInfo = tcpInfo; - this.totalRetrans = totalRetrans; - this.retransmits = retransmits; - this.rtt = rtt; - } - } - - static EpollInfo loadEpollInfo() { - boolean epollAvailable = false; - try { - Class epollClass = Class.forName("io.netty.channel.epoll.Epoll"); - Method isAvailableMethod = epollClass.getDeclaredMethod("isAvailable"); - epollAvailable = (Boolean) isAvailableMethod.invoke(null); - if (epollAvailable) { - Class channelClass = Class.forName("io.netty.channel.epoll.EpollSocketChannel"); - Class infoClass = Class.forName("io.netty.channel.epoll.EpollTcpInfo"); - return new EpollInfo( - channelClass, - infoClass.getDeclaredConstructor(), - channelClass.getMethod("tcpInfo", infoClass), - infoClass.getMethod("totalRetrans"), - infoClass.getMethod("retrans"), - infoClass.getMethod("rtt")); - } - } catch (ReflectiveOperationException e) { - log.log(Level.FINE, "Failed to initialize Epoll tcp_info reflection", e); - } finally { - log.log(Level.FINE, "Epoll available during static init of TcpMetrics:" - + "{0}", epollAvailable); - } - return null; - } + static EpollTcpInfoAccessor epollInfo = EpollTcpInfoAccessor.INSTANCE; private static final long RECORD_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(5); private final MetricRecorder metricRecorder; @@ -102,7 +51,7 @@ static EpollInfo loadEpollInfo() { Object tcpInfo = null; if (epollInfo != null) { try { - tcpInfo = epollInfo.infoConstructor.newInstance(); + tcpInfo = epollInfo.newTcpInfo(); } catch (ReflectiveOperationException e) { log.log(Level.FINE, "Failed to instantiate EpollTcpInfo", e); } @@ -120,7 +69,7 @@ void channelActive(Channel channel) { } private void scheduleNextReport(final Channel channel, boolean isInitial) { - if (epollInfo == null || !epollInfo.channelClass.isInstance(channel) || !channel.isActive()) { + if (epollInfo == null || !epollInfo.isEpollSocketChannel(channel) || !channel.isActive()) { return; } @@ -147,7 +96,7 @@ void channelInactive(Channel channel) { metricRecorder.addLongUpDownCounter(InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT, -1, Collections.emptyList(), labelValues); // Final collection on close - if (epollInfo != null && epollInfo.channelClass.isInstance(channel)) { + if (epollInfo != null && epollInfo.isEpollSocketChannel(channel)) { recordTcpInfo(channel, true); } } @@ -157,7 +106,7 @@ void recordTcpInfo(Channel channel) { } private void recordTcpInfo(Channel channel, boolean isClose) { - if (epollInfo == null || !epollInfo.channelClass.isInstance(channel)) { + if (epollInfo == null || !epollInfo.isEpollSocketChannel(channel)) { return; } List labelValues = getLabelValues(channel); @@ -165,10 +114,10 @@ private void recordTcpInfo(Channel channel, boolean isClose) { long retransmits; long rtt; try { - epollInfo.tcpInfo.invoke(channel, tcpInfo); - totalRetrans = (Long) epollInfo.totalRetrans.invoke(tcpInfo); - retransmits = (Long) epollInfo.retransmits.invoke(tcpInfo); - rtt = (Long) epollInfo.rtt.invoke(tcpInfo); + epollInfo.read(channel, tcpInfo); + totalRetrans = epollInfo.get(tcpInfo, EpollTcpInfoAccessor.Field.TOTAL_RETRANS); + retransmits = epollInfo.get(tcpInfo, EpollTcpInfoAccessor.Field.RETRANSMITS); + rtt = epollInfo.get(tcpInfo, EpollTcpInfoAccessor.Field.RTT); } catch (ReflectiveOperationException e) { log.log(Level.FINE, "Error computing TCP metrics", e); return; diff --git a/netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java b/netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java new file mode 100644 index 00000000000..56440a4d9af --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java @@ -0,0 +1,46 @@ +/* + * 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 com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.epoll.EpollSocketChannel; +import io.netty.channel.epoll.EpollTcpInfo; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class EpollTcpInfoAccessorTest { + + @Test + public void create_resolvesAllFieldsOnNettyEpollClasses() throws Exception { + // Class literals do not trigger epoll's native initialization, so this runs on any platform. + EpollTcpInfoAccessor accessor = + EpollTcpInfoAccessor.create(EpollSocketChannel.class, EpollTcpInfo.class); + + assertThat(accessor.isEpollSocketChannel(new EmbeddedChannel())).isFalse(); + } + + @Test + public void create_missingGetter_throws() { + assertThrows(NoSuchMethodException.class, + () -> EpollTcpInfoAccessor.create(FakeEpollTcpInfo.FakeChannel.class, Object.class)); + } +} diff --git a/netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java b/netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java new file mode 100644 index 00000000000..3ab18b987d6 --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java @@ -0,0 +1,178 @@ +/* + * 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.netty.EpollTcpInfoAccessor.Field; +import io.netty.channel.embedded.EmbeddedChannel; +import java.util.EnumMap; + +/** Mirrors the public API of {@code io.netty.channel.epoll.EpollTcpInfo}. */ +public final class FakeEpollTcpInfo { + private final EnumMap values = new EnumMap<>(Field.class); + + static EpollTcpInfoAccessor newAccessor() throws ReflectiveOperationException { + return EpollTcpInfoAccessor.create(FakeChannel.class, FakeEpollTcpInfo.class); + } + + public void set(Field field, long value) { + values.put(field, value); + } + + public void setValues(long totalRetrans, long retransmits, long rtt) { + set(Field.TOTAL_RETRANS, totalRetrans); + set(Field.RETRANSMITS, retransmits); + set(Field.RTT, rtt); + } + + long get(Field field) { + return values.getOrDefault(field, 0L); + } + + public int state() { + return (int) get(Field.STATE); + } + + public int caState() { + return (int) get(Field.CA_STATE); + } + + public int retransmits() { + return (int) get(Field.RETRANSMITS); + } + + public int probes() { + return (int) get(Field.PROBES); + } + + public int backoff() { + return (int) get(Field.BACKOFF); + } + + public int options() { + return (int) get(Field.OPTIONS); + } + + public int sndWscale() { + return (int) get(Field.SND_WSCALE); + } + + public int rcvWscale() { + return (int) get(Field.RCV_WSCALE); + } + + public long rto() { + return get(Field.RTO); + } + + public long ato() { + return get(Field.ATO); + } + + public long sndMss() { + return get(Field.SND_MSS); + } + + public long rcvMss() { + return get(Field.RCV_MSS); + } + + public long unacked() { + return get(Field.UNACKED); + } + + public long sacked() { + return get(Field.SACKED); + } + + public long lost() { + return get(Field.LOST); + } + + public long retrans() { + return get(Field.RETRANS); + } + + public long fackets() { + return get(Field.FACKETS); + } + + public long lastDataSent() { + return get(Field.LAST_DATA_SENT); + } + + public long lastAckSent() { + return get(Field.LAST_ACK_SENT); + } + + public long lastDataRecv() { + return get(Field.LAST_DATA_RECV); + } + + public long lastAckRecv() { + return get(Field.LAST_ACK_RECV); + } + + public long pmtu() { + return get(Field.PMTU); + } + + public long rcvSsthresh() { + return get(Field.RCV_SSTHRESH); + } + + public long rtt() { + return get(Field.RTT); + } + + public long rttvar() { + return get(Field.RTTVAR); + } + + public long sndSsthresh() { + return get(Field.SND_SSTHRESH); + } + + public long sndCwnd() { + return get(Field.SND_CWND); + } + + public long advmss() { + return get(Field.ADVMSS); + } + + public long reordering() { + return get(Field.REORDERING); + } + + public long totalRetrans() { + return get(Field.TOTAL_RETRANS); + } + + /** Mirrors {@code EpollSocketChannel.tcpInfo(EpollTcpInfo)}, copying from a source info. */ + public static final class FakeChannel extends EmbeddedChannel { + private final FakeEpollTcpInfo source; + + public FakeChannel(FakeEpollTcpInfo source) { + this.source = source; + } + + public void tcpInfo(FakeEpollTcpInfo info) { + info.values.clear(); + info.values.putAll(source.values); + } + } +} diff --git a/netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java b/netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java new file mode 100644 index 00000000000..4ec8b47dd43 --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java @@ -0,0 +1,156 @@ +/* + * 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 com.google.common.truth.Truth.assertThat; +import static com.google.common.truth.TruthJUnit.assume; + +import io.grpc.InternalChannelz.TcpInfo; +import io.grpc.netty.EpollTcpInfoAccessor.Field; +import io.grpc.netty.NettySocketSupport.NativeSocketOptions; +import io.grpc.netty.NettySocketSupport.NettySocketHelperImpl; +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelException; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.epoll.EpollIoHandler; +import io.netty.channel.epoll.EpollServerSocketChannel; +import io.netty.channel.epoll.EpollSocketChannel; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class NettySocketSupportTest { + private static final int TCP_ESTABLISHED = 1; + + @Test + public void epollSocketChannel_populatesAllTcpInfoFields() throws Exception { + FakeEpollTcpInfo source = new FakeEpollTcpInfo(); + long value = 1; + for (Field field : Field.values()) { + source.set(field, value++); + } + NettySocketHelperImpl helper = new NettySocketHelperImpl(FakeEpollTcpInfo.newAccessor()); + + NativeSocketOptions options = + helper.getNativeSocketOptions(new FakeEpollTcpInfo.FakeChannel(source)); + + assertThat(options.otherInfo).isEmpty(); + TcpInfo tcpInfo = options.tcpInfo; + assertThat(tcpInfo.state).isEqualTo((int) source.get(Field.STATE)); + assertThat(tcpInfo.caState).isEqualTo((int) source.get(Field.CA_STATE)); + assertThat(tcpInfo.retransmits).isEqualTo((int) source.get(Field.RETRANSMITS)); + assertThat(tcpInfo.probes).isEqualTo((int) source.get(Field.PROBES)); + assertThat(tcpInfo.backoff).isEqualTo((int) source.get(Field.BACKOFF)); + assertThat(tcpInfo.options).isEqualTo((int) source.get(Field.OPTIONS)); + assertThat(tcpInfo.sndWscale).isEqualTo((int) source.get(Field.SND_WSCALE)); + assertThat(tcpInfo.rcvWscale).isEqualTo((int) source.get(Field.RCV_WSCALE)); + assertThat(tcpInfo.rto).isEqualTo((int) source.get(Field.RTO)); + assertThat(tcpInfo.ato).isEqualTo((int) source.get(Field.ATO)); + assertThat(tcpInfo.sndMss).isEqualTo((int) source.get(Field.SND_MSS)); + assertThat(tcpInfo.rcvMss).isEqualTo((int) source.get(Field.RCV_MSS)); + assertThat(tcpInfo.unacked).isEqualTo((int) source.get(Field.UNACKED)); + assertThat(tcpInfo.sacked).isEqualTo((int) source.get(Field.SACKED)); + assertThat(tcpInfo.lost).isEqualTo((int) source.get(Field.LOST)); + assertThat(tcpInfo.retrans).isEqualTo((int) source.get(Field.RETRANS)); + assertThat(tcpInfo.fackets).isEqualTo((int) source.get(Field.FACKETS)); + assertThat(tcpInfo.lastDataSent).isEqualTo((int) source.get(Field.LAST_DATA_SENT)); + assertThat(tcpInfo.lastAckSent).isEqualTo((int) source.get(Field.LAST_ACK_SENT)); + assertThat(tcpInfo.lastDataRecv).isEqualTo((int) source.get(Field.LAST_DATA_RECV)); + assertThat(tcpInfo.lastAckRecv).isEqualTo((int) source.get(Field.LAST_ACK_RECV)); + assertThat(tcpInfo.pmtu).isEqualTo((int) source.get(Field.PMTU)); + assertThat(tcpInfo.rcvSsthresh).isEqualTo((int) source.get(Field.RCV_SSTHRESH)); + assertThat(tcpInfo.rtt).isEqualTo((int) source.get(Field.RTT)); + assertThat(tcpInfo.rttvar).isEqualTo((int) source.get(Field.RTTVAR)); + assertThat(tcpInfo.sndSsthresh).isEqualTo((int) source.get(Field.SND_SSTHRESH)); + assertThat(tcpInfo.sndCwnd).isEqualTo((int) source.get(Field.SND_CWND)); + assertThat(tcpInfo.advmss).isEqualTo((int) source.get(Field.ADVMSS)); + assertThat(tcpInfo.reordering).isEqualTo((int) source.get(Field.REORDERING)); + } + + @Test + public void nonEpollChannel_returnsNull() throws Exception { + NettySocketHelperImpl helper = new NettySocketHelperImpl(FakeEpollTcpInfo.newAccessor()); + + assertThat(helper.getNativeSocketOptions(new EmbeddedChannel())).isNull(); + } + + @Test + public void epollUnavailable_returnsNull() { + NettySocketHelperImpl helper = new NettySocketHelperImpl(null); + + assertThat(helper.getNativeSocketOptions(new EmbeddedChannel())).isNull(); + } + + @Test + public void closedChannel_tcpInfoFails_returnsNull() throws Exception { + NettySocketHelperImpl helper = new NettySocketHelperImpl( + EpollTcpInfoAccessor.create(ClosedFdChannel.class, FakeEpollTcpInfo.class)); + ClosedFdChannel channel = new ClosedFdChannel(); + channel.close(); + + assertThat(helper.getNativeSocketOptions(channel)).isNull(); + } + + @Test + public void realEpollSocketChannel_reportsEstablishedTcpInfo() throws Exception { + assume().that(Utils.isEpollAvailable()).isTrue(); + + EventLoopGroup group = new MultiThreadIoEventLoopGroup(1, EpollIoHandler.newFactory()); + try { + Channel server = new ServerBootstrap() + .group(group) + .channel(EpollServerSocketChannel.class) + .childHandler(new ChannelInboundHandlerAdapter()) + .bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)) + .sync() + .channel(); + Channel client = new Bootstrap() + .group(group) + .channel(EpollSocketChannel.class) + .handler(new ChannelInboundHandlerAdapter()) + .connect(server.localAddress()) + .sync() + .channel(); + + TcpInfo tcpInfo = Utils.getSocketOptions(client).tcpInfo; + + assertThat(tcpInfo).isNotNull(); + assertThat(tcpInfo.state).isEqualTo(TCP_ESTABLISHED); + assertThat(tcpInfo.sndMss).isGreaterThan(0); + client.close().sync(); + server.close().sync(); + } finally { + group.shutdownGracefully(); + } + } + + public static final class ClosedFdChannel extends EmbeddedChannel { + public void tcpInfo(FakeEpollTcpInfo info) { + if (!isOpen()) { + throw new ChannelException("getsockopt() failed: Bad file descriptor"); + } + } + } +} diff --git a/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java b/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java index f75a98b46df..956fff77cd3 100644 --- a/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java +++ b/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java @@ -57,19 +57,19 @@ public class TcpMetricsTest { @Mock private MetricRecorder metricRecorder; - private ConfigurableFakeWithTcpInfo channel; + private FakeEpollTcpInfo.FakeChannel channel; private TcpMetrics metrics; @Before public void setUp() throws Exception { FakeEpollTcpInfo dummyInfo = new FakeEpollTcpInfo(); - channel = new ConfigurableFakeWithTcpInfo(dummyInfo); + channel = new FakeEpollTcpInfo.FakeChannel(dummyInfo); metrics = new TcpMetrics(metricRecorder); } @After public void tearDown() throws Exception { - TcpMetrics.epollInfo = TcpMetrics.loadEpollInfo(); + TcpMetrics.epollInfo = EpollTcpInfoAccessor.INSTANCE; } @Test @@ -82,48 +82,15 @@ public void metricsInitialization() { assertNotNull(InternalTcpMetrics.MIN_RTT_INSTRUMENT); } - 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() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); channel.writeInbound("dummy"); tracker.channelInactive(channel); @@ -139,21 +106,33 @@ public void tracker_recordTcpInfo_reflectionSuccess() throws Exception { eq(0.005), any(), any()); } + @Test + public void tracker_recurringRetransmits_usesRetransmitsNotRetrans() throws Exception { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); + TcpMetrics tracker = new TcpMetrics(recorder); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + infoSource.set(EpollTcpInfoAccessor.Field.RETRANS, 9); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); + + tracker.channelInactive(channel); + + verify(recorder).addLongCounter( + eq(Objects.requireNonNull(InternalTcpMetrics.RECURRING_RETRANSMITS_INSTRUMENT)), + eq(4L), any(), any()); + } + @Test public void tracker_periodicRecord_doesNotRecordRecurringRetransmits() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -180,18 +159,12 @@ public void tracker_periodicRecord_doesNotRecordRecurringRetransmits() throws Ex @Test public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -207,7 +180,7 @@ public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() t // Let's just create a new channel instance where tcpInfo sets retrans=5. FakeEpollTcpInfo infoSource2 = new FakeEpollTcpInfo(); infoSource2.setValues(130, 5, 5000); - ConfigurableFakeWithTcpInfo channel2 = new ConfigurableFakeWithTcpInfo(infoSource2); + FakeEpollTcpInfo.FakeChannel channel2 = new FakeEpollTcpInfo.FakeChannel(infoSource2); tracker.channelInactive(channel2); @@ -224,18 +197,12 @@ public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() t @Test public void tracker_periodicRecord_reportsDeltaForTotalRetrans() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -278,18 +245,12 @@ public void tracker_periodicRecord_reportsDeltaForTotalRetrans() throws Exceptio @Test public void tracker_periodicRecord_doesNotReportZeroDeltaForTotalRetrans() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -326,21 +287,6 @@ public void tracker_periodicRecord_doesNotReportZeroDeltaForTotalRetrans() throw 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; - } - } - private static class AddressOverrideEmbeddedChannel extends io.netty.channel.embedded.EmbeddedChannel { private final SocketAddress local; @@ -366,17 +312,11 @@ public SocketAddress remoteAddress() { public void tracker_reportsDeltas_correctly() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); // 10 retransmits total infoSource.setValues(10, 2, 1000); @@ -552,18 +492,12 @@ public void channelInactive_decrementsCount_noEpoll_noError() { @Test public void channelActive_schedulesReportTimer() throws Exception { - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); metrics = new TcpMetrics(metricRecorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); metrics.channelActive(channel); @@ -591,18 +525,12 @@ public void channelActive_schedulesReportTimer() throws Exception { @Test public void channelInactive_cancelsReportTimer() throws Exception { - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); metrics = new TcpMetrics(metricRecorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); metrics.channelActive(channel);