Skip to content

Commit 9d55ee5

Browse files
AgraVatorAgraVator
committed
fix: suggested changes
1 parent e8e8a24 commit 9d55ee5

11 files changed

Lines changed: 289 additions & 271 deletions

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -680,7 +680,7 @@ public <T> Builder setArg(Key<T> key, T value) {
680680
* See {@link Args#getMetricRecorder()}. This is an optional field.
681681
*/
682682
public Builder setMetricRecorder(MetricRecorder metricRecorder) {
683-
this.metricRecorder = checkNotNull(metricRecorder);
683+
this.metricRecorder = checkNotNull(metricRecorder, "metricRecorder");
684684
return this;
685685
}
686686

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,13 @@
1818

1919
import static com.google.common.truth.Truth.assertThat;
2020
import static org.junit.Assert.assertEquals;
21-
import static org.mockito.Mockito.mock;
2221

2322
import io.grpc.InternalConfigurator;
2423
import io.grpc.InternalConfiguratorRegistry;
2524
import io.grpc.Metadata;
2625
import io.grpc.MetricRecorder;
2726
import io.grpc.MetricSink;
27+
import io.grpc.NoopMetricSink;
2828
import io.grpc.ServerBuilder;
2929
import io.grpc.ServerCall;
3030
import io.grpc.ServerCallHandler;
@@ -134,9 +134,9 @@ public void getTracerFactories_disableBoth() {
134134

135135
@Test
136136
public void addMetricSink_addsToSinks() {
137-
MetricSink mockSink = mock(MetricSink.class);
138-
builder.addMetricSink(mockSink);
139-
assertThat(builder.metricSinks).containsExactly(mockSink);
137+
MetricSink noopMetricSink = new NoopMetricSink();
138+
builder.addMetricSink(noopMetricSink);
139+
assertThat(builder.metricSinks).containsExactly(noopMetricSink);
140140
}
141141

142142
@Test

‎netty/src/main/java/io/grpc/netty/NettyClientHandler.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,7 @@ class NettyClientHandler extends AbstractNettyHandler {
124124
private final Supplier<Stopwatch> stopwatchFactory;
125125
private final TransportTracer transportTracer;
126126
private final Attributes eagAttributes;
127-
private final TcpMetrics.Tracker tcpMetrics;
127+
private final TcpMetrics tcpMetrics;
128128
private final String authority;
129129
private final InUseStateAggregator<Http2Stream> inUseState =
130130
new InUseStateAggregator<Http2Stream>() {
@@ -357,7 +357,7 @@ public void onStreamClosed(Http2Stream stream) {
357357
}
358358
}
359359
});
360-
this.tcpMetrics = new TcpMetrics.Tracker(metricRecorder);
360+
this.tcpMetrics = new TcpMetrics(metricRecorder);
361361
}
362362

363363
/**

‎netty/src/main/java/io/grpc/netty/NettyServerBuilder.java‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import io.grpc.ExperimentalApi;
3333
import io.grpc.ForwardingServerBuilder;
3434
import io.grpc.Internal;
35-
import io.grpc.MetricSink;
3635
import io.grpc.ServerBuilder;
3736
import io.grpc.ServerCredentials;
3837
import io.grpc.ServerStreamTracer;
@@ -765,13 +764,6 @@ NettyServerBuilder setTransportTracerFactory(TransportTracer.Factory transportTr
765764
return this;
766765
}
767766

768-
@CanIgnoreReturnValue
769-
@Override
770-
public NettyServerBuilder addMetricSink(MetricSink metricSink) {
771-
serverImplBuilder.addMetricSink(metricSink);
772-
return this;
773-
}
774-
775767
@CanIgnoreReturnValue
776768
@Override
777769
public NettyServerBuilder useTransportSecurity(File certChain, File privateKey) {

‎netty/src/main/java/io/grpc/netty/NettyServerHandler.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ class NettyServerHandler extends AbstractNettyHandler {
128128
private final Http2Connection.PropertyKey streamKey;
129129
private final ServerTransportListener transportListener;
130130
private final int maxMessageSize;
131-
private final TcpMetrics.Tracker tcpMetrics;
131+
private final TcpMetrics tcpMetrics;
132132
private final long keepAliveTimeInNanos;
133133
private final long keepAliveTimeoutInNanos;
134134
private final long maxConnectionAgeInNanos;
@@ -369,7 +369,7 @@ public void onStreamClosed(Http2Stream stream) {
369369

370370
checkArgument(maxMessageSize >= 0, "maxMessageSize must be non-negative: %s", maxMessageSize);
371371
this.maxMessageSize = maxMessageSize;
372-
this.tcpMetrics = new TcpMetrics.Tracker(metricRecorder);
372+
this.tcpMetrics = new TcpMetrics(metricRecorder);
373373
this.keepAliveTimeInNanos = keepAliveTimeInNanos;
374374
this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;
375375
this.maxConnectionIdleManager = maxConnectionIdleManager;
@@ -677,8 +677,8 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
677677

678678
@Override
679679
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
680-
tcpMetrics.channelInactive(ctx.channel());
681680
try {
681+
tcpMetrics.channelInactive(ctx.channel());
682682
if (keepAliveManager != null) {
683683
keepAliveManager.onTransportTermination();
684684
}

‎netty/src/main/java/io/grpc/netty/TcpMetrics.java‎

Lines changed: 98 additions & 98 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package io.grpc.netty;
1818

19+
import com.google.common.annotations.VisibleForTesting;
1920
import io.grpc.InternalTcpMetrics;
2021
import io.grpc.MetricRecorder;
2122
import io.netty.channel.Channel;
@@ -66,7 +67,7 @@ static final class EpollInfo {
6667
}
6768
}
6869

69-
private static EpollInfo loadEpollInfo() {
70+
static EpollInfo loadEpollInfo() {
7071
boolean epollAvailable = false;
7172
try {
7273
Class<?> epollClass = Class.forName("io.netty.channel.epoll.Epoll");
@@ -84,123 +85,117 @@ private static EpollInfo loadEpollInfo() {
8485
infoClass.getMethod("retrans"),
8586
infoClass.getMethod("rtt"));
8687
}
87-
} catch (ReflectiveOperationException | RuntimeException e) {
88+
} catch (ReflectiveOperationException e) {
8889
log.log(Level.FINE, "Failed to initialize Epoll tcp_info reflection", e);
89-
} catch (Error e) {
90-
log.log(Level.FINE, "Failed to load native Epoll library", e);
9190
} finally {
9291
log.log(Level.INFO, "Epoll available during static init of TcpMetrics:"
9392
+ "{0}", epollAvailable);
9493
}
9594
return null;
9695
}
9796

98-
static final class Tracker {
99-
private final MetricRecorder metricRecorder;
100-
private final Object tcpInfo;
97+
private static final long RECORD_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(5);
98+
private final MetricRecorder metricRecorder;
99+
private final Object tcpInfo;
100+
private long lastTotalRetrans = 0;
101+
private ScheduledFuture<?> reportTimer;
101102

102-
private long lastTotalRetrans = 0;
103+
TcpMetrics(MetricRecorder metricRecorder) {
104+
this.metricRecorder = metricRecorder;
103105

104-
Tracker(MetricRecorder metricRecorder) {
105-
this.metricRecorder = metricRecorder;
106-
107-
Object tcpInfo = null;
108-
if (epollInfo != null) {
109-
try {
110-
tcpInfo = epollInfo.infoConstructor.newInstance();
111-
} catch (ReflectiveOperationException e) {
112-
log.log(Level.FINE, "Failed to instantiate EpollTcpInfo", e);
113-
}
106+
Object tcpInfo = null;
107+
if (epollInfo != null) {
108+
try {
109+
tcpInfo = epollInfo.infoConstructor.newInstance();
110+
} catch (ReflectiveOperationException e) {
111+
log.log(Level.FINE, "Failed to instantiate EpollTcpInfo", e);
114112
}
115-
this.tcpInfo = tcpInfo;
116113
}
114+
this.tcpInfo = tcpInfo;
115+
}
117116

118-
private static final long RECORD_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(5);
119-
private ScheduledFuture<?> reportTimer;
117+
void channelActive(Channel channel) {
118+
List<String> labelValues = getLabelValues(channel);
119+
metricRecorder.addLongCounter(InternalTcpMetrics.CONNECTIONS_CREATED_INSTRUMENT, 1,
120+
Collections.emptyList(), labelValues);
121+
metricRecorder.addLongUpDownCounter(InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT, 1,
122+
Collections.emptyList(), labelValues);
123+
scheduleNextReport(channel, true);
124+
}
120125

121-
void channelActive(Channel channel) {
122-
List<String> labelValues = getLabelValues(channel);
123-
metricRecorder.addLongCounter(InternalTcpMetrics.CONNECTIONS_CREATED_INSTRUMENT, 1,
124-
Collections.emptyList(), labelValues);
125-
metricRecorder.addLongUpDownCounter(InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT, 1,
126-
Collections.emptyList(), labelValues);
127-
scheduleNextReport(channel, true);
126+
private void scheduleNextReport(final Channel channel, boolean isInitial) {
127+
if (epollInfo == null || !epollInfo.channelClass.isInstance(channel) || !channel.isActive()) {
128+
return;
128129
}
129130

130-
private void scheduleNextReport(final Channel channel, boolean isInitial) {
131-
if (!channel.isActive()) {
132-
return;
131+
// Initial report has a larger jitter range to spread out initial connections.
132+
// Subsequent reports have a smaller jitter range to avoid drift.
133+
double jitter = isInitial
134+
? 0.1 + ThreadLocalRandom.current().nextDouble() // 10% to 110%
135+
: 0.9 + ThreadLocalRandom.current().nextDouble() * 0.2; // 90% to 110%
136+
long rearmingDelay = (long) (RECORD_INTERVAL_MILLIS * jitter);
137+
138+
reportTimer = channel.eventLoop().schedule(() -> {
139+
if (channel.isActive()) {
140+
recordTcpInfo(channel, false);
141+
scheduleNextReport(channel, false); // Re-arm
133142
}
143+
}, rearmingDelay, TimeUnit.MILLISECONDS);
144+
}
134145

135-
double jitter = isInitial
136-
? 0.1 + ThreadLocalRandom.current().nextDouble() // 10% to 110%
137-
: 0.9 + ThreadLocalRandom.current().nextDouble() * 0.2; // 90% to 110%
138-
long rearmingDelay = (long) (RECORD_INTERVAL_MILLIS * jitter);
139-
140-
reportTimer = channel.eventLoop().schedule(() -> {
141-
if (channel.isActive()) {
142-
Tracker.this.recordTcpInfo(channel, false);
143-
scheduleNextReport(channel, false); // Re-arm
144-
}
145-
}, rearmingDelay, TimeUnit.MILLISECONDS);
146+
void channelInactive(Channel channel) {
147+
if (reportTimer != null) {
148+
reportTimer.cancel(false);
146149
}
147-
148-
void channelInactive(Channel channel) {
149-
if (reportTimer != null) {
150-
reportTimer.cancel(false);
151-
}
152-
List<String> labelValues = getLabelValues(channel);
153-
metricRecorder.addLongUpDownCounter(InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT, -1,
154-
Collections.emptyList(), labelValues);
155-
// Final collection on close
150+
List<String> labelValues = getLabelValues(channel);
151+
metricRecorder.addLongUpDownCounter(InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT, -1,
152+
Collections.emptyList(), labelValues);
153+
// Final collection on close
154+
if (epollInfo != null && epollInfo.channelClass.isInstance(channel)) {
156155
recordTcpInfo(channel, true);
157156
}
157+
}
158158

159-
void recordTcpInfo(Channel channel) {
160-
recordTcpInfo(channel, false);
161-
}
159+
void recordTcpInfo(Channel channel) {
160+
recordTcpInfo(channel, false);
161+
}
162162

163-
private void recordTcpInfo(Channel channel, boolean isClose) {
164-
if (epollInfo == null) {
165-
log.log(Level.FINE, "Skipping recordTcpInfo because"
166-
+ "epollInfo is null");
167-
return;
168-
}
169-
if (!epollInfo.channelClass.isInstance(channel)) {
170-
log.log(Level.FINE, "Skipping recordTcpInfo because channel is not an"
171-
+ "instance of epollSocketChannelClass: {0}",
172-
channel.getClass()
173-
.getName());
174-
return;
175-
}
176-
List<String> labelValues = getLabelValues(channel);
177-
long totalRetrans;
178-
long retransmits;
179-
long rtt;
180-
try {
181-
epollInfo.tcpInfo.invoke(channel, tcpInfo);
182-
totalRetrans = (Long) epollInfo.totalRetrans.invoke(tcpInfo);
183-
retransmits = (Long) epollInfo.retransmits.invoke(tcpInfo);
184-
rtt = (Long) epollInfo.rtt.invoke(tcpInfo);
185-
} catch (ReflectiveOperationException e) {
186-
log.log(Level.FINE, "Error computing TCP metrics", e);
187-
return;
188-
}
163+
private void recordTcpInfo(Channel channel, boolean isClose) {
164+
if (epollInfo == null || !epollInfo.channelClass.isInstance(channel)) {
165+
return;
166+
}
167+
List<String> labelValues = getLabelValues(channel);
168+
long totalRetrans;
169+
long retransmits;
170+
long rtt;
171+
try {
172+
epollInfo.tcpInfo.invoke(channel, tcpInfo);
173+
totalRetrans = (Long) epollInfo.totalRetrans.invoke(tcpInfo);
174+
retransmits = (Long) epollInfo.retransmits.invoke(tcpInfo);
175+
rtt = (Long) epollInfo.rtt.invoke(tcpInfo);
176+
} catch (ReflectiveOperationException e) {
177+
log.log(Level.FINE, "Error computing TCP metrics", e);
178+
return;
179+
}
189180

190-
long deltaTotal = totalRetrans - lastTotalRetrans;
191-
if (deltaTotal > 0) {
192-
metricRecorder.addLongCounter(InternalTcpMetrics.PACKETS_RETRANSMITTED_INSTRUMENT,
193-
deltaTotal, Collections.emptyList(), labelValues);
194-
lastTotalRetrans = totalRetrans;
195-
}
196-
if (isClose && retransmits > 0) {
197-
metricRecorder.addLongCounter(InternalTcpMetrics.RECURRING_RETRANSMITS_INSTRUMENT,
198-
retransmits, Collections.emptyList(), labelValues);
199-
}
200-
metricRecorder.recordDoubleHistogram(InternalTcpMetrics.MIN_RTT_INSTRUMENT,
201-
rtt / 1000000.0, // Convert microseconds to seconds
202-
Collections.emptyList(), labelValues);
181+
long deltaTotal = totalRetrans - lastTotalRetrans;
182+
if (deltaTotal > 0) {
183+
metricRecorder.addLongCounter(InternalTcpMetrics.PACKETS_RETRANSMITTED_INSTRUMENT,
184+
deltaTotal, Collections.emptyList(), labelValues);
185+
lastTotalRetrans = totalRetrans;
203186
}
187+
if (isClose && retransmits > 0) {
188+
metricRecorder.addLongCounter(InternalTcpMetrics.RECURRING_RETRANSMITS_INSTRUMENT,
189+
retransmits, Collections.emptyList(), labelValues);
190+
}
191+
metricRecorder.recordDoubleHistogram(InternalTcpMetrics.MIN_RTT_INSTRUMENT,
192+
rtt / 1000000.0, // Convert microseconds to seconds
193+
Collections.emptyList(), labelValues);
194+
}
195+
196+
@VisibleForTesting
197+
ScheduledFuture<?> getReportTimer() {
198+
return reportTimer;
204199
}
205200

206201
private static List<String> getLabelValues(Channel channel) {
@@ -212,20 +207,25 @@ private static List<String> getLabelValues(Channel channel) {
212207
SocketAddress local = channel.localAddress();
213208
if (local instanceof InetSocketAddress) {
214209
InetSocketAddress inetLocal = (InetSocketAddress) local;
215-
localAddress = inetLocal.getAddress().getHostAddress();
210+
if (inetLocal.getAddress() != null) {
211+
localAddress = inetLocal.getAddress().getHostAddress();
212+
} else if (inetLocal.getHostString() != null) {
213+
localAddress = inetLocal.getHostString();
214+
}
216215
localPort = String.valueOf(inetLocal.getPort());
217216
}
218217

219218
SocketAddress remote = channel.remoteAddress();
220219
if (remote instanceof InetSocketAddress) {
221220
InetSocketAddress inetRemote = (InetSocketAddress) remote;
222-
peerAddress = inetRemote.getAddress().getHostAddress();
221+
if (inetRemote.getAddress() != null) {
222+
peerAddress = inetRemote.getAddress().getHostAddress();
223+
} else if (inetRemote.getHostString() != null) {
224+
peerAddress = inetRemote.getHostString();
225+
}
223226
peerPort = String.valueOf(inetRemote.getPort());
224227
}
225228

226229
return Arrays.asList(localAddress, localPort, peerAddress, peerPort);
227230
}
228-
229-
private TcpMetrics() {
230-
}
231231
}

‎netty/src/test/java/io/grpc/netty/NettyServerHandlerTest.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -212,13 +212,11 @@ public void tcpMetrics_recorded() throws Exception {
212212
manualSetUp();
213213
handler().channelActive(ctx());
214214
// Verify that channelActive triggered TcpMetrics
215-
ArgumentCaptor<Long> countCaptor = ArgumentCaptor.forClass(Long.class);
216215
verify(metricRecorder, atLeastOnce()).addLongCounter(
217216
eq(io.grpc.InternalTcpMetrics.CONNECTIONS_CREATED_INSTRUMENT),
218-
countCaptor.capture(),
217+
eq(1L),
219218
any(),
220219
any());
221-
assertEquals(1L, countCaptor.getValue().longValue());
222220
}
223221

224222
@Test

0 commit comments

Comments
 (0)