grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

xds: migrate to restructured xds loadbalancer 

* Deleted:
  + XdsLoadBalancer
  + XdsLbState
  + XdsComms
  + Their test classes (except XdsCommsTest)
* Moved/Renamed:
  + AdsStreamCallback -> LookasideChannelCallback
  + AdsStreamCallback2 -> AdsStreamCallback
  + XdsComms.LocalityInfo -> ClusterLoadAssignmentData.LocalityInfo
  + XdsComms.LbEndpoint -> ClusterLoadAssignmentData.LbEndpoint
  + XdsComms.DropOverload -> ClusterLoadAssignmentData.DropOverload
  + XdsLocality -> ClusterLoadAssignmentData.XdsLocality
* XdsComms2 is not renamed because it's temporary.
* XdsLoadBalancer2 is not renamed because otherwise the diff is hard to read.
This commit is contained in:
ZHANG Dapeng 2019-10-03 15:58:53 -07:00 committed by GitHub
parent 90b3c88fe2
commit dba7e9c36f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 414 additions and 3162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

210
xds/src/main/java/io/grpc/xds/ClusterLoadAssignmentData.java Normal file
View File

@ -0,0 +1,210 @@
/*
* Copyright 2019 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.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
import io.envoyproxy.envoy.api.v2.core.SocketAddress;
import io.grpc.EquivalentAddressGroup;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
/**
* Contains data types for ClusterLoadAssignment.
*/
final class ClusterLoadAssignmentData {
/**
* An {@code XdsLocality} object is simply a POJO representation for {@link
* io.envoyproxy.envoy.api.v2.core.Locality}, with only details needed for {@link
* XdsLoadBalancer2}.
*/
static final class XdsLocality {
private final String region;
private final String zone;
private final String subzone;
/** Must only be used for testing. */
@VisibleForTesting
XdsLocality(String region, String zone, String subzone) {
this.region = region;
this.zone = zone;
this.subzone = subzone;
}
static XdsLocality fromLocalityProto(io.envoyproxy.envoy.api.v2.core.Locality locality) {
return new XdsLocality(
/* region = */ locality.getRegion(),
/* zone = */ locality.getZone(),
/* subzone = */ locality.getSubZone());
}
io.envoyproxy.envoy.api.v2.core.Locality toLocalityProto() {
return io.envoyproxy.envoy.api.v2.core.Locality.newBuilder()
.setRegion(region)
.setZone(zone)
.setSubZone(subzone)
.build();
}
String getRegion() {
return region;
}
String getZone() {
return zone;
}
String getSubzone() {
return subzone;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
XdsLocality locality = (XdsLocality) o;
return Objects.equal(region, locality.region)
&& Objects.equal(zone, locality.zone)
&& Objects.equal(subzone, locality.subzone);
}
@Override
public int hashCode() {
return Objects.hashCode(region, zone, subzone);
}
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("region", region)
.add("zone", zone)
.add("subzone", subzone)
.toString();
}
}
/**
* Information about the locality from EDS response.
*/
static final class LocalityInfo {
final List<EquivalentAddressGroup> eags;
final List<Integer> endPointWeights;
final int localityWeight;
final int priority;
LocalityInfo(Collection<LbEndpoint> lbEndPoints, int localityWeight, int priority) {
List<EquivalentAddressGroup> eags = new ArrayList<>(lbEndPoints.size());
List<Integer> endPointWeights = new ArrayList<>(lbEndPoints.size());
for (LbEndpoint lbEndPoint : lbEndPoints) {
eags.add(lbEndPoint.eag);
endPointWeights.add(lbEndPoint.endPointWeight);
}
this.eags = Collections.unmodifiableList(eags);
this.endPointWeights = Collections.unmodifiableList(new ArrayList<>(endPointWeights));
this.localityWeight = localityWeight;
this.priority = priority;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
LocalityInfo that = (LocalityInfo) o;
return localityWeight == that.localityWeight
&& priority == that.priority
&& Objects.equal(eags, that.eags)
&& Objects.equal(endPointWeights, that.endPointWeights);
}
@Override
public int hashCode() {
return Objects.hashCode(eags, endPointWeights, localityWeight, priority);
}
}
static final class LbEndpoint {
final EquivalentAddressGroup eag;
final int endPointWeight;
LbEndpoint(io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint lbEndpointProto) {
this(
new EquivalentAddressGroup(ImmutableList.of(fromEnvoyProtoAddress(lbEndpointProto))),
lbEndpointProto.getLoadBalancingWeight().getValue());
}
@VisibleForTesting
LbEndpoint(EquivalentAddressGroup eag, int endPointWeight) {
this.eag = eag;
this.endPointWeight = endPointWeight;
}
private static java.net.SocketAddress fromEnvoyProtoAddress(
io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint lbEndpointProto) {
SocketAddress socketAddress = lbEndpointProto.getEndpoint().getAddress().getSocketAddress();
return new InetSocketAddress(socketAddress.getAddress(), socketAddress.getPortValue());
}
}
static final class DropOverload {
final String category;
final int dropsPerMillion;
DropOverload(String category, int dropsPerMillion) {
this.category = category;
this.dropsPerMillion = dropsPerMillion;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
DropOverload that = (DropOverload) o;
return dropsPerMillion == that.dropsPerMillion && Objects.equal(category, that.category);
}
@Override
public int hashCode() {
return Objects.hashCode(category, dropsPerMillion);
}
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("category", category)
.add("dropsPerMillion", dropsPerMillion)
.toString();
}
}
}

4
xds/src/main/java/io/grpc/xds/LoadReportClient.java
View File

