#19986

Committed 18 Sep 2025 06:08PM UTC coverage: 88.539% (-0.008%) from 88.547%

Build # #19986

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Convert ClusterResolverLb to XdsDepManager

No longer need to hard-code pick_first because of gRFC A61.
https://github.com/grpc/proposal/pull/477

Run Details

34664 of 39151 relevant lines covered (88.54%)

0.89 hits per line

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

93.69

/../xds/src/main/java/io/grpc/xds/ClusterResolverLoadBalancer.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.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;

import com.google.common.collect.ImmutableMap;
import io.grpc.Attributes;
import io.grpc.EquivalentAddressGroup;
import io.grpc.HttpConnectProxiedSocketAddress;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.Status;
import io.grpc.StatusOr;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.util.OutlierDetectionLoadBalancer.OutlierDetectionLoadBalancerConfig;
import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig;
import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig.DiscoveryMechanism;
import io.grpc.xds.Endpoints.DropOverload;
import io.grpc.xds.Endpoints.LbEndpoint;
import io.grpc.xds.Endpoints.LocalityLbEndpoints;
import io.grpc.xds.EnvoyServerProtoData.FailurePercentageEjection;
import io.grpc.xds.EnvoyServerProtoData.OutlierDetection;
import io.grpc.xds.EnvoyServerProtoData.SuccessRateEjection;
import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig;
import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig;
import io.grpc.xds.XdsEndpointResource.EdsUpdate;
import io.grpc.xds.client.Locality;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

/**
 * Load balancer for cluster_resolver_experimental LB policy. This LB policy is the child LB policy
 * of the cds_experimental LB policy and the parent LB policy of the priority_experimental LB
 * policy in the xDS load balancing hierarchy. This policy converts endpoints of non-aggregate
 * clusters (e.g., EDS or Logical DNS) and groups endpoints in priorities and localities to be
 * used in the downstream LB policies for fine-grained load balancing purposes.
 */
final class ClusterResolverLoadBalancer extends LoadBalancer {
  private final XdsLogger logger;
  private final LoadBalancerRegistry lbRegistry;
  private final LoadBalancer delegate;
  private ClusterState clusterState;

  ClusterResolverLoadBalancer(Helper helper, LoadBalancerRegistry lbRegistry) {
    this.delegate = lbRegistry.getProvider(PRIORITY_POLICY_NAME).newLoadBalancer(helper);
    this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");
    logger = XdsLogger.withLogId(
        InternalLogId.allocate("cluster-resolver-lb", helper.getAuthority()));
    logger.log(XdsLogLevel.INFO, "Created");
  }

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

  @Override
  public void shutdown() {
    logger.log(XdsLogLevel.INFO, "Shutdown");
    delegate.shutdown();
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);
    ClusterResolverConfig config =
        (ClusterResolverConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
    XdsConfig xdsConfig = resolvedAddresses.getAttributes().get(XdsAttributes.XDS_CONFIG);

    DiscoveryMechanism instance = config.discoveryMechanism;
    String cluster = instance.cluster;
    if (clusterState == null) {
      clusterState = new ClusterState();
    }

    StatusOr<EdsUpdate> edsUpdate = getEdsUpdate(xdsConfig, cluster);
    StatusOr<ClusterResolutionResult> statusOrResult =
        clusterState.edsUpdateToResult(config, instance, edsUpdate);
    if (!statusOrResult.hasValue()) {
      Status status = Status.UNAVAILABLE
          .withDescription(statusOrResult.getStatus().getDescription())
          .withCause(statusOrResult.getStatus().getCause());
      delegate.handleNameResolutionError(status);
      return status;
    }
    ClusterResolutionResult result = statusOrResult.getValue();
    List<EquivalentAddressGroup> addresses = result.addresses;
    if (addresses.isEmpty()) {
      Status status = Status.UNAVAILABLE
          .withDescription("No usable endpoint from cluster: " + cluster);
      delegate.handleNameResolutionError(status);
      return status;
    }
    PriorityLbConfig childConfig =
        new PriorityLbConfig(
            Collections.unmodifiableMap(result.priorityChildConfigs),
            Collections.unmodifiableList(result.priorities));
    return delegate.acceptResolvedAddresses(
        resolvedAddresses.toBuilder()
            .setLoadBalancingPolicyConfig(childConfig)
            .setAddresses(Collections.unmodifiableList(addresses))
            .build());
  }

  private static StatusOr<EdsUpdate> getEdsUpdate(XdsConfig xdsConfig, String cluster) {
    StatusOr<XdsClusterConfig> clusterConfig = xdsConfig.getClusters().get(cluster);
    if (clusterConfig == null) {
      return StatusOr.fromStatus(Status.INTERNAL
          .withDescription("BUG: cluster resolver could not find cluster in xdsConfig"));
    }
    if (!clusterConfig.hasValue()) {
      return StatusOr.fromStatus(clusterConfig.getStatus());
    }
    if (!(clusterConfig.getValue().getChildren() instanceof XdsClusterConfig.EndpointConfig)) {
      return StatusOr.fromStatus(Status.INTERNAL
          .withDescription("BUG: cluster resolver cluster with children of unknown type"));
    }
    XdsClusterConfig.EndpointConfig endpointConfig =
        (XdsClusterConfig.EndpointConfig) clusterConfig.getValue().getChildren();
    return endpointConfig.getEndpoint();
  }

