#20346

Committed 03 Jul 2026 10:23AM UTC coverage: 89.106% (-0.01%) from 89.118%

Build # #20346

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Move backend service label plumbing to CDS (#12882)

Addresses #12431.

This moves backend service metric label plumbing from `xds_cluster_impl`
to `cds`, matching the A75/A89 aggregate cluster behavior.

Previously, `xds_cluster_impl` added the `grpc.lb.backend_service`
pick-details label and propagated `NameResolver.ATTR_BACKEND_SERVICE` to
its child policy. That works for simple clusters, but it is the wrong
layer after A75 because aggregate clusters need the leaf CDS cluster
name, not the aggregate root cluster name.

This change makes CDS responsible for backend service plumbing:

- non-aggregate CDS adds the `grpc.lb.backend_service` pick-details
label using the CDS cluster name
- non-aggregate CDS propagates `NameResolver.ATTR_BACKEND_SERVICE` to
its child policy, so WRR can consume the backend service context
- aggregate root CDS does not add the backend service attribute or
pick-details label
- leaf CDS instances under an aggregate cluster add their own leaf
cluster name
- `xds_cluster_impl` no longer owns backend service pick-details label
or child attribute propagation

The existing `InternalEquivalentAddressGroup.ATTR_BACKEND_SERVICE` path
is left unchanged for endpoint/subchannel metrics.

Tests cover:

- non-aggregate CDS propagating `NameResolver.ATTR_BACKEND_SERVICE`
- non-aggregate CDS adding `grpc.lb.backend_service`
- aggregate root CDS not adding the root cluster name
- aggregate leaf CDS instances propagating leaf cluster names
- `xds_cluster_impl` no longer adding backend service labels or
attributes
- picker tests using a non-null `PickDetailsConsumer`, matching the real
`PickSubchannelArgs` contract

Coverage Stats

37992 of 42637 relevant lines covered (89.11%)

0.89 hits per line

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

93.87

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

/*
 * Copyright 2020 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.xds;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
import static io.grpc.xds.XdsLbPolicies.CDS_POLICY_NAME;
import static io.grpc.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;

import com.google.common.collect.ImmutableMap;
import com.google.common.primitives.UnsignedInts;
import com.google.errorprone.annotations.CheckReturnValue;
import io.grpc.Attributes;
import io.grpc.ConnectivityState;
import io.grpc.EquivalentAddressGroup;
import io.grpc.HttpConnectProxiedSocketAddress;
import io.grpc.InternalEquivalentAddressGroup;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.NameResolver;
import io.grpc.Status;
import io.grpc.StatusOr;
import io.grpc.internal.GrpcUtil;
import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.util.OutlierDetectionLoadBalancer.OutlierDetectionLoadBalancerConfig;
import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig;
import io.grpc.xds.ClusterImplLoadBalancerProvider.ClusterImplConfig;
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.XdsClusterResource.CdsUpdate;
import io.grpc.xds.XdsClusterResource.CdsUpdate.ClusterType;
import io.grpc.xds.XdsConfig.Subscription;
import io.grpc.xds.XdsConfig.XdsClusterConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.AggregateConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
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 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 cds_experimental LB policy. One instance per top-level cluster.
 * The top-level cluster may be a plain EDS/logical-DNS cluster or an aggregate cluster formed
 * by a group of sub-clusters in a tree hierarchy.
 */
final class CdsLoadBalancer2 extends LoadBalancer {
  static boolean pickFirstWeightedShuffling =
      GrpcUtil.getFlag("GRPC_EXPERIMENTAL_PF_WEIGHTED_SHUFFLING", true);

  private final XdsLogger logger;
  private final Helper helper;
  private final LoadBalancerRegistry lbRegistry;
  private final ClusterState clusterState = new ClusterState();
  private final CdsLbHelper cdsLbHelper = new CdsLbHelper();
  private GracefulSwitchLoadBalancer delegate;
  private boolean addBackendServicePickDetailsLabel;
  // Following fields are effectively final.
  private String clusterName;
  private Subscription clusterSubscription;

  CdsLoadBalancer2(Helper helper, LoadBalancerRegistry lbRegistry) {
    this.helper = checkNotNull(helper, "helper");
    this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");
    this.delegate = new GracefulSwitchLoadBalancer(cdsLbHelper);
    logger = XdsLogger.withLogId(InternalLogId.allocate("cds-lb", helper.getAuthority()));
    logger.log(XdsLogLevel.INFO, "Created");
  }

  @Override
  public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
    logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);
    if (this.clusterName == null) {
      CdsConfig config = (CdsConfig) resolvedAddresses.getLoadBalancingPolicyConfig();
      logger.log(XdsLogLevel.INFO, "Config: {0}", config);
      if (config.isDynamic) {
        clusterSubscription = resolvedAddresses.getAttributes()
            .get(XdsAttributes.XDS_CLUSTER_SUBSCRIPT_REGISTRY)
            .subscribeToCluster(config.name);
      }
      this.clusterName = config.name;
    }
    XdsConfig xdsConfig = resolvedAddresses.getAttributes().get(XdsAttributes.XDS_CONFIG);
    StatusOr<XdsClusterConfig> clusterConfigOr = xdsConfig.getClusters().get(clusterName);
    if (clusterConfigOr == null) {
      if (clusterSubscription == null) {
        // Should be impossible, because XdsDependencyManager wouldn't have generated this
        return fail(Status.INTERNAL.withDescription(
            errorPrefix() + "Unable to find non-dynamic cluster"));
      }
      // The dynamic cluster must not have loaded yet
      return Status.OK;
    }
    if (!clusterConfigOr.hasValue()) {
      return fail(clusterConfigOr.getStatus());
    }
    XdsClusterConfig clusterConfig = clusterConfigOr.getValue();

    if (clusterConfig.getChildren() instanceof EndpointConfig) {
      addBackendServicePickDetailsLabel = true;
      StatusOr<EdsUpdate> edsUpdate = getEdsUpdate(xdsConfig, clusterName);
      StatusOr<ClusterResolutionResult> statusOrResult = clusterState.edsUpdateToResult(
          clusterName,
          clusterConfig.getClusterResource(),
          clusterConfig.getClusterResource().lbPolicyConfig(),
          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: " + clusterName);
        delegate.handleNameResolutionError(status);
        return status;
      }
      Object gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
          lbRegistry.getProvider(PRIORITY_POLICY_NAME),
          new PriorityLbConfig(
              Collections.unmodifiableMap(result.priorityChildConfigs),
              Collections.unmodifiableList(result.priorities)));
      return delegate.acceptResolvedAddresses(
          resolvedAddresses.toBuilder()
            .setLoadBalancingPolicyConfig(gracefulConfig)
            .setAddresses(Collections.unmodifiableList(addresses))
            .setAttributes(resolvedAddresses.getAttributes().toBuilder()
                .set(NameResolver.ATTR_BACKEND_SERVICE, clusterName)
                .build())
            .build());
    } else if (clusterConfig.getChildren() instanceof AggregateConfig) {
      addBackendServicePickDetailsLabel = false;
      Map<String, PriorityChildConfig> priorityChildConfigs = new HashMap<>();
      List<String> leafClusters = ((AggregateConfig) clusterConfig.getChildren()).getLeafNames();
      for (String childCluster: leafClusters) {
        priorityChildConfigs.put(childCluster,
                new PriorityChildConfig(
                        GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
                                lbRegistry.getProvider(CDS_POLICY_NAME),
                                new CdsConfig(childCluster)),
                        false));
      }
      Object gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
          lbRegistry.getProvider(PRIORITY_POLICY_NAME),
          new PriorityLoadBalancerProvider.PriorityLbConfig(
              Collections.unmodifiableMap(priorityChildConfigs), leafClusters));
      return delegate.acceptResolvedAddresses(
          resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(gracefulConfig).build());
    } else {
      return fail(Status.INTERNAL.withDescription(
              errorPrefix() + "Unexpected cluster children type: "
                      + clusterConfig.getChildren().getClass()));
    }
  }

  @Override
  public void handleNameResolutionError(Status error) {
    logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);
    if (delegate != null) {
      delegate.handleNameResolutionError(error);
    } else {
      helper.updateBalancingState(
          TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    }
  }

  @Override
  public void shutdown() {
    logger.log(XdsLogLevel.INFO, "Shutdown");
    delegate.shutdown();
    delegate = new GracefulSwitchLoadBalancer(cdsLbHelper);
    addBackendServicePickDetailsLabel = false;
    if (clusterSubscription != null) {
      clusterSubscription.close();
      clusterSubscription = null;
    }
  }

  @CheckReturnValue // don't forget to return up the stack after the fail call
  private Status fail(Status error) {
    delegate.shutdown();
    addBackendServicePickDetailsLabel = false;
    helper.updateBalancingState(
        TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
    return Status.OK; // XdsNameResolver isn't a polling NR, so this value doesn't matter
  }

  private String errorPrefix() {
    return "CdsLb for " + clusterName + ": ";
  }

  private final class CdsLbHelper extends ForwardingLoadBalancerHelper {
    @Override
    public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
      if (addBackendServicePickDetailsLabel) {
        newPicker = new BackendServiceMetricLabelSubchannelPicker(newPicker, clusterName);
      }
      delegate().updateBalancingState(newState, newPicker);
    }

    @Override
    protected Helper delegate() {
      return helper;
    }
  }

  private static final class BackendServiceMetricLabelSubchannelPicker extends SubchannelPicker {
    private final SubchannelPicker delegate;
    private final String backendService;

    private BackendServiceMetricLabelSubchannelPicker(
        SubchannelPicker delegate, String backendService) {
      this.delegate = checkNotNull(delegate, "delegate");
      this.backendService = checkNotNull(backendService, "backendService");
    }

    @Override
    public PickResult pickSubchannel(PickSubchannelArgs args) {
      args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.backend_service", backendService);
      return delegate.pickSubchannel(args);
    }
  }

  /**
   * The number of bits assigned to the fractional part of fixed-point values. We normalize weights
   * to a fixed-point number between 0 and 1, representing that item's proportion of traffic (1 ==
   * 100% of traffic). We reserve at least one bit for the whole number so that we don't need to
   * special case a single item, and so that we can round up very low values without risking uint32
   * overflow of the sum of weights.
   */
  private static final int FIXED_POINT_FRACTIONAL_BITS = 31;

  /** Divide two uint32s and produce a fixed-point uint32 result. */
  private static long fractionToFixedPoint(long numerator, long denominator) {
    long one = 1L << FIXED_POINT_FRACTIONAL_BITS;
    return numerator * one / denominator;
  }

  /** Multiply two uint32 fixed-point numbers, returning a uint32 fixed-point. */
  private static long fixedPointMultiply(long a, long b) {
    return (a * b) >> FIXED_POINT_FRACTIONAL_BITS;
  }

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

  /**
   * 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()
        + "\"}";
  }

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

    StatusOr<ClusterResolutionResult> edsUpdateToResult(
        String clusterName,
        CdsUpdate discovery,
        Object lbConfig,
        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",
            clusterName, 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(clusterName, localityLbEndpoints);
      Map<String, Long> priorityLocalityWeightSums;
      if (pickFirstWeightedShuffling) {
        priorityLocalityWeightSums = new HashMap<>(sortedPriorityNames.size() * 2);
        for (Locality locality : localityLbEndpoints.keySet()) {
          LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);
          String priorityName = localityPriorityNames.get(locality);
          Long sum = priorityLocalityWeightSums.get(priorityName);
          if (sum == null) {
            sum = 0L;
          }
          long weight = UnsignedInts.toLong(localityLbInfo.localityWeight());
          priorityLocalityWeightSums.put(priorityName, sum + weight);
        }
      } else {
        priorityLocalityWeightSums = null;
      }

      for (Locality locality : localityLbEndpoints.keySet()) {
        LocalityLbEndpoints localityLbInfo = localityLbEndpoints.get(locality);
        String priorityName = localityPriorityNames.get(locality);
        String localityName = localityName(locality);
        AddressFilter.PathChain pathChain =
            AddressFilter.createPathChain(Arrays.asList(priorityName, localityName));

        boolean discard = true;
        // These sums _should_ fit in uint32, but XdsEndpointResource isn't actually verifying that
        // is true today. Since we are using long to avoid signedness trouble, the math happens to
        // still work if it turns out the sums exceed uint32.
        long localityWeightSum = 0;
        long endpointWeightSum = 0;
        if (pickFirstWeightedShuffling) {
          localityWeightSum = priorityLocalityWeightSums.get(priorityName);
          for (LbEndpoint endpoint : localityLbInfo.endpoints()) {
            if (endpoint.isHealthy()) {
              endpointWeightSum += UnsignedInts.toLong(endpoint.loadBalancingWeight());
            }
          }
        }
        for (LbEndpoint endpoint : localityLbInfo.endpoints()) {
          if (endpoint.isHealthy()) {
            discard = false;
            long weight;
            if (pickFirstWeightedShuffling) {
              // Combine locality and endpoint weights as defined by gRFC A113
              long localityWeight = fractionToFixedPoint(
                  UnsignedInts.toLong(localityLbInfo.localityWeight()), localityWeightSum);
              long endpointWeight = fractionToFixedPoint(
                  UnsignedInts.toLong(endpoint.loadBalancingWeight()), endpointWeightSum);
              weight = fixedPointMultiply(localityWeight, endpointWeight);
              if (weight == 0) {
                weight = 1;
              }
            } else {
              weight = localityLbInfo.localityWeight();
              if (endpoint.loadBalancingWeight() != 0) {
                weight *= endpoint.loadBalancingWeight();
              }
            }

            Attributes attr =
                endpoint.eag().getAttributes().toBuilder()
                    .set(InternalEquivalentAddressGroup.ATTR_BACKEND_SERVICE, clusterName)
                    .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())
                    .set(AddressFilter.PATH_CHAIN_KEY, pathChain)
                    .build();
            EquivalentAddressGroup eag;
            if (discovery.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);
            }
            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", clusterName);
      }
      sortedPriorityNames.retainAll(prioritizedLocalityWeights.keySet());
      Map<String, PriorityChildConfig> priorityChildConfigs =
          generatePriorityChildConfigs(
              clusterName, discovery, 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(
      String clusterName,
      CdsUpdate 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(
              clusterName, discovery.edsServiceName(), discovery.lrsServerInfo(),
              discovery.maxConcurrentRequests(), dropOverloads, endpointLbConfig,
              discovery.upstreamTlsContext(), 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.clusterType() == ClusterType.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();
  }
}

grpc / grpc-java / #20346

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