#20033

Committed 29 Oct 2025 04:43PM UTC coverage: 88.533% (-0.03%) from 88.561%

Build # #20033

Build Type

push

github

Committed by

web-flow

Commit Message

xds,googleapis: Allow wrapping NameResolver to inject XdsClient (#12450)

Since there is no longer a single global XdsClient, it makes more sense
to allow things like the c2p name resolver to inject its own bootstrap
even if there is one defined in an environment variable.
GoogleCloudToProdNameResolver can now pass an XdsClient instance to
XdsNameResolver, and SharedXdsClientPoolProvider allows
GoogleCloudToProdNameResolver to choose the bootstrap for that one
specific target.

Since XdsNameResolver is no longer in control of the XdsClient pool the
XdsClient instance is now passed to ClusterImplLb. A channel will now
only access the global XdsClient pool exactly once: in the name
resolver.

BootstrapInfo is purposefully being shared across channels, as we really
want to share things like credentials which can have significant memory
use and may have caches which reduce I/O when shared. That is why
SharedXdsClientPoolProvider receives BootstrapInfo instead of
Map<String,?>.

Verifying BootstrapInfo.server() is not empty was moved from
SharedXdsClientPoolProvider to GrpcBootstrapperImpl so avoid
getOrCreate() throwing an exception in only that one case. It might make
sense to move that to BootstrapperImpl, but that will need more
investigation.

A lot of tests needed updating because XdsClientPoolProvider is no
longer responsible for parsing the bootstrap, so we now need bootstraps
even if XdsClientPoolProvider will ignore it.

This also fixes a bug in GoogleCloudToProdNameResolver where it would
initialize the delegate even when it failed to create the bootstrap.
That would certainly cause all RPCs on the channel to fail because of
the missing bootstrap and it defeated the point of `succeeded == false`
and `refresh()` which was supposed to retry contacting the metadata
server.

The server tests were enhanced to give a useful error when
server.start() throws an exception, as otherwise the real error is lost.

b/442819521

Run Details

34966 of 39495 relevant lines covered (88.53%)

0.89 hits per line

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

77.19

/../xds/src/main/java/io/grpc/xds/XdsServerBuilder.java

/*
 * Copyright 2019 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.xds;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static io.grpc.xds.XdsAttributes.ATTR_DRAIN_GRACE_NANOS;
import static io.grpc.xds.XdsAttributes.ATTR_FILTER_CHAIN_SELECTOR_MANAGER;

import com.google.common.annotations.VisibleForTesting;
import com.google.errorprone.annotations.DoNotCall;
import io.grpc.Attributes;
import io.grpc.ExperimentalApi;
import io.grpc.ForwardingServerBuilder;
import io.grpc.Internal;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.ServerCredentials;
import io.grpc.netty.InternalNettyServerBuilder;
import io.grpc.netty.InternalNettyServerCredentials;
import io.grpc.netty.InternalProtocolNegotiator;
import io.grpc.netty.NettyServerBuilder;
import io.grpc.xds.FilterChainMatchingProtocolNegotiators.FilterChainMatchingNegotiatorServerFactory;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

/**
 * A version of {@link ServerBuilder} to create xDS managed servers.
 */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/7514")
public final class XdsServerBuilder extends ForwardingServerBuilder<XdsServerBuilder> {
  private static final long AS_LARGE_AS_INFINITE = TimeUnit.DAYS.toNanos(1000);

  private final NettyServerBuilder delegate;
  private final int port;
  private XdsServingStatusListener xdsServingStatusListener;
  private AtomicBoolean isServerBuilt = new AtomicBoolean(false);
  private final FilterRegistry filterRegistry = FilterRegistry.getDefaultRegistry();
  private XdsClientPoolFactory xdsClientPoolFactory =
          SharedXdsClientPoolProvider.getDefaultProvider();
  private Map<String, ?> bootstrapOverride;
  private long drainGraceTime = 10;
  private TimeUnit drainGraceTimeUnit = TimeUnit.MINUTES;

  private XdsServerBuilder(NettyServerBuilder nettyDelegate, int port) {
    this.delegate = nettyDelegate;
    this.port = port;
    xdsServingStatusListener = new DefaultListener("port:" + port);
  }

  @Override
  @Internal
  protected ServerBuilder<?> delegate() {
    checkState(!isServerBuilt.get(), "Server already built!");
    return delegate;
  }

  /** Set the {@link XdsServingStatusListener} to receive "serving" and "not serving" states. */
  public XdsServerBuilder xdsServingStatusListener(
      XdsServingStatusListener xdsServingStatusListener) {
    this.xdsServingStatusListener =
        checkNotNull(xdsServingStatusListener, "xdsServingStatusListener");
    return this;
  }

  /**
   * Sets the grace time when draining connections with outdated configuration. When an xDS config
   * update changes connection configuration, pre-existing connections stop accepting new RPCs to be
   * replaced by new connections. RPCs on those pre-existing connections have the grace time to
   * complete. RPCs that do not complete in time will be cancelled, allowing the connection to
   * terminate. {@code Long.MAX_VALUE} nano seconds or an unreasonably large value are considered
   * infinite. The default is 10 minutes.
   */
  public XdsServerBuilder drainGraceTime(long drainGraceTime, TimeUnit drainGraceTimeUnit) {
    checkArgument(drainGraceTime >= 0, "drain grace time must be non-negative: %s",
        drainGraceTime);
    checkNotNull(drainGraceTimeUnit, "drainGraceTimeUnit");
    if (drainGraceTimeUnit.toNanos(drainGraceTime) >= AS_LARGE_AS_INFINITE) {
      drainGraceTimeUnit = null;
    }
    this.drainGraceTime = drainGraceTime;
    this.drainGraceTimeUnit = drainGraceTimeUnit;
    return this;
  }

  @DoNotCall("Unsupported. Use forPort(int, ServerCredentials) instead")
  public static ServerBuilder<?> forPort(int port) {
    throw new UnsupportedOperationException(
            "Unsupported call - use forPort(int, ServerCredentials)");
  }

  /** Creates a gRPC server builder for the given port. */
  public static XdsServerBuilder forPort(int port, ServerCredentials serverCredentials) {
    checkNotNull(serverCredentials, "serverCredentials");
    InternalProtocolNegotiator.ServerFactory originalNegotiatorFactory =
            InternalNettyServerCredentials.toNegotiator(serverCredentials);
    ServerCredentials wrappedCredentials = InternalNettyServerCredentials.create(
            new FilterChainMatchingNegotiatorServerFactory(originalNegotiatorFactory));
    NettyServerBuilder nettyDelegate = NettyServerBuilder.forPort(port, wrappedCredentials);
    return new XdsServerBuilder(nettyDelegate, port);
  }

  @Override
  public Server build() {
    checkState(isServerBuilt.compareAndSet(false, true), "Server already built!");
    FilterChainSelectorManager filterChainSelectorManager = new FilterChainSelectorManager();
    Attributes.Builder builder = Attributes.newBuilder()
            .set(ATTR_FILTER_CHAIN_SELECTOR_MANAGER, filterChainSelectorManager);
    if (drainGraceTimeUnit != null) {
      builder.set(ATTR_DRAIN_GRACE_NANOS, drainGraceTimeUnit.toNanos(drainGraceTime));
    }
    InternalNettyServerBuilder.eagAttributes(delegate, builder.build());
    return new XdsServerWrapper("0.0.0.0:" + port, delegate, xdsServingStatusListener,
            filterChainSelectorManager, xdsClientPoolFactory, bootstrapOverride, filterRegistry);
  }

  @VisibleForTesting
  XdsServerBuilder xdsClientPoolFactory(XdsClientPoolFactory xdsClientPoolFactory) {
    this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory");
    return this;
  }

  /**
   * Allows providing bootstrap override, useful for testing.
   */
  public XdsServerBuilder overrideBootstrapForTest(Map<String, ?> bootstrapOverride) {
    this.bootstrapOverride = checkNotNull(bootstrapOverride, "bootstrapOverride");
    if (this.xdsClientPoolFactory == SharedXdsClientPoolProvider.getDefaultProvider()) {
      this.xdsClientPoolFactory = new SharedXdsClientPoolProvider();
    }
    return this;
  }

  /**
   * Returns the delegate {@link NettyServerBuilder} to allow experimental level
   * transport-specific configuration. Note this API will always be experimental.
   */
  public ServerBuilder<?> transportBuilder() {
    return delegate;
  }

  /**
   * Applications can register this listener to receive "serving" and "not serving" states of
   * the server using {@link #xdsServingStatusListener(XdsServingStatusListener)}.
   */
  public interface XdsServingStatusListener {

    /** Callback invoked when server begins serving. */
    void onServing();

    /** Callback invoked when server is forced to be "not serving" due to an error.
     * @param throwable cause of the error
     */
    void onNotServing(Throwable throwable);
  }

  /** Default implementation of {@link XdsServingStatusListener} that logs at WARNING level. */
  private static class DefaultListener implements XdsServingStatusListener {
    private final Logger logger;
    private final String prefix;
    boolean notServingDueToError;

    DefaultListener(String prefix) {
      logger = Logger.getLogger(DefaultListener.class.getName());
      this.prefix = prefix;
    }

    /** Log calls to onServing() following a call to onNotServing() at WARNING level. */
    @Override
    public void onServing() {
      if (notServingDueToError) {
        notServingDueToError = false;
        logger.warning("[" + prefix + "] Entering serving state.");
      }
    }

    @Override
    public void onNotServing(Throwable throwable) {
      logger.warning("[" + prefix + "] " + throwable.getMessage());
      notServingDueToError = true;
    }
  }
}

1	/*
2	* Copyright 2019 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.xds;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static com.google.common.base.Preconditions.checkState;
22	import static io.grpc.xds.XdsAttributes.ATTR_DRAIN_GRACE_NANOS;
23	import static io.grpc.xds.XdsAttributes.ATTR_FILTER_CHAIN_SELECTOR_MANAGER;
24
25	import com.google.common.annotations.VisibleForTesting;
26	import com.google.errorprone.annotations.DoNotCall;
27	import io.grpc.Attributes;
28	import io.grpc.ExperimentalApi;
29	import io.grpc.ForwardingServerBuilder;
30	import io.grpc.Internal;
31	import io.grpc.Server;
32	import io.grpc.ServerBuilder;
33	import io.grpc.ServerCredentials;
34	import io.grpc.netty.InternalNettyServerBuilder;
35	import io.grpc.netty.InternalNettyServerCredentials;
36	import io.grpc.netty.InternalProtocolNegotiator;
37	import io.grpc.netty.NettyServerBuilder;
38	import io.grpc.xds.FilterChainMatchingProtocolNegotiators.FilterChainMatchingNegotiatorServerFactory;
39	import java.util.Map;
40	import java.util.concurrent.TimeUnit;
41	import java.util.concurrent.atomic.AtomicBoolean;
42	import java.util.logging.Logger;
43
44	/**
45	* A version of {@link ServerBuilder} to create xDS managed servers.
46	*/
47	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/7514")
48	public final class XdsServerBuilder extends ForwardingServerBuilder<XdsServerBuilder> {
49	private static final long AS_LARGE_AS_INFINITE = TimeUnit.DAYS.toNanos(1000);	1✔
50
51	private final NettyServerBuilder delegate;
52	private final int port;
53	private XdsServingStatusListener xdsServingStatusListener;
54	private AtomicBoolean isServerBuilt = new AtomicBoolean(false);	1✔
55	private final FilterRegistry filterRegistry = FilterRegistry.getDefaultRegistry();	1✔
56	private XdsClientPoolFactory xdsClientPoolFactory =	1✔
57	SharedXdsClientPoolProvider.getDefaultProvider();	1✔
58	private Map<String, ?> bootstrapOverride;
59	private long drainGraceTime = 10;	1✔
60	private TimeUnit drainGraceTimeUnit = TimeUnit.MINUTES;	1✔
61
62	private XdsServerBuilder(NettyServerBuilder nettyDelegate, int port) {	1✔
63	this.delegate = nettyDelegate;	1✔
64	this.port = port;	1✔
65	xdsServingStatusListener = new DefaultListener("port:" + port);	1✔
66	}	1✔
67
68	@Override
69	@Internal
70	protected ServerBuilder<?> delegate() {
71	checkState(!isServerBuilt.get(), "Server already built!");	1✔
72	return delegate;	1✔
73	}
74
75	/** Set the {@link XdsServingStatusListener} to receive "serving" and "not serving" states. */
76	public XdsServerBuilder xdsServingStatusListener(
77	XdsServingStatusListener xdsServingStatusListener) {
78	this.xdsServingStatusListener =	1✔
79	checkNotNull(xdsServingStatusListener, "xdsServingStatusListener");	1✔
80	return this;	1✔
81	}
82
83	/**
84	* Sets the grace time when draining connections with outdated configuration. When an xDS config
85	* update changes connection configuration, pre-existing connections stop accepting new RPCs to be
86	* replaced by new connections. RPCs on those pre-existing connections have the grace time to
87	* complete. RPCs that do not complete in time will be cancelled, allowing the connection to
88	* terminate. {@code Long.MAX_VALUE} nano seconds or an unreasonably large value are considered
89	* infinite. The default is 10 minutes.
90	*/
91	public XdsServerBuilder drainGraceTime(long drainGraceTime, TimeUnit drainGraceTimeUnit) {
92	checkArgument(drainGraceTime >= 0, "drain grace time must be non-negative: %s",	1✔
93	drainGraceTime);
94	checkNotNull(drainGraceTimeUnit, "drainGraceTimeUnit");	×
95	if (drainGraceTimeUnit.toNanos(drainGraceTime) >= AS_LARGE_AS_INFINITE) {	×
96	drainGraceTimeUnit = null;	×
97	}
98	this.drainGraceTime = drainGraceTime;	×
99	this.drainGraceTimeUnit = drainGraceTimeUnit;	×
100	return this;	×
101	}
102
103	@DoNotCall("Unsupported. Use forPort(int, ServerCredentials) instead")
104	public static ServerBuilder<?> forPort(int port) {
105	throw new UnsupportedOperationException(	×
106	"Unsupported call - use forPort(int, ServerCredentials)");
107	}
108
109	/** Creates a gRPC server builder for the given port. */
110	public static XdsServerBuilder forPort(int port, ServerCredentials serverCredentials) {
111	checkNotNull(serverCredentials, "serverCredentials");	1✔
112	InternalProtocolNegotiator.ServerFactory originalNegotiatorFactory =	1✔
113	InternalNettyServerCredentials.toNegotiator(serverCredentials);	1✔
114	ServerCredentials wrappedCredentials = InternalNettyServerCredentials.create(	1✔
115	new FilterChainMatchingNegotiatorServerFactory(originalNegotiatorFactory));
116	NettyServerBuilder nettyDelegate = NettyServerBuilder.forPort(port, wrappedCredentials);	1✔
117	return new XdsServerBuilder(nettyDelegate, port);	1✔
118	}
119
120	@Override
121	public Server build() {
122	checkState(isServerBuilt.compareAndSet(false, true), "Server already built!");	1✔
123	FilterChainSelectorManager filterChainSelectorManager = new FilterChainSelectorManager();	1✔
124	Attributes.Builder builder = Attributes.newBuilder()	1✔
125	.set(ATTR_FILTER_CHAIN_SELECTOR_MANAGER, filterChainSelectorManager);	1✔
126	if (drainGraceTimeUnit != null) {	1✔
127	builder.set(ATTR_DRAIN_GRACE_NANOS, drainGraceTimeUnit.toNanos(drainGraceTime));	1✔
128	}
129	InternalNettyServerBuilder.eagAttributes(delegate, builder.build());	1✔
130	return new XdsServerWrapper("0.0.0.0:" + port, delegate, xdsServingStatusListener,	1✔
131	filterChainSelectorManager, xdsClientPoolFactory, bootstrapOverride, filterRegistry);
132	}
133
134	@VisibleForTesting
135	XdsServerBuilder xdsClientPoolFactory(XdsClientPoolFactory xdsClientPoolFactory) {
136	this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory");	1✔
137	return this;	1✔
138	}
139
140	/**
141	* Allows providing bootstrap override, useful for testing.
142	*/
143	public XdsServerBuilder overrideBootstrapForTest(Map<String, ?> bootstrapOverride) {
144	this.bootstrapOverride = checkNotNull(bootstrapOverride, "bootstrapOverride");	1✔
145	if (this.xdsClientPoolFactory == SharedXdsClientPoolProvider.getDefaultProvider()) {	1✔
146	this.xdsClientPoolFactory = new SharedXdsClientPoolProvider();	1✔
147	}
148	return this;	1✔
149	}
150
151	/**
152	* Returns the delegate {@link NettyServerBuilder} to allow experimental level
153	* transport-specific configuration. Note this API will always be experimental.
154	*/
155	public ServerBuilder<?> transportBuilder() {
156	return delegate;	×
157	}
158
159	/**
160	* Applications can register this listener to receive "serving" and "not serving" states of
161	* the server using {@link #xdsServingStatusListener(XdsServingStatusListener)}.
162	*/
163	public interface XdsServingStatusListener {
164
165	/** Callback invoked when server begins serving. */
166	void onServing();
167
168	/** Callback invoked when server is forced to be "not serving" due to an error.
169	* @param throwable cause of the error
170	*/
171	void onNotServing(Throwable throwable);
172	}
173
174	/** Default implementation of {@link XdsServingStatusListener} that logs at WARNING level. */
175	private static class DefaultListener implements XdsServingStatusListener {
176	private final Logger logger;
177	private final String prefix;
178	boolean notServingDueToError;
179
180	DefaultListener(String prefix) {	1✔
181	logger = Logger.getLogger(DefaultListener.class.getName());	1✔
182	this.prefix = prefix;	1✔
183	}	1✔
184
185	/** Log calls to onServing() following a call to onNotServing() at WARNING level. */
186	@Override
187	public void onServing() {
188	if (notServingDueToError) {	1✔
189	notServingDueToError = false;	×
190	logger.warning("[" + prefix + "] Entering serving state.");	×
191	}
192	}	1✔
193
194	@Override
195	public void onNotServing(Throwable throwable) {
196	logger.warning("[" + prefix + "] " + throwable.getMessage());	×
197	notServingDueToError = true;	×
198	}	×
199	}
200	}

grpc / grpc-java / #20033

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