#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

98.11

/../xds/src/main/java/io/grpc/xds/WeightedTargetLoadBalancer.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.CONNECTING;
import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;

import com.google.common.collect.ImmutableMap;
import io.grpc.Attributes;
import io.grpc.ConnectivityState;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.Status;
import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.xds.WeightedRandomPicker.WeightedChildPicker;
import io.grpc.xds.WeightedTargetLoadBalancerProvider.WeightedPolicySelection;
import io.grpc.xds.WeightedTargetLoadBalancerProvider.WeightedTargetConfig;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;

/** Load balancer for weighted_target policy. */
final class WeightedTargetLoadBalancer extends LoadBalancer {
  public static final Attributes.Key<String> CHILD_NAME =
      Attributes.Key.create("io.grpc.xds.WeightedTargetLoadBalancer.CHILD_NAME");

  private final XdsLogger logger;
  private final Map<String, GracefulSwitchLoadBalancer> childBalancers = new HashMap<>();
  private final Map<String, ChildHelper> childHelpers = new HashMap<>();
  private final Helper helper;

  private Map<String, WeightedPolicySelection> targets = ImmutableMap.of();
  // Set to true if currently in the process of handling resolved addresses.
  private boolean resolvingAddresses;

  WeightedTargetLoadBalancer(Helper helper) {
    this.helper = checkNotNull(helper, "helper");
    logger = XdsLogger.withLogId(
        InternalLogId.allocate("weighted-target-lb", helper.getAuthority()));
    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    try {
      resolvingAddresses = true;
      return acceptResolvedAddressesInternal(resolvedAddresses);
    } finally {
      resolvingAddresses = false;
    }
  }

  public Status acceptResolvedAddressesInternal(ResolvedAddresses resolvedAddresses) {
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);
    Object lbConfig = resolvedAddresses.getLoadBalancingPolicyConfig();
    checkNotNull(lbConfig, "missing weighted_target lb config");
    WeightedTargetConfig weightedTargetConfig = (WeightedTargetConfig) lbConfig;
    Map<String, WeightedPolicySelection> newTargets = weightedTargetConfig.targets;
    for (String targetName : newTargets.keySet()) {
      if (!targets.containsKey(targetName)) {
        ChildHelper childHelper = new ChildHelper(targetName);
        GracefulSwitchLoadBalancer childBalancer = new GracefulSwitchLoadBalancer(childHelper);
        childHelpers.put(targetName, childHelper);
        childBalancers.put(targetName, childBalancer);
      }
    }
    targets = newTargets;
    Status status = Status.OK;
    for (String targetName : targets.keySet()) {
      Status newStatus = childBalancers.get(targetName).acceptResolvedAddresses(
          resolvedAddresses.toBuilder()
              .setAddresses(AddressFilter.filter(resolvedAddresses.getAddresses(), targetName))
              .setLoadBalancingPolicyConfig(targets.get(targetName).childConfig)
              .setAttributes(resolvedAddresses.getAttributes().toBuilder()
                .set(CHILD_NAME, targetName)
                .build())
              .build());
      if (!newStatus.isOk()) {
        status = newStatus;
      }
    }

