#19191

Committed 03 May 2024 07:42PM UTC coverage: 88.323% (+0.02%) from 88.303%

Build # #19191

Build Type

push

github

Committed by

ejona86

Commit Message

xds: Plumb locality in xds_cluster_impl and weighted_target

As part of gRFC A78:

> To support the locality label in the WRR metrics, we will extend the
> `weighted_target` LB policy (see A28) to define a resolver attribute
> that indicates the name of its child. This attribute will be passed
> down to each of its children with the appropriate value, so that any
> LB policy that sits underneath the `weighted_target` policy will be
> able to use it.

xds_cluster_impl is involved because it uses the child names in the
AddressFilter, which must match the names used by weighted_target.
Instead of using Locality.toString() in multiple policies and assuming
the policies agree, we now have xds_cluster_impl decide the locality's
name and pass it down explicitly. This allows us to change the name
format to match gRFC A78:

> If locality information is available, the value of this label will be
> of the form `{region="${REGION}", zone="${ZONE}",
> sub_zone="${SUB_ZONE}"}`, where `${REGION}`, `${ZONE}`, and
> `${SUB_ZONE}` are replaced with the actual values. If no locality
> information is available, the label will be set to the empty string.

Run Details

31520 of 35687 relevant lines covered (88.32%)

0.88 hits per line

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

88.52

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

/*
 * Copyright 2022 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 static io.grpc.xds.XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME;

import com.google.common.base.MoreObjects;
import io.grpc.Attributes;
import io.grpc.EquivalentAddressGroup;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancerRegistry;
import io.grpc.Status;
import io.grpc.internal.ServiceConfigUtil.PolicySelection;
import io.grpc.util.GracefulSwitchLoadBalancer;
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.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * This load balancer acts as a parent for the {@link WeightedTargetLoadBalancer} and configures
 * it with a child policy in its configuration and locality weights it gets from an attribute in
 * {@link io.grpc.LoadBalancer.ResolvedAddresses}.
 */
final class WrrLocalityLoadBalancer extends LoadBalancer {

  private final XdsLogger logger;
  private final Helper helper;
  private final GracefulSwitchLoadBalancer switchLb;
  private final LoadBalancerRegistry lbRegistry;

  WrrLocalityLoadBalancer(Helper helper) {
    this(helper, LoadBalancerRegistry.getDefaultRegistry());
  }

  WrrLocalityLoadBalancer(Helper helper, LoadBalancerRegistry lbRegistry) {
    this.helper = checkNotNull(helper, "helper");
    this.lbRegistry = lbRegistry;
    switchLb = new GracefulSwitchLoadBalancer(helper);
    logger = XdsLogger.withLogId(
        InternalLogId.allocate("xds-wrr-locality-lb", helper.getAuthority()));
    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);

    // The configuration with the child policy is combined with the locality weights
    // to produce the weighted target LB config.
    WrrLocalityConfig wrrLocalityConfig
        = (WrrLocalityConfig) resolvedAddresses.getLoadBalancingPolicyConfig();

    // A map of locality weights is built up from the locality weight attributes in each address.
    Map<String, Integer> localityWeights = new HashMap<>();
    for (EquivalentAddressGroup eag : resolvedAddresses.getAddresses()) {
      Attributes eagAttrs = eag.getAttributes();
      String locality = eagAttrs.get(InternalXdsAttributes.ATTR_LOCALITY_NAME);
      Integer localityWeight = eagAttrs.get(InternalXdsAttributes.ATTR_LOCALITY_WEIGHT);

      if (locality == null) {
        Status unavailableStatus = Status.UNAVAILABLE.withDescription(
            "wrr_locality error: no locality provided");
        helper.updateBalancingState(TRANSIENT_FAILURE,
            new FixedResultPicker(PickResult.withError(unavailableStatus)));
        return unavailableStatus;
      }
      if (localityWeight == null) {
        Status unavailableStatus = Status.UNAVAILABLE.withDescription(
                "wrr_locality error: no weight provided for locality " + locality);
        helper.updateBalancingState(TRANSIENT_FAILURE,
            new FixedResultPicker(PickResult.withError(unavailableStatus)));
        return unavailableStatus;
      }

      if (!localityWeights.containsKey(locality)) {
        localityWeights.put(locality, localityWeight);
      } else if (!localityWeights.get(locality).equals(localityWeight)) {
        logger.log(XdsLogLevel.WARNING,
            "Locality {0} has both weights {1} and {2}, using weight {1}", locality,
            localityWeights.get(locality), localityWeight);
      }
    }

