#20025

Committed 21 Oct 2025 08:04PM UTC coverage: 88.588% (+0.01%) from 88.575%

Build # #20025

Build Type

push

github

Committed by

ejona86

Commit Message

xds: Remove commented out+broken import from cluster resolver

It was added by mistake in 032d2928e.

Run Details

34948 of 39450 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, 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.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, discovery.backendMetricPropagation);
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 / #20025

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