Skip to content

Commit

Permalink
core: [outlier detection] eject/uneject states to delegate listener. (#…
Browse files Browse the repository at this point in the history 
…9469) (#9476)
  • Loading branch information
@temawi
temawi committed Aug 22, 2022
1 parent 184fdaa commit 3ebdfc7
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 30 deletions.
6 changes: 3 additions & 3 deletions core/src/main/java/io/grpc/util/OutlierDetectionLoadBalancer.java
View file
Expand Up @@ -233,16 +233,16 @@ class OutlierDetectionSubchannel extends ForwardingSubchannel {
private AddressTracker addressTracker;
private boolean ejected;
private ConnectivityStateInfo lastSubchannelState;
private OutlierDetectionSubchannelStateListener subchannelStateListener;
private SubchannelStateListener subchannelStateListener;

OutlierDetectionSubchannel(Subchannel delegate) {
this.delegate = delegate;
}

@Override
public void start(SubchannelStateListener listener) {
subchannelStateListener = new OutlierDetectionSubchannelStateListener(listener);
super.start(subchannelStateListener);
subchannelStateListener = listener;
super.start(new OutlierDetectionSubchannelStateListener(listener));
}

@Override
Expand Down
55 changes: 28 additions & 27 deletions core/src/test/java/io/grpc/util/OutlierDetectionLoadBalancerTest.java
View file
Expand Up @@ -356,7 +356,7 @@ public void successRateNoOutliers() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of());
generateLoad(ImmutableMap.of(), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -380,7 +380,7 @@ public void successRateOneOutlier() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -405,7 +405,7 @@ public void successRateOneOutlier_configChange() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -425,7 +425,7 @@ public void successRateOneOutlier_configChange() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel2, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel2, Status.DEADLINE_EXCEEDED), 8);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -450,7 +450,7 @@ public void successRateOneOutlier_unejected() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
fakeClock.forwardTime(config.intervalNanos + 1, TimeUnit.NANOSECONDS);
Expand All @@ -459,7 +459,7 @@ public void successRateOneOutlier_unejected() {
assertEjectedSubchannels(ImmutableSet.of(servers.get(0).getAddresses().get(0)));

// Now we produce more load, but the subchannel start working and is no longer an outlier.
generateLoad(ImmutableMap.of());
generateLoad(ImmutableMap.of(), 8);

// Move forward in time to a point where the detection timer has fired.
fakeClock.forwardTime(config.maxEjectionTimeNanos + 1, TimeUnit.NANOSECONDS);
Expand All @@ -486,7 +486,7 @@ public void successRateOneOutlier_notEnoughVolume() {
// We produce an outlier, but don't give it enough calls to reach the minimum volume.
generateLoad(
ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED),
ImmutableMap.of(subchannel1, 19));
ImmutableMap.of(subchannel1, 19), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -510,7 +510,7 @@ public void successRateOneOutlier_notEnoughAddressesWithVolume() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -536,7 +536,7 @@ public void successRateOneOutlier_enforcementPercentage() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -563,7 +563,7 @@ public void successRateTwoOutliers() {

generateLoad(ImmutableMap.of(
subchannel1, Status.DEADLINE_EXCEEDED,
subchannel2, Status.DEADLINE_EXCEEDED));
subchannel2, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -591,7 +591,7 @@ public void successRateTwoOutliers_maxEjectionPercentage() {

generateLoad(ImmutableMap.of(
subchannel1, Status.DEADLINE_EXCEEDED,
subchannel2, Status.DEADLINE_EXCEEDED));
subchannel2, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -626,7 +626,7 @@ public void failurePercentageNoOutliers() {
loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

// By default all calls will return OK.
generateLoad(ImmutableMap.of());
generateLoad(ImmutableMap.of(), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -650,7 +650,7 @@ public void failurePercentageOneOutlier() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -674,7 +674,7 @@ public void failurePercentageOneOutlier_notEnoughVolume() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -700,7 +700,7 @@ public void failurePercentageOneOutlier_enforcementPercentage() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -737,7 +737,7 @@ public void successRateAndFailurePercentageThreeOutliers() {
subchannel1, Status.DEADLINE_EXCEEDED,
subchannel2, Status.DEADLINE_EXCEEDED,
subchannel3, Status.DEADLINE_EXCEEDED),
ImmutableMap.of(subchannel3, 1));
ImmutableMap.of(subchannel3, 1), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -766,7 +766,7 @@ public void subchannelUpdateAddress_singleReplaced() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -813,7 +813,7 @@ public void subchannelUpdateAddress_singleReplacedWithMultiple() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of());
generateLoad(ImmutableMap.of(), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -860,7 +860,7 @@ public void subchannelUpdateAddress_multipleReplacedWithSingle() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -922,7 +922,7 @@ public void successRateAndFailurePercentage_noOutliers() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of());
generateLoad(ImmutableMap.of(), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -949,7 +949,7 @@ public void successRateAndFailurePercentage_successRateOutlier() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand All @@ -976,7 +976,7 @@ public void successRateAndFailurePercentage_errorPercentageOutlier() {

loadBalancer.handleResolvedAddresses(buildResolvedAddress(config, servers));

generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED));
generateLoad(ImmutableMap.of(subchannel1, Status.DEADLINE_EXCEEDED), 7);

// Move forward in time to a point where the detection timer has fired.
forwardTime(config);
Expand Down Expand Up @@ -1025,22 +1025,23 @@ private void deliverSubchannelState(Subchannel subchannel, ConnectivityStateInfo
subchannelStateListeners.get(subchannel).onSubchannelState(newState);
}

private void generateLoad(Map<Subchannel, Status> statusMap) {
generateLoad(statusMap, null);
private void generateLoad(Map<Subchannel, Status> statusMap, int expectedStateChanges) {
generateLoad(statusMap, null, expectedStateChanges);
}

// Generates 100 calls, 20 each across the subchannels. Default status is OK.
private void generateLoad(Map<Subchannel, Status> statusMap,
Map<Subchannel, Integer> maxCallsMap) {
Map<Subchannel, Integer> maxCallsMap, int expectedStateChanges) {
deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(READY));
deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY));
deliverSubchannelState(subchannel3, ConnectivityStateInfo.forNonError(READY));
deliverSubchannelState(subchannel4, ConnectivityStateInfo.forNonError(READY));
deliverSubchannelState(subchannel5, ConnectivityStateInfo.forNonError(READY));

verify(mockHelper, times(7)).updateBalancingState(stateCaptor.capture(),
verify(mockHelper, times(expectedStateChanges)).updateBalancingState(stateCaptor.capture(),
pickerCaptor.capture());
SubchannelPicker picker = pickerCaptor.getAllValues().get(6);
SubchannelPicker picker = pickerCaptor.getAllValues()
.get(pickerCaptor.getAllValues().size() - 1);

HashMap<Subchannel, Integer> callCountMap = new HashMap<>();
for (int i = 0; i < 100; i++) {
Expand Down

0 comments on commit 3ebdfc7

Please sign in to comment.