@ -37,7 +37,7 @@ interface LoadReportClient {
* no-op.
*
* <p>This method is not thread-safe and should be called from the same synchronized context
* returned by {@link XdsLoadBalancer.Helper#getSynchronizationContext}.
* returned by {@link XdsLoadBalancer2.Helper#getSynchronizationContext}.
*
* @param callback containing methods to be invoked for passing information received from load
* reporting responses to xDS load balancer.
@ -49,7 +49,7 @@ interface LoadReportClient {
* {@link LoadReportClient} is no-op.
*
* <p>This method is not thread-safe and should be called from the same synchronized context
* returned by {@link XdsLoadBalancer.Helper#getSynchronizationContext}.
* returned by {@link XdsLoadBalancer2.Helper#getSynchronizationContext}.
*/
void stopLoadReporting();

2
xds/src/main/java/io/grpc/xds/LoadReportClientImpl.java
View File

@ -53,7 +53,7 @@ import javax.annotation.concurrent.NotThreadSafe;
* Client of xDS load reporting service.
*
* <p>Methods in this class are expected to be called in the same synchronized context that {@link
* XdsLoadBalancer.Helper#getSynchronizationContext} returns.
* XdsLoadBalancer2.Helper#getSynchronizationContext} returns.
*/
@NotThreadSafe
final class LoadReportClientImpl implements LoadReportClient {

9
xds/src/main/java/io/grpc/xds/LoadStatsStore.java
View File

@ -17,6 +17,7 @@
package io.grpc.xds;
import io.envoyproxy.envoy.api.v2.endpoint.ClusterStats;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import javax.annotation.Nullable;
/**
@ -26,7 +27,7 @@ import javax.annotation.Nullable;
* (i.e., Google backends) are aggregated in locality granularity (i.e., Google cluster) while the
* numbers of dropped calls are aggregated in cluster granularity.
*
* <p>An {@code LoadStatsStore} lives the same span of lifecycle as {@link XdsLoadBalancer} and
* <p>An {@code LoadStatsStore} lives the same span of lifecycle as {@link XdsLoadBalancer2} and
* only tracks loads for localities exposed by remote traffic director. A proper usage should be
*
* <ol>
@ -60,7 +61,7 @@ interface LoadStatsStore {
* reporting.
*
* <p>This method is not thread-safe and should be called from the same synchronized context
* returned by {@link XdsLoadBalancer.Helper#getSynchronizationContext}.
* returned by {@link XdsLoadBalancer2.Helper#getSynchronizationContext}.
*/
ClusterStats generateLoadReport();
@ -72,7 +73,7 @@ interface LoadStatsStore {
* balancer discovery responses before recording loads for those localities.
*
* <p>This method is not thread-safe and should be called from the same synchronized context
* returned by {@link XdsLoadBalancer.Helper#getSynchronizationContext}.
* returned by {@link XdsLoadBalancer2.Helper#getSynchronizationContext}.
*/
void addLocality(XdsLocality locality);
@ -87,7 +88,7 @@ interface LoadStatsStore {
* waste and keep including zero-load upstream locality stats in generated load reports.
*
* <p>This method is not thread-safe and should be called from the same synchronized context
* returned by {@link XdsLoadBalancer.Helper#getSynchronizationContext}.
* returned by {@link XdsLoadBalancer2.Helper#getSynchronizationContext}.
*/
void removeLocality(XdsLocality locality);

1
xds/src/main/java/io/grpc/xds/LoadStatsStoreImpl.java
View File

@ -26,6 +26,7 @@ import io.envoyproxy.envoy.api.v2.endpoint.EndpointLoadMetricStats;
import io.envoyproxy.envoy.api.v2.endpoint.UpstreamLocalityStats;
import io.grpc.xds.ClientLoadCounter.ClientLoadSnapshot;
import io.grpc.xds.ClientLoadCounter.MetricValue;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

5
xds/src/main/java/io/grpc/xds/LocalityStore.java
View File

@ -44,11 +44,12 @@ import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.xds.ClientLoadCounter.LoadRecordingSubchannelPicker;
import io.grpc.xds.ClientLoadCounter.MetricsObservingSubchannelPicker;
import io.grpc.xds.ClientLoadCounter.MetricsRecordingListener;
import io.grpc.xds.ClusterLoadAssignmentData.DropOverload;
import io.grpc.xds.ClusterLoadAssignmentData.LocalityInfo;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import io.grpc.xds.InterLocalityPicker.WeightedChildPicker;
import io.grpc.xds.OrcaOobUtil.OrcaReportingConfig;
import io.grpc.xds.OrcaOobUtil.OrcaReportingHelperWrapper;
import io.grpc.xds.XdsComms.DropOverload;
import io.grpc.xds.XdsComms.LocalityInfo;
import io.grpc.xds.XdsSubchannelPickers.ErrorPicker;
import java.util.ArrayList;
import java.util.Collections;

87
xds/src/main/java/io/grpc/xds/LookasideChannelLb.java
View File

@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload;
import io.envoyproxy.envoy.api.v2.endpoint.LocalityLbEndpoints;
import io.envoyproxy.envoy.type.FractionalPercent;
import io.envoyproxy.envoy.type.FractionalPercent.DenominatorType;
@ -31,10 +30,12 @@ import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.internal.ExponentialBackoffPolicy;
import io.grpc.internal.GrpcUtil;
import io.grpc.xds.ClusterLoadAssignmentData.DropOverload;
import io.grpc.xds.ClusterLoadAssignmentData.LbEndpoint;
import io.grpc.xds.ClusterLoadAssignmentData.LocalityInfo;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import io.grpc.xds.LoadReportClient.LoadReportCallback;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.XdsComms.LbEndpoint;
import io.grpc.xds.XdsComms.LocalityInfo;
import io.grpc.xds.XdsComms2.AdsStreamCallback;
import java.util.ArrayList;
import java.util.List;
@ -50,11 +51,11 @@ final class LookasideChannelLb extends LoadBalancer {
private final XdsComms2 xdsComms2;
LookasideChannelLb(
Helper helper, AdsStreamCallback adsCallback, ManagedChannel lbChannel,
Helper helper, LookasideChannelCallback lookasideChannelCallback, ManagedChannel lbChannel,
LocalityStore localityStore) {
this(
helper,
adsCallback,
lookasideChannelCallback,
lbChannel,
new LoadReportClientImpl(
lbChannel, helper, GrpcUtil.STOPWATCH_SUPPLIER, new ExponentialBackoffPolicy.Provider(),
@ -65,7 +66,7 @@ final class LookasideChannelLb extends LoadBalancer {
@VisibleForTesting
LookasideChannelLb(
Helper helper,
AdsStreamCallback adsCallback,
LookasideChannelCallback lookasideChannelCallback,
ManagedChannel lbChannel,
LoadReportClient lrsClient,
final LocalityStore localityStore) {
@ -79,10 +80,10 @@ final class LookasideChannelLb extends LoadBalancer {
};
this.lrsClient = lrsClient;
AdsStreamCallback2 adsCallback2 = new AdsStreamCallback2Impl(
adsCallback, lrsClient, lrsCallback, localityStore) ;
AdsStreamCallback adsCallback = new AdsStreamCallbackImpl(
lookasideChannelCallback, lrsClient, lrsCallback, localityStore) ;
xdsComms2 = new XdsComms2(
lbChannel, helper, adsCallback2, new ExponentialBackoffPolicy.Provider(),
lbChannel, helper, adsCallback, new ExponentialBackoffPolicy.Provider(),
GrpcUtil.STOPWATCH_SUPPLIER);
}
@ -123,30 +124,18 @@ final class LookasideChannelLb extends LoadBalancer {
lbChannel.shutdown();
}
// TODO(zdapeng): The old AdsStreamCallback will be renamed to LookasideChannelCallback,
// and AdsStreamCallback2 will be renamed to AdsStreamCallback
/**
* Callback on ADS stream events. The callback methods should be called in a proper {@link
* io.grpc.SynchronizationContext}.
*/
interface AdsStreamCallback2 {
void onEdsResponse(ClusterLoadAssignment clusterLoadAssignment);
private static final class AdsStreamCallbackImpl implements AdsStreamCallback {
void onError();
}
private static final class AdsStreamCallback2Impl implements AdsStreamCallback2 {
final AdsStreamCallback adsCallback;
final LookasideChannelCallback lookasideChannelCallback;
final LoadReportClient lrsClient;
final LoadReportCallback lrsCallback;
final LocalityStore localityStore;
boolean firstEdsResponseReceived;
AdsStreamCallback2Impl(
AdsStreamCallback adsCallback, LoadReportClient lrsClient, LoadReportCallback lrsCallback,
LocalityStore localityStore) {
this.adsCallback = adsCallback;
AdsStreamCallbackImpl(
LookasideChannelCallback lookasideChannelCallback, LoadReportClient lrsClient,
LoadReportCallback lrsCallback, LocalityStore localityStore) {
this.lookasideChannelCallback = lookasideChannelCallback;
this.lrsClient = lrsClient;
this.lrsCallback = lrsCallback;
this.localityStore = localityStore;
@ -156,25 +145,22 @@ final class LookasideChannelLb extends LoadBalancer {
public void onEdsResponse(ClusterLoadAssignment clusterLoadAssignment) {
if (!firstEdsResponseReceived) {
firstEdsResponseReceived = true;
adsCallback.onWorking();
lookasideChannelCallback.onWorking();
lrsClient.startLoadReporting(lrsCallback);
}
List<DropOverload> dropOverloadsProto =
List<ClusterLoadAssignment.Policy.DropOverload> dropOverloadsProto =
clusterLoadAssignment.getPolicy().getDropOverloadsList();
ImmutableList.Builder<XdsComms.DropOverload> dropOverloadsBuilder
= ImmutableList.builder();
for (ClusterLoadAssignment.Policy.DropOverload dropOverload
: dropOverloadsProto) {
ImmutableList.Builder<DropOverload> dropOverloadsBuilder = ImmutableList.builder();
for (ClusterLoadAssignment.Policy.DropOverload dropOverload : dropOverloadsProto) {
int rateInMillion = rateInMillion(dropOverload.getDropPercentage());
dropOverloadsBuilder.add(new XdsComms.DropOverload(
dropOverload.getCategory(), rateInMillion));
dropOverloadsBuilder.add(new DropOverload(dropOverload.getCategory(), rateInMillion));
if (rateInMillion == 1000_000) {
adsCallback.onAllDrop();
lookasideChannelCallback.onAllDrop();
break;
}
}
ImmutableList<XdsComms.DropOverload> dropOverloads = dropOverloadsBuilder.build();
ImmutableList<DropOverload> dropOverloads = dropOverloadsBuilder.build();
localityStore.updateDropPercentage(dropOverloads);
List<LocalityLbEndpoints> localities = clusterLoadAssignment.getEndpointsList();
@ -203,7 +189,30 @@ final class LookasideChannelLb extends LoadBalancer {
@Override
public void onError() {
adsCallback.onError();
lookasideChannelCallback.onError();
}
}
/**
* Callback on ADS stream events. The callback methods should be called in a proper {@link
* io.grpc.SynchronizationContext}.
*/
interface LookasideChannelCallback {
/**
* Once the response observer receives the first response.
*/
void onWorking();
/**
* Once an error occurs in ADS stream.
*/
void onError();
/**
* Once receives a response indicating that 100% of calls should be dropped.
*/
void onAllDrop();
}
}

24
xds/src/main/java/io/grpc/xds/LookasideLb.java
View File

@ -30,7 +30,7 @@ import io.grpc.NameResolver.ConfigOrError;
import io.grpc.util.ForwardingLoadBalancer;
import io.grpc.util.GracefulSwitchLoadBalancer;
import io.grpc.xds.LocalityStore.LocalityStoreImpl;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.LookasideChannelLb.LookasideChannelCallback;
import io.grpc.xds.XdsLoadBalancerProvider.XdsConfig;
import java.util.Map;
import java.util.logging.Logger;
@ -38,26 +38,26 @@ import java.util.logging.Logger;
/** Lookaside load balancer that handles balancer name changes. */
final class LookasideLb extends ForwardingLoadBalancer {
private final AdsStreamCallback adsCallback;
private final LookasideChannelCallback lookasideChannelCallback;
private final LookasideChannelLbFactory lookasideChannelLbFactory;
private final GracefulSwitchLoadBalancer lookasideChannelLb;
private final LoadBalancerRegistry lbRegistry;
private String balancerName;
LookasideLb(Helper lookasideLbHelper, AdsStreamCallback adsCallback) {
LookasideLb(Helper lookasideLbHelper, LookasideChannelCallback lookasideChannelCallback) {
this(
lookasideLbHelper, adsCallback, new LookasideChannelLbFactoryImpl(),
lookasideLbHelper, lookasideChannelCallback, new LookasideChannelLbFactoryImpl(),
LoadBalancerRegistry.getDefaultRegistry());
}
@VisibleForTesting
LookasideLb(
Helper lookasideLbHelper,
AdsStreamCallback adsCallback,
LookasideChannelCallback lookasideChannelCallback,
LookasideChannelLbFactory lookasideChannelLbFactory,
LoadBalancerRegistry lbRegistry) {
this.adsCallback = adsCallback;
this.lookasideChannelCallback = lookasideChannelCallback;
this.lookasideChannelLbFactory = lookasideChannelLbFactory;
this.lbRegistry = lbRegistry;
this.lookasideChannelLb = new GracefulSwitchLoadBalancer(lookasideLbHelper);
@ -113,23 +113,25 @@ final class LookasideLb extends ForwardingLoadBalancer {
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
return lookasideChannelLbFactory.newLoadBalancer(helper, adsCallback, balancerName);
return lookasideChannelLbFactory.newLoadBalancer(
helper, lookasideChannelCallback, balancerName);
}
};
}
@VisibleForTesting
interface LookasideChannelLbFactory {
LoadBalancer newLoadBalancer(Helper helper, AdsStreamCallback adsCallback, String balancerName);
LoadBalancer newLoadBalancer(
Helper helper, LookasideChannelCallback lookasideChannelCallback, String balancerName);
}
private static final class LookasideChannelLbFactoryImpl implements LookasideChannelLbFactory {
@Override
public LoadBalancer newLoadBalancer(Helper helper, AdsStreamCallback adsCallback,
String balancerName) {
public LoadBalancer newLoadBalancer(
Helper helper, LookasideChannelCallback lookasideChannelCallback, String balancerName) {
return new LookasideChannelLb(
helper, adsCallback, initLbChannel(helper, balancerName),
helper, lookasideChannelCallback, initLbChannel(helper, balancerName),
new LocalityStoreImpl(helper, LoadBalancerRegistry.getDefaultRegistry()));
}

461
xds/src/main/java/io/grpc/xds/XdsComms.java
View File

@ -1,461 +0,0 @@
/*
* Copyright 2019 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.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Objects;
import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Struct;
import com.google.protobuf.Value;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
import io.envoyproxy.envoy.api.v2.core.Node;
import io.envoyproxy.envoy.api.v2.core.SocketAddress;
import io.envoyproxy.envoy.api.v2.endpoint.LocalityLbEndpoints;
import io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc;
import io.envoyproxy.envoy.type.FractionalPercent;
import io.envoyproxy.envoy.type.FractionalPercent.DenominatorType;
import io.grpc.ChannelLogger.ChannelLogLevel;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer.Helper;
import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.BackoffPolicy;
import io.grpc.stub.StreamObserver;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.CheckForNull;
/**
* ADS client implementation.
*/
final class XdsComms {
private final ManagedChannel channel;
private final Helper helper;
private final BackoffPolicy.Provider backoffPolicyProvider;
private final Supplier<Stopwatch> stopwatchSupplier;
@CheckForNull
private ScheduledHandle adsRpcRetryTimer;
// never null
private BackoffPolicy adsRpcRetryPolicy;
// never null
private AdsStream adsStream;
/**
* Information about the locality from EDS response.
*/
// TODO(zdapeng): move this class out.
static final class LocalityInfo {
final List<EquivalentAddressGroup> eags;
final List<Integer> endPointWeights;
final int localityWeight;
final int priority;
LocalityInfo(Collection<LbEndpoint> lbEndPoints, int localityWeight, int priority) {
List<EquivalentAddressGroup> eags = new ArrayList<>(lbEndPoints.size());
List<Integer> endPointWeights = new ArrayList<>(lbEndPoints.size());
for (LbEndpoint lbEndPoint : lbEndPoints) {
eags.add(lbEndPoint.eag);
endPointWeights.add(lbEndPoint.endPointWeight);
}
this.eags = Collections.unmodifiableList(eags);
this.endPointWeights = Collections.unmodifiableList(new ArrayList<>(endPointWeights));
this.localityWeight = localityWeight;
this.priority = priority;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
LocalityInfo that = (LocalityInfo) o;
return localityWeight == that.localityWeight
&& priority == that.priority
&& Objects.equal(eags, that.eags)
&& Objects.equal(endPointWeights, that.endPointWeights);
}
@Override
public int hashCode() {
return Objects.hashCode(eags, endPointWeights, localityWeight, priority);
}
}
static final class LbEndpoint {
final EquivalentAddressGroup eag;
final int endPointWeight;
LbEndpoint(io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint lbEndpointProto) {
this(
new EquivalentAddressGroup(ImmutableList.of(fromEnvoyProtoAddress(lbEndpointProto))),
lbEndpointProto.getLoadBalancingWeight().getValue());
}
@VisibleForTesting
LbEndpoint(EquivalentAddressGroup eag, int endPointWeight) {
this.eag = eag;
this.endPointWeight = endPointWeight;
}
private static java.net.SocketAddress fromEnvoyProtoAddress(
io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint lbEndpointProto) {
SocketAddress socketAddress = lbEndpointProto.getEndpoint().getAddress().getSocketAddress();
return new InetSocketAddress(socketAddress.getAddress(), socketAddress.getPortValue());
}
}
static final class DropOverload {
final String category;
final int dropsPerMillion;
DropOverload(String category, int dropsPerMillion) {
this.category = category;
this.dropsPerMillion = dropsPerMillion;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
DropOverload that = (DropOverload) o;
return dropsPerMillion == that.dropsPerMillion && Objects.equal(category, that.category);
}
@Override
public int hashCode() {
return Objects.hashCode(category, dropsPerMillion);
}
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("category", category)
.add("dropsPerMillion", dropsPerMillion)
.toString();
}
}
private final class AdsStream {
static final String EDS_TYPE_URL =
"type.googleapis.com/envoy.api.v2.ClusterLoadAssignment";
static final String TRAFFICDIRECTOR_GRPC_HOSTNAME = "TRAFFICDIRECTOR_GRPC_HOSTNAME";
final LocalityStore localityStore;
final AdsStreamCallback adsStreamCallback;
final StreamObserver<DiscoveryRequest> xdsRequestWriter;
final Stopwatch retryStopwatch = stopwatchSupplier.get().start();
final StreamObserver<DiscoveryResponse> xdsResponseReader =
new StreamObserver<DiscoveryResponse>() {
// Must be accessed in SynchronizationContext
boolean firstEdsResponseReceived;
@Override
public void onNext(final DiscoveryResponse value) {
class HandleResponseRunnable implements Runnable {
@Override
public void run() {
String typeUrl = value.getTypeUrl();
if (EDS_TYPE_URL.equals(typeUrl)) {
// Assuming standard mode.
ClusterLoadAssignment clusterLoadAssignment;
try {
// maybe better to run this deserialization task out of syncContext?
clusterLoadAssignment =
value.getResources(0).unpack(ClusterLoadAssignment.class);
} catch (InvalidProtocolBufferException | RuntimeException e) {
cancelRpc("Received invalid EDS response", e);
adsStreamCallback.onError();
scheduleRetry();
return;
}
helper.getChannelLogger().log(
ChannelLogLevel.DEBUG,
"Received an EDS response: {0}", clusterLoadAssignment);
if (!firstEdsResponseReceived) {
firstEdsResponseReceived = true;
adsStreamCallback.onWorking();
}
List<ClusterLoadAssignment.Policy.DropOverload> dropOverloadsProto =
clusterLoadAssignment.getPolicy().getDropOverloadsList();
ImmutableList.Builder<DropOverload> dropOverloadsBuilder
= ImmutableList.builder();
for (ClusterLoadAssignment.Policy.DropOverload dropOverload
: dropOverloadsProto) {
int rateInMillion = rateInMillion(dropOverload.getDropPercentage());
dropOverloadsBuilder.add(new DropOverload(
dropOverload.getCategory(), rateInMillion));
if (rateInMillion == 1000_000) {
adsStreamCallback.onAllDrop();
break;
}
}
ImmutableList<DropOverload> dropOverloads = dropOverloadsBuilder.build();
localityStore.updateDropPercentage(dropOverloads);
List<LocalityLbEndpoints> localities = clusterLoadAssignment.getEndpointsList();
ImmutableMap.Builder<XdsLocality, LocalityInfo> localityEndpointsMapping =
new ImmutableMap.Builder<>();
for (LocalityLbEndpoints localityLbEndpoints : localities) {
io.envoyproxy.envoy.api.v2.core.Locality localityProto =
localityLbEndpoints.getLocality();
XdsLocality locality = XdsLocality.fromLocalityProto(localityProto);
List<LbEndpoint> lbEndPoints = new ArrayList<>();
for (io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint lbEndpoint
: localityLbEndpoints.getLbEndpointsList()) {
lbEndPoints.add(new LbEndpoint(lbEndpoint));
}
int localityWeight = localityLbEndpoints.getLoadBalancingWeight().getValue();
int priority = localityLbEndpoints.getPriority();
if (localityWeight != 0) {
localityEndpointsMapping.put(
locality, new LocalityInfo(lbEndPoints, localityWeight, priority));
}
}
localityStore.updateLocalityStore(localityEndpointsMapping.build());
}
}
}
helper.getSynchronizationContext().execute(new HandleResponseRunnable());
}
@Override
public void onError(Throwable t) {
helper.getSynchronizationContext().execute(
new Runnable() {
@Override
public void run() {
closed = true;
if (cancelled) {
return;
}
adsStreamCallback.onError();
scheduleRetry();
}
});
}
@Override
public void onCompleted() {
onError(Status.INTERNAL.withDescription("Server closed the ADS streaming RPC")
.asException());
}
// run in SynchronizationContext
void scheduleRetry() {
if (channel.isShutdown()) {
return;
}
checkState(
cancelled || closed,
"Scheduling retry while the stream is neither cancelled nor closed");
checkState(
adsRpcRetryTimer == null, "Scheduling retry while a retry is already pending");
class AdsRpcRetryTask implements Runnable {
@Override
public void run() {
adsRpcRetryTimer = null;
refreshAdsStream();
}
}
if (firstEdsResponseReceived) {
// Reset the backoff sequence if balancer has sent the initial response
adsRpcRetryPolicy = backoffPolicyProvider.get();
// Retry immediately
helper.getSynchronizationContext().execute(new AdsRpcRetryTask());
return;
}
adsRpcRetryTimer = helper.getSynchronizationContext().schedule(
new AdsRpcRetryTask(),
adsRpcRetryPolicy.nextBackoffNanos() - retryStopwatch.elapsed(TimeUnit.NANOSECONDS),
TimeUnit.NANOSECONDS,
helper.getScheduledExecutorService());
}
};
boolean cancelled;
boolean closed;
AdsStream(AdsStreamCallback adsStreamCallback, LocalityStore localityStore) {
this.adsStreamCallback = adsStreamCallback;
this.xdsRequestWriter = AggregatedDiscoveryServiceGrpc.newStub(channel).withWaitForReady()
.streamAggregatedResources(xdsResponseReader);
this.localityStore = localityStore;
checkState(adsRpcRetryTimer == null, "Creating AdsStream while retry is pending");
// Assuming standard mode, and send EDS request only
DiscoveryRequest edsRequest =
DiscoveryRequest.newBuilder()
.setNode(Node.newBuilder()
.setMetadata(Struct.newBuilder()
.putFields(
"endpoints_required",
Value.newBuilder().setBoolValue(true).build())))
.setTypeUrl(EDS_TYPE_URL)
// In the future, the right resource name can be obtained from CDS response.
.addResourceNames(helper.getAuthority()).build();
helper.getChannelLogger().log(ChannelLogLevel.DEBUG, "Sending EDS request {0}", edsRequest);
xdsRequestWriter.onNext(edsRequest);
}
AdsStream(AdsStream adsStream) {
this(adsStream.adsStreamCallback, adsStream.localityStore);
}
// run in SynchronizationContext
void cancelRpc(String message, Throwable cause) {
if (cancelled) {
return;
}
cancelled = true;
xdsRequestWriter.onError(
Status.CANCELLED.withDescription(message).withCause(cause).asRuntimeException());
}
}
private static int rateInMillion(FractionalPercent fractionalPercent) {
int numerator = fractionalPercent.getNumerator();
checkArgument(numerator >= 0, "numerator shouldn't be negative in %s", fractionalPercent);
DenominatorType type = fractionalPercent.getDenominator();
switch (type) {
case TEN_THOUSAND:
numerator *= 100;
break;
case HUNDRED:
numerator *= 100_00;
break;
case MILLION:
break;
default:
throw new IllegalArgumentException("unknown denominator type of " + fractionalPercent);
}
if (numerator > 1000_000) {
numerator = 1000_000;
}
return numerator;
}
/**
* Starts a new ADS streaming RPC.
*/
XdsComms(
ManagedChannel channel, Helper helper, AdsStreamCallback adsStreamCallback,
LocalityStore localityStore, BackoffPolicy.Provider backoffPolicyProvider,
Supplier<Stopwatch> stopwatchSupplier) {
this.channel = checkNotNull(channel, "channel");
this.helper = checkNotNull(helper, "helper");
this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier");
this.adsStream = new AdsStream(
checkNotNull(adsStreamCallback, "adsStreamCallback"),
checkNotNull(localityStore, "localityStore"));
this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
this.adsRpcRetryPolicy = backoffPolicyProvider.get();
}
// run in SynchronizationContext
void refreshAdsStream() {
checkState(!channel.isShutdown(), "channel is alreday shutdown");
if (adsStream.closed || adsStream.cancelled) {
cancelRetryTimer();
adsStream = new AdsStream(adsStream);
}
}
// run in SynchronizationContext
// TODO: Change method name to shutdown or shutdownXdsComms if that gives better semantics (
// cancel LB RPC and clean up retry timer).
void shutdownLbRpc(String message) {
adsStream.cancelRpc(message, null);
cancelRetryTimer();
}
// run in SynchronizationContext
private void cancelRetryTimer() {
if (adsRpcRetryTimer != null) {
adsRpcRetryTimer.cancel();
adsRpcRetryTimer = null;
}
}
/**
* Callback on ADS stream events. The callback methods should be called in a proper {@link
* io.grpc.SynchronizationContext}.
*/
interface AdsStreamCallback {
/**
* Once the response observer receives the first response.
*/
void onWorking();
/**
* Once an error occurs in ADS stream.
*/
void onError();
/**
* Once receives a response indicating that 100% of calls should be dropped.
*/
void onAllDrop();
}
}

30
xds/src/main/java/io/grpc/xds/XdsComms2.java
View File

@ -36,7 +36,6 @@ import io.grpc.Status;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.BackoffPolicy;
import io.grpc.stub.StreamObserver;
import io.grpc.xds.LookasideChannelLb.AdsStreamCallback2;
import java.util.concurrent.TimeUnit;
import javax.annotation.CheckForNull;
@ -63,7 +62,7 @@ final class XdsComms2 {
static final String EDS_TYPE_URL =
"type.googleapis.com/envoy.api.v2.ClusterLoadAssignment";
final AdsStreamCallback2 adsStreamCallback2;
final AdsStreamCallback adsStreamCallback;
final StreamObserver<DiscoveryRequest> xdsRequestWriter;
final Stopwatch retryStopwatch = stopwatchSupplier.get().start();
@ -90,7 +89,7 @@ final class XdsComms2 {
value.getResources(0).unpack(ClusterLoadAssignment.class);
} catch (InvalidProtocolBufferException | RuntimeException e) {
cancelRpc("Received invalid EDS response", e);
adsStreamCallback2.onError();
adsStreamCallback.onError();
scheduleRetry();
return;
}
@ -98,7 +97,8 @@ final class XdsComms2 {
helper.getChannelLogger().log(
ChannelLogLevel.DEBUG,
"Received an EDS response: {0}", clusterLoadAssignment);
adsStreamCallback2.onEdsResponse(clusterLoadAssignment);
firstEdsResponseReceived = true;
adsStreamCallback.onEdsResponse(clusterLoadAssignment);
}
}
}
@ -116,7 +116,7 @@ final class XdsComms2 {
if (cancelled) {
return;
}
adsStreamCallback2.onError();
adsStreamCallback.onError();
scheduleRetry();
}
});
@ -168,8 +168,8 @@ final class XdsComms2 {
boolean cancelled;
boolean closed;
AdsStream(AdsStreamCallback2 adsStreamCallback2) {
this.adsStreamCallback2 = adsStreamCallback2;
AdsStream(AdsStreamCallback adsStreamCallback) {
this.adsStreamCallback = adsStreamCallback;
this.xdsRequestWriter = AggregatedDiscoveryServiceGrpc.newStub(channel).withWaitForReady()
.streamAggregatedResources(xdsResponseReader);
@ -190,7 +190,7 @@ final class XdsComms2 {
}
AdsStream(AdsStream adsStream) {
this(adsStream.adsStreamCallback2);
this(adsStream.adsStreamCallback);
}
// run in SynchronizationContext
@ -208,13 +208,13 @@ final class XdsComms2 {
* Starts a new ADS streaming RPC.
*/
XdsComms2(
ManagedChannel channel, Helper helper, AdsStreamCallback2 adsStreamCallback2,
ManagedChannel channel, Helper helper, AdsStreamCallback adsStreamCallback,
BackoffPolicy.Provider backoffPolicyProvider, Supplier<Stopwatch> stopwatchSupplier) {
this.channel = checkNotNull(channel, "channel");
this.helper = checkNotNull(helper, "helper");
this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier");
this.adsStream = new AdsStream(
checkNotNull(adsStreamCallback2, "adsStreamCallback2"));
checkNotNull(adsStreamCallback, "adsStreamCallback"));
this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
this.adsRpcRetryPolicy = backoffPolicyProvider.get();
}
@ -244,4 +244,14 @@ final class XdsComms2 {
adsRpcRetryTimer = null;
}
}
/**
* Callback on ADS stream events. The callback methods should be called in a proper {@link
* io.grpc.SynchronizationContext}.
*/
interface AdsStreamCallback {
void onEdsResponse(ClusterLoadAssignment clusterLoadAssignment);
void onError();
}
}

115
xds/src/main/java/io/grpc/xds/XdsLbState.java
View File

@ -1,115 +0,0 @@
/*
* Copyright 2019 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 io.grpc.Attributes;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.Subchannel;
import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.internal.BackoffPolicy;
import io.grpc.internal.GrpcUtil;
import io.grpc.internal.ServiceConfigUtil.LbConfig;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import java.util.List;
import javax.annotation.Nullable;
/**
* The states of an XDS working session of {@link XdsLoadBalancer}. Created when XdsLoadBalancer
* switches to the current mode. Shutdown and discarded when XdsLoadBalancer switches to another
* mode.
*
* <p>There might be two implementations:
*
* <ul>
* <li>Standard plugin: No child plugin specified in lb config. Lb will send CDS request,
* and then EDS requests. EDS requests request for endpoints.</li>
* <li>Custom plugin: Child plugin specified in lb config. Lb will send EDS directly. EDS requests
* do not request for endpoints.</li>
* </ul>
*/
class XdsLbState {
final String balancerName;
@Nullable
final LbConfig childPolicy;
private final LocalityStore localityStore;
private final Helper helper;
private final ManagedChannel channel;
private final AdsStreamCallback adsStreamCallback;
private final BackoffPolicy.Provider backoffPolicyProvider;
@Nullable
private XdsComms xdsComms;
XdsLbState(
String balancerName,
@Nullable LbConfig childPolicy,
Helper helper,
LocalityStore localityStore,
ManagedChannel channel,
AdsStreamCallback adsStreamCallback,
BackoffPolicy.Provider backoffPolicyProvider) {
this.balancerName = checkNotNull(balancerName, "balancerName");
this.childPolicy = childPolicy;
this.helper = checkNotNull(helper, "helper");
this.localityStore = checkNotNull(localityStore, "localityStore");
this.channel = checkNotNull(channel, "channel");
this.adsStreamCallback = checkNotNull(adsStreamCallback, "adsStreamCallback");
this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
}
final void handleResolvedAddressGroups(
List<EquivalentAddressGroup> servers, Attributes attributes) {
// start XdsComms if not already alive
if (xdsComms != null) {
xdsComms.refreshAdsStream();
} else {
// TODO(zdapeng): pass a helper that has the right ChannelLogger for the oobChannel
xdsComms = new XdsComms(
channel, helper, adsStreamCallback, localityStore, backoffPolicyProvider,
GrpcUtil.STOPWATCH_SUPPLIER);
}
// TODO: maybe update picker
}
final void handleNameResolutionError(Status error) {
// NO-OP?
}
final void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo newState) {
// TODO: maybe update picker
localityStore.handleSubchannelState(subchannel, newState);
}
ManagedChannel shutdownAndReleaseChannel(String message) {
localityStore.reset();
if (xdsComms != null) {
xdsComms.shutdownLbRpc(message);
xdsComms = null;
}
return channel;
}
}

479
xds/src/main/java/io/grpc/xds/XdsLoadBalancer.java
View File

@ -1,479 +0,0 @@
/*
* Copyright 2019 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 com.google.common.base.Preconditions.checkState;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
import static io.grpc.xds.XdsLoadBalancerProvider.XDS_POLICY_NAME;
import static java.util.logging.Level.FINEST;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableList;
import io.grpc.Attributes;
import io.grpc.ChannelLogger.ChannelLogLevel;
import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancerRegistry;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.NameResolver.ConfigOrError;
import io.grpc.Status;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.internal.BackoffPolicy;
import io.grpc.internal.GrpcAttributes;
import io.grpc.internal.ServiceConfigUtil.LbConfig;
import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.xds.LoadReportClient.LoadReportCallback;
import io.grpc.xds.LoadReportClientImpl.LoadReportClientFactory;
import io.grpc.xds.LocalityStore.LocalityStoreImpl;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.XdsLoadBalancerProvider.XdsConfig;
import io.grpc.xds.XdsSubchannelPickers.ErrorPicker;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
import javax.annotation.CheckForNull;
import javax.annotation.Nullable;
/**
* A {@link LoadBalancer} that uses the XDS protocol.
*/
final class XdsLoadBalancer extends LoadBalancer {
private final LocalityStore localityStore;
private final Helper helper;
private final LoadBalancerRegistry lbRegistry;
private final FallbackManager fallbackManager;
private final BackoffPolicy.Provider backoffPolicyProvider;
private final LoadReportClientFactory lrsClientFactory;
@Nullable
private LoadReportClient lrsClient;
@Nullable
private XdsLbState xdsLbState;
private final AdsStreamCallback adsStreamCallback = new AdsStreamCallback() {
@Override
public void onWorking() {
if (fallbackManager.childPolicyHasBeenReady) {
// cancel Fallback-After-Startup timer if there's any
fallbackManager.cancelFallbackTimer();
}
fallbackManager.childBalancerWorked = true;
lrsClient.startLoadReporting(lrsCallback);
}
@Override
public void onError() {
if (!fallbackManager.childBalancerWorked) {
// start Fallback-at-Startup immediately
fallbackManager.useFallbackPolicy();
} else if (fallbackManager.childPolicyHasBeenReady) {
// TODO: schedule a timer for Fallback-After-Startup
} // else: the Fallback-at-Startup timer is still pending, noop and wait
}
@Override
public void onAllDrop() {
fallbackManager.cancelFallback();
}
};
private final LoadReportCallback lrsCallback =
new LoadReportCallback() {
@Override
public void onReportResponse(long reportIntervalNano) {
localityStore.updateOobMetricsReportInterval(reportIntervalNano);
}
};
private LbConfig fallbackPolicy;
XdsLoadBalancer(Helper helper, LoadBalancerRegistry lbRegistry,
BackoffPolicy.Provider backoffPolicyProvider) {
this(helper, lbRegistry, backoffPolicyProvider, LoadReportClientFactory.getInstance(),
new FallbackManager(helper, lbRegistry));
}
private XdsLoadBalancer(Helper helper,
LoadBalancerRegistry lbRegistry,
BackoffPolicy.Provider backoffPolicyProvider,
LoadReportClientFactory lrsClientFactory,
FallbackManager fallbackManager) {
this(helper, lbRegistry, backoffPolicyProvider, lrsClientFactory, fallbackManager,
new LocalityStoreImpl(new LocalityStoreHelper(helper, fallbackManager), lbRegistry));
}
@VisibleForTesting
XdsLoadBalancer(Helper helper,
LoadBalancerRegistry lbRegistry,
BackoffPolicy.Provider backoffPolicyProvider,
LoadReportClientFactory lrsClientFactory,
FallbackManager fallbackManager,
LocalityStore localityStore) {
this.helper = checkNotNull(helper, "helper");
this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");
this.backoffPolicyProvider = checkNotNull(backoffPolicyProvider, "backoffPolicyProvider");
this.lrsClientFactory = checkNotNull(lrsClientFactory, "lrsClientFactory");
this.fallbackManager = checkNotNull(fallbackManager, "fallbackManager");
this.localityStore = checkNotNull(localityStore, "localityStore");
}
private static final class LocalityStoreHelper extends ForwardingLoadBalancerHelper {
final Helper delegate;
final FallbackManager fallbackManager;
LocalityStoreHelper(Helper delegate, FallbackManager fallbackManager) {
this.delegate = checkNotNull(delegate, "delegate");
this.fallbackManager = checkNotNull(fallbackManager, "fallbackManager");
}
@Override
protected Helper delegate() {
return delegate;
}
@Override
public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
if (newState == READY) {
checkState(
fallbackManager.childBalancerWorked,
"channel goes to READY before the load balancer even worked");
fallbackManager.childPolicyHasBeenReady = true;
fallbackManager.cancelFallback();
}
if (!fallbackManager.isInFallbackMode()) {
delegate.getChannelLogger().log(
ChannelLogLevel.INFO, "Picker updated - state: {0}, picker: {1}", newState, newPicker);
delegate.updateBalancingState(newState, newPicker);
}
}
}
@Override
public void handleResolvedAddresses(ResolvedAddresses resolvedAddresses) {
List<EquivalentAddressGroup> servers = resolvedAddresses.getAddresses();
Attributes attributes = resolvedAddresses.getAttributes();
Map<String, ?> newRawLbConfig = checkNotNull(
attributes.get(ATTR_LOAD_BALANCING_CONFIG), "ATTR_LOAD_BALANCING_CONFIG not available");
ConfigOrError cfg =
XdsLoadBalancerProvider.parseLoadBalancingConfigPolicy(newRawLbConfig, lbRegistry);
if (cfg.getError() != null) {
throw cfg.getError().asRuntimeException();
}
XdsConfig xdsConfig = (XdsConfig) cfg.getConfig();
fallbackPolicy = xdsConfig.fallbackPolicy;
fallbackManager.updateFallbackServers(servers, attributes, fallbackPolicy);
fallbackManager.startFallbackTimer();
handleNewConfig(xdsConfig);
xdsLbState.handleResolvedAddressGroups(servers, attributes);
}
private void handleNewConfig(XdsConfig xdsConfig) {
String newBalancerName = xdsConfig.balancerName;
LbConfig childPolicy = xdsConfig.childPolicy;
ManagedChannel lbChannel;
if (xdsLbState == null) {
lbChannel = initLbChannel(helper, newBalancerName);
lrsClient =
lrsClientFactory.createLoadReportClient(lbChannel, helper, backoffPolicyProvider,
localityStore.getLoadStatsStore());
} else if (!newBalancerName.equals(xdsLbState.balancerName)) {
lrsClient.stopLoadReporting();
ManagedChannel oldChannel =
xdsLbState.shutdownAndReleaseChannel(
String.format("Changing balancer name from %s to %s", xdsLbState.balancerName,
newBalancerName));
oldChannel.shutdown();
lbChannel = initLbChannel(helper, newBalancerName);
lrsClient =
lrsClientFactory.createLoadReportClient(lbChannel, helper, backoffPolicyProvider,
localityStore.getLoadStatsStore());
} else if (!Objects.equal(
getPolicyNameOrNull(childPolicy),
getPolicyNameOrNull(xdsLbState.childPolicy))) {
// Changing child policy does not affect load reporting.
lbChannel =
xdsLbState.shutdownAndReleaseChannel(
String.format("Changing child policy from %s to %s", xdsLbState.childPolicy,
childPolicy));
} else { // effectively no change in policy, keep xdsLbState unchanged
return;
}
xdsLbState =
new XdsLbState(newBalancerName, childPolicy, helper, localityStore, lbChannel,
adsStreamCallback, backoffPolicyProvider);
}
private static ManagedChannel initLbChannel(Helper helper, String balancerName) {
ManagedChannel channel;
try {
channel = helper.createResolvingOobChannel(balancerName);
} catch (UnsupportedOperationException uoe) {
// Temporary solution until createResolvingOobChannel is implemented
// FIXME (https://github.com/grpc/grpc-java/issues/5495)
Logger logger = Logger.getLogger(XdsLoadBalancer.class.getName());
if (logger.isLoggable(FINEST)) {
logger.log(
FINEST,
"createResolvingOobChannel() not supported by the helper: " + helper,
uoe);
logger.log(
FINEST,
"creating oob channel for target {0} using default ManagedChannelBuilder",
balancerName);
}
channel = ManagedChannelBuilder.forTarget(balancerName).build();
}
return channel;
}
@Nullable
private static String getPolicyNameOrNull(@Nullable LbConfig config) {
if (config == null) {
return null;
}
return config.getPolicyName();
}
@Override
public void handleNameResolutionError(Status error) {
if (xdsLbState != null) {
xdsLbState.handleNameResolutionError(error);
}
if (fallbackManager.isInFallbackMode()) {
fallbackManager.fallbackBalancer.handleNameResolutionError(error);
}
if (xdsLbState == null && !fallbackManager.isInFallbackMode()) {
helper.updateBalancingState(TRANSIENT_FAILURE, new ErrorPicker(error));
}
}
/**
* This is only for the subchannel that is created by the child/fallback balancer using the
* old API {@link LoadBalancer.Helper#createSubchannel(EquivalentAddressGroup, Attributes)} or
* {@link LoadBalancer.Helper#createSubchannel(List, Attributes)}. Otherwise, it either won't be
* called or won't have any effect.
*/
@Deprecated
@Override
public void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo newState) {
if (fallbackManager.isInFallbackMode()) {
fallbackManager.fallbackBalancer.handleSubchannelState(subchannel, newState);
}
// xdsLbState should never be null here since handleSubchannelState cannot be called while the
// lb is shutdown.
xdsLbState.handleSubchannelState(subchannel, newState);
}
@Override
public void shutdown() {
if (xdsLbState != null) {
lrsClient.stopLoadReporting();
lrsClient = null;
ManagedChannel channel = xdsLbState.shutdownAndReleaseChannel("Client shutdown");
channel.shutdown();
xdsLbState = null;
}
fallbackManager.cancelFallback();
}
@Override
public boolean canHandleEmptyAddressListFromNameResolution() {
return true;
}
@Nullable
XdsLbState getXdsLbStateForTest() {
return xdsLbState;
}
@VisibleForTesting
static final class FallbackManager {
private static final long FALLBACK_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(10); // same as grpclb
private final Helper helper;
private final LoadBalancerRegistry lbRegistry;
private LbConfig fallbackPolicy;
// read-only for outer class
private LoadBalancer fallbackBalancer;
// Scheduled only once. Never reset.
@CheckForNull
private ScheduledHandle fallbackTimer;
private List<EquivalentAddressGroup> fallbackServers = ImmutableList.of();
private Attributes fallbackAttributes;
// allow value write by outer class
private boolean childBalancerWorked;
private boolean childPolicyHasBeenReady;
FallbackManager(Helper helper, LoadBalancerRegistry lbRegistry) {
this.helper = checkNotNull(helper, "helper");
this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");
}
/**
* Fallback mode being on indicates that an update from child LBs will be ignored unless the
* update triggers turning off the fallback mode first.
*/
boolean isInFallbackMode() {
return fallbackBalancer != null;
}
void cancelFallbackTimer() {
if (fallbackTimer != null) {
fallbackTimer.cancel();
}
}
void cancelFallback() {
cancelFallbackTimer();
if (fallbackBalancer != null) {
helper.getChannelLogger().log(
ChannelLogLevel.INFO, "Shutting down XDS fallback balancer");
fallbackBalancer.shutdown();
fallbackBalancer = null;
}
}
void useFallbackPolicy() {
if (fallbackBalancer != null) {
return;
}
cancelFallbackTimer();
helper.getChannelLogger().log(
ChannelLogLevel.INFO, "Using XDS fallback policy");
final class FallbackBalancerHelper extends ForwardingLoadBalancerHelper {
LoadBalancer balancer;
@Override
public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
checkNotNull(balancer, "there is a bug");
if (balancer != fallbackBalancer) {
// ignore updates from a misbehaving shutdown fallback balancer
return;
}
helper.getChannelLogger().log(
ChannelLogLevel.INFO,
"Picker updated - state: {0}, picker: {1}", newState, newPicker);
super.updateBalancingState(newState, newPicker);
}
@Override
protected Helper delegate() {
return helper;
}
}
FallbackBalancerHelper fallbackBalancerHelper = new FallbackBalancerHelper();
fallbackBalancer = lbRegistry.getProvider(fallbackPolicy.getPolicyName())
.newLoadBalancer(fallbackBalancerHelper);
fallbackBalancerHelper.balancer = fallbackBalancer;
propagateFallbackAddresses();
}
void updateFallbackServers(
List<EquivalentAddressGroup> servers, Attributes attributes,
LbConfig fallbackPolicy) {
this.fallbackServers = servers;
this.fallbackAttributes = Attributes.newBuilder()
.setAll(attributes)
.set(ATTR_LOAD_BALANCING_CONFIG, fallbackPolicy.getRawConfigValue())
.build();
LbConfig currentFallbackPolicy = this.fallbackPolicy;
this.fallbackPolicy = fallbackPolicy;
if (fallbackBalancer != null) {
if (fallbackPolicy.getPolicyName().equals(currentFallbackPolicy.getPolicyName())) {
propagateFallbackAddresses();
} else {
fallbackBalancer.shutdown();
fallbackBalancer = null;
useFallbackPolicy();
}
}
}
private void propagateFallbackAddresses() {
String fallbackPolicyName = fallbackPolicy.getPolicyName();
List<EquivalentAddressGroup> servers = fallbackServers;
// Some addresses in the list may be grpclb-v1 balancer addresses, so if the fallback policy
// does not support grpclb-v1 balancer addresses, then we need to exclude them from the list.
if (!fallbackPolicyName.equals("grpclb") && !fallbackPolicyName.equals(XDS_POLICY_NAME)) {
ImmutableList.Builder<EquivalentAddressGroup> backends = ImmutableList.builder();
for (EquivalentAddressGroup eag : fallbackServers) {
if (eag.getAttributes().get(GrpcAttributes.ATTR_LB_ADDR_AUTHORITY) == null) {
backends.add(eag);
}
}
servers = backends.build();
}
// TODO(zhangkun83): FIXME(#5496): this is a temporary hack.
if (servers.isEmpty()
&& !fallbackBalancer.canHandleEmptyAddressListFromNameResolution()) {
fallbackBalancer.handleNameResolutionError(Status.UNAVAILABLE.withDescription(
"NameResolver returned no usable address."
+ " addrs=" + fallbackServers + ", attrs=" + fallbackAttributes));
} else {
// TODO(carl-mastrangelo): propagate the load balancing config policy
fallbackBalancer.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(servers)
.setAttributes(fallbackAttributes)
.build());
}
}
void startFallbackTimer() {
if (fallbackTimer == null) {
class FallbackTask implements Runnable {
@Override
public void run() {
useFallbackPolicy();
}
}
fallbackTimer = helper.getSynchronizationContext().schedule(
new FallbackTask(), FALLBACK_TIMEOUT_MS, TimeUnit.MILLISECONDS,
helper.getScheduledExecutorService());
}
}
}
}

14
xds/src/main/java/io/grpc/xds/XdsLoadBalancer2.java
View File

@ -27,7 +27,7 @@ import io.grpc.LoadBalancer;
import io.grpc.Status;
import io.grpc.SynchronizationContext.ScheduledHandle;
import io.grpc.util.ForwardingLoadBalancerHelper;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.LookasideChannelLb.LookasideChannelCallback;
import java.util.concurrent.TimeUnit;
import javax.annotation.CheckForNull;
import javax.annotation.Nullable;
@ -46,7 +46,7 @@ final class XdsLoadBalancer2 extends LoadBalancer {
private final Helper helper;
private final LoadBalancer lookasideLb;
private final LoadBalancer.Factory fallbackLbFactory;
private final AdsStreamCallback adsCallback = new AdsStreamCallback() {
private final LookasideChannelCallback lookasideChannelCallback = new LookasideChannelCallback() {
@Override
public void onWorking() {
if (childPolicyHasBeenReady) {
@ -93,7 +93,8 @@ final class XdsLoadBalancer2 extends LoadBalancer {
LookasideLbFactory lookasideLbFactory,
LoadBalancer.Factory fallbackLbFactory) {
this.helper = helper;
this.lookasideLb = lookasideLbFactory.newLoadBalancer(new LookasideLbHelper(), adsCallback);
this.lookasideLb = lookasideLbFactory.newLoadBalancer(new LookasideLbHelper(),
lookasideChannelCallback);
this.fallbackLbFactory = fallbackLbFactory;
}
@ -246,13 +247,14 @@ final class XdsLoadBalancer2 extends LoadBalancer {
/** Factory of a look-aside load balancer. The interface itself is for convenience in test. */
@VisibleForTesting
interface LookasideLbFactory {
LoadBalancer newLoadBalancer(Helper helper, AdsStreamCallback adsCallback);
LoadBalancer newLoadBalancer(Helper helper, LookasideChannelCallback lookasideChannelCallback);
}
private static final class LookasideLbFactoryImpl implements LookasideLbFactory {
@Override
public LoadBalancer newLoadBalancer(Helper lookasideLbHelper, AdsStreamCallback adsCallback) {
return new LookasideLb(lookasideLbHelper, adsCallback);
public LoadBalancer newLoadBalancer(
Helper lookasideLbHelper, LookasideChannelCallback lookasideChannelCallback) {
return new LookasideLb(lookasideLbHelper, lookasideChannelCallback);
}
}

4
xds/src/main/java/io/grpc/xds/XdsLoadBalancerProvider.java
View File

@ -29,7 +29,6 @@ import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.NameResolver.ConfigOrError;
import io.grpc.Status;
import io.grpc.internal.ExponentialBackoffPolicy;
import io.grpc.internal.ServiceConfigUtil;
import io.grpc.internal.ServiceConfigUtil.LbConfig;
import java.util.List;
@ -66,8 +65,7 @@ public final class XdsLoadBalancerProvider extends LoadBalancerProvider {
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
return new XdsLoadBalancer(helper, LoadBalancerRegistry.getDefaultRegistry(),
new ExponentialBackoffPolicy.Provider());
return new XdsLoadBalancer2(helper);
}
@Override

94
xds/src/main/java/io/grpc/xds/XdsLocality.java
View File

@ -1,94 +0,0 @@
/*
* Copyright 2019 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.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Objects;
/**
* An {@code XdsLocality} object is simply a POJO representation for {@link
* io.envoyproxy.envoy.api.v2.core.Locality}, with only details needed for {@link XdsLoadBalancer}.
*/
final class XdsLocality {
private final String region;
private final String zone;
private final String subzone;
/** Must only be used for testing. */
@VisibleForTesting
XdsLocality(String region, String zone, String subzone) {
this.region = region;
this.zone = zone;
this.subzone = subzone;
}
static XdsLocality fromLocalityProto(io.envoyproxy.envoy.api.v2.core.Locality locality) {
return new XdsLocality(
/* region = */ locality.getRegion(),
/* zone = */ locality.getZone(),
/* subzone = */ locality.getSubZone());
}
io.envoyproxy.envoy.api.v2.core.Locality toLocalityProto() {
return io.envoyproxy.envoy.api.v2.core.Locality.newBuilder()
.setRegion(region)
.setZone(zone)
.setSubZone(subzone)
.build();
}
String getRegion() {
return region;
}
String getZone() {
return zone;
}
String getSubzone() {
return subzone;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
XdsLocality locality = (XdsLocality) o;
return Objects.equal(region, locality.region)
&& Objects.equal(zone, locality.zone)
&& Objects.equal(subzone, locality.subzone);
}
@Override
public int hashCode() {
return Objects.hashCode(region, zone, subzone);
}
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("region", region)
.add("zone", zone)
.add("subzone", subzone)
.toString();
}
}

11
xds/src/test/java/io/grpc/xds/XdsLocalityTest.java → xds/src/test/java/io/grpc/xds/ClusterLoadAssignmentDataTest.java
View File

@ -20,18 +20,19 @@ import static com.google.common.truth.Truth.assertThat;
import com.google.common.testing.EqualsTester;
import io.envoyproxy.envoy.api.v2.core.Locality;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
/**
* Unit tests for {@link XdsLocality}.
* Unit tests for {@link ClusterLoadAssignmentData}.
*/
@RunWith(JUnit4.class)
public class XdsLocalityTest {
public class ClusterLoadAssignmentDataTest {
@Test
public void convertToAndFromLocalityProto() {
public void xdsLocality_convertToAndFromLocalityProto() {
Locality locality =
Locality.newBuilder()
.setRegion("test_region")
@ -50,7 +51,7 @@ public class XdsLocalityTest {
}
@Test
public void equal() {
public void xdsLocality_equal() {
new EqualsTester()
.addEqualityGroup(
new XdsLocality("region-a", "zone-a", "subzone-a"),
@ -65,7 +66,7 @@ public class XdsLocalityTest {
}
@Test
public void hash() {
public void xdsLocality_hash() {
assertThat(new XdsLocality("region", "zone", "subzone").hashCode())
.isEqualTo(new XdsLocality("region", "zone","subzone").hashCode());
}

285
xds/src/test/java/io/grpc/xds/FallbackManagerTest.java
View File

@ -1,285 +0,0 @@
/*
* Copyright 2019 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.truth.Truth.assertThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import com.google.common.collect.ImmutableList;
import io.grpc.Attributes;
import io.grpc.ChannelLogger;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.ResolvedAddresses;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.Status;
import io.grpc.Status.Code;
import io.grpc.SynchronizationContext;
import io.grpc.internal.FakeClock;
import io.grpc.internal.GrpcAttributes;
import io.grpc.internal.ServiceConfigUtil.LbConfig;
import io.grpc.xds.XdsLoadBalancer.FallbackManager;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
/**
* Unit test for {@link FallbackManager}.
*/
@RunWith(JUnit4.class)
public class FallbackManagerTest {
private static final long FALLBACK_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(10);
private final FakeClock fakeClock = new FakeClock();
private final LoadBalancerRegistry lbRegistry = new LoadBalancerRegistry();
private final LoadBalancerProvider fakeFallbackLbProvider = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return fallbackPolicy.getPolicyName();
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
return fakeFallbackLb;
}
};
private final LoadBalancerProvider fakeRoundRonbinLbProvider = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return "round_robin";
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
return fakeRoundRobinLb;
}
};
private final SynchronizationContext syncContext = new SynchronizationContext(
new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
throw new AssertionError(e);
}
});
@Mock
private Helper helper;
@Mock
private LoadBalancer fakeRoundRobinLb;
@Mock
private LoadBalancer fakeFallbackLb;
@Mock
private ChannelLogger channelLogger;
private FallbackManager fallbackManager;
private LbConfig fallbackPolicy;
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
doReturn(syncContext).when(helper).getSynchronizationContext();
doReturn(fakeClock.getScheduledExecutorService()).when(helper).getScheduledExecutorService();
doReturn(channelLogger).when(helper).getChannelLogger();
fallbackPolicy = new LbConfig("test_policy", new HashMap<String, Void>());
lbRegistry.register(fakeRoundRonbinLbProvider);
lbRegistry.register(fakeFallbackLbProvider);
fallbackManager = new FallbackManager(helper, lbRegistry);
}
@After
public void tearDown() {
assertThat(fakeClock.getPendingTasks()).isEmpty();
}
@Test
public void useFallbackWhenTimeout() {
fallbackManager.startFallbackTimer();
List<EquivalentAddressGroup> eags = ImmutableList.of(
new EquivalentAddressGroup(ImmutableList.<SocketAddress>of(new InetSocketAddress(8080))));
fallbackManager.updateFallbackServers(
eags, Attributes.EMPTY, fallbackPolicy);
assertThat(fallbackManager.isInFallbackMode()).isFalse();
verify(fakeFallbackLb, never())
.handleResolvedAddresses(ArgumentMatchers.any(ResolvedAddresses.class));
fakeClock.forwardTime(FALLBACK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
assertThat(fallbackManager.isInFallbackMode()).isTrue();
verify(fakeFallbackLb).handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(eags)
.setAttributes(
Attributes.newBuilder()
.set(
LoadBalancer.ATTR_LOAD_BALANCING_CONFIG,
fallbackPolicy.getRawConfigValue())
.build())
.build());
}
@Test
public void fallback_handleBackendsEagsOnly() {
fallbackManager.startFallbackTimer();
EquivalentAddressGroup eag0 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8080)));
Attributes attributes = Attributes
.newBuilder()
.set(GrpcAttributes.ATTR_LB_ADDR_AUTHORITY, "this is a balancer address")
.build();
EquivalentAddressGroup eag1 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8081)), attributes);
EquivalentAddressGroup eag2 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8082)));
List<EquivalentAddressGroup> eags = ImmutableList.of(eag0, eag1, eag2);
fallbackManager.updateFallbackServers(
eags, Attributes.EMPTY, fallbackPolicy);
fakeClock.forwardTime(FALLBACK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
assertThat(fallbackManager.isInFallbackMode()).isTrue();
verify(fakeFallbackLb).handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(ImmutableList.of(eag0, eag2))
.setAttributes(
Attributes.newBuilder()
.set(
LoadBalancer.ATTR_LOAD_BALANCING_CONFIG,
fallbackPolicy.getRawConfigValue())
.build())
.build());
}
@Test
public void fallback_handleGrpclbAddresses() {
lbRegistry.deregister(fakeFallbackLbProvider);
fallbackPolicy = new LbConfig("grpclb", new HashMap<String, Void>());
lbRegistry.register(fakeFallbackLbProvider);
fallbackManager.startFallbackTimer();
EquivalentAddressGroup eag0 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8080)));
Attributes attributes = Attributes
.newBuilder()
.set(GrpcAttributes.ATTR_LB_ADDR_AUTHORITY, "this is a balancer address")
.build();
EquivalentAddressGroup eag1 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8081)), attributes);
EquivalentAddressGroup eag2 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8082)));
List<EquivalentAddressGroup> eags = ImmutableList.of(eag0, eag1, eag2);
fallbackManager.updateFallbackServers(
eags, Attributes.EMPTY, fallbackPolicy);
fakeClock.forwardTime(FALLBACK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
assertThat(fallbackManager.isInFallbackMode()).isTrue();
verify(fakeFallbackLb).handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(eags)
.setAttributes(
Attributes.newBuilder()
.set(
LoadBalancer.ATTR_LOAD_BALANCING_CONFIG,
fallbackPolicy.getRawConfigValue())
.build())
.build());
}
@Test
public void fallback_onlyGrpclbAddresses_NoBackendAddress() {
lbRegistry.deregister(fakeFallbackLbProvider);
fallbackPolicy = new LbConfig("not_grpclb", new HashMap<String, Void>());
lbRegistry.register(fakeFallbackLbProvider);
fallbackManager.startFallbackTimer();
Attributes attributes = Attributes
.newBuilder()
.set(GrpcAttributes.ATTR_LB_ADDR_AUTHORITY, "this is a balancer address")
.build();
EquivalentAddressGroup eag1 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8081)), attributes);
EquivalentAddressGroup eag2 = new EquivalentAddressGroup(
ImmutableList.<SocketAddress>of(new InetSocketAddress(8082)), attributes);
List<EquivalentAddressGroup> eags = ImmutableList.of(eag1, eag2);
fallbackManager.updateFallbackServers(
eags, Attributes.EMPTY, fallbackPolicy);
fakeClock.forwardTime(FALLBACK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
assertThat(fallbackManager.isInFallbackMode()).isTrue();
ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
verify(fakeFallbackLb).handleNameResolutionError(statusCaptor.capture());
assertThat(statusCaptor.getValue().getCode()).isEqualTo(Code.UNAVAILABLE);
}
@Test
public void cancelFallback() {
fallbackManager.startFallbackTimer();
List<EquivalentAddressGroup> eags = ImmutableList.of(
new EquivalentAddressGroup(ImmutableList.<SocketAddress>of(new InetSocketAddress(8080))));
fallbackManager.updateFallbackServers(
eags, Attributes.EMPTY, fallbackPolicy);
fallbackManager.cancelFallback();
fakeClock.forwardTime(FALLBACK_TIMEOUT_MS, TimeUnit.MILLISECONDS);
assertThat(fallbackManager.isInFallbackMode()).isFalse();
verify(fakeFallbackLb, never())
.handleResolvedAddresses(ArgumentMatchers.any(ResolvedAddresses.class));
}
}

4
xds/src/test/java/io/grpc/xds/LoadStatsStoreImplTest.java
View File

@ -24,6 +24,7 @@ import io.envoyproxy.envoy.api.v2.endpoint.ClusterStats.DroppedRequests;
import io.envoyproxy.envoy.api.v2.endpoint.EndpointLoadMetricStats;
import io.envoyproxy.envoy.api.v2.endpoint.UpstreamLocalityStats;
import io.grpc.xds.ClientLoadCounter.MetricValue;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@ -71,7 +72,8 @@ public class LoadStatsStoreImplTest {
return res;
}
private static UpstreamLocalityStats buildUpstreamLocalityStats(XdsLocality locality,
private static UpstreamLocalityStats buildUpstreamLocalityStats(
XdsLocality locality,
long callsSucceed,
long callsInProgress,
long callsFailed,

7
xds/src/test/java/io/grpc/xds/LocalityStoreTest.java
View File

@ -62,6 +62,10 @@ import io.grpc.internal.FakeClock.ScheduledTask;
import io.grpc.internal.FakeClock.TaskFilter;
import io.grpc.xds.ClientLoadCounter.LoadRecordingStreamTracerFactory;
import io.grpc.xds.ClientLoadCounter.MetricsRecordingListener;
import io.grpc.xds.ClusterLoadAssignmentData.DropOverload;
import io.grpc.xds.ClusterLoadAssignmentData.LbEndpoint;
import io.grpc.xds.ClusterLoadAssignmentData.LocalityInfo;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import io.grpc.xds.InterLocalityPicker.WeightedChildPicker;
import io.grpc.xds.LocalityStore.LocalityStoreImpl;
import io.grpc.xds.LocalityStore.LocalityStoreImpl.PickerFactory;
@ -69,9 +73,6 @@ import io.grpc.xds.OrcaOobUtil.OrcaOobReportListener;
import io.grpc.xds.OrcaOobUtil.OrcaReportingConfig;
import io.grpc.xds.OrcaOobUtil.OrcaReportingHelperWrapper;
import io.grpc.xds.OrcaPerRequestUtil.OrcaPerRequestReportListener;
import io.grpc.xds.XdsComms.DropOverload;
import io.grpc.xds.XdsComms.LbEndpoint;
import io.grpc.xds.XdsComms.LocalityInfo;
import io.grpc.xds.XdsSubchannelPickers.ErrorPicker;
import java.net.InetSocketAddress;
import java.util.Collection;

53
xds/src/test/java/io/grpc/xds/LookasideChannelLbTest.java
View File

@ -55,10 +55,11 @@ import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.internal.testing.StreamRecorder;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.GrpcCleanupRule;
import io.grpc.xds.ClusterLoadAssignmentData.DropOverload;
import io.grpc.xds.ClusterLoadAssignmentData.LocalityInfo;
import io.grpc.xds.ClusterLoadAssignmentData.XdsLocality;
import io.grpc.xds.LoadReportClient.LoadReportCallback;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.XdsComms.DropOverload;
import io.grpc.xds.XdsComms.LocalityInfo;
import io.grpc.xds.LookasideChannelLb.LookasideChannelCallback;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@ -103,7 +104,7 @@ public class LookasideChannelLbTest {
@Mock
private Helper helper;
@Mock
private AdsStreamCallback adsStreamCallback;
private LookasideChannelCallback lookasideChannelCallback;
@Mock
private LoadReportClient loadReportClient;
@Mock
@ -168,17 +169,17 @@ public class LookasideChannelLbTest {
doReturn(loadStatsStore).when(localityStore).getLoadStatsStore();
lookasideChannelLb = new LookasideChannelLb(
helper, adsStreamCallback, channel, loadReportClient, localityStore);
helper, lookasideChannelCallback, channel, loadReportClient, localityStore);
}
@Test
public void firstAndSecondEdsResponseReceived() {
verify(adsStreamCallback, never()).onWorking();
verify(lookasideChannelCallback, never()).onWorking();
verify(loadReportClient, never()).startLoadReporting(any(LoadReportCallback.class));
// first EDS response
serverResponseWriter.onNext(edsResponse);
verify(adsStreamCallback).onWorking();
verify(lookasideChannelCallback).onWorking();
ArgumentCaptor<LoadReportCallback> loadReportCallbackCaptor =
ArgumentCaptor.forClass(LoadReportCallback.class);
verify(loadReportClient).startLoadReporting(loadReportCallbackCaptor.capture());
@ -186,14 +187,14 @@ public class LookasideChannelLbTest {
// second EDS response
serverResponseWriter.onNext(edsResponse);
verify(adsStreamCallback, times(1)).onWorking();
verify(lookasideChannelCallback, times(1)).onWorking();
verify(loadReportClient, times(1)).startLoadReporting(any(LoadReportCallback.class));
verify(localityStore, never()).updateOobMetricsReportInterval(anyLong());
loadReportCallback.onReportResponse(1234);
verify(localityStore).updateOobMetricsReportInterval(1234);
verify(adsStreamCallback, never()).onError();
verify(lookasideChannelCallback, never()).onError();
lookasideChannelLb.shutdown();
}
@ -235,7 +236,7 @@ public class LookasideChannelLbTest {
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build());
verify(adsStreamCallback, never()).onAllDrop();
verify(lookasideChannelCallback, never()).onAllDrop();
verify(localityStore).updateDropPercentage(ImmutableList.of(
new DropOverload("cat_1", 300_00),
new DropOverload("cat_2", 45_00),
@ -270,12 +271,12 @@ public class LookasideChannelLbTest {
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build());
verify(adsStreamCallback).onAllDrop();
verify(lookasideChannelCallback).onAllDrop();
verify(localityStore).updateDropPercentage(ImmutableList.of(
new DropOverload("cat_1", 300_00),
new DropOverload("cat_2", 100_00_00)));
verify(adsStreamCallback, never()).onError();
verify(lookasideChannelCallback, never()).onError();
lookasideChannelLb.shutdown();
}
@ -348,13 +349,13 @@ public class LookasideChannelLbTest {
XdsLocality locality1 = XdsLocality.fromLocalityProto(localityProto1);
LocalityInfo localityInfo1 = new LocalityInfo(
ImmutableList.of(
new XdsComms.LbEndpoint(endpoint11),
new XdsComms.LbEndpoint(endpoint12)),
new ClusterLoadAssignmentData.LbEndpoint(endpoint11),
new ClusterLoadAssignmentData.LbEndpoint(endpoint12)),
1, 0);
LocalityInfo localityInfo2 = new LocalityInfo(
ImmutableList.of(
new XdsComms.LbEndpoint(endpoint21),
new XdsComms.LbEndpoint(endpoint22)),
new ClusterLoadAssignmentData.LbEndpoint(endpoint21),
new ClusterLoadAssignmentData.LbEndpoint(endpoint22)),
2, 0);
XdsLocality locality2 = XdsLocality.fromLocalityProto(localityProto2);
@ -364,16 +365,16 @@ public class LookasideChannelLbTest {
assertThat(localityEndpointsMappingCaptor.getValue()).containsExactly(
locality1, localityInfo1, locality2, localityInfo2).inOrder();
verify(adsStreamCallback, never()).onError();
verify(lookasideChannelCallback, never()).onError();
lookasideChannelLb.shutdown();
}
@Test
public void verifyRpcErrorPropagation() {
verify(adsStreamCallback, never()).onError();
verify(lookasideChannelCallback, never()).onError();
serverResponseWriter.onError(new RuntimeException());
verify(adsStreamCallback).onError();
verify(lookasideChannelCallback).onError();
}
@Test
@ -397,8 +398,8 @@ public class LookasideChannelLbTest {
// Simulates a syntactically incorrect EDS response.
serverResponseWriter.onNext(DiscoveryResponse.getDefaultInstance());
verify(loadReportClient, never()).startLoadReporting(any(LoadReportCallback.class));
verify(adsStreamCallback, never()).onWorking();
verify(adsStreamCallback, never()).onError();
verify(lookasideChannelCallback, never()).onWorking();
verify(lookasideChannelCallback, never()).onError();
// Simulate a syntactically correct EDS response.
DiscoveryResponse edsResponse =
@ -408,7 +409,7 @@ public class LookasideChannelLbTest {
.build();
serverResponseWriter.onNext(edsResponse);
verify(adsStreamCallback).onWorking();
verify(lookasideChannelCallback).onWorking();
ArgumentCaptor<LoadReportCallback> lrsCallbackCaptor = ArgumentCaptor.forClass(null);
verify(loadReportClient).startLoadReporting(lrsCallbackCaptor.capture());
@ -417,13 +418,15 @@ public class LookasideChannelLbTest {
// Simulate another EDS response from the same remote balancer.
serverResponseWriter.onNext(edsResponse);
verifyNoMoreInteractions(adsStreamCallback, loadReportClient);
verifyNoMoreInteractions(lookasideChannelCallback, loadReportClient);
// Simulate an EDS error response.
serverResponseWriter.onError(Status.ABORTED.asException());
verify(adsStreamCallback).onError();
verify(lookasideChannelCallback).onError();
verifyNoMoreInteractions(adsStreamCallback, loadReportClient);
verifyNoMoreInteractions(lookasideChannelCallback, loadReportClient);
verify(localityStore, times(1)).updateOobMetricsReportInterval(anyLong()); // only once
lookasideChannelLb.shutdown();
}
}

9
xds/src/test/java/io/grpc/xds/LookasideLbTest.java
View File

@ -33,8 +33,8 @@ import io.grpc.LoadBalancer.ResolvedAddresses;
import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.LoadBalancerRegistry;
import io.grpc.internal.JsonParser;
import io.grpc.xds.LookasideChannelLb.LookasideChannelCallback;
import io.grpc.xds.LookasideLb.LookasideChannelLbFactory;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@ -55,7 +55,7 @@ public class LookasideLbTest {
new LookasideChannelLbFactory() {
@Override
public LoadBalancer newLoadBalancer(
Helper helper, AdsStreamCallback adsCallback, String balancerName) {
Helper helper, LookasideChannelCallback lookasideChannelCallback, String balancerName) {
// just return a mock and record helper and balancer.
helpers.add(helper);
LoadBalancer balancer = mock(LoadBalancer.class);
@ -65,7 +65,8 @@ public class LookasideLbTest {
};
private LoadBalancer lookasideLb = new LookasideLb(
helper, mock(AdsStreamCallback.class), lookasideChannelLbFactory, new LoadBalancerRegistry());
helper, mock(LookasideChannelCallback.class), lookasideChannelLbFactory,
new LoadBalancerRegistry());
@Test
@ -163,7 +164,7 @@ public class LookasideLbTest {
public void handleResolvedAddress_createLbChannel()
throws Exception {
// Test balancer created with the default real LookasideChannelLbFactory
lookasideLb = new LookasideLb(helper, mock(AdsStreamCallback.class));
lookasideLb = new LookasideLb(helper, mock(LookasideChannelCallback.class));
String lbConfigRaw11 = "{'balancerName' : 'dns:///balancer1.example.com:8080'}"
.replace("'", "\"");
@SuppressWarnings("unchecked")

217
xds/src/test/java/io/grpc/xds/XdsCommsTest.java
View File

@ -30,12 +30,9 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.Any;
import com.google.protobuf.UInt32Value;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy;
import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
import io.envoyproxy.envoy.api.v2.core.Address;
@ -45,8 +42,6 @@ import io.envoyproxy.envoy.api.v2.endpoint.Endpoint;
import io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint;
import io.envoyproxy.envoy.api.v2.endpoint.LocalityLbEndpoints;
import io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase;
import io.envoyproxy.envoy.type.FractionalPercent;
import io.envoyproxy.envoy.type.FractionalPercent.DenominatorType;
import io.grpc.ChannelLogger;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
@ -62,23 +57,19 @@ import io.grpc.internal.FakeClock;
import io.grpc.internal.testing.StreamRecorder;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.GrpcCleanupRule;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.XdsComms.DropOverload;
import io.grpc.xds.XdsComms.LocalityInfo;
import io.grpc.xds.XdsComms2.AdsStreamCallback;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
/**
* Unit tests for {@link XdsComms}.
* Unit tests for {@link XdsComms2}.
*/
@RunWith(JUnit4.class)
public class XdsCommsTest {
@ -99,15 +90,11 @@ public class XdsCommsTest {
@Mock
private AdsStreamCallback adsStreamCallback;
@Mock
private LocalityStore localityStore;
@Mock
private BackoffPolicy.Provider backoffPolicyProvider;
@Mock
private BackoffPolicy backoffPolicy1;
@Mock
private BackoffPolicy backoffPolicy2;
@Captor
private ArgumentCaptor<ImmutableMap<XdsLocality, LocalityInfo>> localityEndpointsMappingCaptor;
private final FakeClock fakeClock = new FakeClock();
private final SynchronizationContext syncContext = new SynchronizationContext(
@ -123,7 +110,7 @@ public class XdsCommsTest {
private StreamObserver<DiscoveryResponse> responseWriter;
private ManagedChannel channel;
private XdsComms xdsComms;
private XdsComms2 xdsComms;
@Before
public void setUp() throws Exception {
@ -195,14 +182,14 @@ public class XdsCommsTest {
doReturn(backoffPolicy1, backoffPolicy2).when(backoffPolicyProvider).get();
doReturn(10L, 100L, 1000L).when(backoffPolicy1).nextBackoffNanos();
doReturn(20L, 200L).when(backoffPolicy2).nextBackoffNanos();
xdsComms = new XdsComms(
channel, helper, adsStreamCallback, localityStore, backoffPolicyProvider,
xdsComms = new XdsComms2(
channel, helper, adsStreamCallback, backoffPolicyProvider,
fakeClock.getStopwatchSupplier());
}
@Test
public void shutdownLbRpc_verifyChannelNotShutdown() throws Exception {
xdsComms.shutdownLbRpc("shutdown msg1");
xdsComms.shutdownLbRpc();
assertTrue(streamRecorder.awaitCompletion(1, TimeUnit.SECONDS));
assertEquals(Status.Code.CANCELLED, Status.fromThrowable(streamRecorder.getError()).getCode());
assertFalse(channel.isShutdown());
@ -210,13 +197,13 @@ public class XdsCommsTest {
@Test
public void cancel() throws Exception {
xdsComms.shutdownLbRpc("cause1");
xdsComms.shutdownLbRpc();
assertTrue(streamRecorder.awaitCompletion(1, TimeUnit.SECONDS));
assertEquals(Status.Code.CANCELLED, Status.fromThrowable(streamRecorder.getError()).getCode());
}
@Test
public void standardMode_sendEdsRequest_getEdsResponse_withNoDrop() {
public void handleEdsResponse() {
assertThat(streamRecorder.getValues()).hasSize(1);
DiscoveryRequest request = streamRecorder.getValues().get(0);
assertThat(request.getTypeUrl()).isEqualTo(EDS_TYPE_URL);
@ -266,8 +253,7 @@ public class XdsCommsTest {
.setAddress("addr31").setPortValue(31))))
.setLoadBalancingWeight(UInt32Value.of(31))
.build();
DiscoveryResponse edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
ClusterLoadAssignment clusterLoadAssignment = ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto1)
.addLbEndpoints(endpoint11)
@ -282,37 +268,16 @@ public class XdsCommsTest {
.setLocality(localityProto3)
.addLbEndpoints(endpoint3)
.setLoadBalancingWeight(UInt32Value.of(0)))
.build()))
.build();
DiscoveryResponse edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(clusterLoadAssignment))
.setTypeUrl(EDS_TYPE_URL)
.build();
responseWriter.onNext(edsResponse);
verify(adsStreamCallback).onWorking();
verify(adsStreamCallback).onEdsResponse(clusterLoadAssignment);
XdsLocality locality1 = XdsLocality.fromLocalityProto(localityProto1);
LocalityInfo localityInfo1 = new LocalityInfo(
ImmutableList.of(
new XdsComms.LbEndpoint(endpoint11),
new XdsComms.LbEndpoint(endpoint12)),
1,
0);
LocalityInfo localityInfo2 = new LocalityInfo(
ImmutableList.of(
new XdsComms.LbEndpoint(endpoint21),
new XdsComms.LbEndpoint(endpoint22)),
2,
0);
XdsLocality locality2 = XdsLocality.fromLocalityProto(localityProto2);
InOrder inOrder = inOrder(localityStore);
inOrder.verify(localityStore).updateDropPercentage(ImmutableList.<DropOverload>of());
inOrder.verify(localityStore).updateLocalityStore(localityEndpointsMappingCaptor.capture());
assertThat(localityEndpointsMappingCaptor.getValue()).containsExactly(
locality1, localityInfo1, locality2, localityInfo2).inOrder();
edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
ClusterLoadAssignment clusterLoadAssignment2 = ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto2)
.addLbEndpoints(endpoint21)
@ -323,155 +288,17 @@ public class XdsCommsTest {
.addLbEndpoints(endpoint11)
.addLbEndpoints(endpoint12)
.setLoadBalancingWeight(UInt32Value.of(1)))
.build()))
.build();
edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(clusterLoadAssignment2))
.setTypeUrl(EDS_TYPE_URL)
.build();
responseWriter.onNext(edsResponse);
verify(adsStreamCallback, times(1)).onWorking();
verify(adsStreamCallback).onEdsResponse(clusterLoadAssignment2);
verifyNoMoreInteractions(adsStreamCallback);
inOrder.verify(localityStore).updateDropPercentage(ImmutableList.<DropOverload>of());
inOrder.verify(localityStore).updateLocalityStore(localityEndpointsMappingCaptor.capture());
assertThat(localityEndpointsMappingCaptor.getValue()).containsExactly(
locality2, localityInfo2, locality1, localityInfo1).inOrder();
xdsComms.shutdownLbRpc("End test");
}
@Test
public void standardMode_sendEdsRequest_getEdsResponse_withDrops() {
Locality localityProto1 = Locality.newBuilder()
.setRegion("region1").setZone("zone1").setSubZone("subzone1").build();
LbEndpoint endpoint11 = LbEndpoint.newBuilder()
.setEndpoint(Endpoint.newBuilder()
.setAddress(Address.newBuilder()
.setSocketAddress(SocketAddress.newBuilder()
.setAddress("addr11").setPortValue(11))))
.setLoadBalancingWeight(UInt32Value.of(11))
.build();
Locality localityProto2 = Locality.newBuilder()
.setRegion("region2").setZone("zone2").setSubZone("subzone2").build();
LbEndpoint endpoint21 = LbEndpoint.newBuilder()
.setEndpoint(Endpoint.newBuilder()
.setAddress(Address.newBuilder()
.setSocketAddress(SocketAddress.newBuilder()
.setAddress("addr21").setPortValue(21))))
.setLoadBalancingWeight(UInt32Value.of(21))
.build();
DiscoveryResponse edsResponseWithDrops = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto2)
.addLbEndpoints(endpoint21)
.setLoadBalancingWeight(UInt32Value.of(2)))
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto1)
.addLbEndpoints(endpoint11)
.setLoadBalancingWeight(UInt32Value.of(1)))
.setPolicy(Policy.newBuilder()
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("throttle")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(123).setDenominator(DenominatorType.MILLION).build())
.build())
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("lb")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(456).setDenominator(DenominatorType.TEN_THOUSAND).build())
.build())
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("fake_category")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(78).setDenominator(DenominatorType.HUNDRED).build())
.build())
.build())
.build()))
.setTypeUrl(EDS_TYPE_URL)
.build();
responseWriter.onNext(edsResponseWithDrops);
verify(adsStreamCallback).onWorking();
verifyNoMoreInteractions(adsStreamCallback);
InOrder inOrder = inOrder(localityStore);
inOrder.verify(localityStore).updateDropPercentage(ImmutableList.of(
new DropOverload("throttle", 123),
new DropOverload("lb", 456_00),
new DropOverload("fake_category", 78_00_00)));
inOrder.verify(localityStore).updateLocalityStore(localityEndpointsMappingCaptor.capture());
XdsLocality locality1 = XdsLocality.fromLocalityProto(localityProto1);
LocalityInfo localityInfo1 = new LocalityInfo(
ImmutableList.of(new XdsComms.LbEndpoint(endpoint11)), 1, 0);
LocalityInfo localityInfo2 = new LocalityInfo(
ImmutableList.of(new XdsComms.LbEndpoint(endpoint21)), 2, 0);
XdsLocality locality2 = XdsLocality.fromLocalityProto(localityProto2);
assertThat(localityEndpointsMappingCaptor.getValue()).containsExactly(
locality2, localityInfo2, locality1, localityInfo1).inOrder();
DiscoveryResponse edsResponseWithAllDrops = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto2)
.addLbEndpoints(endpoint21)
.setLoadBalancingWeight(UInt32Value.of(2)))
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto1)
.addLbEndpoints(endpoint11)
.setLoadBalancingWeight(UInt32Value.of(1)))
.setPolicy(Policy.newBuilder()
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("throttle")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(123).setDenominator(DenominatorType.MILLION).build())
.build())
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("lb")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(456).setDenominator(DenominatorType.TEN_THOUSAND).build())
.build())
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("fake_category")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(789).setDenominator(DenominatorType.HUNDRED).build())
.build())
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("fake_category_2")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(78).setDenominator(DenominatorType.HUNDRED).build())
.build())
.build())
.build()))
.setTypeUrl(EDS_TYPE_URL)
.build();
responseWriter.onNext(edsResponseWithAllDrops);
verify(adsStreamCallback, times(1)).onWorking();
verify(adsStreamCallback).onAllDrop();
verify(adsStreamCallback, never()).onError();
inOrder.verify(localityStore).updateDropPercentage(ImmutableList.of(
new DropOverload("throttle", 123),
new DropOverload("lb", 456_00),
new DropOverload("fake_category", 1000_000)));
inOrder.verify(localityStore).updateLocalityStore(localityEndpointsMappingCaptor.capture());
assertThat(localityEndpointsMappingCaptor.getValue()).containsExactly(
locality2, localityInfo2, locality1, localityInfo1).inOrder();
xdsComms.shutdownLbRpc("End test");
xdsComms.shutdownLbRpc();
}
@Test
@ -500,7 +327,7 @@ public class XdsCommsTest {
* Verify retry is scheduled. Verify the 6th PRC starts after backoff.
*
* <p>The 6th RPC fails with response observer onError() without receiving initial response.
* Verify retry is scheduled. Call {@link XdsComms#shutdownLbRpc(String)}, verify retry timer is
* Verify retry is scheduled. Call {@link XdsComms2#shutdownLbRpc()}, verify retry timer is
* cancelled.
*/
@Test
@ -653,7 +480,7 @@ public class XdsCommsTest {
assertEquals(1, fakeClock.numPendingTasks(LB_RPC_RETRY_TASK_FILTER));
// Shutdown cancels retry
xdsComms.shutdownLbRpc("shutdown");
xdsComms.shutdownLbRpc();
assertEquals(0, fakeClock.numPendingTasks(LB_RPC_RETRY_TASK_FILTER));
}
@ -666,6 +493,6 @@ public class XdsCommsTest {
xdsComms.refreshAdsStream();
assertEquals(0, fakeClock.numPendingTasks(LB_RPC_RETRY_TASK_FILTER));
xdsComms.shutdownLbRpc("End test");
xdsComms.shutdownLbRpc();
}
}

156
xds/src/test/java/io/grpc/xds/XdsLbStateTest.java
View File

@ -1,156 +0,0 @@
/*
* Copyright 2019 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.truth.Truth.assertThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import com.google.common.collect.ImmutableList;
import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
import io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase;
import io.grpc.Attributes;
import io.grpc.ChannelLogger;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer.Helper;
import io.grpc.ManagedChannel;
import io.grpc.SynchronizationContext;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.internal.BackoffPolicy;
import io.grpc.internal.FakeClock;
import io.grpc.internal.testing.StreamRecorder;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.GrpcCleanupRule;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
/**
* Unit tests for {@link XdsLbState}.
*/
@RunWith(JUnit4.class)
public class XdsLbStateTest {
private static final String BALANCER_NAME = "balancerName";
@Rule
public final GrpcCleanupRule cleanupRule = new GrpcCleanupRule();
@Mock
private Helper helper;
@Mock
private AdsStreamCallback adsStreamCallback;
@Mock
private LocalityStore localityStore;
@Mock
private BackoffPolicy.Provider backoffPolicyProvider;
private final FakeClock fakeClock = new FakeClock();
private final SynchronizationContext syncContext = new SynchronizationContext(
new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
throw new AssertionError(e);
}
});
private final StreamRecorder<DiscoveryRequest> streamRecorder = StreamRecorder.create();
private ManagedChannel channel;
private XdsLbState xdsLbState;
@Before
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
doReturn(syncContext).when(helper).getSynchronizationContext();
doReturn(fakeClock.getScheduledExecutorService()).when(helper).getScheduledExecutorService();
doReturn("fake_authority").when(helper).getAuthority();
doReturn(mock(ChannelLogger.class)).when(helper).getChannelLogger();
String serverName = InProcessServerBuilder.generateName();
AggregatedDiscoveryServiceImplBase serviceImpl = new AggregatedDiscoveryServiceImplBase() {
@Override
public StreamObserver<DiscoveryRequest> streamAggregatedResources(
final StreamObserver<DiscoveryResponse> responseObserver) {
return new StreamObserver<DiscoveryRequest>() {
@Override
public void onNext(DiscoveryRequest value) {
streamRecorder.onNext(value);
}
@Override
public void onError(Throwable t) {
streamRecorder.onError(t);
}
@Override
public void onCompleted() {
streamRecorder.onCompleted();
responseObserver.onCompleted();
}
};
}
};
cleanupRule.register(
InProcessServerBuilder
.forName(serverName)
.addService(serviceImpl)
.directExecutor()
.build()
.start());
channel =
cleanupRule.register(InProcessChannelBuilder.forName(serverName).directExecutor().build());
doReturn(channel).when(helper).createResolvingOobChannel(BALANCER_NAME);
xdsLbState = new XdsLbState(
BALANCER_NAME, null, helper, localityStore, channel, adsStreamCallback,
backoffPolicyProvider);
}
@Test
public void shutdownResetsLocalityStore() {
xdsLbState.shutdownAndReleaseChannel("Client shutdown");
verify(localityStore).reset();
}
@Test
public void shutdownDoesNotTearDownChannel() {
ManagedChannel lbChannel = xdsLbState.shutdownAndReleaseChannel("Client shutdown");
assertThat(lbChannel).isSameInstanceAs(channel);
assertThat(channel.isShutdown()).isFalse();
}
@Test
public void handleResolvedAddressGroupsTriggerEds() throws Exception {
xdsLbState.handleResolvedAddressGroups(
ImmutableList.<EquivalentAddressGroup>of(), Attributes.EMPTY);
assertThat(streamRecorder.firstValue().get().getTypeUrl())
.isEqualTo("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment");
xdsLbState.shutdownAndReleaseChannel("End test");
}
}

23
xds/src/test/java/io/grpc/xds/XdsLoadBalancer2Test.java
View File

@ -40,7 +40,7 @@ import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.internal.FakeClock;
import io.grpc.xds.XdsComms.AdsStreamCallback;
import io.grpc.xds.LookasideChannelLb.LookasideChannelCallback;
import io.grpc.xds.XdsLoadBalancer2.LookasideLbFactory;
import java.util.ArrayList;
import java.util.List;
@ -75,7 +75,7 @@ public class XdsLoadBalancer2Test {
@Mock
private Helper helper;
private LoadBalancer xdsLoadBalancer;
private AdsStreamCallback adsCallback;
private LookasideChannelCallback lookasideChannelCallback;
private Helper lookasideLbHelper;
private final List<LoadBalancer> lookasideLbs = new ArrayList<>();
@ -89,10 +89,11 @@ public class XdsLoadBalancer2Test {
public void setUp() {
LookasideLbFactory lookasideLbFactory = new LookasideLbFactory() {
@Override
public LoadBalancer newLoadBalancer(Helper helper, AdsStreamCallback adsCallback) {
public LoadBalancer newLoadBalancer(
Helper helper, LookasideChannelCallback lookasideChannelCallback) {
// just return a mock and record the input and output
lookasideLbHelper = helper;
XdsLoadBalancer2Test.this.adsCallback = adsCallback;
XdsLoadBalancer2Test.this.lookasideChannelCallback = lookasideChannelCallback;
LoadBalancer lookasideLb = mock(LoadBalancer.class);
lookasideLbs.add(lookasideLb);
return lookasideLb;
@ -141,7 +142,7 @@ public class XdsLoadBalancer2Test {
public void timeoutAtStartup_expectUseFallback_thenBackendReady_expectExitFallback() {
verifyNotInFallbackMode();
fakeClock.forwardTime(9, TimeUnit.SECONDS);
adsCallback.onWorking();
lookasideChannelCallback.onWorking();
verifyNotInFallbackMode();
fakeClock.forwardTime(1, TimeUnit.SECONDS);
verifyInFallbackMode();
@ -161,7 +162,7 @@ public class XdsLoadBalancer2Test {
verifyNotInFallbackMode();
assertThat(fakeClock.getPendingTasks()).hasSize(1);
adsCallback.onWorking();
lookasideChannelCallback.onWorking();
SubchannelPicker subchannelPicker = mock(SubchannelPicker.class);
lookasideLbHelper.updateBalancingState(READY, subchannelPicker);
verify(helper).updateBalancingState(READY, subchannelPicker);
@ -176,7 +177,7 @@ public class XdsLoadBalancer2Test {
verifyNotInFallbackMode();
assertThat(fakeClock.getPendingTasks()).hasSize(1);
adsCallback.onAllDrop();
lookasideChannelCallback.onAllDrop();
assertThat(fakeClock.getPendingTasks()).isEmpty();
verifyNotInFallbackMode();
@ -188,7 +189,7 @@ public class XdsLoadBalancer2Test {
verifyNotInFallbackMode();
assertThat(fakeClock.getPendingTasks()).hasSize(1);
adsCallback.onError();
lookasideChannelCallback.onError();
verifyInFallbackMode();
assertThat(fallbackLbs).hasSize(1);
@ -198,8 +199,8 @@ public class XdsLoadBalancer2Test {
public void lookasideChannelSeeingEdsResponseThenFailsBeforeTimeoutAtStartup() {
verifyNotInFallbackMode();
assertThat(fakeClock.getPendingTasks()).hasSize(1);
adsCallback.onWorking();
adsCallback.onError();
lookasideChannelCallback.onWorking();
lookasideChannelCallback.onError();
verifyNotInFallbackMode();
fakeClock.forwardTime(10, TimeUnit.SECONDS);
@ -220,7 +221,7 @@ public class XdsLoadBalancer2Test {
.build();
xdsLoadBalancer.handleResolvedAddresses(resolvedAddresses);
adsCallback.onError();
lookasideChannelCallback.onError();
LoadBalancer fallbackLb = Iterables.getLast(fallbackLbs);
verify(fallbackLb).handleResolvedAddresses(same(resolvedAddresses));
}

812
xds/src/test/java/io/grpc/xds/XdsLoadBalancerTest.java
View File

@ -1,812 +0,0 @@
/*
* Copyright 2019 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.truth.Truth.assertThat;
import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.READY;
import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
import static io.grpc.LoadBalancer.ATTR_LOAD_BALANCING_CONFIG;
import static io.grpc.xds.XdsSubchannelPickers.BUFFER_PICKER;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import com.google.protobuf.Any;
import com.google.protobuf.UInt32Value;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy;
import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
import io.envoyproxy.envoy.api.v2.core.Address;
import io.envoyproxy.envoy.api.v2.core.Locality;
import io.envoyproxy.envoy.api.v2.core.SocketAddress;
import io.envoyproxy.envoy.api.v2.endpoint.Endpoint;
import io.envoyproxy.envoy.api.v2.endpoint.LbEndpoint;
import io.envoyproxy.envoy.api.v2.endpoint.LocalityLbEndpoints;
import io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase;
import io.envoyproxy.envoy.type.FractionalPercent;
import io.envoyproxy.envoy.type.FractionalPercent.DenominatorType;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.ConnectivityState;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.PickSubchannelArgs;
import io.grpc.LoadBalancer.ResolvedAddresses;
import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.ManagedChannel;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.SynchronizationContext;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.internal.BackoffPolicy;
import io.grpc.internal.FakeClock;
import io.grpc.internal.FakeClock.TaskFilter;
import io.grpc.internal.JsonParser;
import io.grpc.internal.testing.StreamRecorder;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.GrpcCleanupRule;
import io.grpc.xds.XdsSubchannelPickers.ErrorPicker;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
/**
* Unit tests for {@link XdsLoadBalancer}.
*/
@RunWith(JUnit4.class)
public class XdsLoadBalancerTest {
@Rule
public final GrpcCleanupRule cleanupRule = new GrpcCleanupRule();
@Mock
private Helper helper;
@Mock
private LoadBalancer fallbackBalancer1;
@Mock
private LoadBalancer fakeBalancer2;
@Mock
private BackoffPolicy.Provider backoffPolicyProvider;
private XdsLoadBalancer lb;
private final FakeClock fakeClock = new FakeClock();
private final StreamRecorder<DiscoveryRequest> streamRecorder = StreamRecorder.create();
private final LoadBalancerRegistry lbRegistry = new LoadBalancerRegistry();
private Helper fallbackHelper1;
private final LoadBalancerProvider lbProvider1 = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return "fallback_1";
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
fallbackHelper1 = helper;
return fallbackBalancer1;
}
};
private final LoadBalancerProvider lbProvider2 = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return "supported_2";
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
return fakeBalancer2;
}
};
private final Locality localityProto1 = Locality.newBuilder()
.setRegion("region1").setZone("zone1").setSubZone("subzone1").build();
private final LbEndpoint endpoint11 = LbEndpoint.newBuilder()
.setEndpoint(Endpoint.newBuilder()
.setAddress(Address.newBuilder()
.setSocketAddress(SocketAddress.newBuilder()
.setAddress("addr11").setPortValue(11))))
.setLoadBalancingWeight(UInt32Value.of(11))
.build();
private final DiscoveryResponse edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto1)
.addLbEndpoints(endpoint11)
.setLoadBalancingWeight(UInt32Value.of(1)))
.build()))
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build();
private Helper childHelper;
@Mock
private LoadBalancer childBalancer;
private final LoadBalancerProvider roundRobin = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return "round_robin";
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
childHelper = helper;
return childBalancer;
}
};
private final SynchronizationContext syncContext = new SynchronizationContext(
new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
throw new AssertionError(e);
}
});
private final TaskFilter fallbackTaskFilter = new TaskFilter() {
@Override
public boolean shouldAccept(Runnable runnable) {
return runnable.toString().contains("FallbackTask");
}
};
private ManagedChannel oobChannel1;
private ManagedChannel oobChannel2;
private ManagedChannel oobChannel3;
private StreamObserver<DiscoveryResponse> serverResponseWriter;
@Before
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
lbRegistry.register(lbProvider1);
lbRegistry.register(lbProvider2);
lbRegistry.register(roundRobin);
lb = new XdsLoadBalancer(helper, lbRegistry, backoffPolicyProvider);
doReturn(syncContext).when(helper).getSynchronizationContext();
doReturn(fakeClock.getScheduledExecutorService()).when(helper).getScheduledExecutorService();
doReturn(mock(ChannelLogger.class)).when(helper).getChannelLogger();
doReturn("fake_authority").when(helper).getAuthority();
String serverName = InProcessServerBuilder.generateName();
AggregatedDiscoveryServiceImplBase serviceImpl = new AggregatedDiscoveryServiceImplBase() {
@Override
public StreamObserver<DiscoveryRequest> streamAggregatedResources(
final StreamObserver<DiscoveryResponse> responseObserver) {
serverResponseWriter = responseObserver;
return new StreamObserver<DiscoveryRequest>() {
@Override
public void onNext(DiscoveryRequest value) {
streamRecorder.onNext(value);
}
@Override
public void onError(Throwable t) {
streamRecorder.onError(t);
}
@Override
public void onCompleted() {
streamRecorder.onCompleted();
responseObserver.onCompleted();
}
};
}
};
cleanupRule.register(
InProcessServerBuilder
.forName(serverName)
.directExecutor()
.addService(serviceImpl)
.build()
.start());
InProcessChannelBuilder channelBuilder =
InProcessChannelBuilder.forName(serverName).directExecutor();
oobChannel1 = mock(
ManagedChannel.class,
delegatesTo(cleanupRule.register(channelBuilder.build())));
oobChannel2 = mock(
ManagedChannel.class,
delegatesTo(cleanupRule.register(channelBuilder.build())));
oobChannel3 = mock(
ManagedChannel.class,
delegatesTo(cleanupRule.register(channelBuilder.build())));
doReturn(oobChannel1).doReturn(oobChannel2).doReturn(oobChannel3)
.when(helper).createResolvingOobChannel(anyString());
// To write less tedious code for tests, allow fallbackBalancer to handle empty address list.
doReturn(true).when(fallbackBalancer1).canHandleEmptyAddressListFromNameResolution();
}
@After
public void tearDown() {
lb.shutdown();
}
@Test
public void canHandleEmptyAddressListFromNameResolution() {
assertTrue(lb.canHandleEmptyAddressListFromNameResolution());
}
@Test
public void resolverEvent_standardModeToStandardMode() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"unsupported\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
Attributes attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
XdsLbState xdsLbState1 = lb.getXdsLbStateForTest();
assertThat(xdsLbState1.childPolicy).isNull();
verify(helper).createResolvingOobChannel(anyString());
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig2 = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig2).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
XdsLbState xdsLbState2 = lb.getXdsLbStateForTest();
assertThat(xdsLbState2.childPolicy).isNull();
assertThat(xdsLbState2).isSameInstanceAs(xdsLbState1);
// verify oobChannel is unchanged
verify(helper).createResolvingOobChannel(anyString());
// verify ADS stream is unchanged
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
}
@Test
public void resolverEvent_standardModeToCustomMode() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"unsupported\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
Attributes attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
verify(helper).createResolvingOobChannel(anyString());
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"supported_2\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig2 = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig2).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
assertThat(lb.getXdsLbStateForTest().childPolicy).isNotNull();
// verify oobChannel is unchanged
verify(helper).createResolvingOobChannel(anyString());
// verify ADS stream is reset
verify(oobChannel1, times(2))
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
}
@Test
public void resolverEvent_customModeToStandardMode() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"supported_2\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
Attributes attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
verify(helper).createResolvingOobChannel(anyString());
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
assertThat(lb.getXdsLbStateForTest().childPolicy).isNotNull();
lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"unsupported\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig2 = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig2).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
assertThat(lb.getXdsLbStateForTest().childPolicy).isNull();
// verify oobChannel is unchanged
verify(helper).createResolvingOobChannel(anyString());
// verify ADS stream is reset
verify(oobChannel1, times(2))
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
}
@Test
public void resolverEvent_customModeToCustomMode() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"supported_2\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
Attributes attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
assertThat(lb.getXdsLbStateForTest().childPolicy).isNotNull();
verify(helper).createResolvingOobChannel(anyString());
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"fallback_1\" : {\"key\" : \"val\"}}, {\"unfallback_1\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig2 = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig2).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
assertThat(lb.getXdsLbStateForTest().childPolicy).isNotNull();
// verify oobChannel is unchanged
verify(helper).createResolvingOobChannel(anyString());
// verify ADS stream is reset
verify(oobChannel1, times(2))
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
}
@Test
public void resolverEvent_balancerNameChange() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"unsupported\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
Attributes attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
verify(helper).createResolvingOobChannel(anyString());
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8443\","
+ "\"childPolicy\" : [{\"fallback_1\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig2 = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig2).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
assertThat(lb.getXdsLbStateForTest().childPolicy).isNotNull();
// verify oobChannel is unchanged
verify(helper, times(2)).createResolvingOobChannel(anyString());
verify(oobChannel1)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
verify(oobChannel2)
.newCall(ArgumentMatchers.<MethodDescriptor<?, ?>>any(),
ArgumentMatchers.<CallOptions>any());
verifyNoMoreInteractions(oobChannel3);
}
@Test
public void resolutionErrorAtStartup() {
lb.handleNameResolutionError(Status.UNAVAILABLE);
assertNull(childHelper);
assertNull(fallbackHelper1);
verify(helper).updateBalancingState(same(TRANSIENT_FAILURE), isA(ErrorPicker.class));
}
@Test
public void resolutionErrorAtFallback() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
// let fallback timer expire
assertThat(fakeClock.forwardTime(10, TimeUnit.SECONDS)).isEqualTo(1);
ArgumentCaptor<ResolvedAddresses> captor = ArgumentCaptor.forClass(ResolvedAddresses.class);
verify(fallbackBalancer1).handleResolvedAddresses(captor.capture());
assertThat(captor.getValue().getAttributes().get(ATTR_LOAD_BALANCING_CONFIG))
.containsExactly("fallback_1_option", "yes");
Status status = Status.UNAVAILABLE.withDescription("resolution error");
lb.handleNameResolutionError(status);
verify(fallbackBalancer1).handleNameResolutionError(status);
}
@Test
public void fallback_AdsNotWorkingYetTimerExpired() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
assertNull(childHelper);
assertNull(fallbackHelper1);
assertThat(fakeClock.forwardTime(10, TimeUnit.SECONDS)).isEqualTo(1);
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).isEmpty();
assertNull(childHelper);
assertNotNull(fallbackHelper1);
ArgumentCaptor<ResolvedAddresses> captor = ArgumentCaptor.forClass(ResolvedAddresses.class);
verify(fallbackBalancer1).handleResolvedAddresses(captor.capture());
assertThat(captor.getValue().getAttributes().get(ATTR_LOAD_BALANCING_CONFIG))
.containsExactly("fallback_1_option", "yes");
SubchannelPicker picker = mock(SubchannelPicker.class);
fallbackHelper1.updateBalancingState(CONNECTING, picker);
verify(helper).updateBalancingState(CONNECTING, picker);
}
@Test
public void allDropCancelsFallbackTimer() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
DiscoveryResponse edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto1)
.addLbEndpoints(endpoint11)
.setLoadBalancingWeight(UInt32Value.of(1)))
.setPolicy(Policy.newBuilder()
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("throttle")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(100).setDenominator(DenominatorType.HUNDRED).build())
.build()))
.build()))
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build();
serverResponseWriter.onNext(edsResponse);
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).isEmpty();
assertNotNull(childHelper);
assertNull(fallbackHelper1);
verify(fallbackBalancer1, never()).handleResolvedAddresses(any(ResolvedAddresses.class));
}
@Test
public void allDropExitFallbackMode() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
// let the fallback timer expire
assertThat(fakeClock.forwardTime(10, TimeUnit.SECONDS)).isEqualTo(1);
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).isEmpty();
assertNull(childHelper);
assertNotNull(fallbackHelper1);
// receives EDS response with 100% drop
DiscoveryResponse edsResponse = DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.newBuilder()
.addEndpoints(LocalityLbEndpoints.newBuilder()
.setLocality(localityProto1)
.addLbEndpoints(endpoint11)
.setLoadBalancingWeight(UInt32Value.of(1)))
.setPolicy(Policy.newBuilder()
.addDropOverloads(
io.envoyproxy.envoy.api.v2.ClusterLoadAssignment.Policy.DropOverload
.newBuilder()
.setCategory("throttle")
.setDropPercentage(FractionalPercent.newBuilder()
.setNumerator(100).setDenominator(DenominatorType.HUNDRED).build())
.build()))
.build()))
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build();
serverResponseWriter.onNext(edsResponse);
verify(fallbackBalancer1).shutdown();
assertNotNull(childHelper);
ArgumentCaptor<SubchannelPicker> subchannelPickerCaptor =
ArgumentCaptor.forClass(SubchannelPicker.class);
verify(helper).updateBalancingState(same(CONNECTING), subchannelPickerCaptor.capture());
assertThat(subchannelPickerCaptor.getValue().pickSubchannel(mock(PickSubchannelArgs.class))
.isDrop()).isTrue();
}
@Test
public void fallback_ErrorWithoutReceivingEdsResponse() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
assertNull(childHelper);
assertNull(fallbackHelper1);
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).hasSize(1);
serverResponseWriter.onError(new Exception("fake error"));
// goes to fallback-at-startup mode immediately
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).isEmpty();
assertNull(childHelper);
assertNotNull(fallbackHelper1);
// verify fallback balancer is working
ArgumentCaptor<ResolvedAddresses> captor = ArgumentCaptor.forClass(ResolvedAddresses.class);
verify(fallbackBalancer1).handleResolvedAddresses(captor.capture());
assertThat(captor.getValue().getAttributes().get(ATTR_LOAD_BALANCING_CONFIG))
.containsExactly("fallback_1_option", "yes");
SubchannelPicker picker = mock(SubchannelPicker.class);
fallbackHelper1.updateBalancingState(CONNECTING, picker);
verify(helper).updateBalancingState(CONNECTING, picker);
}
@Test
public void fallback_EdsResponseReceivedThenErrorBeforeBackendReady() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
serverResponseWriter.onNext(edsResponse);
assertNotNull(childHelper);
assertNull(fallbackHelper1);
verify(helper).updateBalancingState(CONNECTING, BUFFER_PICKER);
serverResponseWriter.onError(new Exception("fake error"));
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).hasSize(1);
// verify fallback balancer is not started
assertNull(fallbackHelper1);
verify(fallbackBalancer1, never()).handleResolvedAddresses(any(ResolvedAddresses.class));
SubchannelPicker picker1 = mock(SubchannelPicker.class);
childHelper.updateBalancingState(CONNECTING, picker1);
verify(helper, times(2)).updateBalancingState(CONNECTING, BUFFER_PICKER);
childHelper.updateBalancingState(TRANSIENT_FAILURE, picker1);
verify(helper).updateBalancingState(same(TRANSIENT_FAILURE), isA(ErrorPicker.class));
assertThat(fakeClock.forwardTime(10, TimeUnit.SECONDS)).isEqualTo(1);
// verify fallback balancer is working
ArgumentCaptor<ResolvedAddresses> captor = ArgumentCaptor.forClass(ResolvedAddresses.class);
assertNotNull(fallbackHelper1);
verify(fallbackBalancer1).handleResolvedAddresses(captor.capture());
assertThat(captor.getValue().getAttributes().get(ATTR_LOAD_BALANCING_CONFIG))
.containsExactly("fallback_1_option", "yes");
SubchannelPicker picker2 = mock(SubchannelPicker.class);
childHelper.updateBalancingState(CONNECTING, picker2);
// verify childHelper no more delegates updateBalancingState to parent helper
verify(helper, times(3)).updateBalancingState(
any(ConnectivityState.class), any(SubchannelPicker.class));
SubchannelPicker picker3 = mock(SubchannelPicker.class);
fallbackHelper1.updateBalancingState(CONNECTING, picker3);
verify(helper).updateBalancingState(CONNECTING, picker3);
SubchannelPicker picker4 = mock(SubchannelPicker.class);
childHelper.updateBalancingState(READY, picker4);
verify(fallbackBalancer1).shutdown();
verify(helper).updateBalancingState(same(READY), isA(InterLocalityPicker.class));
}
@Test
public void fallback_AdsErrorWithActiveSubchannel() throws Exception {
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallback1Attributes())
.build());
serverResponseWriter.onNext(edsResponse);
assertNotNull(childHelper);
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).hasSize(1);
assertNull(fallbackHelper1);
verify(helper).updateBalancingState(CONNECTING, BUFFER_PICKER);
childHelper.updateBalancingState(READY, mock(SubchannelPicker.class));
verify(helper).updateBalancingState(same(READY), isA(InterLocalityPicker.class));
assertThat(fakeClock.getPendingTasks(fallbackTaskFilter)).isEmpty();
serverResponseWriter.onError(new Exception("fake error"));
assertNull(fallbackHelper1);
verify(fallbackBalancer1, never()).handleResolvedAddresses(
any(ResolvedAddresses.class));
// verify childHelper still delegates updateBalancingState to parent helper
childHelper.updateBalancingState(CONNECTING, mock(SubchannelPicker.class));
verify(helper, times(2)).updateBalancingState(CONNECTING, BUFFER_PICKER);
}
private static Attributes standardModeWithFallback1Attributes() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"fallbackPolicy\" : [{\"fallback_1\" : { \"fallback_1_option\" : \"yes\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
return Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
}
@Test
public void shutdown_cleanupTimers() throws Exception {
String lbConfigRaw = "{ "
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"unsupported\" : {\"key\" : \"val\"}}, {\"unsupported_2\" : {}}],"
+ "\"fallbackPolicy\" : [{\"unsupported\" : {}}, {\"fallback_1\" : {\"key\" : \"val\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
Attributes attrs = Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
lb.handleResolvedAddresses(
ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(attrs)
.build());
assertThat(fakeClock.getPendingTasks()).isNotEmpty();
lb.shutdown();
assertThat(fakeClock.getPendingTasks()).isEmpty();
}
}

