#20392

Committed 05 Aug 2026 08:40AM UTC coverage: 89.156% (-0.02%) from 89.175%

Build # #20392

Build Type

push

github

Committed by

web-flow

Commit Message

Fix: Append child channel configurators instead of overwriting (#12921)

appends child channels instead of overwriting them while propagating to subsequent child channels

Coverage Stats

38322 of 42983 relevant lines covered (89.16%)

0.89 hits per line

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

81.01

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

grpc / grpc-java / #20392

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