#19528

Committed 28 Oct 2024 05:25PM UTC coverage: 84.59% (-0.04%) from 84.627%

Build # #19528

Build Type

push

github

Committed by

web-flow

Commit Message

netty: add soft Metadata size limit enforcement. (#11603)

Run Details

33914 of 40092 relevant lines covered (84.59%)

0.85 hits per line

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

92.22

/../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.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 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) {
    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");
    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);
  }
}

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

grpc / grpc-java / #19528

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