#19518

Committed 23 Oct 2024 07:52PM UTC coverage: 84.616% (-0.06%) from 84.679%

Build # #19518

Build Type

push

github

Committed by

web-flow

Commit Message

inprocess: Support tracing message sizes guarded by flag (#11629)

Run Details

33859 of 40015 relevant lines covered (84.62%)

0.85 hits per line

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

79.03

/../inprocess/src/main/java/io/grpc/inprocess/InProcessChannelBuilder.java

/*
 * Copyright 2015 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.inprocess;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.inprocess.InProcessTransport.isEnabledSupportTracingMessageSizes;

import com.google.errorprone.annotations.DoNotCall;
import io.grpc.ChannelCredentials;
import io.grpc.ChannelLogger;
import io.grpc.ExperimentalApi;
import io.grpc.ForwardingChannelBuilder2;
import io.grpc.Internal;
import io.grpc.ManagedChannelBuilder;
import io.grpc.internal.ClientTransportFactory;
import io.grpc.internal.ConnectionClientTransport;
import io.grpc.internal.GrpcUtil;
import io.grpc.internal.ManagedChannelImplBuilder;
import io.grpc.internal.ManagedChannelImplBuilder.ClientTransportFactoryBuilder;
import io.grpc.internal.SharedResourceHolder;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * Builder for a channel that issues in-process requests. Clients identify the in-process server by
 * its name.
 *
 * <p>The channel is intended to be fully-featured, high performance, and useful in testing.
 *
 * <p>For usage examples, see {@link InProcessServerBuilder}.
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1783")
public final class InProcessChannelBuilder extends
    ForwardingChannelBuilder2<InProcessChannelBuilder> {

  /**
   * Create a channel builder that will connect to the server with the given name.
   *
   * @param name the identity of the server to connect to
   * @return a new builder
   */
  public static InProcessChannelBuilder forName(String name) {
    return forAddress(new InProcessSocketAddress(checkNotNull(name, "name")));
  }

  /**
   * Create a channel builder that will connect to the server referenced by the given target URI.
   * Only intended for use with a custom name resolver.
   *
   * @param target the identity of the server to connect to
   * @return a new builder
   */
  public static InProcessChannelBuilder forTarget(String target) {
    return new InProcessChannelBuilder(null, checkNotNull(target, "target"));
  }

  /**
   * Create a channel builder that will connect to the server referenced by the given address.
   *
   * @param address the address of the server to connect to
   * @return a new builder
   */
  public static InProcessChannelBuilder forAddress(SocketAddress address) {
    return new InProcessChannelBuilder(checkNotNull(address, "address"), null);
  }

  /**
   * Always fails.  Call {@link #forName} instead.
   */
  @DoNotCall("Unsupported. Use forName() instead")
  public static InProcessChannelBuilder forAddress(String name, int port) {
    throw new UnsupportedOperationException("call forName() instead");
  }

  private final ManagedChannelImplBuilder managedChannelImplBuilder;
  private ScheduledExecutorService scheduledExecutorService;
  private int maxInboundMetadataSize = Integer.MAX_VALUE;
  private boolean transportIncludeStatusCause = false;
  private long assumedMessageSize = -1;

  private InProcessChannelBuilder(@Nullable SocketAddress directAddress, @Nullable String target) {

    final class InProcessChannelTransportFactoryBuilder implements ClientTransportFactoryBuilder {
      @Override
      public ClientTransportFactory buildClientTransportFactory() {
        return buildTransportFactory();
      }
    }

    if (directAddress != null) {
      managedChannelImplBuilder = new ManagedChannelImplBuilder(directAddress, "localhost",
          new InProcessChannelTransportFactoryBuilder(), null);
    } else {
      managedChannelImplBuilder = new ManagedChannelImplBuilder(target,
          new InProcessChannelTransportFactoryBuilder(), null);
    }

    // In-process transport should not record its traffic to the stats module.
    // https://github.com/grpc/grpc-java/issues/2284
    managedChannelImplBuilder.setStatsRecordStartedRpcs(false);
    managedChannelImplBuilder.setStatsRecordFinishedRpcs(false);
    managedChannelImplBuilder.setStatsRecordRetryMetrics(false);
    if (!isEnabledSupportTracingMessageSizes) {
      managedChannelImplBuilder.disableRetry();
    }
  }

  @Internal
  @Override
  protected ManagedChannelBuilder<?> delegate() {
    return managedChannelImplBuilder;
  }

  @Override
  public InProcessChannelBuilder maxInboundMessageSize(int max) {
    // TODO(carl-mastrangelo): maybe throw an exception since this not enforced?
    return super.maxInboundMessageSize(max);
  }

  /**
   * Does nothing.
   */
  @Override
  public InProcessChannelBuilder useTransportSecurity() {
    return this;
  }

  /**
   * Does nothing.
   */
  @Override
  public InProcessChannelBuilder usePlaintext() {
    return this;
  }

  /** Does nothing. */
  @Override
  public InProcessChannelBuilder keepAliveTime(long keepAliveTime, TimeUnit timeUnit) {
    return this;
  }

  /** Does nothing. */
  @Override
  public InProcessChannelBuilder keepAliveTimeout(long keepAliveTimeout, TimeUnit timeUnit) {
    return this;
  }

  /** Does nothing. */
  @Override
  public InProcessChannelBuilder keepAliveWithoutCalls(boolean enable) {
    return this;
  }

  /**
   * Provides a custom scheduled executor service.
   *
   * <p>It's an optional parameter. If the user has not provided a scheduled executor service when
   * the channel is built, the builder will use a static cached thread pool.
   *
   * @return this
   *
   * @since 1.11.0
   */
  public InProcessChannelBuilder scheduledExecutorService(
      ScheduledExecutorService scheduledExecutorService) {
    this.scheduledExecutorService =
        checkNotNull(scheduledExecutorService, "scheduledExecutorService");
    return this;
  }

  /**
   * Sets the maximum size of metadata allowed to be received. {@code Integer.MAX_VALUE} disables
   * the enforcement. Defaults to no limit ({@code Integer.MAX_VALUE}).
   *
   * <p>There is potential for performance penalty when this setting is enabled, as the Metadata
   * must actually be serialized. Since the current implementation of Metadata pre-serializes, it's
   * currently negligible. But Metadata is free to change its implementation.
   *
   * @param bytes the maximum size of received metadata
   * @return this
   * @throws IllegalArgumentException if bytes is non-positive
   * @since 1.17.0
   */
  @Override
  public InProcessChannelBuilder maxInboundMetadataSize(int bytes) {
    checkArgument(bytes > 0, "maxInboundMetadataSize must be > 0");
    this.maxInboundMetadataSize = bytes;
    return this;
  }

  /**
   * Sets whether to include the cause with the status that is propagated
   * forward from the InProcessTransport. This was added to make debugging failing
   * tests easier by showing the cause of the status.
   *
   * <p>By default, this is set to false.
   * A default value of false maintains consistency with other transports which strip causal
   * information from the status to avoid leaking information to untrusted clients, and
   * to avoid sharing language-specific information with the client.
   * For the in-process implementation, this is not a concern.
   *
   * @param enable whether to include cause in status
   * @return this
   */
  public InProcessChannelBuilder propagateCauseWithStatus(boolean enable) {
    this.transportIncludeStatusCause = enable;
    return this;
  }

  /**
   * Assumes RPC messages are the specified size. This avoids serializing
   * messages for metrics and retry memory tracking. This can dramatically
   * improve performance when accurate message sizes are not needed and if
   * nothing else needs the serialized message.
   * @param assumedMessageSize length of InProcess transport's messageSize.
   * @return this
   * @throws IllegalArgumentException if assumedMessageSize is negative.
   */
  public InProcessChannelBuilder assumedMessageSize(long assumedMessageSize) {
    checkArgument(assumedMessageSize >= 0, "assumedMessageSize must be >= 0");
    this.assumedMessageSize = assumedMessageSize;
    return this;
  }

  ClientTransportFactory buildTransportFactory() {
    return new InProcessClientTransportFactory(scheduledExecutorService,
            maxInboundMetadataSize, transportIncludeStatusCause, assumedMessageSize);
  }

  void setStatsEnabled(boolean value) {
    this.managedChannelImplBuilder.setStatsEnabled(value);
  }

  /**
   * Creates InProcess transports. Exposed for internal use, as it should be private.
   */
  static final class InProcessClientTransportFactory implements ClientTransportFactory {
    private final ScheduledExecutorService timerService;
    private final boolean useSharedTimer;
    private final int maxInboundMetadataSize;
    private boolean closed;
    private final boolean includeCauseWithStatus;
    private long assumedMessageSize;

    private InProcessClientTransportFactory(
        @Nullable ScheduledExecutorService scheduledExecutorService,
        int maxInboundMetadataSize, boolean includeCauseWithStatus, long assumedMessageSize) {
      useSharedTimer = scheduledExecutorService == null;
      timerService = useSharedTimer
          ? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : scheduledExecutorService;
      this.maxInboundMetadataSize = maxInboundMetadataSize;
      this.includeCauseWithStatus = includeCauseWithStatus;
      this.assumedMessageSize = assumedMessageSize;
    }

    @Override
    public ConnectionClientTransport newClientTransport(
        SocketAddress addr, ClientTransportOptions options, ChannelLogger channelLogger) {
      if (closed) {
        throw new IllegalStateException("The transport factory is closed.");
      }
      // TODO(carl-mastrangelo): Pass channelLogger in.
      return new InProcessTransport(
          addr, maxInboundMetadataSize, options.getAuthority(), options.getUserAgent(),
          options.getEagAttributes(), includeCauseWithStatus, assumedMessageSize);
    }

    @Override
    public ScheduledExecutorService getScheduledExecutorService() {
      return timerService;
    }

    @Override
    public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials channelCreds) {
      return null;
    }

    @Override
    public void close() {
      if (closed) {
        return;
      }
      closed = true;
      if (useSharedTimer) {
        SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timerService);
      }
    }

    @Override
    public Collection<Class<? extends SocketAddress>> getSupportedSocketAddressTypes() {
      return Arrays.asList(InProcessSocketAddress.class, AnonymousInProcessSocketAddress.class);
    }
  }
}