416
xds/src/test/java/io/grpc/xds/XdsLoadBalancerWithLrsTest.java
View File

@ -1,416 +0,0 @@
/*
* Copyright 2019 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.truth.Truth.assertThat;
import static io.grpc.LoadBalancer.ATTR_LOAD_BALANCING_CONFIG;
import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.same;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
import com.google.protobuf.Any;
import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
import io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase;
import io.grpc.Attributes;
import io.grpc.ChannelLogger;
import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer.ResolvedAddresses;
import io.grpc.LoadBalancerProvider;
import io.grpc.LoadBalancerRegistry;
import io.grpc.ManagedChannel;
import io.grpc.Status;
import io.grpc.Status.Code;
import io.grpc.SynchronizationContext;
import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.internal.BackoffPolicy;
import io.grpc.internal.FakeClock;
import io.grpc.internal.JsonParser;
import io.grpc.internal.testing.StreamRecorder;
import io.grpc.stub.StreamObserver;
import io.grpc.testing.GrpcCleanupRule;
import io.grpc.xds.LoadReportClient.LoadReportCallback;
import io.grpc.xds.LoadReportClientImpl.LoadReportClientFactory;
import io.grpc.xds.XdsLoadBalancer.FallbackManager;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
/**
* Unit tests for {@link XdsLoadBalancer}, especially for interactions between
* {@link XdsLoadBalancer} and {@link LoadReportClient}.
*/
@RunWith(JUnit4.class)
public class XdsLoadBalancerWithLrsTest {
private static final String SERVICE_AUTHORITY = "test authority";
@Rule
public final GrpcCleanupRule cleanupRule = new GrpcCleanupRule();
private final SynchronizationContext syncContext = new SynchronizationContext(
new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
throw new AssertionError(e);
}
});
@Mock
private Helper helper;
@Mock
private BackoffPolicy.Provider backoffPolicyProvider;
@Mock
private LocalityStore localityStore;
@Mock
private LoadReportClientFactory lrsClientFactory;
@Mock
private LoadReportClient lrsClient;
@Mock
private LoadStatsStore loadStatsStore;
@Mock
private LoadBalancer fallbackBalancer;
@Mock
private LoadBalancer mockBalancer;
private final FakeClock fakeClock = new FakeClock();
private final LoadBalancerRegistry lbRegistry = new LoadBalancerRegistry();
private final StreamRecorder<DiscoveryRequest> streamRecorder = StreamRecorder.create();
private final LoadBalancerProvider fallBackLbProvider = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return "fallback";
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
fallBackLbHelper = helper;
return fallbackBalancer;
}
};
private final LoadBalancerProvider lbProvider = new LoadBalancerProvider() {
@Override
public boolean isAvailable() {
return true;
}
@Override
public int getPriority() {
return 5;
}
@Override
public String getPolicyName() {
return "supported";
}
@Override
public LoadBalancer newLoadBalancer(Helper helper) {
return mockBalancer;
}
};
private Helper fallBackLbHelper;
private StreamObserver<DiscoveryResponse> serverResponseWriter;
private ManagedChannel oobChannel1;
private ManagedChannel oobChannel2;
private ManagedChannel oobChannel3;
private LoadBalancer xdsLoadBalancer;
@Before
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
String serverName = InProcessServerBuilder.generateName();
AggregatedDiscoveryServiceImplBase serviceImpl = new AggregatedDiscoveryServiceImplBase() {
@Override
public StreamObserver<DiscoveryRequest> streamAggregatedResources(
final StreamObserver<DiscoveryResponse> responseObserver) {
serverResponseWriter = responseObserver;
return new StreamObserver<DiscoveryRequest>() {
@Override
public void onNext(DiscoveryRequest value) {
streamRecorder.onNext(value);
}
@Override
public void onError(Throwable t) {
streamRecorder.onError(t);
}
@Override
public void onCompleted() {
streamRecorder.onCompleted();
responseObserver.onCompleted();
}
};
}
};
cleanupRule.register(
InProcessServerBuilder
.forName(serverName)
.directExecutor()
.addService(serviceImpl)
.build()
.start());
InProcessChannelBuilder channelBuilder =
InProcessChannelBuilder.forName(serverName).directExecutor();
oobChannel1 = mock(
ManagedChannel.class,
delegatesTo(cleanupRule.register(channelBuilder.build())));
oobChannel2 = mock(
ManagedChannel.class,
delegatesTo(cleanupRule.register(channelBuilder.build())));
oobChannel3 = mock(
ManagedChannel.class,
delegatesTo(cleanupRule.register(channelBuilder.build())));
lbRegistry.register(fallBackLbProvider);
lbRegistry.register(lbProvider);
when(helper.getSynchronizationContext()).thenReturn(syncContext);
when(helper.getScheduledExecutorService()).thenReturn(fakeClock.getScheduledExecutorService());
when(helper.getAuthority()).thenReturn(SERVICE_AUTHORITY);
when(helper.getChannelLogger()).thenReturn(mock(ChannelLogger.class));
when(helper.createResolvingOobChannel(anyString()))
.thenReturn(oobChannel1, oobChannel2, oobChannel3);
when(localityStore.getLoadStatsStore()).thenReturn(loadStatsStore);
when(lrsClientFactory.createLoadReportClient(any(ManagedChannel.class), any(Helper.class),
any(BackoffPolicy.Provider.class), any(LoadStatsStore.class))).thenReturn(lrsClient);
xdsLoadBalancer =
new XdsLoadBalancer(helper, lbRegistry, backoffPolicyProvider, lrsClientFactory,
new FallbackManager(helper, lbRegistry), localityStore);
}
@After
public void tearDown() {
xdsLoadBalancer.shutdown();
}
/**
* Tests load reporting is initiated after receiving the first valid EDS response from the traffic
* director, then its operation is independent of load balancing until xDS load balancer is
* shutdown.
*/
@Test
public void reportLoadAfterReceivingFirstEdsResponseUntilShutdown() throws Exception {
xdsLoadBalancer.handleResolvedAddresses(ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallbackAttributes())
.build());
verify(lrsClientFactory)
.createLoadReportClient(same(oobChannel1), same(helper), same(backoffPolicyProvider),
same(loadStatsStore));
assertThat(streamRecorder.getValues()).hasSize(1);
// Let fallback timer elapse and xDS load balancer enters fallback mode on startup.
assertThat(fakeClock.getPendingTasks()).hasSize(1);
assertThat(fallBackLbHelper).isNull();
fakeClock.forwardTime(10, TimeUnit.SECONDS);
assertThat(fallBackLbHelper).isNotNull();
verify(lrsClient, never()).startLoadReporting(any(LoadReportCallback.class));
// Simulates a syntactically incorrect EDS response.
serverResponseWriter.onNext(DiscoveryResponse.getDefaultInstance());
verify(lrsClient, never()).startLoadReporting(any(LoadReportCallback.class));
ArgumentCaptor<LoadReportCallback> lrsCallbackCaptor = ArgumentCaptor.forClass(null);
// Simulate a syntactically correct EDS response.
DiscoveryResponse edsResponse =
DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.getDefaultInstance()))
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build();
serverResponseWriter.onNext(edsResponse);
verify(lrsClient).startLoadReporting(lrsCallbackCaptor.capture());
lrsCallbackCaptor.getValue().onReportResponse(19543);
verify(localityStore).updateOobMetricsReportInterval(19543);
// Simulate another EDS response from the same remote balancer.
serverResponseWriter.onNext(edsResponse);
// Simulate an EDS error response.
serverResponseWriter.onError(Status.ABORTED.asException());
// Shutdown xDS load balancer.
xdsLoadBalancer.shutdown();
verify(lrsClient).stopLoadReporting();
verifyNoMoreInteractions(lrsClientFactory, lrsClient);
}
/**
* Tests load report client sends load to new traffic director when xDS load balancer talks to
* the remote balancer.
*/
@Test
@SuppressWarnings("unchecked")
public void reportLoadToNewTrafficDirectorAfterBalancerNameChange() throws Exception {
InOrder inOrder = inOrder(lrsClientFactory, lrsClient);
xdsLoadBalancer.handleResolvedAddresses(ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallbackAttributes())
.build());
inOrder.verify(lrsClientFactory)
.createLoadReportClient(same(oobChannel1), same(helper), same(backoffPolicyProvider),
same(loadStatsStore));
assertThat(streamRecorder.getValues()).hasSize(1);
inOrder.verify(lrsClient, never()).startLoadReporting(any(LoadReportCallback.class));
// Simulate receiving a new service config with balancer name changed before xDS protocol is
// established.
Map<String, ?> newLbConfig =
(Map<String, ?>) JsonParser.parse(
"{\"balancerName\" : \"dns:///another.balancer.example.com:8080\","
+ "\"fallbackPolicy\" : [{\"fallback\" : { \"fallback_option\" : \"yes\"}}]}");
xdsLoadBalancer.handleResolvedAddresses(ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, newLbConfig).build())
.build());
assertThat(oobChannel1.isShutdown()).isTrue();
assertThat(streamRecorder.getValues()).hasSize(2);
inOrder.verify(lrsClientFactory)
.createLoadReportClient(same(oobChannel2), same(helper), same(backoffPolicyProvider),
same(loadStatsStore));
// Simulate a syntactically correct EDS response.
DiscoveryResponse edsResponse =
DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.getDefaultInstance()))
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build();
serverResponseWriter.onNext(edsResponse);
inOrder.verify(lrsClient).startLoadReporting(any(LoadReportCallback.class));
// Simulate receiving a new service config with balancer name changed.
newLbConfig = (Map<String, ?>) JsonParser.parse(
"{\"balancerName\" : \"dns:///third.balancer.example.com:8080\","
+ "\"fallbackPolicy\" : [{\"fallback\" : { \"fallback_option\" : \"yes\"}}]}");
xdsLoadBalancer.handleResolvedAddresses(ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, newLbConfig).build())
.build());
assertThat(oobChannel2.isShutdown()).isTrue();
assertThat(streamRecorder.getValues()).hasSize(3);
inOrder.verify(lrsClient).stopLoadReporting();
inOrder.verify(lrsClientFactory)
.createLoadReportClient(same(oobChannel3), same(helper), same(backoffPolicyProvider),
same(loadStatsStore));
serverResponseWriter.onNext(edsResponse);
inOrder.verify(lrsClient).startLoadReporting(any(LoadReportCallback.class));
inOrder.verifyNoMoreInteractions();
}
/**
* Tests the case that load reporting is not interrupted when child balancing policy changes,
* even though xDS balancer refreshes discovery RPC with the traffic director.
*/
@Test
public void loadReportNotAffectedWhenChildPolicyChanges() throws Exception {
xdsLoadBalancer.handleResolvedAddresses(ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(standardModeWithFallbackAttributes())
.build());
verify(lrsClientFactory)
.createLoadReportClient(same(oobChannel1), same(helper), same(backoffPolicyProvider),
same(loadStatsStore));
assertThat(streamRecorder.getValues()).hasSize(1);
// Simulate a syntactically correct EDS response.
DiscoveryResponse edsResponse =
DiscoveryResponse.newBuilder()
.addResources(Any.pack(ClusterLoadAssignment.getDefaultInstance()))
.setTypeUrl("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment")
.build();
serverResponseWriter.onNext(edsResponse);
verify(lrsClient).startLoadReporting(any(LoadReportCallback.class));
// Simulate receiving a new service config with child policy changed.
@SuppressWarnings("unchecked")
Map<String, ?> newLbConfig =
(Map<String, ?>) JsonParser.parse(
"{\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"childPolicy\" : [{\"supported\" : {\"key\" : \"val\"}}],"
+ "\"fallbackPolicy\" : [{\"fallback\" : { \"fallback_option\" : \"yes\"}}]}");
xdsLoadBalancer.handleResolvedAddresses(ResolvedAddresses.newBuilder()
.setAddresses(Collections.<EquivalentAddressGroup>emptyList())
.setAttributes(Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, newLbConfig).build())
.build());
assertThat(oobChannel1.isShutdown()).isFalse();
assertThat(Status.fromThrowable(streamRecorder.getError()).getCode())
.isEqualTo(Code.CANCELLED);
assertThat(streamRecorder.getValues()).hasSize(2);
verify(lrsClient, never()).stopLoadReporting();
verifyNoMoreInteractions(lrsClientFactory, lrsClient);
}
private static Attributes standardModeWithFallbackAttributes() throws Exception {
String lbConfigRaw = "{"
+ "\"balancerName\" : \"dns:///balancer.example.com:8080\","
+ "\"fallbackPolicy\" : [{\"fallback\" : { \"fallback_option\" : \"yes\"}}]"
+ "}";
@SuppressWarnings("unchecked")
Map<String, ?> lbConfig = (Map<String, ?>) JsonParser.parse(lbConfigRaw);
return Attributes.newBuilder().set(ATTR_LOAD_BALANCING_CONFIG, lbConfig).build();
}
}