#20349

Committed 07 Jul 2026 10:24AM UTC coverage: 89.111% (-0.01%) from 89.122%

Build # #20349

Build Type

push

github

Committed by

web-flow

Commit Message

enable child channel plugins (#12578)

Implements https://github.com/grpc/proposal/pull/529

Coverage Stats

38030 of 42677 relevant lines covered (89.11%)

0.89 hits per line

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

78.33

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

  @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,
        this.channelConfigurator);
  }

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

grpc / grpc-java / #20349

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