    // Weighted target LB expects a WeightedPolicySelection for each locality as it will create a
    // child LB for each.
    Map<String, WeightedPolicySelection> weightedPolicySelections = new HashMap<>();
    for (String locality : localityWeights.keySet()) {
      weightedPolicySelections.put(locality,
          new WeightedPolicySelection(localityWeights.get(locality),
              wrrLocalityConfig.childPolicy));
    }

    switchLb.switchTo(lbRegistry.getProvider(WEIGHTED_TARGET_POLICY_NAME));
    switchLb.handleResolvedAddresses(
        resolvedAddresses.toBuilder()
            .setLoadBalancingPolicyConfig(new WeightedTargetConfig(weightedPolicySelections))
            .build());

    return Status.OK;
  }

  @Override
  public void handleNameResolutionError(Status error) {
    logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
    switchLb.handleNameResolutionError(error);
  }

  @Override
  public void shutdown() {
    switchLb.shutdown();
  }

  /**
   * The LB config for {@link WrrLocalityLoadBalancer}.
   */
  static final class WrrLocalityConfig {

    final PolicySelection childPolicy;

    WrrLocalityConfig(PolicySelection childPolicy) {
      this.childPolicy = childPolicy;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }
      WrrLocalityConfig that = (WrrLocalityConfig) o;
      return Objects.equals(childPolicy, that.childPolicy);
    }

    @Override
    public int hashCode() {
      return Objects.hashCode(childPolicy);
    }

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

