#20002

Committed 29 Sep 2025 04:21PM UTC coverage: 88.592% (+0.02%) from 88.575%

Build # #20002

Build Type

push

github

Committed by

web-flow

Commit Message

xds: xDS based SNI setting and SAN validation (#12378)

When using xDS credentials make SNI for the Tls handshake to be
configured via xDS, rather than use the channel authority as the SNI,
and make SAN validation to be able to use the SNI sent when so
instructed via xDS.

Implements A101.

Run Details

34877 of 39368 relevant lines covered (88.59%)

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

grpc / grpc-java / #20002

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