#19507

Committed 11 Oct 2024 04:58AM UTC coverage: 84.658% (-0.008%) from 84.666%

Build # #19507

Build Type

push

github

Committed by

web-flow

Commit Message

inprocess: Support tracing message sizes (#11406)

Run Details

33814 of 39942 relevant lines covered (84.66%)

0.85 hits per line

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

78.33

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

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

grpc / grpc-java / #19507

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