#20230

Committed 31 Mar 2026 09:55AM UTC coverage: 88.734% (+0.01%) from 88.72%

Build # #20230

Build Type

push

github

Committed by

web-flow

Commit Message

openTelemetry: add tcp metrics (#12652)

Implements [A80](https://github.com/grpc/proposal/pull/519)

Run Details

35697 of 40229 relevant lines covered (88.73%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.31

/../netty/src/main/java/io/grpc/netty/NettyServerTransport.java

/*
 * Copyright 2014 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 com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
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;
import io.grpc.internal.ServerTransportListener;
import io.grpc.internal.TransportTracer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import java.io.IOException;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * The Netty-based server transport.
 */
class NettyServerTransport implements ServerTransport {
  // connectionLog is for connection related messages only
  private static final Logger connectionLog = Logger.getLogger(
      String.format("%s.connections", NettyServerTransport.class.getName()));

  // Some exceptions are not very useful and add too much noise to the log
  private static final ImmutableList<String> QUIET_EXCEPTIONS = ImmutableList.of(
      "NativeIoException" /* Netty exceptions */);

  private final InternalLogId logId;
  private final Channel channel;
  private final ChannelPromise channelUnused;
  private final ProtocolNegotiator protocolNegotiator;
  private final int maxStreams;
  // only accessed from channel event loop
  private NettyServerHandler grpcHandler;
  private ServerTransportListener listener;
  private boolean terminated;
  private final boolean autoFlowControl;
  private final int flowControlWindow;
  private final int maxMessageSize;
  private final int maxHeaderListSize;
  private final int softLimitHeaderListSize;
  private final long keepAliveTimeInNanos;
  private final long keepAliveTimeoutInNanos;
  private final long maxConnectionIdleInNanos;
  private final long maxConnectionAgeInNanos;
  private final long maxConnectionAgeGraceInNanos;
  private final boolean permitKeepAliveWithoutCalls;
  private final long permitKeepAliveTimeInNanos;
  private final int maxRstCount;
  private final long maxRstPeriodNanos;
  private final Attributes eagAttributes;
  private final MetricRecorder metricRecorder;
  private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
  private final TransportTracer transportTracer;

  NettyServerTransport(
      Channel channel,
      ChannelPromise channelUnused,
      ProtocolNegotiator protocolNegotiator,
      List<? extends ServerStreamTracer.Factory> streamTracerFactories,
      TransportTracer transportTracer,
      int maxStreams,
      boolean autoFlowControl,
      int flowControlWindow,
      int maxMessageSize,
      int maxHeaderListSize,
      int softLimitHeaderListSize,
      long keepAliveTimeInNanos,
      long keepAliveTimeoutInNanos,
      long maxConnectionIdleInNanos,
      long maxConnectionAgeInNanos,
      long maxConnectionAgeGraceInNanos,
      boolean permitKeepAliveWithoutCalls,
      long permitKeepAliveTimeInNanos,
      int maxRstCount,
      long maxRstPeriodNanos,
      Attributes eagAttributes,
      MetricRecorder metricRecorder) {
    this.channel = Preconditions.checkNotNull(channel, "channel");
    this.channelUnused = channelUnused;
    this.protocolNegotiator = Preconditions.checkNotNull(protocolNegotiator, "protocolNegotiator");
    this.streamTracerFactories =
        Preconditions.checkNotNull(streamTracerFactories, "streamTracerFactories");
    this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
    this.maxStreams = maxStreams;
    this.autoFlowControl = autoFlowControl;
    this.flowControlWindow = flowControlWindow;
    this.maxMessageSize = maxMessageSize;
    this.maxHeaderListSize = maxHeaderListSize;
    this.softLimitHeaderListSize = softLimitHeaderListSize;
    this.keepAliveTimeInNanos = keepAliveTimeInNanos;
    this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;
    this.maxConnectionIdleInNanos = maxConnectionIdleInNanos;
    this.maxConnectionAgeInNanos = maxConnectionAgeInNanos;
    this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;
    this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;
    this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;
    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);
  }

  public void start(ServerTransportListener listener) {
    Preconditions.checkState(this.listener == null, "Handler already registered");
    this.listener = listener;

    // Create the Netty handler for the pipeline.
    grpcHandler = createHandler(listener, channelUnused);

    // Notify when the channel closes.
    final class TerminationNotifier implements ChannelFutureListener {
      boolean done;

      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        if (!done) {
          done = true;
          notifyTerminated(grpcHandler.connectionError());
        }
      }
    }

    ChannelHandler negotiationHandler = protocolNegotiator.newHandler(grpcHandler);
    ChannelHandler bufferingHandler = new WriteBufferingAndExceptionHandler(negotiationHandler);

    ChannelFutureListener terminationNotifier = new TerminationNotifier();
    channelUnused.addListener(terminationNotifier);
    channel.closeFuture().addListener(terminationNotifier);

    channel.pipeline().addLast(bufferingHandler);
  }

  @Override
  public ScheduledExecutorService getScheduledExecutorService() {
    return channel.eventLoop();
  }

  @Override
  public void shutdown() {
    if (channel.isOpen()) {
      channel.close();
    }
  }

  @Override
  public void shutdownNow(Status reason) {
    if (channel.isOpen()) {
      channel.writeAndFlush(new ForcefulCloseCommand(reason));
    }
  }

  @Override
  public InternalLogId getLogId() {
    return logId;
  }

  /**
   * For testing purposes only.
   */
  Channel channel() {
    return channel;
  }

  /**
   * Accepts a throwable and returns the appropriate logging level. Uninteresting exceptions
   * should not clutter the log.
   */
  @VisibleForTesting
  static Level getLogLevel(Throwable t) {
    if (t.getClass().equals(IOException.class)
        || t.getClass().equals(SocketException.class)
        || QUIET_EXCEPTIONS.contains(t.getClass().getSimpleName())) {
      return Level.FINE;
    }
    return Level.INFO;
  }

  private void notifyTerminated(Throwable t) {
    if (t != null) {
      connectionLog.log(getLogLevel(t), "Transport failed", t);
    }
    if (!terminated) {
      terminated = true;
      listener.transportTerminated();
    }
  }

  @Override
  public ListenableFuture<SocketStats> getStats() {
    final SettableFuture<SocketStats> result = SettableFuture.create();
    if (channel.eventLoop().inEventLoop()) {
      // This is necessary, otherwise we will block forever if we get the future from inside
      // the event loop.
      result.set(getStatsHelper(channel));
      return result;
    }
    channel.eventLoop().submit(
        new Runnable() {
          @Override
          public void run() {
            result.set(getStatsHelper(channel));
          }
        })
        .addListener(
            new GenericFutureListener<Future<Object>>() {
              @Override
              public void operationComplete(Future<Object> future) throws Exception {
                if (!future.isSuccess()) {
                  result.setException(future.cause());
                }
              }
            });
    return result;
  }

  private SocketStats getStatsHelper(Channel ch) {
    Preconditions.checkState(ch.eventLoop().inEventLoop());
    return new SocketStats(
        transportTracer.getStats(),
        channel.localAddress(),
        channel.remoteAddress(),
        Utils.getSocketOptions(ch),
        grpcHandler == null ? null : grpcHandler.getSecurityInfo());

  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
        .add("logId", logId.getId())
        .add("channel", channel)
        .toString();
  }

  /**
   * Creates the Netty handler to be used in the channel pipeline.
   */
  private NettyServerHandler createHandler(
      ServerTransportListener transportListener, ChannelPromise channelUnused) {
    return NettyServerHandler.newHandler(
        transportListener,
        channelUnused,
        streamTracerFactories,
        transportTracer,
        maxStreams,
        autoFlowControl,
        flowControlWindow,
        maxHeaderListSize,
        softLimitHeaderListSize,
        maxMessageSize,
        keepAliveTimeInNanos,
        keepAliveTimeoutInNanos,
        maxConnectionIdleInNanos,
        maxConnectionAgeInNanos,
        maxConnectionAgeGraceInNanos,
        permitKeepAliveWithoutCalls,
        permitKeepAliveTimeInNanos,
        maxRstCount,
        maxRstPeriodNanos,
        eagAttributes,
        metricRecorder);
  }
}

1	/*
2	* Copyright 2014 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.netty;
18
19	import com.google.common.annotations.VisibleForTesting;
20	import com.google.common.base.MoreObjects;
21	import com.google.common.base.Preconditions;
22	import com.google.common.collect.ImmutableList;
23	import com.google.common.util.concurrent.ListenableFuture;
24	import com.google.common.util.concurrent.SettableFuture;
25	import io.grpc.Attributes;
26	import io.grpc.InternalChannelz.SocketStats;
27	import io.grpc.InternalLogId;
28	import io.grpc.MetricRecorder;
29	import io.grpc.ServerStreamTracer;
30	import io.grpc.Status;
31	import io.grpc.internal.ServerTransport;
32	import io.grpc.internal.ServerTransportListener;
33	import io.grpc.internal.TransportTracer;
34	import io.netty.channel.Channel;
35	import io.netty.channel.ChannelFuture;
36	import io.netty.channel.ChannelFutureListener;
37	import io.netty.channel.ChannelHandler;
38	import io.netty.channel.ChannelPromise;
39	import io.netty.util.concurrent.Future;
40	import io.netty.util.concurrent.GenericFutureListener;
41	import java.io.IOException;
42	import java.net.SocketAddress;
43	import java.net.SocketException;
44	import java.util.List;
45	import java.util.concurrent.ScheduledExecutorService;
46	import java.util.logging.Level;
47	import java.util.logging.Logger;
48
49	/**
50	* The Netty-based server transport.
51	*/
52	class NettyServerTransport implements ServerTransport {
53	// connectionLog is for connection related messages only
54	private static final Logger connectionLog = Logger.getLogger(	1✔
55	String.format("%s.connections", NettyServerTransport.class.getName()));	1✔
56
57	// Some exceptions are not very useful and add too much noise to the log
58	private static final ImmutableList<String> QUIET_EXCEPTIONS = ImmutableList.of(	1✔
59	"NativeIoException" /* Netty exceptions */);
60
61	private final InternalLogId logId;
62	private final Channel channel;
63	private final ChannelPromise channelUnused;
64	private final ProtocolNegotiator protocolNegotiator;
65	private final int maxStreams;
66	// only accessed from channel event loop
67	private NettyServerHandler grpcHandler;
68	private ServerTransportListener listener;
69	private boolean terminated;
70	private final boolean autoFlowControl;
71	private final int flowControlWindow;
72	private final int maxMessageSize;
73	private final int maxHeaderListSize;
74	private final int softLimitHeaderListSize;
75	private final long keepAliveTimeInNanos;
76	private final long keepAliveTimeoutInNanos;
77	private final long maxConnectionIdleInNanos;
78	private final long maxConnectionAgeInNanos;
79	private final long maxConnectionAgeGraceInNanos;
80	private final boolean permitKeepAliveWithoutCalls;
81	private final long permitKeepAliveTimeInNanos;
82	private final int maxRstCount;
83	private final long maxRstPeriodNanos;
84	private final Attributes eagAttributes;
85	private final MetricRecorder metricRecorder;
86	private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
87	private final TransportTracer transportTracer;
88
89	NettyServerTransport(
90	Channel channel,
91	ChannelPromise channelUnused,
92	ProtocolNegotiator protocolNegotiator,
93	List<? extends ServerStreamTracer.Factory> streamTracerFactories,
94	TransportTracer transportTracer,
95	int maxStreams,
96	boolean autoFlowControl,
97	int flowControlWindow,
98	int maxMessageSize,
99	int maxHeaderListSize,
100	int softLimitHeaderListSize,
101	long keepAliveTimeInNanos,
102	long keepAliveTimeoutInNanos,
103	long maxConnectionIdleInNanos,
104	long maxConnectionAgeInNanos,
105	long maxConnectionAgeGraceInNanos,
106	boolean permitKeepAliveWithoutCalls,
107	long permitKeepAliveTimeInNanos,
108	int maxRstCount,
109	long maxRstPeriodNanos,
110	Attributes eagAttributes,
111	MetricRecorder metricRecorder) {	1✔
112	this.channel = Preconditions.checkNotNull(channel, "channel");	1✔
113	this.channelUnused = channelUnused;	1✔
114	this.protocolNegotiator = Preconditions.checkNotNull(protocolNegotiator, "protocolNegotiator");	1✔
115	this.streamTracerFactories =	1✔
116	Preconditions.checkNotNull(streamTracerFactories, "streamTracerFactories");	1✔
117	this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");	1✔
118	this.maxStreams = maxStreams;	1✔
119	this.autoFlowControl = autoFlowControl;	1✔
120	this.flowControlWindow = flowControlWindow;	1✔
121	this.maxMessageSize = maxMessageSize;	1✔
122	this.maxHeaderListSize = maxHeaderListSize;	1✔
123	this.softLimitHeaderListSize = softLimitHeaderListSize;	1✔
124	this.keepAliveTimeInNanos = keepAliveTimeInNanos;	1✔
125	this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;	1✔
126	this.maxConnectionIdleInNanos = maxConnectionIdleInNanos;	1✔
127	this.maxConnectionAgeInNanos = maxConnectionAgeInNanos;	1✔
128	this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;	1✔
129	this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;	1✔
130	this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;	1✔
131	this.maxRstCount = maxRstCount;	1✔
132	this.maxRstPeriodNanos = maxRstPeriodNanos;	1✔
133	this.eagAttributes = Preconditions.checkNotNull(eagAttributes, "eagAttributes");	1✔
134	this.metricRecorder = metricRecorder;	1✔
135	SocketAddress remote = channel.remoteAddress();	1✔
136	this.logId = InternalLogId.allocate(getClass(), remote != null ? remote.toString() : null);	1✔
137	}	1✔
138
139	public void start(ServerTransportListener listener) {
140	Preconditions.checkState(this.listener == null, "Handler already registered");	1✔
141	this.listener = listener;	1✔
142
143	// Create the Netty handler for the pipeline.
144	grpcHandler = createHandler(listener, channelUnused);	1✔
145
146	// Notify when the channel closes.
147	final class TerminationNotifier implements ChannelFutureListener {	1✔
148	boolean done;
149
150	@Override
151	public void operationComplete(ChannelFuture future) throws Exception {
152	if (!done) {	1✔
153	done = true;	1✔
154	notifyTerminated(grpcHandler.connectionError());	1✔
155	}
156	}	1✔
157	}
158
159	ChannelHandler negotiationHandler = protocolNegotiator.newHandler(grpcHandler);	1✔
160	ChannelHandler bufferingHandler = new WriteBufferingAndExceptionHandler(negotiationHandler);	1✔
161
162	ChannelFutureListener terminationNotifier = new TerminationNotifier();	1✔
163	channelUnused.addListener(terminationNotifier);	1✔
164	channel.closeFuture().addListener(terminationNotifier);	1✔
165
166	channel.pipeline().addLast(bufferingHandler);	1✔
167	}	1✔
168
169	@Override
170	public ScheduledExecutorService getScheduledExecutorService() {
171	return channel.eventLoop();	1✔
172	}
173
174	@Override
175	public void shutdown() {
176	if (channel.isOpen()) {	1✔
177	channel.close();	1✔
178	}
179	}	1✔
180
181	@Override
182	public void shutdownNow(Status reason) {
183	if (channel.isOpen()) {	1✔
184	channel.writeAndFlush(new ForcefulCloseCommand(reason));	1✔
185	}
186	}	1✔
187
188	@Override
189	public InternalLogId getLogId() {
190	return logId;	1✔
191	}
192
193	/**
194	* For testing purposes only.
195	*/
196	Channel channel() {
197	return channel;	1✔
198	}
199
200	/**
201	* Accepts a throwable and returns the appropriate logging level. Uninteresting exceptions
202	* should not clutter the log.
203	*/
204	@VisibleForTesting
205	static Level getLogLevel(Throwable t) {
206	if (t.getClass().equals(IOException.class)	1✔
207	\|\| t.getClass().equals(SocketException.class)	1✔
208	\|\| QUIET_EXCEPTIONS.contains(t.getClass().getSimpleName())) {	1✔
209	return Level.FINE;	1✔
210	}
211	return Level.INFO;	1✔
212	}
213
214	private void notifyTerminated(Throwable t) {
215	if (t != null) {	1✔
216	connectionLog.log(getLogLevel(t), "Transport failed", t);	1✔
217	}
218	if (!terminated) {	1✔
219	terminated = true;	1✔
220	listener.transportTerminated();	1✔
221	}
222	}	1✔
223
224	@Override
225	public ListenableFuture<SocketStats> getStats() {
226	final SettableFuture<SocketStats> result = SettableFuture.create();	1✔
227	if (channel.eventLoop().inEventLoop()) {	1✔
228	// This is necessary, otherwise we will block forever if we get the future from inside
229	// the event loop.
230	result.set(getStatsHelper(channel));	×
231	return result;	×
232	}
233	channel.eventLoop().submit(	1✔
234	new Runnable() {	1✔
235	@Override
236	public void run() {
237	result.set(getStatsHelper(channel));	1✔
238	}	1✔
239	})
240	.addListener(	1✔
241	new GenericFutureListener<Future<Object>>() {	1✔
242	@Override
243	public void operationComplete(Future<Object> future) throws Exception {
244	if (!future.isSuccess()) {	1✔
245	result.setException(future.cause());	×
246	}
247	}	1✔
248	});
249	return result;	1✔
250	}
251
252	private SocketStats getStatsHelper(Channel ch) {
253	Preconditions.checkState(ch.eventLoop().inEventLoop());	1✔
254	return new SocketStats(	1✔
255	transportTracer.getStats(),	1✔
256	channel.localAddress(),	1✔
257	channel.remoteAddress(),	1✔
258	Utils.getSocketOptions(ch),	1✔
259	grpcHandler == null ? null : grpcHandler.getSecurityInfo());	1✔
260
261	}
262
263	@Override
264	public String toString() {
265	return MoreObjects.toStringHelper(this)	×
266	.add("logId", logId.getId())	×
267	.add("channel", channel)	×
268	.toString();	×
269	}
270
271	/**
272	* Creates the Netty handler to be used in the channel pipeline.
273	*/
274	private NettyServerHandler createHandler(
275	ServerTransportListener transportListener, ChannelPromise channelUnused) {
276	return NettyServerHandler.newHandler(	1✔
277	transportListener,
278	channelUnused,
279	streamTracerFactories,
280	transportTracer,
281	maxStreams,
282	autoFlowControl,
283	flowControlWindow,
284	maxHeaderListSize,
285	softLimitHeaderListSize,
286	maxMessageSize,
287	keepAliveTimeInNanos,
288	keepAliveTimeoutInNanos,
289	maxConnectionIdleInNanos,
290	maxConnectionAgeInNanos,
291	maxConnectionAgeGraceInNanos,
292	permitKeepAliveWithoutCalls,
293	permitKeepAliveTimeInNanos,
294	maxRstCount,
295	maxRstPeriodNanos,
296	eagAttributes,
297	metricRecorder);
298	}
299	}

grpc / grpc-java / #20230

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous