#20378

Committed 29 Jul 2026 05:42PM UTC coverage: 89.156% (-0.02%) from 89.175%

Build # #20378

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Supports injecting custom ldsResourceNameResolvers (#12925)

Also add support for creating XdsServerBuilder with SocketAddresses.

Coverage Stats

38296 of 42954 relevant lines covered (89.16%)

0.89 hits per line

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

79.73

/../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.ChannelConfigurator;
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.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.logging.Logger;
import javax.annotation.Nullable;

/**
 * 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;
  @Nullable private Function<String, String> ldsResourceNameResolver;
  private long drainGraceTime = 10;
  private TimeUnit drainGraceTimeUnit = TimeUnit.MINUTES;
  private ChannelConfigurator channelConfigurator = builder -> { };


  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;
  }

  /**
   * Sets the configurator that will be stored in the server built by this builder.
   *
   * <p>This configurator will subsequently be used to configure any child channels
   * created by that server.
   *
   * @param channelConfigurator the configurator to store in the channel.
   * @return this
   */
  public XdsServerBuilder childChannelConfigurator(ChannelConfigurator channelConfigurator) {
    this.channelConfigurator = checkNotNull(channelConfigurator, "channelConfigurator");
    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);
  }

  /** Creates a gRPC server builder for the given address. */
  public static XdsServerBuilder forAddress(
      SocketAddress address, ServerCredentials serverCredentials) {
    checkNotNull(address, "address");
    checkNotNull(serverCredentials, "serverCredentials");
    InternalProtocolNegotiator.ServerFactory originalNegotiatorFactory =
        InternalNettyServerCredentials.toNegotiator(serverCredentials);
    ServerCredentials wrappedCredentials =
        InternalNettyServerCredentials.create(
            new FilterChainMatchingNegotiatorServerFactory(originalNegotiatorFactory));
    NettyServerBuilder nettyDelegate = NettyServerBuilder.forAddress(address, wrappedCredentials);
    int port = 0;
    if (address instanceof InetSocketAddress) {
      InetSocketAddress inetSocketAddress = (InetSocketAddress) address;
      port = inetSocketAddress.getPort();
    }
    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,
        ldsResourceNameResolver,
        filterRegistry,
        this.channelConfigurator);
  }

  /**
   * Provides a function that takes the listening address and returns the LDS resource name. When
   * provided, this overrides the server_listener_resource_name_template in the bootstrap.
   */
  public XdsServerBuilder ldsResourceNameResolver(
      Function<String, String> ldsResourceNameResolver) {
    this.ldsResourceNameResolver = checkNotNull(ldsResourceNameResolver, "ldsResourceNameResolver");
    return this;
  }

  @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.ChannelConfigurator;
29	import io.grpc.ExperimentalApi;
30	import io.grpc.ForwardingServerBuilder;
31	import io.grpc.Internal;
32	import io.grpc.Server;
33	import io.grpc.ServerBuilder;
34	import io.grpc.ServerCredentials;
35	import io.grpc.netty.InternalNettyServerBuilder;
36	import io.grpc.netty.InternalNettyServerCredentials;
37	import io.grpc.netty.InternalProtocolNegotiator;
38	import io.grpc.netty.NettyServerBuilder;
39	import io.grpc.xds.FilterChainMatchingProtocolNegotiators.FilterChainMatchingNegotiatorServerFactory;
40	import java.net.InetSocketAddress;
41	import java.net.SocketAddress;
42	import java.util.Map;
43	import java.util.concurrent.TimeUnit;
44	import java.util.concurrent.atomic.AtomicBoolean;
45	import java.util.function.Function;
46	import java.util.logging.Logger;
47	import javax.annotation.Nullable;
48
49	/**
50	* A version of {@link ServerBuilder} to create xDS managed servers.
51	*/
52	@ExperimentalApi("https://github.com/grpc/grpc-java/issues/7514")
53	public final class XdsServerBuilder extends ForwardingServerBuilder<XdsServerBuilder> {
54	private static final long AS_LARGE_AS_INFINITE = TimeUnit.DAYS.toNanos(1000);	1✔
55
56	private final NettyServerBuilder delegate;
57	private final int port;
58	private XdsServingStatusListener xdsServingStatusListener;
59	private AtomicBoolean isServerBuilt = new AtomicBoolean(false);	1✔
60	private final FilterRegistry filterRegistry = FilterRegistry.getDefaultRegistry();	1✔
61	private XdsClientPoolFactory xdsClientPoolFactory =	1✔
62	SharedXdsClientPoolProvider.getDefaultProvider();	1✔
63	private Map<String, ?> bootstrapOverride;
64	@Nullable private Function<String, String> ldsResourceNameResolver;
65	private long drainGraceTime = 10;	1✔
66	private TimeUnit drainGraceTimeUnit = TimeUnit.MINUTES;	1✔
67	private ChannelConfigurator channelConfigurator = builder -> { };	1✔
68
69
70	private XdsServerBuilder(NettyServerBuilder nettyDelegate, int port) {	1✔
71	this.delegate = nettyDelegate;	1✔
72	this.port = port;	1✔
73	xdsServingStatusListener = new DefaultListener("port:" + port);	1✔
74	}	1✔
75
76	@Override
77	@Internal
78	protected ServerBuilder<?> delegate() {
79	checkState(!isServerBuilt.get(), "Server already built!");	1✔
80	return delegate;	1✔
81	}
82
83	/** Set the {@link XdsServingStatusListener} to receive "serving" and "not serving" states. */
84	public XdsServerBuilder xdsServingStatusListener(
85	XdsServingStatusListener xdsServingStatusListener) {
86	this.xdsServingStatusListener =	1✔
87	checkNotNull(xdsServingStatusListener, "xdsServingStatusListener");	1✔
88	return this;	1✔
89	}
90
91	/**
92	* Sets the grace time when draining connections with outdated configuration. When an xDS config
93	* update changes connection configuration, pre-existing connections stop accepting new RPCs to be
94	* replaced by new connections. RPCs on those pre-existing connections have the grace time to
95	* complete. RPCs that do not complete in time will be cancelled, allowing the connection to
96	* terminate. {@code Long.MAX_VALUE} nano seconds or an unreasonably large value are considered
97	* infinite. The default is 10 minutes.
98	*/
99	public XdsServerBuilder drainGraceTime(long drainGraceTime, TimeUnit drainGraceTimeUnit) {
100	checkArgument(drainGraceTime >= 0, "drain grace time must be non-negative: %s",	1✔
101	drainGraceTime);
102	checkNotNull(drainGraceTimeUnit, "drainGraceTimeUnit");	×
103	if (drainGraceTimeUnit.toNanos(drainGraceTime) >= AS_LARGE_AS_INFINITE) {	×
104	drainGraceTimeUnit = null;	×
105	}
106	this.drainGraceTime = drainGraceTime;	×
107	this.drainGraceTimeUnit = drainGraceTimeUnit;	×
108	return this;	×
109	}
110
111	/**
112	* Sets the configurator that will be stored in the server built by this builder.
113	*
114	* <p>This configurator will subsequently be used to configure any child channels
115	* created by that server.
116	*
117	* @param channelConfigurator the configurator to store in the channel.
118	* @return this
119	*/
120	public XdsServerBuilder childChannelConfigurator(ChannelConfigurator channelConfigurator) {
121	this.channelConfigurator = checkNotNull(channelConfigurator, "channelConfigurator");	1✔
122	return this;	1✔
123	}
124
125	@DoNotCall("Unsupported. Use forPort(int, ServerCredentials) instead")
126	public static ServerBuilder<?> forPort(int port) {
127	throw new UnsupportedOperationException(	×
128	"Unsupported call - use forPort(int, ServerCredentials)");
129	}
130
131	/** Creates a gRPC server builder for the given port. */
132	public static XdsServerBuilder forPort(int port, ServerCredentials serverCredentials) {
133	checkNotNull(serverCredentials, "serverCredentials");	1✔
134	InternalProtocolNegotiator.ServerFactory originalNegotiatorFactory =	1✔
135	InternalNettyServerCredentials.toNegotiator(serverCredentials);	1✔
136	ServerCredentials wrappedCredentials = InternalNettyServerCredentials.create(	1✔
137	new FilterChainMatchingNegotiatorServerFactory(originalNegotiatorFactory));
138	NettyServerBuilder nettyDelegate = NettyServerBuilder.forPort(port, wrappedCredentials);	1✔
139	return new XdsServerBuilder(nettyDelegate, port);	1✔
140	}
141
142	/** Creates a gRPC server builder for the given address. */
143	public static XdsServerBuilder forAddress(
144	SocketAddress address, ServerCredentials serverCredentials) {
145	checkNotNull(address, "address");	1✔
146	checkNotNull(serverCredentials, "serverCredentials");	1✔
147	InternalProtocolNegotiator.ServerFactory originalNegotiatorFactory =	1✔
148	InternalNettyServerCredentials.toNegotiator(serverCredentials);	1✔
149	ServerCredentials wrappedCredentials =	1✔
150	InternalNettyServerCredentials.create(	1✔
151	new FilterChainMatchingNegotiatorServerFactory(originalNegotiatorFactory));
152	NettyServerBuilder nettyDelegate = NettyServerBuilder.forAddress(address, wrappedCredentials);	1✔
153	int port = 0;	1✔
154	if (address instanceof InetSocketAddress) {	1✔
155	InetSocketAddress inetSocketAddress = (InetSocketAddress) address;	1✔
156	port = inetSocketAddress.getPort();	1✔
157	}
158	return new XdsServerBuilder(nettyDelegate, port);	1✔
159	}
160
161	@Override
162	public Server build() {
163	checkState(isServerBuilt.compareAndSet(false, true), "Server already built!");	1✔
164	FilterChainSelectorManager filterChainSelectorManager = new FilterChainSelectorManager();	1✔
165	Attributes.Builder builder = Attributes.newBuilder()	1✔
166	.set(ATTR_FILTER_CHAIN_SELECTOR_MANAGER, filterChainSelectorManager);	1✔
167	if (drainGraceTimeUnit != null) {	1✔
168	builder.set(ATTR_DRAIN_GRACE_NANOS, drainGraceTimeUnit.toNanos(drainGraceTime));	1✔
169	}
170	InternalNettyServerBuilder.eagAttributes(delegate, builder.build());	1✔
171	return new XdsServerWrapper(	1✔
172	"0.0.0.0:" + port,
173	delegate,
174	xdsServingStatusListener,
175	filterChainSelectorManager,
176	xdsClientPoolFactory,
177	bootstrapOverride,
178	ldsResourceNameResolver,
179	filterRegistry,
180	this.channelConfigurator);
181	}
182
183	/**
184	* Provides a function that takes the listening address and returns the LDS resource name. When
185	* provided, this overrides the server_listener_resource_name_template in the bootstrap.
186	*/
187	public XdsServerBuilder ldsResourceNameResolver(
188	Function<String, String> ldsResourceNameResolver) {
189	this.ldsResourceNameResolver = checkNotNull(ldsResourceNameResolver, "ldsResourceNameResolver");	×
190	return this;	×
191	}
192
193	@VisibleForTesting
194	XdsServerBuilder xdsClientPoolFactory(XdsClientPoolFactory xdsClientPoolFactory) {
195	this.xdsClientPoolFactory = checkNotNull(xdsClientPoolFactory, "xdsClientPoolFactory");	1✔
196	return this;	1✔
197	}
198
199	/**
200	* Allows providing bootstrap override, useful for testing.
201	*/
202	public XdsServerBuilder overrideBootstrapForTest(Map<String, ?> bootstrapOverride) {
203	this.bootstrapOverride = checkNotNull(bootstrapOverride, "bootstrapOverride");	1✔
204	if (this.xdsClientPoolFactory == SharedXdsClientPoolProvider.getDefaultProvider()) {	1✔
205	this.xdsClientPoolFactory = new SharedXdsClientPoolProvider();	1✔
206	}
207	return this;	1✔
208	}
209
210	/**
211	* Returns the delegate {@link NettyServerBuilder} to allow experimental level
212	* transport-specific configuration. Note this API will always be experimental.
213	*/
214	public ServerBuilder<?> transportBuilder() {
215	return delegate;	×
216	}
217
218	/**
219	* Applications can register this listener to receive "serving" and "not serving" states of
220	* the server using {@link #xdsServingStatusListener(XdsServingStatusListener)}.
221	*/
222	public interface XdsServingStatusListener {
223
224	/** Callback invoked when server begins serving. */
225	void onServing();
226
227	/** Callback invoked when server is forced to be "not serving" due to an error.
228	* @param throwable cause of the error
229	*/
230	void onNotServing(Throwable throwable);
231	}
232
233	/** Default implementation of {@link XdsServingStatusListener} that logs at WARNING level. */
234	private static class DefaultListener implements XdsServingStatusListener {
235	private final Logger logger;
236	private final String prefix;
237	boolean notServingDueToError;
238
239	DefaultListener(String prefix) {	1✔
240	logger = Logger.getLogger(DefaultListener.class.getName());	1✔
241	this.prefix = prefix;	1✔
242	}	1✔
243
244	/** Log calls to onServing() following a call to onNotServing() at WARNING level. */
245	@Override
246	public void onServing() {
247	if (notServingDueToError) {	1✔
248	notServingDueToError = false;	×
249	logger.warning("[" + prefix + "] Entering serving state.");	×
250	}
251	}	1✔
252
253	@Override
254	public void onNotServing(Throwable throwable) {
255	logger.warning("[" + prefix + "] " + throwable.getMessage());	×
256	notServingDueToError = true;	×
257	}	×
258	}
259	}

grpc / grpc-java / #20378

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