1	/*
2	* Copyright 2022 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	import static io.grpc.xds.XdsLbPolicies.WEIGHTED_TARGET_POLICY_NAME;
22
23	import com.google.common.base.MoreObjects;
24	import io.grpc.Attributes;
25	import io.grpc.EquivalentAddressGroup;
26	import io.grpc.InternalLogId;
27	import io.grpc.LoadBalancer;
28	import io.grpc.LoadBalancerRegistry;
29	import io.grpc.Status;
30	import io.grpc.internal.ServiceConfigUtil.PolicySelection;
31	import io.grpc.util.GracefulSwitchLoadBalancer;
32	import io.grpc.xds.WeightedTargetLoadBalancerProvider.WeightedPolicySelection;
33	import io.grpc.xds.WeightedTargetLoadBalancerProvider.WeightedTargetConfig;
34	import io.grpc.xds.client.XdsLogger;
35	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
36	import java.util.HashMap;
37	import java.util.Map;
38	import java.util.Objects;
39
40	/**
41	* This load balancer acts as a parent for the {@link WeightedTargetLoadBalancer} and configures
42	* it with a child policy in its configuration and locality weights it gets from an attribute in
43	* {@link io.grpc.LoadBalancer.ResolvedAddresses}.
44	*/
45	final class WrrLocalityLoadBalancer extends LoadBalancer {
46
47	private final XdsLogger logger;
48	private final Helper helper;
49	private final GracefulSwitchLoadBalancer switchLb;
50	private final LoadBalancerRegistry lbRegistry;
51
52	WrrLocalityLoadBalancer(Helper helper) {
53	this(helper, LoadBalancerRegistry.getDefaultRegistry());	1✔
54	}	1✔
55
56	WrrLocalityLoadBalancer(Helper helper, LoadBalancerRegistry lbRegistry) {	1✔
57	this.helper = checkNotNull(helper, "helper");	1✔
58	this.lbRegistry = lbRegistry;	1✔
59	switchLb = new GracefulSwitchLoadBalancer(helper);	1✔
60	logger = XdsLogger.withLogId(	1✔
61	InternalLogId.allocate("xds-wrr-locality-lb", helper.getAuthority()));	1✔
62	logger.log(XdsLogLevel.INFO, "Created");	1✔
63	}	1✔
64
65	@Override
66	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
67	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
68
69	// The configuration with the child policy is combined with the locality weights
70	// to produce the weighted target LB config.
71	WrrLocalityConfig wrrLocalityConfig	1✔
72	= (WrrLocalityConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
73
74	// A map of locality weights is built up from the locality weight attributes in each address.
75	Map<String, Integer> localityWeights = new HashMap<>();	1✔
76	for (EquivalentAddressGroup eag : resolvedAddresses.getAddresses()) {	1✔
77	Attributes eagAttrs = eag.getAttributes();	1✔
78	String locality = eagAttrs.get(InternalXdsAttributes.ATTR_LOCALITY_NAME);	1✔
79	Integer localityWeight = eagAttrs.get(InternalXdsAttributes.ATTR_LOCALITY_WEIGHT);	1✔
80
81	if (locality == null) {	1✔
82	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	×
83	"wrr_locality error: no locality provided");
84	helper.updateBalancingState(TRANSIENT_FAILURE,	×
85	new FixedResultPicker(PickResult.withError(unavailableStatus)));	×
86	return unavailableStatus;	×
87	}
88	if (localityWeight == null) {	1✔
89	Status unavailableStatus = Status.UNAVAILABLE.withDescription(	1✔
90	"wrr_locality error: no weight provided for locality " + locality);
91	helper.updateBalancingState(TRANSIENT_FAILURE,	1✔
92	new FixedResultPicker(PickResult.withError(unavailableStatus)));	1✔
93	return unavailableStatus;	1✔
94	}
95
96	if (!localityWeights.containsKey(locality)) {	1✔
97	localityWeights.put(locality, localityWeight);	1✔
98	} else if (!localityWeights.get(locality).equals(localityWeight)) {	×
99	logger.log(XdsLogLevel.WARNING,	×
100	"Locality {0} has both weights {1} and {2}, using weight {1}", locality,
101	localityWeights.get(locality), localityWeight);	×
102	}
103	}	1✔
104
105	// Weighted target LB expects a WeightedPolicySelection for each locality as it will create a
106	// child LB for each.
107	Map<String, WeightedPolicySelection> weightedPolicySelections = new HashMap<>();	1✔
108	for (String locality : localityWeights.keySet()) {	1✔
109	weightedPolicySelections.put(locality,	1✔
110	new WeightedPolicySelection(localityWeights.get(locality),	1✔
111	wrrLocalityConfig.childPolicy));
112	}	1✔
113
114	switchLb.switchTo(lbRegistry.getProvider(WEIGHTED_TARGET_POLICY_NAME));	1✔
115	switchLb.handleResolvedAddresses(	1✔
116	resolvedAddresses.toBuilder()	1✔
117	.setLoadBalancingPolicyConfig(new WeightedTargetConfig(weightedPolicySelections))	1✔
118	.build());	1✔
119
120	return Status.OK;	1✔
121	}
122
123	@Override
124	public void handleNameResolutionError(Status error) {
125	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
126	switchLb.handleNameResolutionError(error);	1✔
127	}	1✔
128
129	@Override
130	public void shutdown() {
131	switchLb.shutdown();	1✔
132	}	1✔
133
134	/**
135	* The LB config for {@link WrrLocalityLoadBalancer}.
136	*/
137	static final class WrrLocalityConfig {
138
139	final PolicySelection childPolicy;
140
141	WrrLocalityConfig(PolicySelection childPolicy) {	1✔
142	this.childPolicy = childPolicy;	1✔
143	}	1✔
144
145	@Override
146	public boolean equals(Object o) {
147	if (this == o) {	1✔
148	return true;	1✔
149	}
150	if (o == null \|\| getClass() != o.getClass()) {	1✔
151	return false;	1✔
152	}
153	WrrLocalityConfig that = (WrrLocalityConfig) o;	1✔
154	return Objects.equals(childPolicy, that.childPolicy);	1✔
155	}
156
157	@Override
158	public int hashCode() {
159	return Objects.hashCode(childPolicy);	1✔
160	}
161
162	@Override
163	public String toString() {
164	return MoreObjects.toStringHelper(this).add("childPolicy", childPolicy).toString();	1✔
165	}
166	}
167	}

grpc / grpc-java / #19191

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