#20015

Committed 13 Oct 2025 07:57AM UTC coverage: 88.57% (+0.02%) from 88.552%

Build # #20015

Build Type

push

github

Committed by

web-flow

Commit Message

xds: ORCA to LRS propagation changes (#12203)

Implements gRFC A85 (https://github.com/grpc/proposal/pull/454).

Run Details

34925 of 39432 relevant lines covered (88.57%)

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

grpc / grpc-java / #20015

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