#20365

Committed 27 Jul 2026 06:04AM UTC coverage: 89.199% (+0.07%) from 89.125%

Build # #20365

Build Type

push

github

Committed by

web-flow

Commit Message

core: Implement LB Delay Observability (Proposal A121) (#12807)

This PR implements **Attempt-Level RPC Delay Observability** across the core channel transport, built-in load balancers, xDS policies, and the OpenTelemetry telemetry plugin, aligned with [gRPC Proposal A121](https://github.com/grpc/proposal/pull/556).

Coverage Stats

38276 of 42911 relevant lines covered (89.2%)

0.89 hits per line

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

99.04

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

/*
 * Copyright 2020 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.checkNotNull;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import io.grpc.ConnectivityState;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.util.MultiChildLoadBalancer;
import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * The top-level load balancing policy for use in XDS.
 * This policy does not immediately delete its children.  Instead, it marks them deactivated
 * and starts a timer for deletion.  If a subsequent address update restores the child, then it is
 * simply reactivated instead of built from scratch.  This is necessary because XDS can frequently
 * remove and then add back a server as machines are rebooted or repurposed for load management.
 *
 * <p>Note that this LB does not automatically reconnect children who go into IDLE status
 */
class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {

  // 15 minutes is long enough for a reboot and the services to restart while not so long that
  // many children are waiting for cleanup.
  @VisibleForTesting
  public static final int DELAYED_CHILD_DELETION_TIME_MINUTES = 15;
  protected final SynchronizationContext syncContext;
  private final ScheduledExecutorService timeService;
  private final XdsLogger logger;
  private ResolvedAddresses lastResolvedAddresses;

  ClusterManagerLoadBalancer(Helper helper) {
    super(helper);
    this.syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");
    this.timeService = checkNotNull(helper.getScheduledExecutorService(), "timeService");
    logger = XdsLogger.withLogId(
        InternalLogId.allocate("cluster_manager-lb", helper.getAuthority()));

    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  protected ChildLbState createChildLbState(Object key) {
    return new ClusterManagerLbState(key, GracefulSwitchLoadBalancerFactory.INSTANCE);
  }

  @Override
  protected Map<Object, ResolvedAddresses> createChildAddressesMap(
      ResolvedAddresses resolvedAddresses) {
    lastResolvedAddresses = resolvedAddresses;

    ClusterManagerConfig config = (ClusterManagerConfig)
        resolvedAddresses.getLoadBalancingPolicyConfig();
    logger.log(
        XdsLogLevel.INFO,
        "Received cluster_manager lb config: child names={0}", config.childPolicies.keySet());
    Map<Object, ResolvedAddresses> childAddresses = new HashMap<>();

    // Reactivate children with config; deactivate children without config
    for (ChildLbState rawState : getChildLbStates()) {
      ClusterManagerLbState state = (ClusterManagerLbState) rawState;
      if (config.childPolicies.containsKey(state.getKey())) {
        // Active child
        if (state.deletionTimer != null) {
          state.reactivateChild();
        }
      } else {
        // Inactive child
        if (state.deletionTimer == null) {
          state.deactivateChild();
        }
        if (state.deletionTimer.isPending()) {
          childAddresses.put(state.getKey(), null); // Preserve child, without config update
        }
      }
    }

    for (Map.Entry<String, Object> childPolicy : config.childPolicies.entrySet()) {
      ResolvedAddresses addresses = resolvedAddresses.toBuilder()
          .setLoadBalancingPolicyConfig(childPolicy.getValue())
          .build();
      childAddresses.put(childPolicy.getKey(), addresses);
    }
    return childAddresses;
  }

  /**
   * Using the state of all children will calculate the current connectivity state,
   * update currentConnectivityState, generate a picker and then call
   * {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
   */
  @Override
  protected void updateOverallBalancingState() {
    ConnectivityState overallState = null;
    final Map<Object, SubchannelPicker> childPickers = new HashMap<>();
    for (ChildLbState childLbState : getChildLbStates()) {
      if (((ClusterManagerLbState) childLbState).deletionTimer != null) {
        continue;
      }
      childPickers.put(childLbState.getKey(), childLbState.getCurrentPicker());
      overallState = aggregateState(overallState, childLbState.getCurrentState());
    }

    if (overallState != null) {
      getHelper().updateBalancingState(overallState, getSubchannelPicker(childPickers));
      currentConnectivityState = overallState;
    }
  }

  protected SubchannelPicker getSubchannelPicker(Map<Object, SubchannelPicker> childPickers) {
    return new SubchannelPicker() {
      @Override
      public PickResult pickSubchannel(PickSubchannelArgs args) {
        String clusterName =
            args.getCallOptions().getOption(XdsNameResolver.CLUSTER_SELECTION_KEY);
        SubchannelPicker childPicker = childPickers.get(clusterName);
        if (childPicker == null) {
          return
              PickResult.withError(
                  Status.UNAVAILABLE.withDescription("CDS encountered error: unable to find "
                      + "available subchannel for cluster " + clusterName));
        }
        PickResult childResult = childPicker.pickSubchannel(args);
        if (!childResult.hasResult() && childResult.getDelayType() != null) {
          String reason = "xds_cluster_manager: child '" + clusterName + "': "
              + (childResult.getDelayReason() != null ? childResult.getDelayReason() : "");
          return PickResult.withNoResult(childResult.getDelayType(), reason);
        }
        return childResult;
      }

      @Override
      public String toString() {
        return MoreObjects.toStringHelper(this).add("pickers", childPickers).toString();
      }
    };
  }

  @Override
  public void handleNameResolutionError(Status error) {
    logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
    boolean gotoTransientFailure = true;
    for (ChildLbState state : getChildLbStates()) {
      if (((ClusterManagerLbState) state).deletionTimer == null) {
        gotoTransientFailure = false;
        state.getLb().handleNameResolutionError(error);
      }
    }
    if (gotoTransientFailure) {
      getHelper().updateBalancingState(
          TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    }
  }

  /**
   * This differs from the base class in the use of the deletion timer.  When it is deactivated,
   * rather than immediately calling shutdown it starts a timer.  If shutdown or reactivate
   * are called before the timer fires, the timer is canceled.  Otherwise, time timer calls shutdown
   * and removes the child from the petiole policy when it is triggered.
   */
  private class ClusterManagerLbState extends ChildLbState {
    @Nullable
    ScheduledHandle deletionTimer;

    public ClusterManagerLbState(Object key, LoadBalancer.Factory policyFactory) {
      super(key, policyFactory);
    }

    @Override
    protected ChildLbStateHelper createChildHelper() {
      return new ClusterManagerChildHelper();
    }

    @Override
    protected void shutdown() {
      if (deletionTimer != null) {
        deletionTimer.cancel();
        deletionTimer = null;
      }
      super.shutdown();
    }

    void reactivateChild() {
      assert deletionTimer != null;
      deletionTimer.cancel();
      deletionTimer = null;
      logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());
    }

    void deactivateChild() {
      assert deletionTimer == null;

      class DeletionTask implements Runnable {

        @Override
        public void run() {
          acceptResolvedAddresses(lastResolvedAddresses);
        }
      }

      deletionTimer =
          syncContext.schedule(
              new DeletionTask(),
              DELAYED_CHILD_DELETION_TIME_MINUTES,
              TimeUnit.MINUTES,
              timeService);
      logger.log(XdsLogLevel.DEBUG, "Child balancer {0} deactivated", getKey());
    }

    private class ClusterManagerChildHelper extends ChildLbStateHelper {
      @Override
      public void updateBalancingState(final ConnectivityState newState,
                                       final SubchannelPicker newPicker) {
        if (getCurrentState() == ConnectivityState.SHUTDOWN) {
          return;
        }

        // Subchannel picker and state are saved, but will only be propagated to the channel
        // when the child instance exits deactivated state.
        setCurrentState(newState);
        setCurrentPicker(newPicker);
        // If we are already in the process of resolving addresses, the overall balancing state
        // will be updated at the end of it, and we don't need to trigger that update here.
        if (deletionTimer == null && !resolvingAddresses) {
          updateOverallBalancingState();
        }
      }
    }
  }

  static final class GracefulSwitchLoadBalancerFactory extends LoadBalancer.Factory {
    static final LoadBalancer.Factory INSTANCE = new GracefulSwitchLoadBalancerFactory();

    @Override
    public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
      return new GracefulSwitchLoadBalancer(helper);
    }
  }
}

1	/*
2	* Copyright 2020 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.checkNotNull;
20	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
21
22	import com.google.common.annotations.VisibleForTesting;
23	import com.google.common.base.MoreObjects;
24	import io.grpc.ConnectivityState;
25	import io.grpc.InternalLogId;
26	import io.grpc.LoadBalancer;
27	import io.grpc.Status;
28	import io.grpc.SynchronizationContext;
29	import io.grpc.SynchronizationContext.ScheduledHandle;
30	import io.grpc.util.GracefulSwitchLoadBalancer;
31	import io.grpc.util.MultiChildLoadBalancer;
32	import io.grpc.xds.ClusterManagerLoadBalancerProvider.ClusterManagerConfig;
33	import io.grpc.xds.client.XdsLogger;
34	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
35	import java.util.HashMap;
36	import java.util.Map;
37	import java.util.Map.Entry;
38	import java.util.concurrent.ScheduledExecutorService;
39	import java.util.concurrent.TimeUnit;
40	import javax.annotation.Nullable;
41
42	/**
43	* The top-level load balancing policy for use in XDS.
44	* This policy does not immediately delete its children. Instead, it marks them deactivated
45	* and starts a timer for deletion. If a subsequent address update restores the child, then it is
46	* simply reactivated instead of built from scratch. This is necessary because XDS can frequently
47	* remove and then add back a server as machines are rebooted or repurposed for load management.
48	*
49	* <p>Note that this LB does not automatically reconnect children who go into IDLE status
50	*/
51	class ClusterManagerLoadBalancer extends MultiChildLoadBalancer {
52
53	// 15 minutes is long enough for a reboot and the services to restart while not so long that
54	// many children are waiting for cleanup.
55	@VisibleForTesting
56	public static final int DELAYED_CHILD_DELETION_TIME_MINUTES = 15;
57	protected final SynchronizationContext syncContext;
58	private final ScheduledExecutorService timeService;
59	private final XdsLogger logger;
60	private ResolvedAddresses lastResolvedAddresses;
61
62	ClusterManagerLoadBalancer(Helper helper) {
63	super(helper);	1✔
64	this.syncContext = checkNotNull(helper.getSynchronizationContext(), "syncContext");	1✔
65	this.timeService = checkNotNull(helper.getScheduledExecutorService(), "timeService");	1✔
66	logger = XdsLogger.withLogId(	1✔
67	InternalLogId.allocate("cluster_manager-lb", helper.getAuthority()));	1✔
68
69	logger.log(XdsLogLevel.INFO, "Created");	1✔
70	}	1✔
71
72	@Override
73	protected ChildLbState createChildLbState(Object key) {
74	return new ClusterManagerLbState(key, GracefulSwitchLoadBalancerFactory.INSTANCE);	1✔
75	}
76
77	@Override
78	protected Map<Object, ResolvedAddresses> createChildAddressesMap(
79	ResolvedAddresses resolvedAddresses) {
80	lastResolvedAddresses = resolvedAddresses;	1✔
81
82	ClusterManagerConfig config = (ClusterManagerConfig)	1✔
83	resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
84	logger.log(	1✔
85	XdsLogLevel.INFO,
86	"Received cluster_manager lb config: child names={0}", config.childPolicies.keySet());	1✔
87	Map<Object, ResolvedAddresses> childAddresses = new HashMap<>();	1✔
88
89	// Reactivate children with config; deactivate children without config
90	for (ChildLbState rawState : getChildLbStates()) {	1✔
91	ClusterManagerLbState state = (ClusterManagerLbState) rawState;	1✔
92	if (config.childPolicies.containsKey(state.getKey())) {	1✔
93	// Active child
94	if (state.deletionTimer != null) {	1✔
95	state.reactivateChild();	1✔
96	}
97	} else {
98	// Inactive child
99	if (state.deletionTimer == null) {	1✔
100	state.deactivateChild();	1✔
101	}
102	if (state.deletionTimer.isPending()) {	1✔
103	childAddresses.put(state.getKey(), null); // Preserve child, without config update	1✔
104	}
105	}
106	}	1✔
107
108	for (Map.Entry<String, Object> childPolicy : config.childPolicies.entrySet()) {	1✔
109	ResolvedAddresses addresses = resolvedAddresses.toBuilder()	1✔
110	.setLoadBalancingPolicyConfig(childPolicy.getValue())	1✔
111	.build();	1✔
112	childAddresses.put(childPolicy.getKey(), addresses);	1✔
113	}	1✔
114	return childAddresses;	1✔
115	}
116
117	/**
118	* Using the state of all children will calculate the current connectivity state,
119	* update currentConnectivityState, generate a picker and then call
120	* {@link Helper#updateBalancingState(ConnectivityState, SubchannelPicker)}.
121	*/
122	@Override
123	protected void updateOverallBalancingState() {
124	ConnectivityState overallState = null;	1✔
125	final Map<Object, SubchannelPicker> childPickers = new HashMap<>();	1✔
126	for (ChildLbState childLbState : getChildLbStates()) {	1✔
127	if (((ClusterManagerLbState) childLbState).deletionTimer != null) {	1✔
128	continue;	1✔
129	}
130	childPickers.put(childLbState.getKey(), childLbState.getCurrentPicker());	1✔
131	overallState = aggregateState(overallState, childLbState.getCurrentState());	1✔
132	}	1✔
133
134	if (overallState != null) {	1✔
135	getHelper().updateBalancingState(overallState, getSubchannelPicker(childPickers));	1✔
136	currentConnectivityState = overallState;	1✔
137	}
138	}	1✔
139
140	protected SubchannelPicker getSubchannelPicker(Map<Object, SubchannelPicker> childPickers) {
141	return new SubchannelPicker() {	1✔
142	@Override
143	public PickResult pickSubchannel(PickSubchannelArgs args) {
144	String clusterName =	1✔
145	args.getCallOptions().getOption(XdsNameResolver.CLUSTER_SELECTION_KEY);	1✔
146	SubchannelPicker childPicker = childPickers.get(clusterName);	1✔
147	if (childPicker == null) {	1✔
148	return	1✔
149	PickResult.withError(	1✔
150	Status.UNAVAILABLE.withDescription("CDS encountered error: unable to find "	1✔
151	+ "available subchannel for cluster " + clusterName));
152	}
153	PickResult childResult = childPicker.pickSubchannel(args);	1✔
154	if (!childResult.hasResult() && childResult.getDelayType() != null) {	1✔
155	String reason = "xds_cluster_manager: child '" + clusterName + "': "	1✔
156	+ (childResult.getDelayReason() != null ? childResult.getDelayReason() : "");	1✔
157	return PickResult.withNoResult(childResult.getDelayType(), reason);	1✔
158	}
159	return childResult;	1✔
160	}
161
162	@Override
163	public String toString() {
164	return MoreObjects.toStringHelper(this).add("pickers", childPickers).toString();	×
165	}
166	};
167	}
168
169	@Override
170	public void handleNameResolutionError(Status error) {
171	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
172	boolean gotoTransientFailure = true;	1✔
173	for (ChildLbState state : getChildLbStates()) {	1✔
174	if (((ClusterManagerLbState) state).deletionTimer == null) {	1✔
175	gotoTransientFailure = false;	1✔
176	state.getLb().handleNameResolutionError(error);	1✔
177	}
178	}	1✔
179	if (gotoTransientFailure) {	1✔
180	getHelper().updateBalancingState(	1✔
181	TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
182	}
183	}	1✔
184
185	/**
186	* This differs from the base class in the use of the deletion timer. When it is deactivated,
187	* rather than immediately calling shutdown it starts a timer. If shutdown or reactivate
188	* are called before the timer fires, the timer is canceled. Otherwise, time timer calls shutdown
189	* and removes the child from the petiole policy when it is triggered.
190	*/
191	private class ClusterManagerLbState extends ChildLbState {	1✔
192	@Nullable
193	ScheduledHandle deletionTimer;
194
195	public ClusterManagerLbState(Object key, LoadBalancer.Factory policyFactory) {	1✔
196	super(key, policyFactory);	1✔
197	}	1✔
198
199	@Override
200	protected ChildLbStateHelper createChildHelper() {
201	return new ClusterManagerChildHelper();	1✔
202	}
203
204	@Override
205	protected void shutdown() {
206	if (deletionTimer != null) {	1✔
207	deletionTimer.cancel();	1✔
208	deletionTimer = null;	1✔
209	}
210	super.shutdown();	1✔
211	}	1✔
212
213	void reactivateChild() {
214	assert deletionTimer != null;	1✔
215	deletionTimer.cancel();	1✔
216	deletionTimer = null;	1✔
217	logger.log(XdsLogLevel.DEBUG, "Child balancer {0} reactivated", getKey());	1✔
218	}	1✔
219
220	void deactivateChild() {
221	assert deletionTimer == null;	1✔
222
223	class DeletionTask implements Runnable {	1✔
224
225	@Override
226	public void run() {
227	acceptResolvedAddresses(lastResolvedAddresses);	1✔
228	}	1✔
229	}
230
231	deletionTimer =	1✔
232	syncContext.schedule(	1✔
233	new DeletionTask(),
234	DELAYED_CHILD_DELETION_TIME_MINUTES,
235	TimeUnit.MINUTES,
236	timeService);	1✔
237	logger.log(XdsLogLevel.DEBUG, "Child balancer {0} deactivated", getKey());	1✔
238	}	1✔
239
240	private class ClusterManagerChildHelper extends ChildLbStateHelper {	1✔
241	@Override
242	public void updateBalancingState(final ConnectivityState newState,
243	final SubchannelPicker newPicker) {
244	if (getCurrentState() == ConnectivityState.SHUTDOWN) {	1✔
245	return;	1✔
246	}
247
248	// Subchannel picker and state are saved, but will only be propagated to the channel
249	// when the child instance exits deactivated state.
250	setCurrentState(newState);	1✔
251	setCurrentPicker(newPicker);	1✔
252	// If we are already in the process of resolving addresses, the overall balancing state
253	// will be updated at the end of it, and we don't need to trigger that update here.
254	if (deletionTimer == null && !resolvingAddresses) {	1✔
255	updateOverallBalancingState();	1✔
256	}
257	}	1✔
258	}
259	}
260
261	static final class GracefulSwitchLoadBalancerFactory extends LoadBalancer.Factory {	1✔
262	static final LoadBalancer.Factory INSTANCE = new GracefulSwitchLoadBalancerFactory();	1✔
263
264	@Override
265	public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
266	return new GracefulSwitchLoadBalancer(helper);	1✔
267	}
268	}
269	}

grpc / grpc-java / #20365

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