    // Cleanup removed targets.
    // TODO(zdapeng): cache removed target for 15 minutes.
    for (String targetName : childBalancers.keySet()) {
      if (!targets.containsKey(targetName)) {
        childBalancers.get(targetName).shutdown();
      }
    }
    childBalancers.keySet().retainAll(targets.keySet());
    childHelpers.keySet().retainAll(targets.keySet());
    updateOverallBalancingState();
    return status;
  }

  @Override
  public void handleNameResolutionError(Status error) {
    logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
    if (childBalancers.isEmpty()) {
      helper.updateBalancingState(
          TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    }
    for (LoadBalancer childBalancer : childBalancers.values()) {
      childBalancer.handleNameResolutionError(error);
    }
  }

  @Override
  @Deprecated
  @SuppressWarnings("InlineMeSuggester")
  public boolean canHandleEmptyAddressListFromNameResolution() {
    return true;
  }

  @Override
  public void shutdown() {
    logger.log(XdsLogLevel.INFO, "Shutdown");
    for (LoadBalancer childBalancer : childBalancers.values()) {
      childBalancer.shutdown();
    }
    childBalancers.clear();
  }

  private void updateOverallBalancingState() {
    List<WeightedChildPicker> childPickers = new ArrayList<>();

    ConnectivityState overallState = null;
    List<WeightedChildPicker> errorPickers = new ArrayList<>();
    for (String name : targets.keySet()) {
      ChildHelper childHelper = childHelpers.get(name);
      ConnectivityState childState = childHelper.currentState;
      overallState = aggregateState(overallState, childState);
      int weight = targets.get(name).weight;
      if (READY == childState) {
        childPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));
      } else if (TRANSIENT_FAILURE == childState) {
        errorPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));
      }
    }

    SubchannelPicker picker;
    if (childPickers.isEmpty()) {
      if (overallState == TRANSIENT_FAILURE) {
        picker = new WeightedRandomPicker(errorPickers);
      } else {
        picker = new FixedResultPicker(
            PickResult.withNoResult("connecting", "weighted_target: connecting"));
      }
    } else {
      picker = new WeightedRandomPicker(childPickers);
    }

    if (overallState != null) {
      helper.updateBalancingState(overallState, picker);
    }
  }

  @Nullable
  private static ConnectivityState aggregateState(
      @Nullable ConnectivityState overallState, ConnectivityState childState) {
    if (overallState == null) {
      return childState;
    }
    if (overallState == READY || childState == READY) {
      return READY;
    }
    if (overallState == CONNECTING || childState == CONNECTING) {
      return CONNECTING;
    }
    if (overallState == IDLE || childState == IDLE) {
      return IDLE;
    }
    return overallState;
  }

  private final class ChildHelper extends ForwardingLoadBalancerHelper {
    String name;
    ConnectivityState currentState = CONNECTING;
    SubchannelPicker currentPicker = new FixedResultPicker(
        PickResult.withNoResult("connecting", "weighted_target: initializing"));

    private ChildHelper(String name) {
      this.name = name;
    }

    @Override
    public void updateBalancingState(final ConnectivityState newState,
        final SubchannelPicker newPicker) {
      currentState = newState;
      currentPicker = 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 (!resolvingAddresses && childBalancers.containsKey(name)) {
        updateOverallBalancingState();
      }
    }

    @Override
    protected Helper delegate() {
      return 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.CONNECTING;
21	import static io.grpc.ConnectivityState.IDLE;
22	import static io.grpc.ConnectivityState.READY;
23	import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
24
25	import com.google.common.collect.ImmutableMap;
26	import io.grpc.Attributes;
27	import io.grpc.ConnectivityState;
28	import io.grpc.InternalLogId;
29	import io.grpc.LoadBalancer;
30	import io.grpc.Status;
31	import io.grpc.util.ForwardingLoadBalancerHelper;
32	import io.grpc.util.GracefulSwitchLoadBalancer;
33	import io.grpc.xds.WeightedRandomPicker.WeightedChildPicker;
34	import io.grpc.xds.WeightedTargetLoadBalancerProvider.WeightedPolicySelection;
35	import io.grpc.xds.WeightedTargetLoadBalancerProvider.WeightedTargetConfig;
36	import io.grpc.xds.client.XdsLogger;
37	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
38	import java.util.ArrayList;
39	import java.util.HashMap;
40	import java.util.List;
41	import java.util.Map;
42	import javax.annotation.Nullable;
43
44	/** Load balancer for weighted_target policy. */
45	final class WeightedTargetLoadBalancer extends LoadBalancer {
46	public static final Attributes.Key<String> CHILD_NAME =	1✔
47	Attributes.Key.create("io.grpc.xds.WeightedTargetLoadBalancer.CHILD_NAME");	1✔
48
49	private final XdsLogger logger;
50	private final Map<String, GracefulSwitchLoadBalancer> childBalancers = new HashMap<>();	1✔
51	private final Map<String, ChildHelper> childHelpers = new HashMap<>();	1✔
52	private final Helper helper;
53
54	private Map<String, WeightedPolicySelection> targets = ImmutableMap.of();	1✔
55	// Set to true if currently in the process of handling resolved addresses.
56	private boolean resolvingAddresses;
57
58	WeightedTargetLoadBalancer(Helper helper) {	1✔
59	this.helper = checkNotNull(helper, "helper");	1✔
60	logger = XdsLogger.withLogId(	1✔
61	InternalLogId.allocate("weighted-target-lb", helper.getAuthority()));	1✔
62	logger.log(XdsLogLevel.INFO, "Created");	1✔
63	}	1✔
64
65	@Override
66	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
67	try {
68	resolvingAddresses = true;	1✔
69	return acceptResolvedAddressesInternal(resolvedAddresses);	1✔
70	} finally {
71	resolvingAddresses = false;	1✔
72	}
73	}
74
75	public Status acceptResolvedAddressesInternal(ResolvedAddresses resolvedAddresses) {
76	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
77	Object lbConfig = resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
78	checkNotNull(lbConfig, "missing weighted_target lb config");	1✔
79	WeightedTargetConfig weightedTargetConfig = (WeightedTargetConfig) lbConfig;	1✔
80	Map<String, WeightedPolicySelection> newTargets = weightedTargetConfig.targets;	1✔
81	for (String targetName : newTargets.keySet()) {	1✔
82	if (!targets.containsKey(targetName)) {	1✔
83	ChildHelper childHelper = new ChildHelper(targetName);	1✔
84	GracefulSwitchLoadBalancer childBalancer = new GracefulSwitchLoadBalancer(childHelper);	1✔
85	childHelpers.put(targetName, childHelper);	1✔
86	childBalancers.put(targetName, childBalancer);	1✔
87	}
88	}	1✔
89	targets = newTargets;	1✔
90	Status status = Status.OK;	1✔
91	for (String targetName : targets.keySet()) {	1✔
92	Status newStatus = childBalancers.get(targetName).acceptResolvedAddresses(	1✔
93	resolvedAddresses.toBuilder()	1✔
94	.setAddresses(AddressFilter.filter(resolvedAddresses.getAddresses(), targetName))	1✔
95	.setLoadBalancingPolicyConfig(targets.get(targetName).childConfig)	1✔
96	.setAttributes(resolvedAddresses.getAttributes().toBuilder()	1✔
97	.set(CHILD_NAME, targetName)	1✔
98	.build())	1✔
99	.build());	1✔
100	if (!newStatus.isOk()) {	1✔
101	status = newStatus;	1✔
102	}
103	}	1✔
104
105	// Cleanup removed targets.
106	// TODO(zdapeng): cache removed target for 15 minutes.
107	for (String targetName : childBalancers.keySet()) {	1✔
108	if (!targets.containsKey(targetName)) {	1✔
109	childBalancers.get(targetName).shutdown();	1✔
110	}
111	}	1✔
112	childBalancers.keySet().retainAll(targets.keySet());	1✔
113	childHelpers.keySet().retainAll(targets.keySet());	1✔
114	updateOverallBalancingState();	1✔
115	return status;	1✔
116	}
117
118	@Override
119	public void handleNameResolutionError(Status error) {
120	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
121	if (childBalancers.isEmpty()) {	1✔
122	helper.updateBalancingState(	1✔
123	TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
124	}
125	for (LoadBalancer childBalancer : childBalancers.values()) {	1✔
126	childBalancer.handleNameResolutionError(error);	1✔
127	}	1✔
128	}	1✔
129
130	@Override
131	@Deprecated
132	@SuppressWarnings("InlineMeSuggester")
133	public boolean canHandleEmptyAddressListFromNameResolution() {
134	return true;	×
135	}
136
137	@Override
138	public void shutdown() {
139	logger.log(XdsLogLevel.INFO, "Shutdown");	1✔
140	for (LoadBalancer childBalancer : childBalancers.values()) {	1✔
141	childBalancer.shutdown();	1✔
142	}	1✔
143	childBalancers.clear();	1✔
144	}	1✔
145
146	private void updateOverallBalancingState() {
147	List<WeightedChildPicker> childPickers = new ArrayList<>();	1✔
148
149	ConnectivityState overallState = null;	1✔
150	List<WeightedChildPicker> errorPickers = new ArrayList<>();	1✔
151	for (String name : targets.keySet()) {	1✔
152	ChildHelper childHelper = childHelpers.get(name);	1✔
153	ConnectivityState childState = childHelper.currentState;	1✔
154	overallState = aggregateState(overallState, childState);	1✔
155	int weight = targets.get(name).weight;	1✔
156	if (READY == childState) {	1✔
157	childPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));	1✔
158	} else if (TRANSIENT_FAILURE == childState) {	1✔
159	errorPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));	1✔
160	}
161	}	1✔
162
163	SubchannelPicker picker;
164	if (childPickers.isEmpty()) {	1✔
165	if (overallState == TRANSIENT_FAILURE) {	1✔
166	picker = new WeightedRandomPicker(errorPickers);	1✔
167	} else {
168	picker = new FixedResultPicker(	1✔
169	PickResult.withNoResult("connecting", "weighted_target: connecting"));	1✔
170	}
171	} else {
172	picker = new WeightedRandomPicker(childPickers);	1✔
173	}
174
175	if (overallState != null) {	1✔
176	helper.updateBalancingState(overallState, picker);	1✔
177	}
178	}	1✔
179
180	@Nullable
181	private static ConnectivityState aggregateState(
182	@Nullable ConnectivityState overallState, ConnectivityState childState) {
183	if (overallState == null) {	1✔
184	return childState;	1✔
185	}
186	if (overallState == READY \|\| childState == READY) {	1✔
187	return READY;	1✔
188	}
189	if (overallState == CONNECTING \|\| childState == CONNECTING) {	1✔
190	return CONNECTING;	1✔
191	}
192	if (overallState == IDLE \|\| childState == IDLE) {	1✔
193	return IDLE;	×
194	}
195	return overallState;	1✔
196	}
197
198	private final class ChildHelper extends ForwardingLoadBalancerHelper {
199	String name;
200	ConnectivityState currentState = CONNECTING;	1✔
201	SubchannelPicker currentPicker = new FixedResultPicker(	1✔
202	PickResult.withNoResult("connecting", "weighted_target: initializing"));	1✔
203
204	private ChildHelper(String name) {	1✔
205	this.name = name;	1✔
206	}	1✔
207
208	@Override
209	public void updateBalancingState(final ConnectivityState newState,
210	final SubchannelPicker newPicker) {
211	currentState = newState;	1✔
212	currentPicker = newPicker;	1✔
213
214	// If we are already in the process of resolving addresses, the overall balancing state
215	// will be updated at the end of it, and we don't need to trigger that update here.
216	if (!resolvingAddresses && childBalancers.containsKey(name)) {	1✔
217	updateOverallBalancingState();	1✔
218	}
219	}	1✔
220
221	@Override
222	protected Helper delegate() {
223	return helper;	1✔
224	}
225	}
226	}

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