#20136

Committed 06 Jan 2026 05:27AM UTC coverage: 88.693% (+0.01%) from 88.681%

Build # #20136

Build Type

push

github

Committed by

web-flow

Commit Message

core: Implement oobChannel with resolvingOobChannel

The most important part of this change is to ensure that CallCredentials
are not propagated to the OOB channel. Because the authority of the OOB
channel doesn't match the parent channel, we must ensure that any bearer
tokens are not sent to the different server. However, this was not a
problem because resolvingOobChannel has the same constraint. (RLS has a
different constraint, but we were able to let RLS manage that itself.)

This commit does change the behavior of channelz, shutdown, and metrics
for the OOB channel. Previously the OOB channel was registered with
channelz, but it is only a TODO for resolving channel. Channel shutdown
no longer shuts down the OOB channel and it no longer waits for the OOB
channel to terminate before becoming terminated itself. That is also a
pre-existing TODO. Since ManagedChannelImplBuilder is now being used,
global configurators and census are enabled. The proper behavior here is
still being determined, but we would want it to be the same for
resolving OOB channel and OOB channel.

The OOB channel used to refresh the name resolution when the subchannel
went IDLE or TF. That is an older behavior from back when regular
subchannels would also cause the name resolver to refresh. Now-a-days
that goes though the LB tree. gRPC-LB already refreshes name resolution
when its RPC closes, so no longer doing it automatically should be fine.

balancerRpcExecutorPool no longer has its lifetime managed by the child.
It'd be easiest to not use it at all from OOB channel, which wouldn't
actually change the regular behavior, as channels already use the same
executor by default. However, the tests are making use of the executor
being injected, so some propagation needs to be preserved.

Lots of OOB channel tests were deleted, but these were either testing
OobChannel, which is now gone, or things like channelz, which are known
to no longer work like before.

Run Details

35361 of 39869 relevant lines covered (88.69%)

0.89 hits per line

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

98.75

/../core/src/main/java/io/grpc/internal/PickFirstLoadBalancer.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.internal;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.SHUTDOWN;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nullable;

/**
 * A {@link LoadBalancer} that provides no load-balancing over the addresses from the {@link
 * io.grpc.NameResolver}.  The channel's default behavior is used, which is walking down the address
 * list and sticking to the first that works.
 */
final class PickFirstLoadBalancer extends LoadBalancer {
  private final Helper helper;
  private Subchannel subchannel;
  private ConnectivityState currentState = IDLE;

  PickFirstLoadBalancer(Helper helper) {
    this.helper = checkNotNull(helper, "helper");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();
    if (servers.isEmpty()) {
      Status unavailableStatus = Status.UNAVAILABLE.withDescription(
              "NameResolver returned no usable address. addrs=" + resolvedAddresses.getAddresses()
                      + ", attrs=" + resolvedAddresses.getAttributes());
      handleNameResolutionError(unavailableStatus);
      return unavailableStatus;
    }

    // We can optionally be configured to shuffle the address list. This can help better distribute
    // the load.
    if (resolvedAddresses.getLoadBalancingPolicyConfig() instanceof PickFirstLoadBalancerConfig) {
      PickFirstLoadBalancerConfig config
          = (PickFirstLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
      if (config.shuffleAddressList != null && config.shuffleAddressList) {
        servers = new ArrayList<EquivalentAddressGroup>(servers);
        Collections.shuffle(servers,
            config.randomSeed != null ? new Random(config.randomSeed) : new Random());
      }
    }

    if (subchannel == null) {
      final Subchannel subchannel = helper.createSubchannel(
          CreateSubchannelArgs.newBuilder()
              .setAddresses(servers)
              .build());
      subchannel.start(new SubchannelStateListener() {
          @Override
          public void onSubchannelState(ConnectivityStateInfo stateInfo) {
            processSubchannelState(subchannel, stateInfo);
          }
        });
      this.subchannel = subchannel;

      // The channel state does not get updated when doing name resolving today, so for the moment
      // let LB report CONNECTION and call subchannel.requestConnection() immediately.
      updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));
      subchannel.requestConnection();
    } else {
      subchannel.updateAddresses(servers);
    }