1	/*
2	* Copyright 2015 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.inprocess;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static io.grpc.inprocess.InProcessTransport.isEnabledSupportTracingMessageSizes;
22
23	import com.google.errorprone.annotations.DoNotCall;
24	import io.grpc.ChannelCredentials;
25	import io.grpc.ChannelLogger;
26	import io.grpc.ExperimentalApi;
27	import io.grpc.ForwardingChannelBuilder2;
28	import io.grpc.Internal;
29	import io.grpc.ManagedChannelBuilder;
30	import io.grpc.internal.ClientTransportFactory;
31	import io.grpc.internal.ConnectionClientTransport;
32	import io.grpc.internal.GrpcUtil;
33	import io.grpc.internal.ManagedChannelImplBuilder;
34	import io.grpc.internal.ManagedChannelImplBuilder.ClientTransportFactoryBuilder;
35	import io.grpc.internal.SharedResourceHolder;
36	import java.net.SocketAddress;
37	import java.util.Arrays;
38	import java.util.Collection;
39	import java.util.concurrent.ScheduledExecutorService;
40	import java.util.concurrent.TimeUnit;
41	import javax.annotation.Nullable;
42
43	/**
44	* Builder for a channel that issues in-process requests. Clients identify the in-process server by
45	* its name.
46	*
47	* <p>The channel is intended to be fully-featured, high performance, and useful in testing.
48	*
49	* <p>For usage examples, see {@link InProcessServerBuilder}.
50	*/
51	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1783")
52	public final class InProcessChannelBuilder extends
53	ForwardingChannelBuilder2<InProcessChannelBuilder> {
54
55	/**
56	* Create a channel builder that will connect to the server with the given name.
57	*
58	* @param name the identity of the server to connect to
59	* @return a new builder
60	*/
61	public static InProcessChannelBuilder forName(String name) {
62	return forAddress(new InProcessSocketAddress(checkNotNull(name, "name")));	1✔
63	}
64
65	/**
66	* Create a channel builder that will connect to the server referenced by the given target URI.
67	* Only intended for use with a custom name resolver.
68	*
69	* @param target the identity of the server to connect to
70	* @return a new builder
71	*/
72	public static InProcessChannelBuilder forTarget(String target) {
73	return new InProcessChannelBuilder(null, checkNotNull(target, "target"));	1✔
74	}
75
76	/**
77	* Create a channel builder that will connect to the server referenced by the given address.
78	*
79	* @param address the address of the server to connect to
80	* @return a new builder
81	*/
82	public static InProcessChannelBuilder forAddress(SocketAddress address) {
83	return new InProcessChannelBuilder(checkNotNull(address, "address"), null);	1✔
84	}
85
86	/**
87	* Always fails. Call {@link #forName} instead.
88	*/
89	@DoNotCall("Unsupported. Use forName() instead")
90	public static InProcessChannelBuilder forAddress(String name, int port) {
91	throw new UnsupportedOperationException("call forName() instead");	×
92	}
93
94	private final ManagedChannelImplBuilder managedChannelImplBuilder;
95	private ScheduledExecutorService scheduledExecutorService;
96	private int maxInboundMetadataSize = Integer.MAX_VALUE;	1✔
97	private boolean transportIncludeStatusCause = false;	1✔
98	private long assumedMessageSize = -1;	1✔
99
100	private InProcessChannelBuilder(@Nullable SocketAddress directAddress, @Nullable String target) {	1✔
101
102	final class InProcessChannelTransportFactoryBuilder implements ClientTransportFactoryBuilder {	1✔
103	@Override
104	public ClientTransportFactory buildClientTransportFactory() {
105	return buildTransportFactory();	1✔
106	}
107	}
108
109	if (directAddress != null) {	1✔
110	managedChannelImplBuilder = new ManagedChannelImplBuilder(directAddress, "localhost",	1✔
111	new InProcessChannelTransportFactoryBuilder(), null);
112	} else {
113	managedChannelImplBuilder = new ManagedChannelImplBuilder(target,	1✔
114	new InProcessChannelTransportFactoryBuilder(), null);
115	}
116
117	// In-process transport should not record its traffic to the stats module.
118	// https://github.com/grpc/grpc-java/issues/2284
119	managedChannelImplBuilder.setStatsRecordStartedRpcs(false);	1✔
120	managedChannelImplBuilder.setStatsRecordFinishedRpcs(false);	1✔
121	managedChannelImplBuilder.setStatsRecordRetryMetrics(false);	1✔
122	if (!isEnabledSupportTracingMessageSizes) {	1✔
123	managedChannelImplBuilder.disableRetry();	1✔
124	}
125	}	1✔
126
127	@Internal
128	@Override
129	protected ManagedChannelBuilder<?> delegate() {
130	return managedChannelImplBuilder;	1✔
131	}
132
133	@Override
134	public InProcessChannelBuilder maxInboundMessageSize(int max) {
135	// TODO(carl-mastrangelo): maybe throw an exception since this not enforced?
136	return super.maxInboundMessageSize(max);	×
137	}
138
139	/**
140	* Does nothing.
141	*/
142	@Override
143	public InProcessChannelBuilder useTransportSecurity() {
144	return this;	×
145	}
146
147	/**
148	* Does nothing.
149	*/
150	@Override
151	public InProcessChannelBuilder usePlaintext() {
152	return this;	×
153	}
154
155	/** Does nothing. */
156	@Override
157	public InProcessChannelBuilder keepAliveTime(long keepAliveTime, TimeUnit timeUnit) {
158	return this;	×
159	}
160
161	/** Does nothing. */
162	@Override
163	public InProcessChannelBuilder keepAliveTimeout(long keepAliveTimeout, TimeUnit timeUnit) {
164	return this;	×
165	}
166
167	/** Does nothing. */
168	@Override
169	public InProcessChannelBuilder keepAliveWithoutCalls(boolean enable) {
170	return this;	×
171	}
172
173	/**
174	* Provides a custom scheduled executor service.
175	*
176	* <p>It's an optional parameter. If the user has not provided a scheduled executor service when
177	* the channel is built, the builder will use a static cached thread pool.
178	*
179	* @return this
180	*
181	* @since 1.11.0
182	*/
183	public InProcessChannelBuilder scheduledExecutorService(
184	ScheduledExecutorService scheduledExecutorService) {
185	this.scheduledExecutorService =	1✔
186	checkNotNull(scheduledExecutorService, "scheduledExecutorService");	1✔
187	return this;	1✔
188	}
189
190	/**
191	* Sets the maximum size of metadata allowed to be received. {@code Integer.MAX_VALUE} disables
192	* the enforcement. Defaults to no limit ({@code Integer.MAX_VALUE}).
193	*
194	* <p>There is potential for performance penalty when this setting is enabled, as the Metadata
195	* must actually be serialized. Since the current implementation of Metadata pre-serializes, it's
196	* currently negligible. But Metadata is free to change its implementation.
197	*
198	* @param bytes the maximum size of received metadata
199	* @return this
200	* @throws IllegalArgumentException if bytes is non-positive
201	* @since 1.17.0
202	*/
203	@Override
204	public InProcessChannelBuilder maxInboundMetadataSize(int bytes) {
205	checkArgument(bytes > 0, "maxInboundMetadataSize must be > 0");	×
206	this.maxInboundMetadataSize = bytes;	×
207	return this;	×
208	}
209
210	/**
211	* Sets whether to include the cause with the status that is propagated
212	* forward from the InProcessTransport. This was added to make debugging failing
213	* tests easier by showing the cause of the status.
214	*
215	* <p>By default, this is set to false.
216	* A default value of false maintains consistency with other transports which strip causal
217	* information from the status to avoid leaking information to untrusted clients, and
218	* to avoid sharing language-specific information with the client.
219	* For the in-process implementation, this is not a concern.
220	*
221	* @param enable whether to include cause in status
222	* @return this
223	*/
224	public InProcessChannelBuilder propagateCauseWithStatus(boolean enable) {
225	this.transportIncludeStatusCause = enable;	1✔
226	return this;	1✔
227	}
228
229	/**
230	* Assumes RPC messages are the specified size. This avoids serializing
231	* messages for metrics and retry memory tracking. This can dramatically
232	* improve performance when accurate message sizes are not needed and if
233	* nothing else needs the serialized message.
234	* @param assumedMessageSize length of InProcess transport's messageSize.
235	* @return this
236	* @throws IllegalArgumentException if assumedMessageSize is negative.
237	*/
238	public InProcessChannelBuilder assumedMessageSize(long assumedMessageSize) {
239	checkArgument(assumedMessageSize >= 0, "assumedMessageSize must be >= 0");	×
240	this.assumedMessageSize = assumedMessageSize;	×
241	return this;	×
242	}
243
244	ClientTransportFactory buildTransportFactory() {
245	return new InProcessClientTransportFactory(scheduledExecutorService,	1✔
246	maxInboundMetadataSize, transportIncludeStatusCause, assumedMessageSize);
247	}
248
249	void setStatsEnabled(boolean value) {
250	this.managedChannelImplBuilder.setStatsEnabled(value);	1✔
251	}	1✔
252
253	/**
254	* Creates InProcess transports. Exposed for internal use, as it should be private.
255	*/
256	static final class InProcessClientTransportFactory implements ClientTransportFactory {
257	private final ScheduledExecutorService timerService;
258	private final boolean useSharedTimer;
259	private final int maxInboundMetadataSize;
260	private boolean closed;
261	private final boolean includeCauseWithStatus;
262	private long assumedMessageSize;
263
264	private InProcessClientTransportFactory(
265	@Nullable ScheduledExecutorService scheduledExecutorService,
266	int maxInboundMetadataSize, boolean includeCauseWithStatus, long assumedMessageSize) {	1✔
267	useSharedTimer = scheduledExecutorService == null;	1✔
268	timerService = useSharedTimer	1✔
269	? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : scheduledExecutorService;	1✔
270	this.maxInboundMetadataSize = maxInboundMetadataSize;	1✔
271	this.includeCauseWithStatus = includeCauseWithStatus;	1✔
272	this.assumedMessageSize = assumedMessageSize;	1✔
273	}	1✔
274
275	@Override
276	public ConnectionClientTransport newClientTransport(
277	SocketAddress addr, ClientTransportOptions options, ChannelLogger channelLogger) {
278	if (closed) {	1✔
279	throw new IllegalStateException("The transport factory is closed.");	1✔
280	}
281	// TODO(carl-mastrangelo): Pass channelLogger in.
282	return new InProcessTransport(	1✔
283	addr, maxInboundMetadataSize, options.getAuthority(), options.getUserAgent(),	1✔
284	options.getEagAttributes(), includeCauseWithStatus, assumedMessageSize);	1✔
285	}
286
287	@Override
288	public ScheduledExecutorService getScheduledExecutorService() {
289	return timerService;	1✔
290	}
291
292	@Override
293	public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials channelCreds) {
294	return null;	1✔
295	}
296
297	@Override
298	public void close() {
299	if (closed) {	1✔
300	return;	1✔
301	}
302	closed = true;	1✔
303	if (useSharedTimer) {	1✔
304	SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timerService);	1✔
305	}
306	}	1✔
307
308	@Override
309	public Collection<Class<? extends SocketAddress>> getSupportedSocketAddressTypes() {
310	return Arrays.asList(InProcessSocketAddress.class, AnonymousInProcessSocketAddress.class);	1✔
311	}
312	}
313	}

grpc / grpc-java / #19518

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