#20233

Committed 02 Apr 2026 12:18PM UTC coverage: 88.792% (+0.008%) from 88.784%

Build # #20233

Build Type

push

github

Committed by

web-flow

Commit Message

[xds] Implement A114: WRR support for custom backend metrics (#12645)

### Description
This PR implements [gRFC A114: WRR Support for Custom Backend
Metrics](https://github.com/grpc/proposal/pull/536).

It updates the `weighted_round_robin` policy to allow users to configure
which backend metrics drive the load balancing weights.

### Key Changes
* **Configuration**: Supports the new
`metric_names_for_computing_utilization` field in
`WeightedRoundRobinLbConfig`.
* **Weight Calculation**: Implements logic to resolve custom metrics
(including map lookups like `named_metrics.foo`) when
`application_utilization` is absent.
* **Refactor**: Centralizes the complex metric lookup and validation
logic (checking for NaN, <= 0, etc.) into a new internal utility
`MetricReportUtils`.
* **Testing**: Verifies correct precedence: `application_utilization` >
`custom_metrics` (max valid value) > `cpu_utilization`.

Coverage Stats

36007 of 40552 relevant lines covered (88.79%)

0.89 hits per line

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

91.24

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

/*
 * Copyright 2022 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 com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Struct;
import com.google.protobuf.util.Durations;
import com.google.protobuf.util.JsonFormat;
import io.envoyproxy.envoy.config.cluster.v3.Cluster;
import io.envoyproxy.envoy.config.cluster.v3.Cluster.LeastRequestLbConfig;
import io.envoyproxy.envoy.config.cluster.v3.Cluster.RingHashLbConfig;
import io.envoyproxy.envoy.config.cluster.v3.LoadBalancingPolicy;
import io.envoyproxy.envoy.config.cluster.v3.LoadBalancingPolicy.Policy;
import io.envoyproxy.envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin;
import io.envoyproxy.envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest;
import io.envoyproxy.envoy.extensions.load_balancing_policies.pick_first.v3.PickFirst;
import io.envoyproxy.envoy.extensions.load_balancing_policies.ring_hash.v3.RingHash;
import io.envoyproxy.envoy.extensions.load_balancing_policies.round_robin.v3.RoundRobin;
import io.envoyproxy.envoy.extensions.load_balancing_policies.wrr_locality.v3.WrrLocality;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancerRegistry;
import io.grpc.internal.JsonParser;
import io.grpc.xds.LoadBalancerConfigFactory.LoadBalancingPolicyConverter.MaxRecursionReachedException;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
import java.io.IOException;
import java.util.Map;

/**
 * Creates service config JSON  load balancer config objects for a given xDS Cluster message.
 * Supports both the "legacy" configuration style and the new, more advanced one that utilizes the
 * xDS "typed extension" mechanism.
 *
 * <p>Legacy configuration is done by setting the lb_policy enum field and any supporting
 * configuration fields needed by the particular policy.
 *
 * <p>The new approach is to set the load_balancing_policy field that contains both the policy
 * selection as well as any supporting configuration data. Providing a list of acceptable policies
 * is also supported. Note that if this field is used, it will override any configuration set using
 * the legacy approach. The new configuration approach is explained in detail in the <a href="
 * https://github.com/grpc/proposal/blob/master/A52-xds-custom-lb-policies.md">Custom LB Policies
 * gRFC</a>
 */
class LoadBalancerConfigFactory {

  private static final XdsLogger logger = XdsLogger.withLogId(
      InternalLogId.allocate("xds-client-lbconfig-factory", null));

  static final String ROUND_ROBIN_FIELD_NAME = "round_robin";

  static final String RING_HASH_FIELD_NAME = "ring_hash_experimental";
  static final String MIN_RING_SIZE_FIELD_NAME = "minRingSize";
  static final String MAX_RING_SIZE_FIELD_NAME = "maxRingSize";

  static final String LEAST_REQUEST_FIELD_NAME = "least_request_experimental";
  static final String CHOICE_COUNT_FIELD_NAME = "choiceCount";

  static final String WRR_LOCALITY_FIELD_NAME = "wrr_locality_experimental";
  static final String CHILD_POLICY_FIELD = "childPolicy";

  static final String BLACK_OUT_PERIOD = "blackoutPeriod";

  static final String WEIGHT_EXPIRATION_PERIOD = "weightExpirationPeriod";

  static final String OOB_REPORTING_PERIOD = "oobReportingPeriod";

  static final String ENABLE_OOB_LOAD_REPORT = "enableOobLoadReport";

  static final String WEIGHT_UPDATE_PERIOD = "weightUpdatePeriod";

  static final String PICK_FIRST_FIELD_NAME = "pick_first";
  static final String SHUFFLE_ADDRESS_LIST_FIELD_NAME = "shuffleAddressList";

  static final String ERROR_UTILIZATION_PENALTY = "errorUtilizationPenalty";
  static final String METRIC_NAMES_FOR_COMPUTING_UTILIZATION = "metricNamesForComputingUtilization";

  /**
   * Factory method for creating a new {link LoadBalancerConfigConverter} for a given xDS {@link
   * Cluster}.
   *
   * @throws ResourceInvalidException If the {@link Cluster} has an invalid LB configuration.
   */
  static ImmutableMap<String, ?> newConfig(Cluster cluster, boolean enableLeastRequest)
      throws ResourceInvalidException {
    // The new load_balancing_policy will always be used if it is set, but for backward
    // compatibility we will fall back to using the old lb_policy field if the new field is not set.
    if (cluster.hasLoadBalancingPolicy()) {
      try {
        return LoadBalancingPolicyConverter.convertToServiceConfig(cluster.getLoadBalancingPolicy(),
            0);
      } catch (MaxRecursionReachedException e) {
        throw new ResourceInvalidException("Maximum LB config recursion depth reached", e);
      }
    } else {
      return LegacyLoadBalancingPolicyConverter.convertToServiceConfig(cluster, enableLeastRequest);
    }
  }

  /**
   * Builds a service config JSON object for the ring_hash load balancer config based on the given
   * config values.
   */
  private static ImmutableMap<String, ?> buildRingHashConfig(Long minRingSize, Long maxRingSize) {
    ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();
    if (minRingSize != null) {
      configBuilder.put(MIN_RING_SIZE_FIELD_NAME, minRingSize.doubleValue());
    }
    if (maxRingSize != null) {
      configBuilder.put(MAX_RING_SIZE_FIELD_NAME, maxRingSize.doubleValue());
    }
    return ImmutableMap.of(RING_HASH_FIELD_NAME, configBuilder.buildOrThrow());
  }

  /**
   * Builds a service config JSON object for the weighted_round_robin load balancer config based on
   * the given config values.
   */
  private static ImmutableMap<String, ?> buildWrrConfig(String blackoutPeriod,
      String weightExpirationPeriod, String oobReportingPeriod, Boolean enableOobLoadReport,
      String weightUpdatePeriod, Float errorUtilizationPenalty,
      ImmutableList<String> metricNamesForComputingUtilization) {
    ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();
    if (blackoutPeriod != null) {
      configBuilder.put(BLACK_OUT_PERIOD, blackoutPeriod);
    }
    if (weightExpirationPeriod != null) {
      configBuilder.put(WEIGHT_EXPIRATION_PERIOD, weightExpirationPeriod);
    }
    if (oobReportingPeriod != null) {
      configBuilder.put(OOB_REPORTING_PERIOD, oobReportingPeriod);
    }
    if (enableOobLoadReport != null) {
      configBuilder.put(ENABLE_OOB_LOAD_REPORT, enableOobLoadReport);
    }
    if (weightUpdatePeriod != null) {
      configBuilder.put(WEIGHT_UPDATE_PERIOD, weightUpdatePeriod);
    }
    if (errorUtilizationPenalty != null) {
      configBuilder.put(ERROR_UTILIZATION_PENALTY, errorUtilizationPenalty);
    }
    if (metricNamesForComputingUtilization != null
        && !metricNamesForComputingUtilization.isEmpty()) {
      configBuilder.put(METRIC_NAMES_FOR_COMPUTING_UTILIZATION, metricNamesForComputingUtilization);
    }
    return ImmutableMap.of(WeightedRoundRobinLoadBalancerProvider.SCHEME,
        configBuilder.buildOrThrow());
  }

  /**
   * Builds a service config JSON object for the least_request load balancer config based on the
   * given config values.
   */
  private static ImmutableMap<String, ?> buildLeastRequestConfig(Integer choiceCount) {
    ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();
    if (choiceCount != null) {
      configBuilder.put(CHOICE_COUNT_FIELD_NAME, choiceCount.doubleValue());
    }
    return ImmutableMap.of(LEAST_REQUEST_FIELD_NAME, configBuilder.buildOrThrow());
  }

  /**
   * Builds a service config JSON wrr_locality by wrapping another policy config.
   */
  private static ImmutableMap<String, ?> buildWrrLocalityConfig(
      ImmutableMap<String, ?> childConfig) {
    return ImmutableMap.<String, Object>builder().put(WRR_LOCALITY_FIELD_NAME,
        ImmutableMap.of(CHILD_POLICY_FIELD, ImmutableList.of(childConfig))).buildOrThrow();
  }

  /**
   * Builds an empty service config JSON config object for round robin (it is not configurable).
   */
  private static ImmutableMap<String, ?> buildRoundRobinConfig() {
    return ImmutableMap.of(ROUND_ROBIN_FIELD_NAME, ImmutableMap.of());
  }

  /**
   * Builds a service config JSON object for the pick_first load balancer config based on the
   * given config values.
   */
  private static ImmutableMap<String, ?> buildPickFirstConfig(boolean shuffleAddressList) {
    ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();
    configBuilder.put(SHUFFLE_ADDRESS_LIST_FIELD_NAME, shuffleAddressList);
    return ImmutableMap.of(PICK_FIRST_FIELD_NAME, configBuilder.buildOrThrow());
  }

  /**
   * Responsible for converting from a {@code envoy.config.cluster.v3.LoadBalancingPolicy} proto
   * message to a gRPC service config format.
   */
  static class LoadBalancingPolicyConverter {

    private static final int MAX_RECURSION = 16;

    /**
     * Converts a {@link LoadBalancingPolicy} object to a service config JSON object.
     */
    private static ImmutableMap<String, ?> convertToServiceConfig(
        LoadBalancingPolicy loadBalancingPolicy, int recursionDepth)
        throws ResourceInvalidException, MaxRecursionReachedException {
      if (recursionDepth > MAX_RECURSION) {
        throw new MaxRecursionReachedException();
      }
      ImmutableMap<String, ?> serviceConfig = null;

      for (Policy policy : loadBalancingPolicy.getPoliciesList()) {
        Any typedConfig = policy.getTypedExtensionConfig().getTypedConfig();
        try {
          if (typedConfig.is(RingHash.class)) {
            serviceConfig = convertRingHashConfig(typedConfig.unpack(RingHash.class));
          } else if (typedConfig.is(WrrLocality.class)) {
            serviceConfig = convertWrrLocalityConfig(typedConfig.unpack(WrrLocality.class),
                recursionDepth);
          } else if (typedConfig.is(RoundRobin.class)) {
            serviceConfig = convertRoundRobinConfig();
          } else if (typedConfig.is(LeastRequest.class)) {
            serviceConfig = convertLeastRequestConfig(typedConfig.unpack(LeastRequest.class));
          } else if (typedConfig.is(ClientSideWeightedRoundRobin.class)) {
            serviceConfig = convertWeightedRoundRobinConfig(
                typedConfig.unpack(ClientSideWeightedRoundRobin.class));
          } else if (typedConfig.is(PickFirst.class)) {
            serviceConfig = convertPickFirstConfig(typedConfig.unpack(PickFirst.class));
          } else if (typedConfig.is(com.github.xds.type.v3.TypedStruct.class)) {
            serviceConfig = convertCustomConfig(
                typedConfig.unpack(com.github.xds.type.v3.TypedStruct.class));
          } else if (typedConfig.is(com.github.udpa.udpa.type.v1.TypedStruct.class)) {
            serviceConfig = convertCustomConfig(
                typedConfig.unpack(com.github.udpa.udpa.type.v1.TypedStruct.class));
          }

          // TODO: support least_request once it is added to the envoy protos.
        } catch (InvalidProtocolBufferException e) {
          throw new ResourceInvalidException(
              "Unable to unpack typedConfig for: " + typedConfig.getTypeUrl(), e);
        }
        // The service config is expected to have a single root entry, where the name of that entry
        // is the name of the policy. A Load balancer with this name must exist in the registry.
        if (serviceConfig == null || LoadBalancerRegistry.getDefaultRegistry()
            .getProvider(Iterables.getOnlyElement(serviceConfig.keySet())) == null) {
          logger.log(XdsLogLevel.WARNING, "Policy {0} not found in the LB registry, skipping",
              typedConfig.getTypeUrl());
          continue;
        } else {
          return serviceConfig;
        }
      }

      // If we could not find a Policy that we could both convert as well as find a provider for
      // then we have an invalid LB policy configuration.
      throw new ResourceInvalidException("Invalid LoadBalancingPolicy: " + loadBalancingPolicy);
    }

    /**
     * Converts a ring_hash {@link Any} configuration to service config format.
     */
    private static ImmutableMap<String, ?> convertRingHashConfig(RingHash ringHash)
        throws ResourceInvalidException {
      // The hash function needs to be validated here as it is not exposed in the returned
      // configuration for later validation.
      if (RingHash.HashFunction.XX_HASH != ringHash.getHashFunction()) {
        throw new ResourceInvalidException(
            "Invalid ring hash function: " + ringHash.getHashFunction());
      }

      return buildRingHashConfig(
          ringHash.hasMinimumRingSize() ? ringHash.getMinimumRingSize().getValue() : null,
          ringHash.hasMaximumRingSize() ? ringHash.getMaximumRingSize().getValue() : null);
    }

    private static ImmutableMap<String, ?> convertWeightedRoundRobinConfig(
        ClientSideWeightedRoundRobin wrr) throws ResourceInvalidException {
      try {
        return buildWrrConfig(
            wrr.hasBlackoutPeriod() ? Durations.toString(wrr.getBlackoutPeriod()) : null,
            wrr.hasWeightExpirationPeriod()
                ? Durations.toString(wrr.getWeightExpirationPeriod()) : null,
            wrr.hasOobReportingPeriod() ? Durations.toString(wrr.getOobReportingPeriod()) : null,
            wrr.hasEnableOobLoadReport() ? wrr.getEnableOobLoadReport().getValue() : null,
            wrr.hasWeightUpdatePeriod() ? Durations.toString(wrr.getWeightUpdatePeriod()) : null,
            wrr.hasErrorUtilizationPenalty() ? wrr.getErrorUtilizationPenalty().getValue() : null,
            ImmutableList.copyOf(wrr.getMetricNamesForComputingUtilizationList()));
      } catch (IllegalArgumentException ex) {
        throw new ResourceInvalidException("Invalid duration in weighted round robin config: "
            + ex.getMessage());
      }
    }

    /**
     * Converts a wrr_locality {@link Any} configuration to service config format.
     */
    private static ImmutableMap<String, ?> convertWrrLocalityConfig(WrrLocality wrrLocality,
        int recursionDepth)
        throws ResourceInvalidException, MaxRecursionReachedException {
      return buildWrrLocalityConfig(
          convertToServiceConfig(wrrLocality.getEndpointPickingPolicy(), recursionDepth + 1));
    }

    /**
     * "Converts" a round_robin configuration to service config format.
     */
    private static ImmutableMap<String, ?> convertRoundRobinConfig() {
      return buildRoundRobinConfig();
    }

    /**
     * "Converts" a pick_first configuration to service config format.
     */
    private static ImmutableMap<String, ?> convertPickFirstConfig(PickFirst pickFirst) {
      return buildPickFirstConfig(pickFirst.getShuffleAddressList());
    }

    /**
     * Converts a least_request {@link Any} configuration to service config format.
     */
    private static ImmutableMap<String, ?> convertLeastRequestConfig(LeastRequest leastRequest)
        throws ResourceInvalidException {
      return buildLeastRequestConfig(
          leastRequest.hasChoiceCount() ? leastRequest.getChoiceCount().getValue() : null);
    }

    /**
     * Converts a custom TypedStruct LB config to service config format.
     */
    @SuppressWarnings("unchecked")
    private static ImmutableMap<String, ?> convertCustomConfig(
        com.github.xds.type.v3.TypedStruct configTypedStruct)
        throws ResourceInvalidException {
      return ImmutableMap.of(parseCustomConfigTypeName(configTypedStruct.getTypeUrl()),
          (Map<String, ?>) parseCustomConfigJson(configTypedStruct.getValue()));
    }

    /**
     * Converts a custom UDPA (legacy) TypedStruct LB config to service config format.
     */
    @SuppressWarnings("unchecked")
    private static ImmutableMap<String, ?> convertCustomConfig(
        com.github.udpa.udpa.type.v1.TypedStruct configTypedStruct)
        throws ResourceInvalidException {
      return ImmutableMap.of(parseCustomConfigTypeName(configTypedStruct.getTypeUrl()),
          (Map<String, ?>) parseCustomConfigJson(configTypedStruct.getValue()));
    }

    /**
     * Print the config Struct into JSON and then parse that into our internal representation.
     */
    private static Object parseCustomConfigJson(Struct configStruct)
        throws ResourceInvalidException {
      Object rawJsonConfig = null;
      try {
        rawJsonConfig = JsonParser.parse(JsonFormat.printer().print(configStruct));
      } catch (IOException e) {
        throw new ResourceInvalidException("Unable to parse custom LB config JSON", e);
      }

      if (!(rawJsonConfig instanceof Map)) {
        throw new ResourceInvalidException("Custom LB config does not contain a JSON object");
      }
      return rawJsonConfig;
    }


    private static String parseCustomConfigTypeName(String customConfigTypeName) {
      if (customConfigTypeName.contains("/")) {
        customConfigTypeName = customConfigTypeName.substring(
            customConfigTypeName.lastIndexOf("/") + 1);
      }
      return customConfigTypeName;
    }

    // Used to signal that the LB config goes too deep.
    static class MaxRecursionReachedException extends Exception {
      static final long serialVersionUID = 1L;
    }
  }

  /**
   * Builds a JSON LB configuration based on the old style of using the xDS Cluster proto message.
   * The lb_policy field is used to select the policy and configuration is extracted from various
   * policy specific fields in Cluster.
   */
  static class LegacyLoadBalancingPolicyConverter {

    /**
     * Factory method for creating a new {link LoadBalancerConfigConverter} for a given xDS {@link
     * Cluster}.
     *
     * @throws ResourceInvalidException If the {@link Cluster} has an invalid LB configuration.
     */
    static ImmutableMap<String, ?> convertToServiceConfig(Cluster cluster,
        boolean enableLeastRequest) throws ResourceInvalidException {
      switch (cluster.getLbPolicy()) {
        case RING_HASH:
          return convertRingHashConfig(cluster);
        case ROUND_ROBIN:
          return buildWrrLocalityConfig(buildRoundRobinConfig());
        case LEAST_REQUEST:
          if (enableLeastRequest) {
            return buildWrrLocalityConfig(convertLeastRequestConfig(cluster));
          }
          break;
        default:
      }
      throw new ResourceInvalidException(
          "Cluster " + cluster.getName() + ": unsupported lb policy: " + cluster.getLbPolicy());
    }

    /**
     * Creates a new ring_hash service config JSON object based on the old {@link RingHashLbConfig}
     * config message.
     */
    private static ImmutableMap<String, ?> convertRingHashConfig(Cluster cluster)
        throws ResourceInvalidException {
      RingHashLbConfig lbConfig = cluster.getRingHashLbConfig();

      // The hash function needs to be validated here as it is not exposed in the returned
      // configuration for later validation.
      if (lbConfig.getHashFunction() != RingHashLbConfig.HashFunction.XX_HASH) {
        throw new ResourceInvalidException(
            "Cluster " + cluster.getName() + ": invalid ring hash function: " + lbConfig);
      }

      return buildRingHashConfig(
          lbConfig.hasMinimumRingSize() ? (Long) lbConfig.getMinimumRingSize().getValue() : null,
          lbConfig.hasMaximumRingSize() ? (Long) lbConfig.getMaximumRingSize().getValue() : null);
    }

    /**
     * Creates a new least_request service config JSON object based on the old {@link
     * LeastRequestLbConfig} config message.
     */
    private static ImmutableMap<String, ?> convertLeastRequestConfig(Cluster cluster) {
      LeastRequestLbConfig lbConfig = cluster.getLeastRequestLbConfig();
      return buildLeastRequestConfig(
          lbConfig.hasChoiceCount() ? (Integer) lbConfig.getChoiceCount().getValue() : null);
    }
  }
}

1	/*
2	* Copyright 2022 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 com.google.common.collect.ImmutableList;
20	import com.google.common.collect.ImmutableMap;
21	import com.google.common.collect.Iterables;
22	import com.google.protobuf.Any;
23	import com.google.protobuf.InvalidProtocolBufferException;
24	import com.google.protobuf.Struct;
25	import com.google.protobuf.util.Durations;
26	import com.google.protobuf.util.JsonFormat;
27	import io.envoyproxy.envoy.config.cluster.v3.Cluster;
28	import io.envoyproxy.envoy.config.cluster.v3.Cluster.LeastRequestLbConfig;
29	import io.envoyproxy.envoy.config.cluster.v3.Cluster.RingHashLbConfig;
30	import io.envoyproxy.envoy.config.cluster.v3.LoadBalancingPolicy;
31	import io.envoyproxy.envoy.config.cluster.v3.LoadBalancingPolicy.Policy;
32	import io.envoyproxy.envoy.extensions.load_balancing_policies.client_side_weighted_round_robin.v3.ClientSideWeightedRoundRobin;
33	import io.envoyproxy.envoy.extensions.load_balancing_policies.least_request.v3.LeastRequest;
34	import io.envoyproxy.envoy.extensions.load_balancing_policies.pick_first.v3.PickFirst;
35	import io.envoyproxy.envoy.extensions.load_balancing_policies.ring_hash.v3.RingHash;
36	import io.envoyproxy.envoy.extensions.load_balancing_policies.round_robin.v3.RoundRobin;
37	import io.envoyproxy.envoy.extensions.load_balancing_policies.wrr_locality.v3.WrrLocality;
38	import io.grpc.InternalLogId;
39	import io.grpc.LoadBalancerRegistry;
40	import io.grpc.internal.JsonParser;
41	import io.grpc.xds.LoadBalancerConfigFactory.LoadBalancingPolicyConverter.MaxRecursionReachedException;
42	import io.grpc.xds.client.XdsLogger;
43	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
44	import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
45	import java.io.IOException;
46	import java.util.Map;
47
48	/**
49	* Creates service config JSON load balancer config objects for a given xDS Cluster message.
50	* Supports both the "legacy" configuration style and the new, more advanced one that utilizes the
51	* xDS "typed extension" mechanism.
52	*
53	* <p>Legacy configuration is done by setting the lb_policy enum field and any supporting
54	* configuration fields needed by the particular policy.
55	*
56	* <p>The new approach is to set the load_balancing_policy field that contains both the policy
57	* selection as well as any supporting configuration data. Providing a list of acceptable policies
58	* is also supported. Note that if this field is used, it will override any configuration set using
59	* the legacy approach. The new configuration approach is explained in detail in the <a href="
60	* https://github.com/grpc/proposal/blob/master/A52-xds-custom-lb-policies.md">Custom LB Policies
61	* gRFC</a>
62	*/
63	class LoadBalancerConfigFactory {	×
64
65	private static final XdsLogger logger = XdsLogger.withLogId(	1✔
66	InternalLogId.allocate("xds-client-lbconfig-factory", null));	1✔
67
68	static final String ROUND_ROBIN_FIELD_NAME = "round_robin";
69
70	static final String RING_HASH_FIELD_NAME = "ring_hash_experimental";
71	static final String MIN_RING_SIZE_FIELD_NAME = "minRingSize";
72	static final String MAX_RING_SIZE_FIELD_NAME = "maxRingSize";
73
74	static final String LEAST_REQUEST_FIELD_NAME = "least_request_experimental";
75	static final String CHOICE_COUNT_FIELD_NAME = "choiceCount";
76
77	static final String WRR_LOCALITY_FIELD_NAME = "wrr_locality_experimental";
78	static final String CHILD_POLICY_FIELD = "childPolicy";
79
80	static final String BLACK_OUT_PERIOD = "blackoutPeriod";
81
82	static final String WEIGHT_EXPIRATION_PERIOD = "weightExpirationPeriod";
83
84	static final String OOB_REPORTING_PERIOD = "oobReportingPeriod";
85
86	static final String ENABLE_OOB_LOAD_REPORT = "enableOobLoadReport";
87
88	static final String WEIGHT_UPDATE_PERIOD = "weightUpdatePeriod";
89
90	static final String PICK_FIRST_FIELD_NAME = "pick_first";
91	static final String SHUFFLE_ADDRESS_LIST_FIELD_NAME = "shuffleAddressList";
92
93	static final String ERROR_UTILIZATION_PENALTY = "errorUtilizationPenalty";
94	static final String METRIC_NAMES_FOR_COMPUTING_UTILIZATION = "metricNamesForComputingUtilization";
95
96	/**
97	* Factory method for creating a new {link LoadBalancerConfigConverter} for a given xDS {@link
98	* Cluster}.
99	*
100	* @throws ResourceInvalidException If the {@link Cluster} has an invalid LB configuration.
101	*/
102	static ImmutableMap<String, ?> newConfig(Cluster cluster, boolean enableLeastRequest)
103	throws ResourceInvalidException {
104	// The new load_balancing_policy will always be used if it is set, but for backward
105	// compatibility we will fall back to using the old lb_policy field if the new field is not set.
106	if (cluster.hasLoadBalancingPolicy()) {	1✔
107	try {
108	return LoadBalancingPolicyConverter.convertToServiceConfig(cluster.getLoadBalancingPolicy(),	1✔
109	0);
110	} catch (MaxRecursionReachedException e) {	1✔
111	throw new ResourceInvalidException("Maximum LB config recursion depth reached", e);	1✔
112	}
113	} else {
114	return LegacyLoadBalancingPolicyConverter.convertToServiceConfig(cluster, enableLeastRequest);	1✔
115	}
116	}
117
118	/**
119	* Builds a service config JSON object for the ring_hash load balancer config based on the given
120	* config values.
121	*/
122	private static ImmutableMap<String, ?> buildRingHashConfig(Long minRingSize, Long maxRingSize) {
123	ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();	1✔
124	if (minRingSize != null) {	1✔
125	configBuilder.put(MIN_RING_SIZE_FIELD_NAME, minRingSize.doubleValue());	1✔
126	}
127	if (maxRingSize != null) {	1✔
128	configBuilder.put(MAX_RING_SIZE_FIELD_NAME, maxRingSize.doubleValue());	1✔
129	}
130	return ImmutableMap.of(RING_HASH_FIELD_NAME, configBuilder.buildOrThrow());	1✔
131	}
132
133	/**
134	* Builds a service config JSON object for the weighted_round_robin load balancer config based on
135	* the given config values.
136	*/
137	private static ImmutableMap<String, ?> buildWrrConfig(String blackoutPeriod,
138	String weightExpirationPeriod, String oobReportingPeriod, Boolean enableOobLoadReport,
139	String weightUpdatePeriod, Float errorUtilizationPenalty,
140	ImmutableList<String> metricNamesForComputingUtilization) {
141	ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();	1✔
142	if (blackoutPeriod != null) {	1✔
143	configBuilder.put(BLACK_OUT_PERIOD, blackoutPeriod);	1✔
144	}
145	if (weightExpirationPeriod != null) {	1✔
146	configBuilder.put(WEIGHT_EXPIRATION_PERIOD, weightExpirationPeriod);	×
147	}
148	if (oobReportingPeriod != null) {	1✔
149	configBuilder.put(OOB_REPORTING_PERIOD, oobReportingPeriod);	×
150	}
151	if (enableOobLoadReport != null) {	1✔
152	configBuilder.put(ENABLE_OOB_LOAD_REPORT, enableOobLoadReport);	1✔
153	}
154	if (weightUpdatePeriod != null) {	1✔
155	configBuilder.put(WEIGHT_UPDATE_PERIOD, weightUpdatePeriod);	×
156	}
157	if (errorUtilizationPenalty != null) {	1✔
158	configBuilder.put(ERROR_UTILIZATION_PENALTY, errorUtilizationPenalty);	1✔
159	}
160	if (metricNamesForComputingUtilization != null	1✔
161	&& !metricNamesForComputingUtilization.isEmpty()) {	1✔
162	configBuilder.put(METRIC_NAMES_FOR_COMPUTING_UTILIZATION, metricNamesForComputingUtilization);	1✔
163	}
164	return ImmutableMap.of(WeightedRoundRobinLoadBalancerProvider.SCHEME,	1✔
165	configBuilder.buildOrThrow());	1✔
166	}
167
168	/**
169	* Builds a service config JSON object for the least_request load balancer config based on the
170	* given config values.
171	*/
172	private static ImmutableMap<String, ?> buildLeastRequestConfig(Integer choiceCount) {
173	ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();	1✔
174	if (choiceCount != null) {	1✔
175	configBuilder.put(CHOICE_COUNT_FIELD_NAME, choiceCount.doubleValue());	1✔
176	}
177	return ImmutableMap.of(LEAST_REQUEST_FIELD_NAME, configBuilder.buildOrThrow());	1✔
178	}
179
180	/**
181	* Builds a service config JSON wrr_locality by wrapping another policy config.
182	*/
183	private static ImmutableMap<String, ?> buildWrrLocalityConfig(
184	ImmutableMap<String, ?> childConfig) {
185	return ImmutableMap.<String, Object>builder().put(WRR_LOCALITY_FIELD_NAME,	1✔
186	ImmutableMap.of(CHILD_POLICY_FIELD, ImmutableList.of(childConfig))).buildOrThrow();	1✔
187	}
188
189	/**
190	* Builds an empty service config JSON config object for round robin (it is not configurable).
191	*/
192	private static ImmutableMap<String, ?> buildRoundRobinConfig() {
193	return ImmutableMap.of(ROUND_ROBIN_FIELD_NAME, ImmutableMap.of());	1✔
194	}
195
196	/**
197	* Builds a service config JSON object for the pick_first load balancer config based on the
198	* given config values.
199	*/
200	private static ImmutableMap<String, ?> buildPickFirstConfig(boolean shuffleAddressList) {
201	ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();	1✔
202	configBuilder.put(SHUFFLE_ADDRESS_LIST_FIELD_NAME, shuffleAddressList);	1✔
203	return ImmutableMap.of(PICK_FIRST_FIELD_NAME, configBuilder.buildOrThrow());	1✔
204	}
205
206	/**
207	* Responsible for converting from a {@code envoy.config.cluster.v3.LoadBalancingPolicy} proto
208	* message to a gRPC service config format.
209	*/
210	static class LoadBalancingPolicyConverter {	×
211
212	private static final int MAX_RECURSION = 16;
213
214	/**
215	* Converts a {@link LoadBalancingPolicy} object to a service config JSON object.
216	*/
217	private static ImmutableMap<String, ?> convertToServiceConfig(
218	LoadBalancingPolicy loadBalancingPolicy, int recursionDepth)
219	throws ResourceInvalidException, MaxRecursionReachedException {
220	if (recursionDepth > MAX_RECURSION) {	1✔
221	throw new MaxRecursionReachedException();	1✔
222	}
223	ImmutableMap<String, ?> serviceConfig = null;	1✔
224
225	for (Policy policy : loadBalancingPolicy.getPoliciesList()) {	1✔
226	Any typedConfig = policy.getTypedExtensionConfig().getTypedConfig();	1✔
227	try {
228	if (typedConfig.is(RingHash.class)) {	1✔
229	serviceConfig = convertRingHashConfig(typedConfig.unpack(RingHash.class));	1✔
230	} else if (typedConfig.is(WrrLocality.class)) {	1✔
231	serviceConfig = convertWrrLocalityConfig(typedConfig.unpack(WrrLocality.class),	1✔
232	recursionDepth);
233	} else if (typedConfig.is(RoundRobin.class)) {	1✔
234	serviceConfig = convertRoundRobinConfig();	1✔
235	} else if (typedConfig.is(LeastRequest.class)) {	1✔
236	serviceConfig = convertLeastRequestConfig(typedConfig.unpack(LeastRequest.class));	1✔
237	} else if (typedConfig.is(ClientSideWeightedRoundRobin.class)) {	1✔
238	serviceConfig = convertWeightedRoundRobinConfig(	1✔
239	typedConfig.unpack(ClientSideWeightedRoundRobin.class));	1✔
240	} else if (typedConfig.is(PickFirst.class)) {	1✔
241	serviceConfig = convertPickFirstConfig(typedConfig.unpack(PickFirst.class));	1✔
242	} else if (typedConfig.is(com.github.xds.type.v3.TypedStruct.class)) {	1✔
243	serviceConfig = convertCustomConfig(	1✔
244	typedConfig.unpack(com.github.xds.type.v3.TypedStruct.class));	1✔
245	} else if (typedConfig.is(com.github.udpa.udpa.type.v1.TypedStruct.class)) {	1✔
246	serviceConfig = convertCustomConfig(	1✔
247	typedConfig.unpack(com.github.udpa.udpa.type.v1.TypedStruct.class));	1✔
248	}
249
250	// TODO: support least_request once it is added to the envoy protos.
251	} catch (InvalidProtocolBufferException e) {	×
252	throw new ResourceInvalidException(	×
253	"Unable to unpack typedConfig for: " + typedConfig.getTypeUrl(), e);	×
254	}	1✔
255	// The service config is expected to have a single root entry, where the name of that entry
256	// is the name of the policy. A Load balancer with this name must exist in the registry.
257	if (serviceConfig == null \|\| LoadBalancerRegistry.getDefaultRegistry()	1✔
258	.getProvider(Iterables.getOnlyElement(serviceConfig.keySet())) == null) {	1✔
259	logger.log(XdsLogLevel.WARNING, "Policy {0} not found in the LB registry, skipping",	1✔
260	typedConfig.getTypeUrl());	1✔
261	continue;	1✔
262	} else {
263	return serviceConfig;	1✔
264	}
265	}
266
267	// If we could not find a Policy that we could both convert as well as find a provider for
268	// then we have an invalid LB policy configuration.
269	throw new ResourceInvalidException("Invalid LoadBalancingPolicy: " + loadBalancingPolicy);	1✔
270	}
271
272	/**
273	* Converts a ring_hash {@link Any} configuration to service config format.
274	*/
275	private static ImmutableMap<String, ?> convertRingHashConfig(RingHash ringHash)
276	throws ResourceInvalidException {
277	// The hash function needs to be validated here as it is not exposed in the returned
278	// configuration for later validation.
279	if (RingHash.HashFunction.XX_HASH != ringHash.getHashFunction()) {	1✔
280	throw new ResourceInvalidException(	1✔
281	"Invalid ring hash function: " + ringHash.getHashFunction());	1✔
282	}
283
284	return buildRingHashConfig(	1✔
285	ringHash.hasMinimumRingSize() ? ringHash.getMinimumRingSize().getValue() : null,	1✔
286	ringHash.hasMaximumRingSize() ? ringHash.getMaximumRingSize().getValue() : null);	1✔
287	}
288
289	private static ImmutableMap<String, ?> convertWeightedRoundRobinConfig(
290	ClientSideWeightedRoundRobin wrr) throws ResourceInvalidException {
291	try {
292	return buildWrrConfig(	1✔
293	wrr.hasBlackoutPeriod() ? Durations.toString(wrr.getBlackoutPeriod()) : null,	1✔
294	wrr.hasWeightExpirationPeriod()	1✔
295	? Durations.toString(wrr.getWeightExpirationPeriod()) : null,	1✔
296	wrr.hasOobReportingPeriod() ? Durations.toString(wrr.getOobReportingPeriod()) : null,	1✔
297	wrr.hasEnableOobLoadReport() ? wrr.getEnableOobLoadReport().getValue() : null,	1✔
298	wrr.hasWeightUpdatePeriod() ? Durations.toString(wrr.getWeightUpdatePeriod()) : null,	1✔
299	wrr.hasErrorUtilizationPenalty() ? wrr.getErrorUtilizationPenalty().getValue() : null,	1✔
300	ImmutableList.copyOf(wrr.getMetricNamesForComputingUtilizationList()));	1✔
301	} catch (IllegalArgumentException ex) {	1✔
302	throw new ResourceInvalidException("Invalid duration in weighted round robin config: "	1✔
303	+ ex.getMessage());	1✔
304	}
305	}
306
307	/**
308	* Converts a wrr_locality {@link Any} configuration to service config format.
309	*/
310	private static ImmutableMap<String, ?> convertWrrLocalityConfig(WrrLocality wrrLocality,
311	int recursionDepth)
312	throws ResourceInvalidException, MaxRecursionReachedException {
313	return buildWrrLocalityConfig(	1✔
314	convertToServiceConfig(wrrLocality.getEndpointPickingPolicy(), recursionDepth + 1));	1✔
315	}
316
317	/**
318	* "Converts" a round_robin configuration to service config format.
319	*/
320	private static ImmutableMap<String, ?> convertRoundRobinConfig() {
321	return buildRoundRobinConfig();	1✔
322	}
323
324	/**
325	* "Converts" a pick_first configuration to service config format.
326	*/
327	private static ImmutableMap<String, ?> convertPickFirstConfig(PickFirst pickFirst) {
328	return buildPickFirstConfig(pickFirst.getShuffleAddressList());	1✔
329	}
330
331	/**
332	* Converts a least_request {@link Any} configuration to service config format.
333	*/
334	private static ImmutableMap<String, ?> convertLeastRequestConfig(LeastRequest leastRequest)
335	throws ResourceInvalidException {
336	return buildLeastRequestConfig(	1✔
337	leastRequest.hasChoiceCount() ? leastRequest.getChoiceCount().getValue() : null);	1✔
338	}
339
340	/**
341	* Converts a custom TypedStruct LB config to service config format.
342	*/
343	@SuppressWarnings("unchecked")
344	private static ImmutableMap<String, ?> convertCustomConfig(
345	com.github.xds.type.v3.TypedStruct configTypedStruct)
346	throws ResourceInvalidException {
347	return ImmutableMap.of(parseCustomConfigTypeName(configTypedStruct.getTypeUrl()),	1✔
348	(Map<String, ?>) parseCustomConfigJson(configTypedStruct.getValue()));	1✔
349	}
350
351	/**
352	* Converts a custom UDPA (legacy) TypedStruct LB config to service config format.
353	*/
354	@SuppressWarnings("unchecked")
355	private static ImmutableMap<String, ?> convertCustomConfig(
356	com.github.udpa.udpa.type.v1.TypedStruct configTypedStruct)
357	throws ResourceInvalidException {
358	return ImmutableMap.of(parseCustomConfigTypeName(configTypedStruct.getTypeUrl()),	1✔
359	(Map<String, ?>) parseCustomConfigJson(configTypedStruct.getValue()));	1✔
360	}
361
362	/**
363	* Print the config Struct into JSON and then parse that into our internal representation.
364	*/
365	private static Object parseCustomConfigJson(Struct configStruct)
366	throws ResourceInvalidException {
367	Object rawJsonConfig = null;	1✔
368	try {
369	rawJsonConfig = JsonParser.parse(JsonFormat.printer().print(configStruct));	1✔
370	} catch (IOException e) {	×
371	throw new ResourceInvalidException("Unable to parse custom LB config JSON", e);	×
372	}	1✔
373
374	if (!(rawJsonConfig instanceof Map)) {	1✔
375	throw new ResourceInvalidException("Custom LB config does not contain a JSON object");	×
376	}
377	return rawJsonConfig;	1✔
378	}
379
380
381	private static String parseCustomConfigTypeName(String customConfigTypeName) {
382	if (customConfigTypeName.contains("/")) {	1✔
383	customConfigTypeName = customConfigTypeName.substring(	1✔
384	customConfigTypeName.lastIndexOf("/") + 1);	1✔
385	}
386	return customConfigTypeName;	1✔
387	}
388
389	// Used to signal that the LB config goes too deep.
390	static class MaxRecursionReachedException extends Exception {	1✔
391	static final long serialVersionUID = 1L;
392	}
393	}
394
395	/**
396	* Builds a JSON LB configuration based on the old style of using the xDS Cluster proto message.
397	* The lb_policy field is used to select the policy and configuration is extracted from various
398	* policy specific fields in Cluster.
399	*/
400	static class LegacyLoadBalancingPolicyConverter {	×
401
402	/**
403	* Factory method for creating a new {link LoadBalancerConfigConverter} for a given xDS {@link
404	* Cluster}.
405	*
406	* @throws ResourceInvalidException If the {@link Cluster} has an invalid LB configuration.
407	*/
408	static ImmutableMap<String, ?> convertToServiceConfig(Cluster cluster,
409	boolean enableLeastRequest) throws ResourceInvalidException {
410	switch (cluster.getLbPolicy()) {	1✔
411	case RING_HASH:
412	return convertRingHashConfig(cluster);	1✔
413	case ROUND_ROBIN:
414	return buildWrrLocalityConfig(buildRoundRobinConfig());	1✔
415	case LEAST_REQUEST:
416	if (enableLeastRequest) {	1✔
417	return buildWrrLocalityConfig(convertLeastRequestConfig(cluster));	1✔
418	}
419	break;
420	default:
421	}
422	throw new ResourceInvalidException(	1✔
423	"Cluster " + cluster.getName() + ": unsupported lb policy: " + cluster.getLbPolicy());	1✔
424	}
425
426	/**
427	* Creates a new ring_hash service config JSON object based on the old {@link RingHashLbConfig}
428	* config message.
429	*/
430	private static ImmutableMap<String, ?> convertRingHashConfig(Cluster cluster)
431	throws ResourceInvalidException {
432	RingHashLbConfig lbConfig = cluster.getRingHashLbConfig();	1✔
433
434	// The hash function needs to be validated here as it is not exposed in the returned
435	// configuration for later validation.
436	if (lbConfig.getHashFunction() != RingHashLbConfig.HashFunction.XX_HASH) {	1✔
437	throw new ResourceInvalidException(	1✔
438	"Cluster " + cluster.getName() + ": invalid ring hash function: " + lbConfig);	1✔
439	}
440
441	return buildRingHashConfig(	1✔
442	lbConfig.hasMinimumRingSize() ? (Long) lbConfig.getMinimumRingSize().getValue() : null,	1✔
443	lbConfig.hasMaximumRingSize() ? (Long) lbConfig.getMaximumRingSize().getValue() : null);	1✔
444	}
445
446	/**
447	* Creates a new least_request service config JSON object based on the old {@link
448	* LeastRequestLbConfig} config message.
449	*/
450	private static ImmutableMap<String, ?> convertLeastRequestConfig(Cluster cluster) {
451	LeastRequestLbConfig lbConfig = cluster.getLeastRequestLbConfig();	1✔
452	return buildLeastRequestConfig(	1✔
453	lbConfig.hasChoiceCount() ? (Integer) lbConfig.getChoiceCount().getValue() : null);	1✔
454	}
455	}
456	}

grpc / grpc-java / #20233

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