  private final class ClusterState {
    private Map<Locality, String> localityPriorityNames = Collections.emptyMap();
    int priorityNameGenId = 1;

    StatusOr<ClusterResolutionResult> edsUpdateToResult(
        ClusterResolverConfig config, DiscoveryMechanism discovery, StatusOr<EdsUpdate> updateOr) {
      if (!updateOr.hasValue()) {
        return StatusOr.fromStatus(updateOr.getStatus());
      }
      EdsUpdate update = updateOr.getValue();
      logger.log(XdsLogLevel.DEBUG, "Received endpoint update {0}", update);
      if (logger.isLoggable(XdsLogLevel.INFO)) {
        logger.log(XdsLogLevel.INFO, "Cluster {0}: {1} localities, {2} drop categories",
            discovery.cluster, update.localityLbEndpointsMap.size(),
            update.dropPolicies.size());
      }
      Map<Locality, LocalityLbEndpoints> localityLbEndpoints =
          update.localityLbEndpointsMap;
      List<DropOverload> dropOverloads = update.dropPolicies;
      List<EquivalentAddressGroup> addresses = new ArrayList<>();
      Map<String, Map<Locality, Integer>> prioritizedLocalityWeights = new HashMap<>();
      List<String> sortedPriorityNames =
          generatePriorityNames(discovery.cluster, localityLbEndpoints);
      for (Locality locality : localityLbEndpoints.keySet()) {
        LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);
        String priorityName = localityPriorityNames.get(locality);
        boolean discard = true;
        for (LbEndpoint endpoint : localityLbInfo.endpoints()) {
          if (endpoint.isHealthy()) {
            discard = false;
            long weight = localityLbInfo.localityWeight();
            if (endpoint.loadBalancingWeight() != 0) {
              weight *= endpoint.loadBalancingWeight();
            }
            String localityName = localityName(locality);
            Attributes attr =
                endpoint.eag().getAttributes().toBuilder()
                    .set(XdsAttributes.ATTR_LOCALITY, locality)
                    .set(EquivalentAddressGroup.ATTR_LOCALITY_NAME, localityName)
                    .set(XdsAttributes.ATTR_LOCALITY_WEIGHT,
                        localityLbInfo.localityWeight())
                    .set(XdsAttributes.ATTR_SERVER_WEIGHT, weight)
                    .set(XdsAttributes.ATTR_ADDRESS_NAME, endpoint.hostname())
                    .build();
            EquivalentAddressGroup eag;
            if (config.isHttp11ProxyAvailable()) {
              List<SocketAddress> rewrittenAddresses = new ArrayList<>();
              for (SocketAddress addr : endpoint.eag().getAddresses()) {
                rewrittenAddresses.add(rewriteAddress(
                    addr, endpoint.endpointMetadata(), localityLbInfo.localityMetadata()));
              }
              eag = new EquivalentAddressGroup(rewrittenAddresses, attr);
            } else {
              eag = new EquivalentAddressGroup(endpoint.eag().getAddresses(), attr);
            }
            eag = AddressFilter.setPathFilter(eag, Arrays.asList(priorityName, localityName));
            addresses.add(eag);
          }
        }
        if (discard) {
          logger.log(XdsLogLevel.INFO,
              "Discard locality {0} with 0 healthy endpoints", locality);
          continue;
        }
        if (!prioritizedLocalityWeights.containsKey(priorityName)) {
          prioritizedLocalityWeights.put(priorityName, new HashMap<Locality, Integer>());
        }
        prioritizedLocalityWeights.get(priorityName).put(
            locality, localityLbInfo.localityWeight());
      }
      if (prioritizedLocalityWeights.isEmpty()) {
        // Will still update the result, as if the cluster resource is revoked.
        logger.log(XdsLogLevel.INFO,
            "Cluster {0} has no usable priority/locality/endpoint", discovery.cluster);
      }
      sortedPriorityNames.retainAll(prioritizedLocalityWeights.keySet());
      Map<String, PriorityChildConfig> priorityChildConfigs =
          generatePriorityChildConfigs(
              discovery, config.lbConfig, lbRegistry,
              prioritizedLocalityWeights, dropOverloads);
      return StatusOr.fromValue(new ClusterResolutionResult(addresses, priorityChildConfigs,
          sortedPriorityNames));
    }

    private SocketAddress rewriteAddress(SocketAddress addr,
        ImmutableMap<String, Object> endpointMetadata,
        ImmutableMap<String, Object> localityMetadata) {
      if (!(addr instanceof InetSocketAddress)) {
        return addr;
      }

      SocketAddress proxyAddress;
      try {
        proxyAddress = (SocketAddress) endpointMetadata.get(
            "envoy.http11_proxy_transport_socket.proxy_address");
        if (proxyAddress == null) {
          proxyAddress = (SocketAddress) localityMetadata.get(
              "envoy.http11_proxy_transport_socket.proxy_address");
        }
      } catch (ClassCastException e) {
        return addr;
      }

      if (proxyAddress == null) {
        return addr;
      }

      return HttpConnectProxiedSocketAddress.newBuilder()
          .setTargetAddress((InetSocketAddress) addr)
          .setProxyAddress(proxyAddress)
          .build();
    }

