#20015

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

Build # #20015

Build Type

push

github

Committed by

web-flow

Commit Message

xds: ORCA to LRS propagation changes (#12203)

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

Run Details

34925 of 39432 relevant lines covered (88.57%)

0.89 hits per line

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

88.76

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

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

package io.grpc.xds;

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

import com.google.errorprone.annotations.CheckReturnValue;
import io.grpc.InternalLogId;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancerRegistry;
import io.grpc.NameResolver;
import io.grpc.Status;
import io.grpc.StatusOr;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig;
import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig;
import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig.DiscoveryMechanism;
import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
import io.grpc.xds.XdsClusterResource.CdsUpdate;
import io.grpc.xds.XdsClusterResource.CdsUpdate.ClusterType;
import io.grpc.xds.XdsConfig.Subscription;
import io.grpc.xds.XdsConfig.XdsClusterConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.AggregateConfig;
import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
import io.grpc.xds.client.XdsLogger;
import io.grpc.xds.client.XdsLogger.XdsLogLevel;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Load balancer for cds_experimental LB policy. One instance per top-level cluster.
 * The top-level cluster may be a plain EDS/logical-DNS cluster or an aggregate cluster formed
 * by a group of sub-clusters in a tree hierarchy.
 */
final class CdsLoadBalancer2 extends LoadBalancer {
  private final XdsLogger logger;
  private final Helper helper;
  private final LoadBalancerRegistry lbRegistry;
  private GracefulSwitchLoadBalancer delegate;
  // Following fields are effectively final.
  private String clusterName;
  private Subscription clusterSubscription;

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

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

    NameResolver.ConfigOrError configOrError;
    Object gracefulConfig;
    if (clusterConfig.getChildren() instanceof EndpointConfig) {
      // The LB policy config is provided in service_config.proto/JSON format.
      configOrError =
              GracefulSwitchLoadBalancer.parseLoadBalancingPolicyConfig(
                      Arrays.asList(clusterConfig.getClusterResource().lbPolicyConfig()),
                      lbRegistry);
      if (configOrError.getError() != null) {
        // Should be impossible, because XdsClusterResource validated this
        return fail(Status.INTERNAL.withDescription(
                errorPrefix() + "Unable to parse the LB config: " + configOrError.getError()));
      }
      CdsUpdate result = clusterConfig.getClusterResource();
      DiscoveryMechanism instance;
      if (result.clusterType() == ClusterType.EDS) {
        instance = DiscoveryMechanism.forEds(
            clusterName,
            result.edsServiceName(),
            result.lrsServerInfo(),
            result.maxConcurrentRequests(),
            result.upstreamTlsContext(),
            result.filterMetadata(),
            result.outlierDetection(),
            result.backendMetricPropagation());
      } else {
        instance = DiscoveryMechanism.forLogicalDns(
            clusterName,
            result.dnsHostName(),
            result.lrsServerInfo(),
            result.maxConcurrentRequests(),
            result.upstreamTlsContext(),
            result.filterMetadata(),
            result.backendMetricPropagation());
      }
      gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
          lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME),
          new ClusterResolverConfig(
              instance,
              configOrError.getConfig(),
              clusterConfig.getClusterResource().isHttp11ProxyAvailable()));
    } else if (clusterConfig.getChildren() instanceof AggregateConfig) {
      Map<String, PriorityChildConfig> priorityChildConfigs = new HashMap<>();
      List<String> leafClusters = ((AggregateConfig) clusterConfig.getChildren()).getLeafNames();
      for (String childCluster: leafClusters) {
        priorityChildConfigs.put(childCluster,
                new PriorityChildConfig(
                        GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
                                lbRegistry.getProvider(CDS_POLICY_NAME),
                                new CdsConfig(childCluster)),
                        false));
      }
      gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
          lbRegistry.getProvider(PRIORITY_POLICY_NAME),
          new PriorityLoadBalancerProvider.PriorityLbConfig(
              Collections.unmodifiableMap(priorityChildConfigs), leafClusters));
    } else {
      return fail(Status.INTERNAL.withDescription(
              errorPrefix() + "Unexpected cluster children type: "
                      + clusterConfig.getChildren().getClass()));
    }

    return delegate.acceptResolvedAddresses(
        resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(gracefulConfig).build());
  }

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

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

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

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

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.ConnectivityState.TRANSIENT_FAILURE;
21	import static io.grpc.xds.XdsLbPolicies.CDS_POLICY_NAME;
22	import static io.grpc.xds.XdsLbPolicies.CLUSTER_RESOLVER_POLICY_NAME;
23	import static io.grpc.xds.XdsLbPolicies.PRIORITY_POLICY_NAME;
24
25	import com.google.errorprone.annotations.CheckReturnValue;
26	import io.grpc.InternalLogId;
27	import io.grpc.LoadBalancer;
28	import io.grpc.LoadBalancerRegistry;
29	import io.grpc.NameResolver;
30	import io.grpc.Status;
31	import io.grpc.StatusOr;
32	import io.grpc.util.GracefulSwitchLoadBalancer;
33	import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig;
34	import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig;
35	import io.grpc.xds.ClusterResolverLoadBalancerProvider.ClusterResolverConfig.DiscoveryMechanism;
36	import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
37	import io.grpc.xds.XdsClusterResource.CdsUpdate;
38	import io.grpc.xds.XdsClusterResource.CdsUpdate.ClusterType;
39	import io.grpc.xds.XdsConfig.Subscription;
40	import io.grpc.xds.XdsConfig.XdsClusterConfig;
41	import io.grpc.xds.XdsConfig.XdsClusterConfig.AggregateConfig;
42	import io.grpc.xds.XdsConfig.XdsClusterConfig.EndpointConfig;
43	import io.grpc.xds.client.XdsLogger;
44	import io.grpc.xds.client.XdsLogger.XdsLogLevel;
45	import java.util.Arrays;
46	import java.util.Collections;
47	import java.util.HashMap;
48	import java.util.List;
49	import java.util.Map;
50
51	/**
52	* Load balancer for cds_experimental LB policy. One instance per top-level cluster.
53	* The top-level cluster may be a plain EDS/logical-DNS cluster or an aggregate cluster formed
54	* by a group of sub-clusters in a tree hierarchy.
55	*/
56	final class CdsLoadBalancer2 extends LoadBalancer {
57	private final XdsLogger logger;
58	private final Helper helper;
59	private final LoadBalancerRegistry lbRegistry;
60	private GracefulSwitchLoadBalancer delegate;
61	// Following fields are effectively final.
62	private String clusterName;
63	private Subscription clusterSubscription;
64
65	CdsLoadBalancer2(Helper helper, LoadBalancerRegistry lbRegistry) {	1✔
66	this.helper = checkNotNull(helper, "helper");	1✔
67	this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");	1✔
68	this.delegate = new GracefulSwitchLoadBalancer(helper);	1✔
69	logger = XdsLogger.withLogId(InternalLogId.allocate("cds-lb", helper.getAuthority()));	1✔
70	logger.log(XdsLogLevel.INFO, "Created");	1✔
71	}	1✔
72
73	@Override
74	public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
75	logger.log(XdsLogLevel.DEBUG, "Received resolution result: {0}", resolvedAddresses);	1✔
76	if (this.clusterName == null) {	1✔
77	CdsConfig config = (CdsConfig) resolvedAddresses.getLoadBalancingPolicyConfig();	1✔
78	logger.log(XdsLogLevel.INFO, "Config: {0}", config);	1✔
79	if (config.isDynamic) {	1✔
80	clusterSubscription = resolvedAddresses.getAttributes()	1✔
81	.get(XdsAttributes.XDS_CLUSTER_SUBSCRIPT_REGISTRY)	1✔
82	.subscribeToCluster(config.name);	1✔
83	}
84	this.clusterName = config.name;	1✔
85	}
86	XdsConfig xdsConfig = resolvedAddresses.getAttributes().get(XdsAttributes.XDS_CONFIG);	1✔
87	StatusOr<XdsClusterConfig> clusterConfigOr = xdsConfig.getClusters().get(clusterName);	1✔
88	if (clusterConfigOr == null) {	1✔
89	if (clusterSubscription == null) {	1✔
90	// Should be impossible, because XdsDependencyManager wouldn't have generated this
91	return fail(Status.INTERNAL.withDescription(	×
92	errorPrefix() + "Unable to find non-dynamic cluster"));	×
93	}
94	// The dynamic cluster must not have loaded yet
95	return Status.OK;	1✔
96	}
97	if (!clusterConfigOr.hasValue()) {	1✔
98	return fail(clusterConfigOr.getStatus());	1✔
99	}
100	XdsClusterConfig clusterConfig = clusterConfigOr.getValue();	1✔
101
102	NameResolver.ConfigOrError configOrError;
103	Object gracefulConfig;
104	if (clusterConfig.getChildren() instanceof EndpointConfig) {	1✔
105	// The LB policy config is provided in service_config.proto/JSON format.
106	configOrError =	1✔
107	GracefulSwitchLoadBalancer.parseLoadBalancingPolicyConfig(	1✔
108	Arrays.asList(clusterConfig.getClusterResource().lbPolicyConfig()),	1✔
109	lbRegistry);
110	if (configOrError.getError() != null) {	1✔
111	// Should be impossible, because XdsClusterResource validated this
112	return fail(Status.INTERNAL.withDescription(	×
113	errorPrefix() + "Unable to parse the LB config: " + configOrError.getError()));	×
114	}
115	CdsUpdate result = clusterConfig.getClusterResource();	1✔
116	DiscoveryMechanism instance;
117	if (result.clusterType() == ClusterType.EDS) {	1✔
118	instance = DiscoveryMechanism.forEds(	1✔
119	clusterName,
120	result.edsServiceName(),	1✔
121	result.lrsServerInfo(),	1✔
122	result.maxConcurrentRequests(),	1✔
123	result.upstreamTlsContext(),	1✔
124	result.filterMetadata(),	1✔
125	result.outlierDetection(),	1✔
126	result.backendMetricPropagation());	1✔
127	} else {
128	instance = DiscoveryMechanism.forLogicalDns(	1✔
129	clusterName,
130	result.dnsHostName(),	1✔
131	result.lrsServerInfo(),	1✔
132	result.maxConcurrentRequests(),	1✔
133	result.upstreamTlsContext(),	1✔
134	result.filterMetadata(),	1✔
135	result.backendMetricPropagation());	1✔
136	}
137	gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(	1✔
138	lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME),	1✔
139	new ClusterResolverConfig(
140	instance,
141	configOrError.getConfig(),	1✔
142	clusterConfig.getClusterResource().isHttp11ProxyAvailable()));	1✔
143	} else if (clusterConfig.getChildren() instanceof AggregateConfig) {	1✔
144	Map<String, PriorityChildConfig> priorityChildConfigs = new HashMap<>();	1✔
145	List<String> leafClusters = ((AggregateConfig) clusterConfig.getChildren()).getLeafNames();	1✔
146	for (String childCluster: leafClusters) {	1✔
147	priorityChildConfigs.put(childCluster,	1✔
148	new PriorityChildConfig(
149	GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(	1✔
150	lbRegistry.getProvider(CDS_POLICY_NAME),	1✔
151	new CdsConfig(childCluster)),
152	false));
153	}	1✔
154	gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(	1✔
155	lbRegistry.getProvider(PRIORITY_POLICY_NAME),	1✔
156	new PriorityLoadBalancerProvider.PriorityLbConfig(
157	Collections.unmodifiableMap(priorityChildConfigs), leafClusters));	1✔
158	} else {	1✔
159	return fail(Status.INTERNAL.withDescription(	×
160	errorPrefix() + "Unexpected cluster children type: "	×
161	+ clusterConfig.getChildren().getClass()));	×
162	}
163
164	return delegate.acceptResolvedAddresses(	1✔
165	resolvedAddresses.toBuilder().setLoadBalancingPolicyConfig(gracefulConfig).build());	1✔
166	}
167
168	@Override
169	public void handleNameResolutionError(Status error) {
170	logger.log(XdsLogLevel.WARNING, "Received name resolution error: {0}", error);	1✔
171	if (delegate != null) {	1✔
172	delegate.handleNameResolutionError(error);	1✔
173	} else {
174	helper.updateBalancingState(	×
175	TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	×
176	}
177	}	1✔
178
179	@Override
180	public void shutdown() {
181	logger.log(XdsLogLevel.INFO, "Shutdown");	1✔
182	delegate.shutdown();	1✔
183	delegate = new GracefulSwitchLoadBalancer(helper);	1✔
184	if (clusterSubscription != null) {	1✔
185	clusterSubscription.close();	1✔
186	clusterSubscription = null;	1✔
187	}
188	}	1✔
189
190	@CheckReturnValue // don't forget to return up the stack after the fail call
191	private Status fail(Status error) {
192	delegate.shutdown();	1✔
193	helper.updateBalancingState(	1✔
194	TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));	1✔
195	return Status.OK; // XdsNameResolver isn't a polling NR, so this value doesn't matter	1✔
196	}
197
198	private String errorPrefix() {
199	return "CdsLb for " + clusterName + ": ";	×
200	}
201	}

grpc / grpc-java / #20015

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

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