xds: suppress the duplicate errors at the client side (#9185)

Fixes #8994

This change suppresses duplicate errors on the xds client side by
skipping calling watchers for the duplicates. This also prevents
generating new child names for the same broken EDS resource (EDS
resource has no localities) that is being re-send by the control plane.

RELEASE NOTES:
- xds: handle receipt of duplicated errors at the client side by
suppressing them in the same manner the duplicate updates are suppressed

Author: mswierq

internal/xds/clients/xdsclient/authority.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -370,24 +371,29 @@ func (a *authority) handleADSResourceUpdate(serverConfig *ServerConfig, rType Re
 
 		// On error, keep previous version of the resource. But update status
 		// and error.
-		if uErr.Err != nil {
+		if err := uErr.Err; err != nil {
 			if a.metricsReporter != nil {
 				a.metricsReporter.ReportMetric(&metrics.ResourceUpdateInvalid{
 					ServerURI: serverConfig.ServerIdentifier.ServerURI, ResourceType: rType.TypeName,
 				})
 			}
-			state.md.ErrState = md.ErrState
-			state.md.Status = md.Status
-			for watcher := range state.watchers {
-				watcher := watcher
-				err := uErr.Err
-				watcherCnt.Add(1)
-				if state.cache == nil {
-					funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.ResourceError(err, done) })
-				} else {
-					funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.AmbientError(err, done) })
+
+			// Notify watchers only if this is not a duplicated error from the previous update.
+			if errState := state.md.ErrState; errState == nil || errState.Err == nil || !strings.Contains(state.md.ErrState.Err.Error(), err.Error()) {
+				for watcher := range state.watchers {
+					watcherCnt.Add(1)
+					if state.cache == nil {
+						funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.ResourceError(err, done) })
+					} else {
+						funcsToSchedule = append(funcsToSchedule, func(context.Context) { watcher.AmbientError(err, done) })
+					}
 				}
 			}
+
+			// Update error state.
+			state.md.ErrState = md.ErrState
+			state.md.Status = md.Status
+
 			continue
 		}
 

internal/xds/clients/xdsclient/test/ads_stream_ack_nack_test.go

@@ -506,3 +506,170 @@ 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) {
+	// The first NACK is expected upon receiving the invalid resource. The
+	// second NACK is a duplicate, sent because the management server resends
+	// the invalid resource in a loop as long as the resource update has not
+	// changed to a valid one.  The third NACK is waited for to verify that the
+	// duplicated error is still suppressed as expected.
+	const wantNACKCount = 3
+
+	nacksReceivedCh := make(chan struct{})
+	nackCount := 0
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
+			if req.GetTypeUrl() != xdsresource.V3ListenerURL {
+				return nil
+			}
+			if req.GetErrorDetail() != nil {
+				nackCount++
+				if nackCount == wantNACKCount {
+					close(nacksReceivedCh)
+				}
+			}
+			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))
+
+	watcher := newListenerWatcher()
+	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 := watcher.resourceErrCh.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout when waiting for an endpoints resource error from the management server: %v", err)
+	}
+	const wantNackErr = "no RouteSpecifier"
+	if gotErr := v.(listenerUpdateErrTuple).resourceErr; !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 wantNACKCount NACKs,
+	// indicating that the NACK resend loop has run multiple times.
+	select {
+	case <-nacksReceivedCh:
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for %d NACKs to be received by the server: %v", wantNACKCount, ctx.Err())
+	}
+
+	// Verify that no duplicate error update was written to the error channel,
+	// proving that subsequent duplicate error updates were suppressed.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	if v, err := watcher.resourceErrCh.Receive(sCtx); err == nil {
+		t.Fatalf("Unexpected duplicate error received by watcher: %v", v)
+	}
+}
+
+// TestADS_NACKError_DuplicateSuppression_ConcatenatedErrorChange verifies that
+// when multiple invalid resources cause a concatenated error stored in
+// metadata, and a subsequent update carries a duplicate error for only a subset
+// of resources (changing the concatenated error string), the client still
+// suppresses the duplicate error notification to the watcher.
+func (s) TestADS_NACKError_DuplicateSuppression_ConcatenatedErrorChange(t *testing.T) {
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{})
+
+	const listenerName1 = "listener-1"
+	const listenerName2 = "listener-2"
+	nodeID := uuid.New().String()
+
+	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
+	client := createXDSClient(t, mgmtServer.Address, nodeID, grpctransport.NewBuilder(configs))
+
+	watcher1 := newListenerWatcher()
+	ldsCancel1 := client.WatchResource(xdsresource.V3ListenerURL, listenerName1, watcher1)
+	defer ldsCancel1()
+
+	watcher2 := newListenerWatcher()
+	ldsCancel2 := client.WatchResource(xdsresource.V3ListenerURL, listenerName2, watcher2)
+	defer ldsCancel2()
+
+	// Update 1: Management server sends two invalid listener resources.
+	resources1 := e2e.UpdateOptions{
+		NodeID: nodeID,
+		Listeners: []*v3listenerpb.Listener{
+			badListenerResource(t, listenerName1),
+			badListenerResource(t, listenerName2),
+		},
+		SkipValidation: true,
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	if err := mgmtServer.Update(ctx, resources1); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources1, err)
+	}
+
+	// Verify that watcher1 receives the initial error for listener 1.
+	v1, err := watcher1.resourceErrCh.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout waiting for listener 1 error from management server: %v", err)
+	}
+	const wantNackErr = "no RouteSpecifier"
+	if gotErr := v1.(listenerUpdateErrTuple).resourceErr; !strings.Contains(gotErr.Error(), wantNackErr) {
+		t.Fatalf("Update 1 listener 1 received with error: %v, want %q", gotErr, wantNackErr)
+	}
+
+	// Verify that watcher2 receives the initial error for listener 2.
+	v2, err := watcher2.resourceErrCh.Receive(ctx)
+	if err != nil {
+		t.Fatalf("Timeout waiting for listener 2 error from management server: %v", err)
+	}
+
+	if gotErr := v2.(listenerUpdateErrTuple).resourceErr; !strings.Contains(gotErr.Error(), wantNackErr) {
+		t.Fatalf("Update 1 listener 2 received with error: %v, want %q", gotErr, wantNackErr)
+	}
+
+	// Update 2: Management server sends listener 1 still invalid, but listener
+	// 2 now valid.  This changes the concatenated error stored in metadata
+	// (from Error1+Error2 to just Error1).
+	resources2 := e2e.UpdateOptions{
+		NodeID: nodeID,
+		Listeners: []*v3listenerpb.Listener{
+			badListenerResource(t, listenerName1),
+			e2e.DefaultClientListener(listenerName2, "route-config-2"),
+		},
+		SkipValidation: true,
+	}
+
+	if err := mgmtServer.Update(ctx, resources2); err != nil {
+		t.Fatalf("Failed to update management server with resources: %v, err: %v", resources2, err)
+	}
+
+	// Wait for watcher2 to receive the valid resource update for listener 2.
+	if _, err := watcher2.updateCh.Receive(ctx); err != nil {
+		t.Fatalf("Timeout waiting for listener 2 resource update: %v", err)
+	}
+
+	// Verify that no duplicate error update was written to watcher 1 error
+	// channel, proving that the duplicate error on listener 1 was suppressed
+	// despite the concatenated error changing.
+	sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout)
+	defer sCancel()
+	if v, err := watcher1.resourceErrCh.Receive(sCtx); err == nil {
+		t.Fatalf("Unexpected duplicate error received by watcher 1: %v", v)
+	}
+}