    private List<String> generatePriorityNames(String name,
        Map<Locality, LocalityLbEndpoints> localityLbEndpoints) {
      TreeMap<Integer, List<Locality>> todo = new TreeMap<>();
      for (Locality locality : localityLbEndpoints.keySet()) {
        int priority = localityLbEndpoints.get(locality).priority();
        if (!todo.containsKey(priority)) {
          todo.put(priority, new ArrayList<>());
        }
        todo.get(priority).add(locality);
      }
      Map<Locality, String> newNames = new HashMap<>();
      Set<String> usedNames = new HashSet<>();
      List<String> ret = new ArrayList<>();
      for (Integer priority: todo.keySet()) {
        String foundName = "";
        for (Locality locality : todo.get(priority)) {
          if (localityPriorityNames.containsKey(locality)
              && usedNames.add(localityPriorityNames.get(locality))) {
            foundName = localityPriorityNames.get(locality);
            break;
          }
        }
        if ("".equals(foundName)) {
          foundName = priorityName(name, priorityNameGenId++);
        }
        for (Locality locality : todo.get(priority)) {
          newNames.put(locality, foundName);
        }
        ret.add(foundName);
      }
      localityPriorityNames = newNames;
      return ret;
    }
  }

  private static class ClusterResolutionResult {
    // Endpoint addresses.
    private final List<EquivalentAddressGroup> addresses;
    // Config (include load balancing policy/config) for each priority in the cluster.
    private final Map<String, PriorityChildConfig> priorityChildConfigs;
    // List of priority names ordered in descending priorities.
    private final List<String> priorities;

    ClusterResolutionResult(List<EquivalentAddressGroup> addresses,
        Map<String, PriorityChildConfig> configs, List<String> priorities) {
      this.addresses = addresses;
      this.priorityChildConfigs = configs;
      this.priorities = priorities;
    }
  }

  /**
   * Generates configs to be used in the priority LB policy for priorities in a cluster.
   *
   * <p>priority LB -> cluster_impl LB (one per priority) -> (weighted_target LB
   * -> round_robin / least_request_experimental (one per locality)) / ring_hash_experimental
   */
  private static Map<String, PriorityChildConfig> generatePriorityChildConfigs(
      DiscoveryMechanism discovery,
      Object endpointLbConfig,
      LoadBalancerRegistry lbRegistry,
      Map<String, Map<Locality, Integer>> prioritizedLocalityWeights,
      List<DropOverload> dropOverloads) {
    Map<String, PriorityChildConfig> configs = new HashMap<>();
    for (String priority : prioritizedLocalityWeights.keySet()) {
      ClusterImplConfig clusterImplConfig =
          new ClusterImplConfig(
              discovery.cluster, discovery.edsServiceName, discovery.lrsServerInfo,
              discovery.maxConcurrentRequests, dropOverloads, endpointLbConfig,
              discovery.tlsContext, discovery.filterMetadata);
      LoadBalancerProvider clusterImplLbProvider =
          lbRegistry.getProvider(XdsLbPolicies.CLUSTER_IMPL_POLICY_NAME);
      Object priorityChildPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
          clusterImplLbProvider, clusterImplConfig);

      // If outlier detection has been configured we wrap the child policy in the outlier detection
      // load balancer.
      if (discovery.outlierDetection != null) {
        LoadBalancerProvider outlierDetectionProvider = lbRegistry.getProvider(
            "outlier_detection_experimental");
        priorityChildPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
            outlierDetectionProvider,
            buildOutlierDetectionLbConfig(discovery.outlierDetection, priorityChildPolicy));
      }

      boolean isEds = discovery.type == DiscoveryMechanism.Type.EDS;
      PriorityChildConfig priorityChildConfig =
          new PriorityChildConfig(priorityChildPolicy, isEds /* ignoreReresolution */);
      configs.put(priority, priorityChildConfig);
    }
    return configs;
  }

  /**
   * Converts {@link OutlierDetection} that represents the xDS configuration to {@link
   * OutlierDetectionLoadBalancerConfig} that the {@link io.grpc.util.OutlierDetectionLoadBalancer}
   * understands.
   */
  private static OutlierDetectionLoadBalancerConfig buildOutlierDetectionLbConfig(
      OutlierDetection outlierDetection, Object childConfig) {
    OutlierDetectionLoadBalancerConfig.Builder configBuilder
        = new OutlierDetectionLoadBalancerConfig.Builder();

    configBuilder.setChildConfig(childConfig);

    if (outlierDetection.intervalNanos() != null) {
      configBuilder.setIntervalNanos(outlierDetection.intervalNanos());
    }
    if (outlierDetection.baseEjectionTimeNanos() != null) {
      configBuilder.setBaseEjectionTimeNanos(outlierDetection.baseEjectionTimeNanos());
    }
    if (outlierDetection.maxEjectionTimeNanos() != null) {
      configBuilder.setMaxEjectionTimeNanos(outlierDetection.maxEjectionTimeNanos());
    }
    if (outlierDetection.maxEjectionPercent() != null) {
      configBuilder.setMaxEjectionPercent(outlierDetection.maxEjectionPercent());
    }

    SuccessRateEjection successRate = outlierDetection.successRateEjection();
    if (successRate != null) {
      OutlierDetectionLoadBalancerConfig.SuccessRateEjection.Builder
          successRateConfigBuilder = new OutlierDetectionLoadBalancerConfig
          .SuccessRateEjection.Builder();

      if (successRate.stdevFactor() != null) {
        successRateConfigBuilder.setStdevFactor(successRate.stdevFactor());
      }
      if (successRate.enforcementPercentage() != null) {
        successRateConfigBuilder.setEnforcementPercentage(successRate.enforcementPercentage());
      }
      if (successRate.minimumHosts() != null) {
        successRateConfigBuilder.setMinimumHosts(successRate.minimumHosts());
      }
      if (successRate.requestVolume() != null) {
        successRateConfigBuilder.setRequestVolume(successRate.requestVolume());
      }

      configBuilder.setSuccessRateEjection(successRateConfigBuilder.build());
    }

    FailurePercentageEjection failurePercentage = outlierDetection.failurePercentageEjection();
    if (failurePercentage != null) {
      OutlierDetectionLoadBalancerConfig.FailurePercentageEjection.Builder
          failurePercentageConfigBuilder = new OutlierDetectionLoadBalancerConfig
          .FailurePercentageEjection.Builder();

      if (failurePercentage.threshold() != null) {
        failurePercentageConfigBuilder.setThreshold(failurePercentage.threshold());
      }
      if (failurePercentage.enforcementPercentage() != null) {
        failurePercentageConfigBuilder.setEnforcementPercentage(
            failurePercentage.enforcementPercentage());
      }
      if (failurePercentage.minimumHosts() != null) {
        failurePercentageConfigBuilder.setMinimumHosts(failurePercentage.minimumHosts());
      }
      if (failurePercentage.requestVolume() != null) {
        failurePercentageConfigBuilder.setRequestVolume(failurePercentage.requestVolume());
      }

      configBuilder.setFailurePercentageEjection(failurePercentageConfigBuilder.build());
    }

    return configBuilder.build();
  }

  /**
   * Generates a string that represents the priority in the LB policy config. The string is unique
   * across priorities in all clusters and priorityName(c, p1) < priorityName(c, p2) iff p1 < p2.
   * The ordering is undefined for priorities in different clusters.
   */
  private static String priorityName(String cluster, int priority) {
    return cluster + "[child" + priority + "]";
  }

  /**
   * Generates a string that represents the locality in the LB policy config. The string is unique
   * across all localities in all clusters.
   */
  private static String localityName(Locality locality) {
    return "{region=\"" + locality.region()
        + "\", zone=\"" + locality.zone()
        + "\", sub_zone=\"" + locality.subZone()
        + "\"}";
  }
}

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.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;
21
22	import com.google.common.collect.ImmutableMap;
23	import io.grpc.Attributes;
24	import io.grpc.EquivalentAddressGroup;
25	import io.grpc.HttpConnectProxiedSocketAddress;
26	import io.grpc.InternalLogId;
27	import io.grpc.LoadBalancer;
28	import io.grpc.LoadBalancerProvider;
29	import io.grpc.LoadBalancerRegistry;
30	import io.grpc.Status;
31	import io.grpc.StatusOr;
32	import io.grpc.util.GracefulSwitchLoadBalancer;
33	import io.grpc.util.OutlierDetectionLoadBalancer.OutlierDetectionLoadBalancerConfig;
34	import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
35	import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig;
36	import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig.DiscoveryMechanism;
37	import io.grpc.xds.Endpoints.DropOverload;
38	import io.grpc.xds.Endpoints.LbEndpoint;
39	import io.grpc.xds.Endpoints.LocalityLbEndpoints;
40	import io.grpc.xds.EnvoyServerProtoData.FailurePercentageEjection;
41	import io.grpc.xds.EnvoyServerProtoData.OutlierDetection;
42	import io.grpc.xds.EnvoyServerProtoData.SuccessRateEjection;
43	import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig;
44	import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
45	import io.grpc.xds.XdsConfig.XdsClusterConfig;
46	import io.grpc.xds.XdsEndpointResource.EdsUpdate;
47	import io.grpc.xds.client.Locality;
48	import io.grpc.xds.client.XdsLogger;
49	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
50	import java.net.InetSocketAddress;
51	import java.net.SocketAddress;
52	import java.util.ArrayList;
53	import java.util.Arrays;
54	import java.util.Collections;
55	import java.util.HashMap;
56	import java.util.HashSet;
57	import java.util.List;
58	import java.util.Map;
59	import java.util.Set;
60	import java.util.TreeMap;
61
62	/**
63	* Load balancer for cluster_resolver_experimental LB policy. This LB policy is the child LB policy
64	* of the cds_experimental LB policy and the parent LB policy of the priority_experimental LB
65	* policy in the xDS load balancing hierarchy. This policy converts endpoints of non-aggregate
66	* clusters (e.g., EDS or Logical DNS) and groups endpoints in priorities and localities to be
67	* used in the downstream LB policies for fine-grained load balancing purposes.
68	*/
69	final class ClusterResolverLoadBalancer extends LoadBalancer {
70	private final XdsLogger logger;
71	private final LoadBalancerRegistry lbRegistry;
72	private final LoadBalancer delegate;
73	private ClusterState clusterState;
74
75	ClusterResolverLoadBalancer(Helper helper, LoadBalancerRegistry lbRegistry) {	1✔
76	this.delegate = lbRegistry.getProvider(PRIORITY_POLICY_NAME).newLoadBalancer(helper);	1✔
77	this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");	1✔
78	logger = XdsLogger.withLogId(	1✔
79	InternalLogId.allocate("cluster-resolver-lb", helper.getAuthority()));	1✔
80	logger.log(XdsLogLevel.INFO, "Created");	1✔
81	}	1✔
82
83	@Override
84	public void handleNameResolutionError(Status error) {
85	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	×
86	delegate.handleNameResolutionError(error);	×
87	}	×
88
89	@Override
90	public void shutdown() {
91	logger.log(XdsLogLevel.INFO, "Shutdown");	1✔
92	delegate.shutdown();	1✔
93	}	1✔
94
95	@Override
96	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
97	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
98	ClusterResolverConfig config =	1✔
99	(ClusterResolverConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
100	XdsConfig xdsConfig = resolvedAddresses.getAttributes().get(XdsAttributes.XDS_CONFIG);	1✔
101
102	DiscoveryMechanism instance = config.discoveryMechanism;	1✔
103	String cluster = instance.cluster;	1✔
104	if (clusterState == null) {	1✔
105	clusterState = new ClusterState();	1✔
106	}
107
108	StatusOr<EdsUpdate> edsUpdate = getEdsUpdate(xdsConfig, cluster);	1✔
109	StatusOr<ClusterResolutionResult> statusOrResult =	1✔
110	clusterState.edsUpdateToResult(config, instance, edsUpdate);	1✔
111	if (!statusOrResult.hasValue()) {	1✔
112	Status status = Status.UNAVAILABLE	1✔
113	.withDescription(statusOrResult.getStatus().getDescription())	1✔
114	.withCause(statusOrResult.getStatus().getCause());	1✔
115	delegate.handleNameResolutionError(status);	1✔
116	return status;	1✔
117	}
118	ClusterResolutionResult result = statusOrResult.getValue();	1✔
119	List<EquivalentAddressGroup> addresses = result.addresses;	1✔
120	if (addresses.isEmpty()) {	1✔
121	Status status = Status.UNAVAILABLE	1✔
122	.withDescription("No usable endpoint from cluster: " + cluster);	1✔
123	delegate.handleNameResolutionError(status);	1✔
124	return status;	1✔
125	}
126	PriorityLbConfig childConfig =	1✔
127	new PriorityLbConfig(
128	Collections.unmodifiableMap(result.priorityChildConfigs),	1✔
129	Collections.unmodifiableList(result.priorities));	1✔
130	return delegate.acceptResolvedAddresses(	1✔
131	resolvedAddresses.toBuilder()	1✔
132	.setLoadBalancingPolicyConfig(childConfig)	1✔
133	.setAddresses(Collections.unmodifiableList(addresses))	1✔
134	.build());	1✔
135	}
136
137	private static StatusOr<EdsUpdate> getEdsUpdate(XdsConfig xdsConfig, String cluster) {
138	StatusOr<XdsClusterConfig> clusterConfig = xdsConfig.getClusters().get(cluster);	1✔
139	if (clusterConfig == null) {	1✔
140	return StatusOr.fromStatus(Status.INTERNAL	×
141	.withDescription("BUG: cluster resolver could not find cluster in xdsConfig"));	×
142	}
143	if (!clusterConfig.hasValue()) {	1✔
144	return StatusOr.fromStatus(clusterConfig.getStatus());	×
145	}
146	if (!(clusterConfig.getValue().getChildren() instanceof XdsClusterConfig.EndpointConfig)) {	1✔
147	return StatusOr.fromStatus(Status.INTERNAL	×
148	.withDescription("BUG: cluster resolver cluster with children of unknown type"));	×
149	}
150	XdsClusterConfig.EndpointConfig endpointConfig =	1✔
151	(XdsClusterConfig.EndpointConfig) clusterConfig.getValue().getChildren();	1✔
152	return endpointConfig.getEndpoint();	1✔
153	}
154
155	private final class ClusterState {	1✔
156	private Map<Locality, String> localityPriorityNames = Collections.emptyMap();	1✔
157	int priorityNameGenId = 1;	1✔
158
159	StatusOr<ClusterResolutionResult> edsUpdateToResult(
160	ClusterResolverConfig config, DiscoveryMechanism discovery, StatusOr<EdsUpdate> updateOr) {
161	if (!updateOr.hasValue()) {	1✔
162	return StatusOr.fromStatus(updateOr.getStatus());	1✔
163	}
164	EdsUpdate update = updateOr.getValue();	1✔
165	logger.log(XdsLogLevel.DEBUG, "Received endpoint update {0}", update);	1✔
166	if (logger.isLoggable(XdsLogLevel.INFO)) {	1✔
167	logger.log(XdsLogLevel.INFO, "Cluster {0}: {1} localities, {2} drop categories",	×
168	discovery.cluster, update.localityLbEndpointsMap.size(),	×
169	update.dropPolicies.size());	×
170	}
171	Map<Locality, LocalityLbEndpoints> localityLbEndpoints =	1✔
172	update.localityLbEndpointsMap;
173	List<DropOverload> dropOverloads = update.dropPolicies;	1✔
174	List<EquivalentAddressGroup> addresses = new ArrayList<>();	1✔
175	Map<String, Map<Locality, Integer>> prioritizedLocalityWeights = new HashMap<>();	1✔
176	List<String> sortedPriorityNames =	1✔
177	generatePriorityNames(discovery.cluster, localityLbEndpoints);	1✔
178	for (Locality locality : localityLbEndpoints.keySet()) {	1✔
179	LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);	1✔
180	String priorityName = localityPriorityNames.get(locality);	1✔
181	boolean discard = true;	1✔
182	for (LbEndpoint endpoint : localityLbInfo.endpoints()) {	1✔
183	if (endpoint.isHealthy()) {	1✔
184	discard = false;	1✔
185	long weight = localityLbInfo.localityWeight();	1✔
186	if (endpoint.loadBalancingWeight() != 0) {	1✔
187	weight *= endpoint.loadBalancingWeight();	1✔
188	}
189	String localityName = localityName(locality);	1✔
190	Attributes attr =	1✔
191	endpoint.eag().getAttributes().toBuilder()	1✔
192	.set(XdsAttributes.ATTR_LOCALITY, locality)	1✔
193	.set(EquivalentAddressGroup.ATTR_LOCALITY_NAME, localityName)	1✔
194	.set(XdsAttributes.ATTR_LOCALITY_WEIGHT,	1✔
195	localityLbInfo.localityWeight())	1✔
196	.set(XdsAttributes.ATTR_SERVER_WEIGHT, weight)	1✔
197	.set(XdsAttributes.ATTR_ADDRESS_NAME, endpoint.hostname())	1✔
198	.build();	1✔
199	EquivalentAddressGroup eag;
200	if (config.isHttp11ProxyAvailable()) {	1✔
201	List<SocketAddress> rewrittenAddresses = new ArrayList<>();	1✔
202	for (SocketAddress addr : endpoint.eag().getAddresses()) {	1✔
203	rewrittenAddresses.add(rewriteAddress(	1✔
204	addr, endpoint.endpointMetadata(), localityLbInfo.localityMetadata()));	1✔
205	}	1✔
206	eag = new EquivalentAddressGroup(rewrittenAddresses, attr);	1✔
207	} else {	1✔
208	eag = new EquivalentAddressGroup(endpoint.eag().getAddresses(), attr);	1✔
209	}
210	eag = AddressFilter.setPathFilter(eag, Arrays.asList(priorityName, localityName));	1✔
211	addresses.add(eag);	1✔
212	}
213	}	1✔
214	if (discard) {	1✔
215	logger.log(XdsLogLevel.INFO,	1✔
216	"Discard locality {0} with 0 healthy endpoints", locality);
217	continue;	1✔
218	}
219	if (!prioritizedLocalityWeights.containsKey(priorityName)) {	1✔
220	prioritizedLocalityWeights.put(priorityName, new HashMap<Locality, Integer>());	1✔
221	}
222	prioritizedLocalityWeights.get(priorityName).put(	1✔
223	locality, localityLbInfo.localityWeight());	1✔
224	}	1✔
225	if (prioritizedLocalityWeights.isEmpty()) {	1✔
226	// Will still update the result, as if the cluster resource is revoked.
227	logger.log(XdsLogLevel.INFO,	1✔
228	"Cluster {0} has no usable priority/locality/endpoint", discovery.cluster);
229	}
230	sortedPriorityNames.retainAll(prioritizedLocalityWeights.keySet());	1✔
231	Map<String, PriorityChildConfig> priorityChildConfigs =	1✔
232	generatePriorityChildConfigs(	1✔
233	discovery, config.lbConfig, lbRegistry,	1✔
234	prioritizedLocalityWeights, dropOverloads);
235	return StatusOr.fromValue(new ClusterResolutionResult(addresses, priorityChildConfigs,	1✔
236	sortedPriorityNames));
237	}
238
239	private SocketAddress rewriteAddress(SocketAddress addr,
240	ImmutableMap<String, Object> endpointMetadata,
241	ImmutableMap<String, Object> localityMetadata) {
242	if (!(addr instanceof InetSocketAddress)) {	1✔
243	return addr;	×
244	}
245
246	SocketAddress proxyAddress;
247	try {
248	proxyAddress = (SocketAddress) endpointMetadata.get(	1✔
249	"envoy.http11_proxy_transport_socket.proxy_address");
250	if (proxyAddress == null) {	1✔
251	proxyAddress = (SocketAddress) localityMetadata.get(	1✔
252	"envoy.http11_proxy_transport_socket.proxy_address");
253	}
254	} catch (ClassCastException e) {	×
255	return addr;	×
256	}	1✔
257
258	if (proxyAddress == null) {	1✔
259	return addr;	1✔
260	}
261
262	return HttpConnectProxiedSocketAddress.newBuilder()	1✔
263	.setTargetAddress((InetSocketAddress) addr)	1✔
264	.setProxyAddress(proxyAddress)	1✔
265	.build();	1✔
266	}
267
268	private List<String> generatePriorityNames(String name,
269	Map<Locality, LocalityLbEndpoints> localityLbEndpoints) {
270	TreeMap<Integer, List<Locality>> todo = new TreeMap<>();	1✔
271	for (Locality locality : localityLbEndpoints.keySet()) {	1✔
272	int priority = localityLbEndpoints.get(locality).priority();	1✔
273	if (!todo.containsKey(priority)) {	1✔
274	todo.put(priority, new ArrayList<>());	1✔
275	}
276	todo.get(priority).add(locality);	1✔
277	}	1✔
278	Map<Locality, String> newNames = new HashMap<>();	1✔
279	Set<String> usedNames = new HashSet<>();	1✔
280	List<String> ret = new ArrayList<>();	1✔
281	for (Integer priority: todo.keySet()) {	1✔
282	String foundName = "";	1✔
283	for (Locality locality : todo.get(priority)) {	1✔
284	if (localityPriorityNames.containsKey(locality)	1✔
285	&& usedNames.add(localityPriorityNames.get(locality))) {	1✔
286	foundName = localityPriorityNames.get(locality);	1✔
287	break;	1✔
288	}
289	}	1✔
290	if ("".equals(foundName)) {	1✔
291	foundName = priorityName(name, priorityNameGenId++);	1✔
292	}
293	for (Locality locality : todo.get(priority)) {	1✔
294	newNames.put(locality, foundName);	1✔
295	}	1✔
296	ret.add(foundName);	1✔
297	}	1✔
298	localityPriorityNames = newNames;	1✔
299	return ret;	1✔
300	}
301	}
302
303	private static class ClusterResolutionResult {
304	// Endpoint addresses.
305	private final List<EquivalentAddressGroup> addresses;
306	// Config (include load balancing policy/config) for each priority in the cluster.
307	private final Map<String, PriorityChildConfig> priorityChildConfigs;
308	// List of priority names ordered in descending priorities.
309	private final List<String> priorities;
310
311	ClusterResolutionResult(List<EquivalentAddressGroup> addresses,
312	Map<String, PriorityChildConfig> configs, List<String> priorities) {	1✔
313	this.addresses = addresses;	1✔
314	this.priorityChildConfigs = configs;	1✔
315	this.priorities = priorities;	1✔
316	}	1✔
317	}
318
319	/**
320	* Generates configs to be used in the priority LB policy for priorities in a cluster.
321	*
322	* <p>priority LB -> cluster_impl LB (one per priority) -> (weighted_target LB
323	* -> round_robin / least_request_experimental (one per locality)) / ring_hash_experimental
324	*/
325	private static Map<String, PriorityChildConfig> generatePriorityChildConfigs(
326	DiscoveryMechanism discovery,
327	Object endpointLbConfig,
328	LoadBalancerRegistry lbRegistry,
329	Map<String, Map<Locality, Integer>> prioritizedLocalityWeights,
330	List<DropOverload> dropOverloads) {
331	Map<String, PriorityChildConfig> configs = new HashMap<>();	1✔
332	for (String priority : prioritizedLocalityWeights.keySet()) {	1✔
333	ClusterImplConfig clusterImplConfig =	1✔
334	new ClusterImplConfig(
335	discovery.cluster, discovery.edsServiceName, discovery.lrsServerInfo,
336	discovery.maxConcurrentRequests, dropOverloads, endpointLbConfig,
337	discovery.tlsContext, discovery.filterMetadata);
338	LoadBalancerProvider clusterImplLbProvider =	1✔
339	lbRegistry.getProvider(XdsLbPolicies.CLUSTER_IMPL_POLICY_NAME);	1✔
340	Object priorityChildPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(	1✔
341	clusterImplLbProvider, clusterImplConfig);
342
343	// If outlier detection has been configured we wrap the child policy in the outlier detection
344	// load balancer.
345	if (discovery.outlierDetection != null) {	1✔
346	LoadBalancerProvider outlierDetectionProvider = lbRegistry.getProvider(	1✔
347	"outlier_detection_experimental");
348	priorityChildPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(	1✔
349	outlierDetectionProvider,
350	buildOutlierDetectionLbConfig(discovery.outlierDetection, priorityChildPolicy));	1✔
351	}
352
353	boolean isEds = discovery.type == DiscoveryMechanism.Type.EDS;	1✔
354	PriorityChildConfig priorityChildConfig =	1✔
355	new PriorityChildConfig(priorityChildPolicy, isEds /* ignoreReresolution */);
356	configs.put(priority, priorityChildConfig);	1✔
357	}	1✔
358	return configs;	1✔
359	}
360
361	/**
362	* Converts {@link OutlierDetection} that represents the xDS configuration to {@link
363	* OutlierDetectionLoadBalancerConfig} that the {@link io.grpc.util.OutlierDetectionLoadBalancer}
364	* understands.
365	*/
366	private static OutlierDetectionLoadBalancerConfig buildOutlierDetectionLbConfig(
367	OutlierDetection outlierDetection, Object childConfig) {
368	OutlierDetectionLoadBalancerConfig.Builder configBuilder	1✔
369	= new OutlierDetectionLoadBalancerConfig.Builder();
370
371	configBuilder.setChildConfig(childConfig);	1✔
372
373	if (outlierDetection.intervalNanos() != null) {	1✔
374	configBuilder.setIntervalNanos(outlierDetection.intervalNanos());	1✔
375	}
376	if (outlierDetection.baseEjectionTimeNanos() != null) {	1✔
377	configBuilder.setBaseEjectionTimeNanos(outlierDetection.baseEjectionTimeNanos());	1✔
378	}
379	if (outlierDetection.maxEjectionTimeNanos() != null) {	1✔
380	configBuilder.setMaxEjectionTimeNanos(outlierDetection.maxEjectionTimeNanos());	1✔
381	}
382	if (outlierDetection.maxEjectionPercent() != null) {	1✔
383	configBuilder.setMaxEjectionPercent(outlierDetection.maxEjectionPercent());	1✔
384	}
385
386	SuccessRateEjection successRate = outlierDetection.successRateEjection();	1✔
387	if (successRate != null) {	1✔
388	OutlierDetectionLoadBalancerConfig.SuccessRateEjection.Builder
389	successRateConfigBuilder = new OutlierDetectionLoadBalancerConfig	1✔
390	.SuccessRateEjection.Builder();
391
392	if (successRate.stdevFactor() != null) {	1✔
393	successRateConfigBuilder.setStdevFactor(successRate.stdevFactor());	1✔
394	}
395	if (successRate.enforcementPercentage() != null) {	1✔
396	successRateConfigBuilder.setEnforcementPercentage(successRate.enforcementPercentage());	1✔
397	}
398	if (successRate.minimumHosts() != null) {	1✔
399	successRateConfigBuilder.setMinimumHosts(successRate.minimumHosts());	1✔
400	}
401	if (successRate.requestVolume() != null) {	1✔
402	successRateConfigBuilder.setRequestVolume(successRate.requestVolume());	1✔
403	}
404
405	configBuilder.setSuccessRateEjection(successRateConfigBuilder.build());	1✔
406	}
407
408	FailurePercentageEjection failurePercentage = outlierDetection.failurePercentageEjection();	1✔
409	if (failurePercentage != null) {	1✔
410	OutlierDetectionLoadBalancerConfig.FailurePercentageEjection.Builder
411	failurePercentageConfigBuilder = new OutlierDetectionLoadBalancerConfig	1✔
412	.FailurePercentageEjection.Builder();
413
414	if (failurePercentage.threshold() != null) {	1✔
415	failurePercentageConfigBuilder.setThreshold(failurePercentage.threshold());	1✔
416	}
417	if (failurePercentage.enforcementPercentage() != null) {	1✔
418	failurePercentageConfigBuilder.setEnforcementPercentage(	1✔
419	failurePercentage.enforcementPercentage());	1✔
420	}
421	if (failurePercentage.minimumHosts() != null) {	1✔
422	failurePercentageConfigBuilder.setMinimumHosts(failurePercentage.minimumHosts());	1✔
423	}
424	if (failurePercentage.requestVolume() != null) {	1✔
425	failurePercentageConfigBuilder.setRequestVolume(failurePercentage.requestVolume());	1✔
426	}
427
428	configBuilder.setFailurePercentageEjection(failurePercentageConfigBuilder.build());	1✔
429	}
430
431	return configBuilder.build();	1✔
432	}
433
434	/**
435	* Generates a string that represents the priority in the LB policy config. The string is unique
436	* across priorities in all clusters and priorityName(c, p1) < priorityName(c, p2) iff p1 < p2.
437	* The ordering is undefined for priorities in different clusters.
438	*/
439	private static String priorityName(String cluster, int priority) {
440	return cluster + "[child" + priority + "]";	1✔
441	}
442
443	/**
444	* Generates a string that represents the locality in the LB policy config. The string is unique
445	* across all localities in all clusters.
446	*/
447	private static String localityName(Locality locality) {
448	return "{region=\"" + locality.region()	1✔
449	+ "\", zone=\"" + locality.zone()	1✔
450	+ "\", sub_zone=\"" + locality.subZone()	1✔
451	+ "\"}";
452	}
453	}

grpc / grpc-java / #19986

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