rls: only reset backoff on recovery from TRANSIENT_FAILURE #8720
+161
−74
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -26,13 +26,15 @@ import ( | |
| "fmt" | ||
| "os" | ||
| "regexp" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/google/go-cmp/cmp" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/balancer" | ||
| "google.golang.org/grpc/codes" | ||
| "google.golang.org/grpc/connectivity" | ||
| "google.golang.org/grpc/credentials" | ||
| "google.golang.org/grpc/internal" | ||
| rlspb "google.golang.org/grpc/internal/proto/grpc_lookup_v1" | ||
|
|
@@ -463,3 +465,124 @@ func (s) TestNewControlChannelUnsupportedCredsBundle(t *testing.T) { | |
| t.Fatal("newControlChannel succeeded when expected to fail") | ||
| } | ||
| } | ||
|
|
||
| // TestControlChannelConnectivityStateTransitions verifies that the control | ||
| // channel only resets backoff when recovering from TRANSIENT_FAILURE, not | ||
| // when going through benign state changes like READY → IDLE → READY. | ||
| func (s) TestControlChannelConnectivityStateTransitions(t *testing.T) { | ||
| tests := []struct { | ||
| name string | ||
| description string | ||
| states []connectivity.State | ||
| wantCallbackCount int | ||
| }{ | ||
| { | ||
| name: "ready_after_transient_failure", | ||
| description: "ready after transient failure triggers callback to reset the timer.", | ||
| states: []connectivity.State{ | ||
| connectivity.TransientFailure, | ||
| connectivity.Ready, | ||
| }, | ||
| wantCallbackCount: 1, | ||
| }, | ||
| { | ||
| name: "ready_after_idle", | ||
| description: "ready after idle does not trigger callback", | ||
| states: []connectivity.State{ | ||
| connectivity.Idle, | ||
| connectivity.Ready, | ||
| }, | ||
| wantCallbackCount: 0, | ||
| }, | ||
| { | ||
| name: "multiple_failures", | ||
| description: "multiple failures trigger callback each time", | ||
| states: []connectivity.State{ | ||
| connectivity.TransientFailure, | ||
| connectivity.Ready, | ||
| connectivity.TransientFailure, | ||
| connectivity.Ready, | ||
| }, | ||
| wantCallbackCount: 2, | ||
| }, | ||
| { | ||
| name: "idle_between_failures", | ||
| description: "idle between failures doesn't affect callback", | ||
| states: []connectivity.State{ | ||
| connectivity.TransientFailure, | ||
| connectivity.Idle, | ||
| connectivity.Ready, | ||
| }, | ||
| wantCallbackCount: 1, | ||
| }, | ||
| } | ||
|
|
||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
|
Member
There was a problem hiding this comment. I think there might be some way to improve the test. Maybe we can use waitGroups , but I will defer to @easwars for his opinion on this.
Contributor
There was a problem hiding this comment. To be able to test this in an e2e style, we would need to do make it possible for the test to see the connectivity state changes on the control channel, but without adding hooks into the test code. I propose the following:
var newConnectivityStateSubscriber = connStateSubscriber
func connStateSubscriber(sub grpcsync.Subscriber) grpcsync.Subscriber {
return sub
}
ctrlCh.unsubscribe = internal.SubscribeToConnectivityStateChanges.(func(cc *grpc.ClientConn, s grpcsync.Subscriber) func())(ctrlCh.cc, newConnectivityStateSubscriber(ctrlCh))
type wrappingConnectivityStateSubscriber struct {
delegate grpcsync.Subscriber
connStateCh chan connectivity.State
}
func (w *wrappingConnectivityStateSubscriber) OnMessage(msg any) {
w.delegate.OnMessage(msg)
w.connStateCh <- msg.(connectivity.State)
}
// Override the connectivity state subscriber.
wrappedSubscriber := &wrappingConnectivityStateSubscriber{connStateCh: make(chan connectivity.State, 1)}
origConnectivityStateSubscriber := newConnectivityStateSubscriber
newConnectivityStateSubscriber = func(delegate grpcsync.Subscriber) grpcsync.Subscriber {
wrappedSubscriber.delegate = delegate
return wrappedSubscriber
}
defer func() { newConnectivityStateSubscriber = origConnectivityStateSubscriber }()
// Make sure an RLS request is sent out.
verifyRLSRequest(t, rlsReqCh, true)
// Verify that the control channel moves to READY.
wantStates := []connectivity.State{
connectivity.Connecting,
connectivity.Ready,
}
for _, wantState := range wantStates {
select {
case gotState := <-wrappedSubscriber.connStateCh:
if gotState != wantState {
t.Fatalf("Unexpected connectivity state: got %v, want %v", gotState, wantState)
}
case <-ctx.Done():
t.Fatalf("Timeout waiting for RLS control channel to become %q", wantState)
}
}
// Stop the RLS server.
lis.Stop()
// Verify that the control channel moves to IDLE.
wantStates = []connectivity.State{
connectivity.Idle,
}
for _, wantState := range wantStates {
select {
case gotState := <-wrappedSubscriber.connStateCh:
if gotState != wantState {
t.Fatalf("Unexpected connectivity state: got %v, want %v", gotState, wantState)
}
case <-ctx.Done():
t.Fatalf("Timeout waiting for RLS control channel to become %q", wantState)
}
}
// Make another RPC similar to the first one. Since the above cache entry
// would have expired by now, this should trigger another RLS request. And
// since the RLS server is down, RLS request will fail and the cache entry
// will enter backoff, and we have overridden the default backoff strategy to
// return a value which will keep this entry in backoff for the whole duration
// of the test.
makeTestRPCAndVerifyError(ctx, t, cc, codes.Unavailable, nil)
// Verify that the control channel moves to TRANSIENT_FAILURE.
wantStates = []connectivity.State{
connectivity.Connecting,
connectivity.TransientFailure,
}
for _, wantState := range wantStates {
select {
case gotState := <-wrappedSubscriber.connStateCh:
if gotState != wantState {
t.Fatalf("Unexpected connectivity state: got %v, want %v", gotState, wantState)
}
case <-ctx.Done():
t.Fatalf("Timeout waiting for RLS control channel to become %q", wantState)
}
}
// Restart the RLS server.
lis.Restart()The above will test the READY --> TF --> READY transition. For the READY --> IDLE --> READY, we need to restart the RLS server once the control channel goes IDLE, and then wait for it to go READY before attempting another RPC and verifying that backoffs are not reset. Let me know what you think about this approach. Thanks |
||
| // Start an RLS server | ||
| rlsServer, _ := rlstest.SetupFakeRLSServer(t, nil) | ||
|
|
||
| // Setup callback to count invocations | ||
| var mu sync.Mutex | ||
| var callbackCount int | ||
| // Buffered channel large enough to never block | ||
| callbackInvoked := make(chan struct{}, 100) | ||
| callback := func() { | ||
| mu.Lock() | ||
| callbackCount++ | ||
| mu.Unlock() | ||
| // Send to channel - should never block with large buffer | ||
| callbackInvoked <- struct{}{} | ||
| } | ||
|
|
||
| // Create control channel | ||
| ctrlCh, err := newControlChannel(rlsServer.Address, "", defaultTestTimeout, balancer.BuildOptions{}, callback) | ||
| if err != nil { | ||
| t.Fatalf("Failed to create control channel: %v", err) | ||
| } | ||
| defer ctrlCh.close() | ||
|
|
||
| // Inject all test states | ||
| for _, state := range tt.states { | ||
| ctrlCh.OnMessage(state) | ||
| } | ||
|
|
||
| // Wait for all expected callbacks with timeout | ||
| callbackTimeout := time.NewTimer(defaultTestTimeout) | ||
| defer callbackTimeout.Stop() | ||
|
|
||
| receivedCallbacks := 0 | ||
| for receivedCallbacks < tt.wantCallbackCount { | ||
| select { | ||
| case <-callbackInvoked: | ||
| receivedCallbacks++ | ||
| case <-callbackTimeout.C: | ||
| mu.Lock() | ||
| got := callbackCount | ||
| mu.Unlock() | ||
| t.Fatalf("Timeout waiting for callbacks: expected %d, received %d via channel, callback count is %d", tt.wantCallbackCount, receivedCallbacks, got) | ||
| } | ||
| } | ||
|
|
||
| // Verify final callback count matches expected | ||
| mu.Lock() | ||
| gotCallbackCount := callbackCount | ||
| mu.Unlock() | ||
|
|
||
| if gotCallbackCount != tt.wantCallbackCount { | ||
| t.Errorf("Got %d callback invocations, want %d", gotCallbackCount, tt.wantCallbackCount) | ||
| } | ||
|
|
||
| // Ensure no extra callbacks are invoked | ||
| select { | ||
| case <-callbackInvoked: | ||
| mu.Lock() | ||
| final := callbackCount | ||
| mu.Unlock() | ||
| t.Fatalf("Received more callbacks than expected: got %d, want %d", final, tt.wantCallbackCount) | ||
| case <-time.After(50 * time.Millisecond): | ||
| // Expected: no more callbacks | ||
| } | ||
| }) | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.