    return Status.OK;
  }

  @Override
  public void handleNameResolutionError(Status error) {
    if (subchannel != null) {
      subchannel.shutdown();
      subchannel = null;
    }

    // NB(lukaszx0) Whether we should propagate the error unconditionally is arguable. It's fine
    // for time being.
    updateBalancingState(TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
  }

  private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
    ConnectivityState newState = stateInfo.getState();
    if (newState == SHUTDOWN) {
      return;
    }
    if (newState == TRANSIENT_FAILURE || newState == IDLE) {
      helper.refreshNameResolution();
    }

    // If we are transitioning from a TRANSIENT_FAILURE to CONNECTING or IDLE we ignore this state
    // transition and still keep the LB in TRANSIENT_FAILURE state. This is referred to as "sticky
    // transient failure". Only a subchannel state change to READY will get the LB out of
    // TRANSIENT_FAILURE. If the state is IDLE we additionally request a new connection so that we
    // keep retrying for a connection.
    if (currentState == TRANSIENT_FAILURE) {
      if (newState == CONNECTING) {
        return;
      } else if (newState == IDLE) {
        requestConnection();
        return;
      }
    }

    SubchannelPicker picker;
    switch (newState) {
      case IDLE:
        picker = new RequestConnectionPicker();
        break;
      case CONNECTING:
        // It's safe to use RequestConnectionPicker here, so when coming from IDLE we could leave
        // the current picker in-place. But ignoring the potential optimization is simpler.
        picker = new FixedResultPicker(PickResult.withNoResult());
        break;
      case READY:
        picker = new FixedResultPicker(PickResult.withSubchannel(subchannel));
        break;
      case TRANSIENT_FAILURE:
        picker = new FixedResultPicker(PickResult.withError(stateInfo.getStatus()));
        break;
      default:
        throw new IllegalArgumentException("Unsupported state:" + newState);
    }

    updateBalancingState(newState, picker);
  }

  private void updateBalancingState(ConnectivityState state, SubchannelPicker picker) {
    currentState = state;
    helper.updateBalancingState(state, picker);
  }

  @Override
  public void shutdown() {
    if (subchannel != null) {
      subchannel.shutdown();
    }
  }

  @Override
  public void requestConnection() {
    if (subchannel != null) {
      subchannel.requestConnection();
    }
  }

  /** Picker that requests connection during the first pick, and returns noResult. */
  private final class RequestConnectionPicker extends SubchannelPicker {
    private final AtomicBoolean connectionRequested = new AtomicBoolean(false);

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      if (connectionRequested.compareAndSet(false, true)) {
        helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);
      }
      return PickResult.withNoResult();
    }
  }

  public static final class PickFirstLoadBalancerConfig {

    @Nullable
    public final Boolean shuffleAddressList;

    // For testing purposes only, not meant to be parsed from a real config.
    @Nullable final Long randomSeed;

    public PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList) {
      this(shuffleAddressList, null);
    }

    PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList, @Nullable Long randomSeed) {
      this.shuffleAddressList = shuffleAddressList;
      this.randomSeed = randomSeed;
    }
  }
}

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.internal;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20	import static io.grpc.ConnectivityState.CONNECTING;
21	import static io.grpc.ConnectivityState.IDLE;
22	import static io.grpc.ConnectivityState.SHUTDOWN;
23	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
24
25	import io.grpc.ConnectivityState;
26	import io.grpc.ConnectivityStateInfo;
27	import io.grpc.EquivalentAddressGroup;
28	import io.grpc.LoadBalancer;
29	import io.grpc.Status;
30	import java.util.ArrayList;
31	import java.util.Collections;
32	import java.util.List;
33	import java.util.Random;
34	import java.util.concurrent.atomic.AtomicBoolean;
35	import javax.annotation.Nullable;
36
37	/**
38	* A {@link LoadBalancer} that provides no load-balancing over the addresses from the {@link
39	* io.grpc.NameResolver}. The channel's default behavior is used, which is walking down the address
40	* list and sticking to the first that works.
41	*/
42	final class PickFirstLoadBalancer extends LoadBalancer {
43	private final Helper helper;
44	private Subchannel subchannel;
45	private ConnectivityState currentState = IDLE;	1✔
46
47	PickFirstLoadBalancer(Helper helper) {	1✔
48	this.helper = checkNotNull(helper, "helper");	1✔
49	}	1✔
50
51	@Override
52	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
53	List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();	1✔
54	if (servers.isEmpty()) {	1✔
55	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
56	"NameResolver returned no usable address. addrs=" + resolvedAddresses.getAddresses()	1✔
57	+ ", attrs=" + resolvedAddresses.getAttributes());	1✔
58	handleNameResolutionError(unavailableStatus);	1✔
59	return unavailableStatus;	1✔
60	}
61
62	// We can optionally be configured to shuffle the address list. This can help better distribute
63	// the load.
64	if (resolvedAddresses.getLoadBalancingPolicyConfig() instanceof PickFirstLoadBalancerConfig) {	1✔
65	PickFirstLoadBalancerConfig config	1✔
66	= (PickFirstLoadBalancerConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
67	if (config.shuffleAddressList != null && config.shuffleAddressList) {	1✔
68	servers = new ArrayList<EquivalentAddressGroup>(servers);	1✔
69	Collections.shuffle(servers,	1✔
70	config.randomSeed != null ? new Random(config.randomSeed) : new Random());	1✔
71	}
72	}
73
74	if (subchannel == null) {	1✔
75	final Subchannel subchannel = helper.createSubchannel(	1✔
76	CreateSubchannelArgs.newBuilder()	1✔
77	.setAddresses(servers)	1✔
78	.build());	1✔
79	subchannel.start(new SubchannelStateListener() {	1✔
80	@Override
81	public void onSubchannelState(ConnectivityStateInfo stateInfo) {
82	processSubchannelState(subchannel, stateInfo);	1✔
83	}	1✔
84	});
85	this.subchannel = subchannel;	1✔
86
87	// The channel state does not get updated when doing name resolving today, so for the moment
88	// let LB report CONNECTION and call subchannel.requestConnection() immediately.
89	updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));	1✔
90	subchannel.requestConnection();	1✔
91	} else {	1✔
92	subchannel.updateAddresses(servers);	1✔
93	}
94
95	return Status.OK;	1✔
96	}
97
98	@Override
99	public void handleNameResolutionError(Status error) {
100	if (subchannel != null) {	1✔
101	subchannel.shutdown();	1✔
102	subchannel = null;	1✔
103	}
104
105	// NB(lukaszx0) Whether we should propagate the error unconditionally is arguable. It's fine
106	// for time being.
107	updateBalancingState(TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
108	}	1✔
109
110	private void processSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
111	ConnectivityState newState = stateInfo.getState();	1✔
112	if (newState == SHUTDOWN) {	1✔
113	return;	1✔
114	}
115	if (newState == TRANSIENT_FAILURE \|\| newState == IDLE) {	1✔
116	helper.refreshNameResolution();	1✔
117	}
118
119	// If we are transitioning from a TRANSIENT_FAILURE to CONNECTING or IDLE we ignore this state
120	// transition and still keep the LB in TRANSIENT_FAILURE state. This is referred to as "sticky
121	// transient failure". Only a subchannel state change to READY will get the LB out of
122	// TRANSIENT_FAILURE. If the state is IDLE we additionally request a new connection so that we
123	// keep retrying for a connection.
124	if (currentState == TRANSIENT_FAILURE) {	1✔
125	if (newState == CONNECTING) {	1✔
126	return;	1✔
127	} else if (newState == IDLE) {	1✔
128	requestConnection();	1✔
129	return;	1✔
130	}
131	}
132
133	SubchannelPicker picker;
134	switch (newState) {	1✔
135	case IDLE:
136	picker = new RequestConnectionPicker();	1✔
137	break;	1✔
138	case CONNECTING:
139	// It's safe to use RequestConnectionPicker here, so when coming from IDLE we could leave
140	// the current picker in-place. But ignoring the potential optimization is simpler.
141	picker = new FixedResultPicker(PickResult.withNoResult());	1✔
142	break;	1✔
143	case READY:
144	picker = new FixedResultPicker(PickResult.withSubchannel(subchannel));	1✔
145	break;	1✔
146	case TRANSIENT_FAILURE:
147	picker = new FixedResultPicker(PickResult.withError(stateInfo.getStatus()));	1✔
148	break;	1✔
149	default:
150	throw new IllegalArgumentException("Unsupported state:" + newState);	×
151	}
152
153	updateBalancingState(newState, picker);	1✔
154	}	1✔
155
156	private void updateBalancingState(ConnectivityState state, SubchannelPicker picker) {
157	currentState = state;	1✔
158	helper.updateBalancingState(state, picker);	1✔
159	}	1✔
160
161	@Override
162	public void shutdown() {
163	if (subchannel != null) {	1✔
164	subchannel.shutdown();	1✔
165	}
166	}	1✔
167
168	@Override
169	public void requestConnection() {
170	if (subchannel != null) {	1✔
171	subchannel.requestConnection();	1✔
172	}
173	}	1✔
174
175	/** Picker that requests connection during the first pick, and returns noResult. */
176	private final class RequestConnectionPicker extends SubchannelPicker {	1✔
177	private final AtomicBoolean connectionRequested = new AtomicBoolean(false);	1✔
178
179	@Override
180	public PickResult pickSubchannel(PickSubchannelArgs args) {
181	if (connectionRequested.compareAndSet(false, true)) {	1✔
182	helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);	1✔
183	}
184	return PickResult.withNoResult();	1✔
185	}
186	}
187
188	public static final class PickFirstLoadBalancerConfig {
189
190	@Nullable
191	public final Boolean shuffleAddressList;
192
193	// For testing purposes only, not meant to be parsed from a real config.
194	@Nullable final Long randomSeed;
195
196	public PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList) {
197	this(shuffleAddressList, null);	1✔
198	}	1✔
199
200	PickFirstLoadBalancerConfig(@Nullable Boolean shuffleAddressList, @Nullable Long randomSeed) {	1✔
201	this.shuffleAddressList = shuffleAddressList;	1✔
202	this.randomSeed = randomSeed;	1✔
203	}	1✔
204	}
205	}

grpc / grpc-java / #20136

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