diff --git a/internal/xds/clients/xdsclient/authority.go b/internal/xds/clients/xdsclient/authority.go index 3b5666bf1bb7..087ba8da5004 100644 --- a/internal/xds/clients/xdsclient/authority.go +++ b/internal/xds/clients/xdsclient/authority.go @@ -22,6 +22,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "sync/atomic" @@ -376,8 +377,12 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName, }) } + isDuplicateErr := isDuplicateResourceErr(state.md.ErrState, uErr.Err) state.md.ErrState = md.ErrState state.md.Status = md.Status + if isDuplicateErr { + continue + } for watcher := range state.watchers { watcher := watcher err := uErr.Err @@ -987,3 +992,10 @@ func cacheState(r *resourceState) string { return "requested" } } + +func isDuplicateResourceErr(errState *xdsresource.UpdateErrorMetadata, resourceErr error) bool { + if errState == nil || errState.Err == nil || resourceErr == nil { + return false + } + return strings.Contains(errState.Err.Error(), resourceErr.Error()) +} diff --git a/internal/xds/clients/xdsclient/test/ads_stream_ack_nack_test.go b/internal/xds/clients/xdsclient/test/ads_stream_ack_nack_test.go index 75639b4c63b0..cdb4c2c935a7 100644 --- a/internal/xds/clients/xdsclient/test/ads_stream_ack_nack_test.go +++ b/internal/xds/clients/xdsclient/test/ads_stream_ack_nack_test.go @@ -22,6 +22,7 @@ import ( "context" "fmt" "strings" + "sync/atomic" "testing" "time" @@ -506,3 +507,92 @@ func (s) TestADS_ACK_NACK_ResourceIsNotRequestedAnymore(t *testing.T) { t.Fatal(err) } } + +// TestADS_NACKError_DuplicateSuppression verifies that when a duplicate +// invalid LDS resource is sent by the server, the client suppresses the +// duplicate error notification to the watcher. +func (s) TestADS_NACKError_DuplicateSuppression(t *testing.T) { + nackReceivedCh := make(chan struct{}) + var nackCount atomic.Int32 + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{ + OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error { + if req.GetTypeUrl() != xdsresource.V3ListenerURL { + return nil + } + if req.GetErrorDetail() != nil { + if count := nackCount.Add(1); count == 3 { + close(nackReceivedCh) + } + } + return nil + }, + }) + + const listenerName = "listener" + nodeID := uuid.New().String() + + // Create an xDS client pointing to the above server. + configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}} + client := createXDSClient(t, mgmtServer.Address, nodeID, grpctransport.NewBuilder(configs)) + + // Use a custom watcher that counts invocations. + errCh := testutils.NewChannelWithSize(1) + watcher := &countingListenerWatcher{errCh: errCh} + ldsCancel := client.WatchResource(xdsresource.V3ListenerURL, listenerName, watcher) + defer ldsCancel() + + resources := e2e.UpdateOptions{ + NodeID: nodeID, + Listeners: []*v3listenerpb.Listener{badListenerResource(t, listenerName)}, + SkipValidation: true, + } + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + if err := mgmtServer.Update(ctx, resources); err != nil { + t.Fatalf("Failed to update management server with resources: %v, err: %v", resources, err) + } + + // Verify that the watcher receives the initial error. + v, err := errCh.Receive(ctx) + if err != nil { + t.Fatalf("timeout when waiting for an endpoints resource error from the management server: %v", err) + } + gotErr := v.(error) + wantNackErr := "no RouteSpecifier" + if !strings.Contains(gotErr.Error(), wantNackErr) { + t.Fatalf("update received with error: %v, want %q", gotErr, wantNackErr) + } + + // Wait for the management server to receive at least 3 NACKs, indicating that + // the NACK resend loop has run multiple times. + select { + case <-nackReceivedCh: + case <-ctx.Done(): + t.Fatalf("timeout waiting for 3 NACKs to be received by the server: %v", ctx.Err()) + } + + // Verify that the count of error callbacks is still exactly 1, proving that + // subsequent duplicate error updates were suppressed. + if count := watcher.errCount.Load(); count != 1 { + t.Fatalf("Error callback invoked %d times, want 1", count) + } +} + +type countingListenerWatcher struct { + errCh *testutils.Channel + errCount atomic.Int32 +} + +func (*countingListenerWatcher) ResourceChanged(_ xdsclient.ResourceData, done func()) { + done() +} +func (c *countingListenerWatcher) ResourceError(err error, done func()) { + c.errCount.Add(1) + c.errCh.Send(err) + done() +} +func (c *countingListenerWatcher) AmbientError(err error, done func()) { + c.errCount.Add(1) + c.errCh.Send(err) + done() +}