#19802

Committed 08 May 2025 06:04AM UTC coverage: 88.611% (-0.005%) from 88.616%

Build # #19802

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Use acceptResolvedAddresses() for WeightedTarget children (#12053)

Convert the tests to use acceptResolvedAddresses() as well.

Run Details

34780 of 39250 relevant lines covered (88.61%)

0.89 hits per line

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

98.08

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

    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	public boolean canHandleEmptyAddressListFromNameResolution() {
132	return true;	×
133	}
134
135	@Override
136	public void shutdown() {
137	logger.log(XdsLogLevel.INFO, "Shutdown");	1✔
138	for (LoadBalancer childBalancer : childBalancers.values()) {	1✔
139	childBalancer.shutdown();	1✔
140	}	1✔
141	childBalancers.clear();	1✔
142	}	1✔
143
144	private void updateOverallBalancingState() {
145	List<WeightedChildPicker> childPickers = new ArrayList<>();	1✔
146
147	ConnectivityState overallState = null;	1✔
148	List<WeightedChildPicker> errorPickers = new ArrayList<>();	1✔
149	for (String name : targets.keySet()) {	1✔
150	ChildHelper childHelper = childHelpers.get(name);	1✔
151	ConnectivityState childState = childHelper.currentState;	1✔
152	overallState = aggregateState(overallState, childState);	1✔
153	int weight = targets.get(name).weight;	1✔
154	if (READY == childState) {	1✔
155	childPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));	1✔
156	} else if (TRANSIENT_FAILURE == childState) {	1✔
157	errorPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));	1✔
158	}
159	}	1✔
160
161	SubchannelPicker picker;
162	if (childPickers.isEmpty()) {	1✔
163	if (overallState == TRANSIENT_FAILURE) {	1✔
164	picker = new WeightedRandomPicker(errorPickers);	1✔
165	} else {
166	picker = new FixedResultPicker(PickResult.withNoResult());	1✔
167	}
168	} else {
169	picker = new WeightedRandomPicker(childPickers);	1✔
170	}
171
172	if (overallState != null) {	1✔
173	helper.updateBalancingState(overallState, picker);	1✔
174	}
175	}	1✔
176
177	@Nullable
178	private static ConnectivityState aggregateState(
179	@Nullable ConnectivityState overallState, ConnectivityState childState) {
180	if (overallState == null) {	1✔
181	return childState;	1✔
182	}
183	if (overallState == READY \|\| childState == READY) {	1✔
184	return READY;	1✔
185	}
186	if (overallState == CONNECTING \|\| childState == CONNECTING) {	1✔
187	return CONNECTING;	1✔
188	}
189	if (overallState == IDLE \|\| childState == IDLE) {	1✔
190	return IDLE;	×
191	}
192	return overallState;	1✔
193	}
194
195	private final class ChildHelper extends ForwardingLoadBalancerHelper {
196	String name;
197	ConnectivityState currentState = CONNECTING;	1✔
198	SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());	1✔
199
200	private ChildHelper(String name) {	1✔
201	this.name = name;	1✔
202	}	1✔
203
204	@Override
205	public void updateBalancingState(final ConnectivityState newState,
206	final SubchannelPicker newPicker) {
207	currentState = newState;	1✔
208	currentPicker = newPicker;	1✔
209
210	// If we are already in the process of resolving addresses, the overall balancing state
211	// will be updated at the end of it, and we don't need to trigger that update here.
212	if (!resolvingAddresses && childBalancers.containsKey(name)) {	1✔
213	updateOverallBalancingState();	1✔
214	}
215	}	1✔
216
217	@Override
218	protected Helper delegate() {
219	return helper;	1✔
220	}
221	}
222	}

grpc / grpc-java / #19802

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