xdsclient: Add error type for NACKed resources (#8117)

arjan-bal · web-flow · commit e0ac3acff4e8 · 2025-02-25T11:15:16.000+05:30
diff --git a/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go b/xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
@@ -923,7 +923,7 @@ func (s) TestResolverError(t *testing.T) {
 	}
 
 	// Push a resource-not-found-error this time around.
-	resolverErr = xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error")
+	resolverErr = xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error")
 	r.ReportError(resolverErr)
 
 	// Wait for the CDS resource to be not requested anymore, or the connection
diff --git a/xds/internal/xdsclient/channel.go b/xds/internal/xdsclient/channel.go
@@ -287,7 +287,7 @@ func decodeResponse(opts *xdsresource.DecodeOptions, rType xdsresource.Type, res
 		perResourceErrors[name] = err
 		// Add place holder in the map so we know this resource name was in
 		// the response.
-		ret[name] = ads.DataAndErrTuple{Err: err}
+		ret[name] = ads.DataAndErrTuple{Err: xdsresource.NewError(xdsresource.ErrorTypeNACKed, err.Error())}
 	}
 
 	if len(topLevelErrors) == 0 && len(perResourceErrors) == 0 {
@@ -299,7 +299,7 @@ func decodeResponse(opts *xdsresource.DecodeOptions, rType xdsresource.Type, res
 	errRet := combineErrors(rType.TypeName(), topLevelErrors, perResourceErrors)
 	md.ErrState = &xdsresource.UpdateErrorMetadata{
 		Version:   resp.Version,
-		Err:       errRet,
+		Err:       xdsresource.NewError(xdsresource.ErrorTypeNACKed, errRet.Error()),
 		Timestamp: timestamp,
 	}
 	return ret, md, errRet
diff --git a/xds/internal/xdsclient/clientimpl_watchers.go b/xds/internal/xdsclient/clientimpl_watchers.go
@@ -34,7 +34,7 @@ type wrappingWatcher struct {
 }
 
 func (w *wrappingWatcher) OnError(err error, done xdsresource.OnDoneFunc) {
-	w.ResourceWatcher.OnError(fmt.Errorf("[xDS node id: %v]: %v", w.nodeID, err), done)
+	w.ResourceWatcher.OnError(fmt.Errorf("[xDS node id: %v]: %w", w.nodeID, err), done)
 }
 
 // WatchResource uses xDS to discover the resource associated with the provided
diff --git a/xds/internal/xdsclient/tests/ads_stream_ack_nack_test.go b/xds/internal/xdsclient/tests/ads_stream_ack_nack_test.go
@@ -20,7 +20,6 @@ package xdsclient_test
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -162,7 +161,7 @@ func (s) TestADS_ACK_NACK_Simple(t *testing.T) {
 	}
 	gotResp = r.(*v3discoverypb.DiscoveryResponse)
 
-	wantNackErr := errors.New("unexpected http connection manager resource type")
+	wantNackErr := xdsresource.NewError(xdsresource.ErrorTypeNACKed, "unexpected http connection manager resource type")
 	if err := verifyListenerUpdate(ctx, lw.updateCh, listenerUpdateErrTuple{err: wantNackErr}); err != nil {
 		t.Fatal(err)
 	}
@@ -309,7 +308,7 @@ func (s) TestADS_NACK_InvalidFirstResponse(t *testing.T) {
 	gotResp := r.(*v3discoverypb.DiscoveryResponse)
 
 	// Verify that the error is propagated to the watcher.
-	var wantNackErr = errors.New("unexpected http connection manager resource type")
+	var wantNackErr = xdsresource.NewError(xdsresource.ErrorTypeNACKed, "unexpected http connection manager resource type")
 	if err := verifyListenerUpdate(ctx, lw.updateCh, listenerUpdateErrTuple{err: wantNackErr}); err != nil {
 		t.Fatal(err)
 	}
diff --git a/xds/internal/xdsclient/tests/authority_test.go b/xds/internal/xdsclient/tests/authority_test.go
@@ -370,6 +370,6 @@ func (cw *clusterWatcherV2) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc
 	// resends resources which are NACKed by the xDS client, using a `Replace()`
 	// here simplifies tests that want access to the most recently received
 	// error.
-	cw.resourceNotFoundCh.Replace(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response"))
+	cw.resourceNotFoundCh.Replace(xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response"))
 	onDone()
 }
diff --git a/xds/internal/xdsclient/tests/cds_watchers_test.go b/xds/internal/xdsclient/tests/cds_watchers_test.go
@@ -82,7 +82,7 @@ func (cw *clusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
 }
 
 func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	cw.updateCh.Replace(clusterUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response")})
+	cw.updateCh.Replace(clusterUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response")})
 	onDone()
 }
 
@@ -670,7 +670,7 @@ func (s) TestCDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	// Verify that an empty update with the expected error is received.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")
+	wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")
 	if err := verifyClusterUpdate(ctx, cw.updateCh, clusterUpdateErrTuple{err: wantErr}); err != nil {
 		t.Fatal(err)
 	}
@@ -847,7 +847,7 @@ func (s) TestCDSWatch_ResourceRemoved(t *testing.T) {
 
 	// The first watcher should receive a resource removed error, while the
 	// second watcher should not receive an update.
-	if err := verifyClusterUpdate(ctx, cw1.updateCh, clusterUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")}); err != nil {
+	if err := verifyClusterUpdate(ctx, cw1.updateCh, clusterUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")}); err != nil {
 		t.Fatal(err)
 	}
 	if err := verifyNoClusterUpdate(ctx, cw2.updateCh); err != nil {
diff --git a/xds/internal/xdsclient/tests/eds_watchers_test.go b/xds/internal/xdsclient/tests/eds_watchers_test.go
@@ -91,7 +91,7 @@ func (ew *endpointsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
 }
 
 func (ew *endpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	ew.updateCh.Replace(endpointsUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Endpoints not found in received response")})
+	ew.updateCh.Replace(endpointsUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Endpoints not found in received response")})
 	onDone()
 }
 
@@ -757,7 +757,7 @@ func (s) TestEDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	// Verify that an empty update with the expected error is received.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")
+	wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")
 	if err := verifyEndpointsUpdate(ctx, ew.updateCh, endpointsUpdateErrTuple{err: wantErr}); err != nil {
 		t.Fatal(err)
 	}
diff --git a/xds/internal/xdsclient/tests/lds_watchers_test.go b/xds/internal/xdsclient/tests/lds_watchers_test.go
@@ -86,7 +86,7 @@ func (lw *listenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
 }
 
 func (lw *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	lw.updateCh.Replace(listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
+	lw.updateCh.Replace(listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
 	onDone()
 }
 
@@ -111,7 +111,7 @@ func (lw *listenerWatcherMultiple) OnError(err error, onDone xdsresource.OnDoneF
 }
 
 func (lw *listenerWatcherMultiple) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	lw.updateCh.Send(listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
+	lw.updateCh.Send(listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})
 	onDone()
 }
 
@@ -135,10 +135,6 @@ func badListenerResource(t *testing.T, name string) *v3listenerpb.Listener {
 	}
 }
 
-// xdsClient is expected to produce an error containing this string when an
-// update is received containing a listener created using `badListenerResource`.
-const wantListenerNACKErr = "no RouteSpecifier"
-
 // verifyNoListenerUpdate verifies that no listener update is received on the
 // provided update channel, and returns an error if an update is received.
 //
@@ -195,6 +191,21 @@ func verifyListenerError(ctx context.Context, updateCh *testutils.Channel, wantE
 	return nil
 }
 
+func verifyErrorType(ctx context.Context, updateCh *testutils.Channel, wantErrType xdsresource.ErrorType, wantNodeID string) error {
+	u, err := updateCh.Receive(ctx)
+	if err != nil {
+		return fmt.Errorf("timeout when waiting for a listener error from the management server: %v", err)
+	}
+	gotErr := u.(listenerUpdateErrTuple).err
+	if got, want := xdsresource.ErrType(gotErr), wantErrType; got != want {
+		return fmt.Errorf("update received with error %v of type: %v, want %v", gotErr, got, want)
+	}
+	if !strings.Contains(gotErr.Error(), wantNodeID) {
+		return fmt.Errorf("update received with error: %v, want error with node ID: %q", gotErr, wantNodeID)
+	}
+	return nil
+}
+
 // TestLDSWatch covers the case where a single watcher exists for a single
 // listener resource. The test verifies the following scenarios:
 //  1. An update from the management server containing the resource being
@@ -726,7 +737,7 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	// Verify that an empty update with the expected error is received.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")
+	wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")
 	if err := verifyListenerUpdate(ctx, lw.updateCh, listenerUpdateErrTuple{err: wantErr}); err != nil {
 		t.Fatal(err)
 	}
@@ -900,7 +911,7 @@ func (s) TestLDSWatch_ResourceRemoved(t *testing.T) {
 	// The first watcher should receive a resource removed error, while the
 	// second watcher should not see an update.
 	if err := verifyListenerUpdate(ctx, lw1.updateCh, listenerUpdateErrTuple{
-		err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, ""),
+		err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, ""),
 	}); err != nil {
 		t.Fatal(err)
 	}
@@ -1002,7 +1013,7 @@ func (s) TestLDSWatch_NewWatcherForRemovedResource(t *testing.T) {
 	}
 
 	// The existing watcher should receive a resource removed error.
-	updateError := listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")}
+	updateError := listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")}
 	if err := verifyListenerUpdate(ctx, lw1.updateCh, updateError); err != nil {
 		t.Fatal(err)
 	}
@@ -1061,15 +1072,15 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {
 	}
 
 	// Verify that the expected error is propagated to the existing watcher.
-	if err := verifyListenerError(ctx, lw.updateCh, wantListenerNACKErr, nodeID); err != nil {
+	if err := verifyErrorType(ctx, lw.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {
 		t.Fatal(err)
 	}
 
 	// Verify that the expected error is propagated to the new watcher as well.
 	lw2 := newListenerWatcher()
 	ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)
 	defer ldsCancel2()
-	if err := verifyListenerError(ctx, lw2.updateCh, wantListenerNACKErr, nodeID); err != nil {
+	if err := verifyErrorType(ctx, lw2.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -1141,7 +1152,7 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 	}
 
 	// Verify that the expected error is propagated to the existing watcher.
-	if err := verifyListenerError(ctx, lw1.updateCh, wantListenerNACKErr, nodeID); err != nil {
+	if err := verifyErrorType(ctx, lw1.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {
 		t.Fatal(err)
 	}
 
@@ -1154,7 +1165,7 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {
 		t.Fatal(err)
 	}
 	// Verify that the expected error is propagated to the existing watcher.
-	if err := verifyListenerError(ctx, lw2.updateCh, wantListenerNACKErr, nodeID); err != nil {
+	if err := verifyErrorType(ctx, lw2.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -1232,7 +1243,7 @@ func (s) TestLDSWatch_PartialValid(t *testing.T) {
 	// Verify that the expected error is propagated to the watcher which
 	// requested for the bad resource.
 	// Verify that the expected error is propagated to the existing watcher.
-	if err := verifyListenerError(ctx, lw1.updateCh, wantListenerNACKErr, nodeID); err != nil {
+	if err := verifyErrorType(ctx, lw1.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {
 		t.Fatal(err)
 	}
 
diff --git a/xds/internal/xdsclient/tests/misc_watchers_test.go b/xds/internal/xdsclient/tests/misc_watchers_test.go
@@ -90,7 +90,7 @@ func (rw *testRouteConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFu
 }
 
 func (rw *testRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})
+	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})
 	onDone()
 }
 
diff --git a/xds/internal/xdsclient/tests/rds_watchers_test.go b/xds/internal/xdsclient/tests/rds_watchers_test.go
@@ -81,7 +81,7 @@ func (rw *routeConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc)
 }
 
 func (rw *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
-	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})
+	rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})
 	onDone()
 }
 
@@ -759,7 +759,7 @@ func (s) TestRDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 	// Verify that an empty update with the expected error is received.
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")
+	wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")
 	if err := verifyRouteConfigUpdate(ctx, rw.updateCh, routeConfigUpdateErrTuple{err: wantErr}); err != nil {
 		t.Fatal(err)
 	}
diff --git a/xds/internal/xdsclient/transport/ads/ads_stream.go b/xds/internal/xdsclient/transport/ads/ads_stream.go
@@ -664,7 +664,7 @@ func (s *StreamImpl) onError(err error, msgReceived bool) {
 	// connection hitting its max connection age limit.
 	// (see [gRFC A9](https://github.com/grpc/proposal/blob/master/A9-server-side-conn-mgt.md)).
 	if msgReceived {
-		err = xdsresource.NewErrorf(xdsresource.ErrTypeStreamFailedAfterRecv, "%s", err.Error())
+		err = xdsresource.NewError(xdsresource.ErrTypeStreamFailedAfterRecv, err.Error())
 	}
 
 	s.eventHandler.OnADSStreamError(err)
diff --git a/xds/internal/xdsclient/xdsresource/errors.go b/xds/internal/xdsclient/xdsresource/errors.go
@@ -18,7 +18,10 @@
 
 package xdsresource
 
-import "fmt"
+import (
+	"errors"
+	"fmt"
+)
 
 // ErrorType is the type of the error that the watcher will receive from the xds
 // client.
@@ -40,6 +43,9 @@ const (
 	// ErrTypeStreamFailedAfterRecv indicates an ADS stream error, after
 	// successful receipt of at least one message from the server.
 	ErrTypeStreamFailedAfterRecv
+	// ErrorTypeNACKed indicates that configuration provided by the xDS management
+	// server was NACKed.
+	ErrorTypeNACKed
 )
 
 type xdsClientError struct {
@@ -57,9 +63,16 @@ func NewErrorf(t ErrorType, format string, args ...any) error {
 	return &xdsClientError{t: t, desc: fmt.Sprintf(format, args...)}
 }
 
+// NewError creates an xDS client error. The callbacks are called with this
+// error, to pass additional information about the error.
+func NewError(t ErrorType, message string) error {
+	return NewErrorf(t, "%s", message)
+}
+
 // ErrType returns the error's type.
 func ErrType(e error) ErrorType {
-	if xe, ok := e.(*xdsClientError); ok {
+	var xe *xdsClientError
+	if ok := errors.As(e, &xe); ok {
 		return xe.t
 	}
 	return ErrorTypeUnknown

Original file line number	Diff line number	Diff line change
`@@ -923,7 +923,7 @@ func (s) TestResolverError(t *testing.T) {`
`923`	`923`	`}`
`924`	`924`
`925`	`925`	`// Push a resource-not-found-error this time around.`
`926`		`- resolverErr = xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error")`
	`926`	`+ resolverErr = xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "xds resource not found error")`
`927`	`927`	`r.ReportError(resolverErr)`
`928`	`928`
`929`	`929`	`// Wait for the CDS resource to be not requested anymore, or the connection`
Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,7 @@ type wrappingWatcher struct {`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`func (w *wrappingWatcher) OnError(err error, done xdsresource.OnDoneFunc) {`
`37`		`- w.ResourceWatcher.OnError(fmt.Errorf("[xDS node id: %v]: %v", w.nodeID, err), done)`
	`37`	`+ w.ResourceWatcher.OnError(fmt.Errorf("[xDS node id: %v]: %w", w.nodeID, err), done)`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`// WatchResource uses xDS to discover the resource associated with the provided`
Original file line number	Diff line number	Diff line change
`@@ -370,6 +370,6 @@ func (cw *clusterWatcherV2) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc`
`370`	`370`	// resends resources which are NACKed by the xDS client, using a `Replace()`
`371`	`371`	`// here simplifies tests that want access to the most recently received`
`372`	`372`	`// error.`
`373`		`- cw.resourceNotFoundCh.Replace(xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response"))`
	`373`	`+ cw.resourceNotFoundCh.Replace(xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response"))`
`374`	`374`	`onDone()`
`375`	`375`	`}`
Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ func (cw *clusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {`
`85`		`- cw.updateCh.Replace(clusterUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response")})`
	`85`	`+ cw.updateCh.Replace(clusterUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Cluster not found in received response")})`
`86`	`86`	`onDone()`
`87`	`87`	`}`
`88`	`88`
`@@ -670,7 +670,7 @@ func (s) TestCDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {`
`670`	`670`	`// Verify that an empty update with the expected error is received.`
`671`	`671`	`ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)`
`672`	`672`	`defer cancel()`
`673`		`- wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")`
	`673`	`+ wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")`
`674`	`674`	`if err := verifyClusterUpdate(ctx, cw.updateCh, clusterUpdateErrTuple{err: wantErr}); err != nil {`
`675`	`675`	`t.Fatal(err)`
`676`	`676`	`}`
`@@ -847,7 +847,7 @@ func (s) TestCDSWatch_ResourceRemoved(t *testing.T) {`
`847`	`847`
`848`	`848`	`// The first watcher should receive a resource removed error, while the`
`849`	`849`	`// second watcher should not receive an update.`
`850`		`- if err := verifyClusterUpdate(ctx, cw1.updateCh, clusterUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")}); err != nil {`
	`850`	`+ if err := verifyClusterUpdate(ctx, cw1.updateCh, clusterUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")}); err != nil {`
`851`	`851`	`t.Fatal(err)`
`852`	`852`	`}`
`853`	`853`	`if err := verifyNoClusterUpdate(ctx, cw2.updateCh); err != nil {`
Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,7 @@ func (ew *endpointsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {`
`91`	`91`	`}`
`92`	`92`
`93`	`93`	`func (ew *endpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {`
`94`		`- ew.updateCh.Replace(endpointsUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Endpoints not found in received response")})`
	`94`	`+ ew.updateCh.Replace(endpointsUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Endpoints not found in received response")})`
`95`	`95`	`onDone()`
`96`	`96`	`}`
`97`	`97`
`@@ -757,7 +757,7 @@ func (s) TestEDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {`
`757`	`757`	`// Verify that an empty update with the expected error is received.`
`758`	`758`	`ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)`
`759`	`759`	`defer cancel()`
`760`		`- wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")`
	`760`	`+ wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")`
`761`	`761`	`if err := verifyEndpointsUpdate(ctx, ew.updateCh, endpointsUpdateErrTuple{err: wantErr}); err != nil {`
`762`	`762`	`t.Fatal(err)`
`763`	`763`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ func (lw *listenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {`
`86`	`86`	`}`
`87`	`87`
`88`	`88`	`func (lw *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {`
`89`		`- lw.updateCh.Replace(listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})`
	`89`	`+ lw.updateCh.Replace(listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})`
`90`	`90`	`onDone()`
`91`	`91`	`}`
`92`	`92`
`@@ -111,7 +111,7 @@ func (lw *listenerWatcherMultiple) OnError(err error, onDone xdsresource.OnDoneF`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`func (lw *listenerWatcherMultiple) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {`
`114`		`- lw.updateCh.Send(listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})`
	`114`	`+ lw.updateCh.Send(listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "Listener not found in received response")})`
`115`	`115`	`onDone()`
`116`	`116`	`}`
`117`	`117`
`@@ -135,10 +135,6 @@ func badListenerResource(t testing.T, name string) v3listenerpb.Listener {`
`135`	`135`	`}`
`136`	`136`	`}`
`137`	`137`
`138`		`-// xdsClient is expected to produce an error containing this string when an`
`139`		-// update is received containing a listener created using `badListenerResource`.
`140`		`-const wantListenerNACKErr = "no RouteSpecifier"`
`141`		`-`
`142`	`138`	`// verifyNoListenerUpdate verifies that no listener update is received on the`
`143`	`139`	`// provided update channel, and returns an error if an update is received.`
`144`	`140`	`//`
`@@ -195,6 +191,21 @@ func verifyListenerError(ctx context.Context, updateCh *testutils.Channel, wantE`
`195`	`191`	`return nil`
`196`	`192`	`}`
`197`	`193`
	`194`	`+func verifyErrorType(ctx context.Context, updateCh *testutils.Channel, wantErrType xdsresource.ErrorType, wantNodeID string) error {`
	`195`	`+ u, err := updateCh.Receive(ctx)`
	`196`	`+ if err != nil {`
	`197`	`+ return fmt.Errorf("timeout when waiting for a listener error from the management server: %v", err)`
	`198`	`+ }`
	`199`	`+ gotErr := u.(listenerUpdateErrTuple).err`
	`200`	`+ if got, want := xdsresource.ErrType(gotErr), wantErrType; got != want {`
	`201`	`+ return fmt.Errorf("update received with error %v of type: %v, want %v", gotErr, got, want)`
	`202`	`+ }`
	`203`	`+ if !strings.Contains(gotErr.Error(), wantNodeID) {`
	`204`	`+ return fmt.Errorf("update received with error: %v, want error with node ID: %q", gotErr, wantNodeID)`
	`205`	`+ }`
	`206`	`+ return nil`
	`207`	`+}`
	`208`	`+`
`198`	`209`	`// TestLDSWatch covers the case where a single watcher exists for a single`
`199`	`210`	`// listener resource. The test verifies the following scenarios:`
`200`	`211`	`// 1. An update from the management server containing the resource being`
`@@ -726,7 +737,7 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {`
`726`	`737`	`// Verify that an empty update with the expected error is received.`
`727`	`738`	`ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)`
`728`	`739`	`defer cancel()`
`729`		`- wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")`
	`740`	`+ wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")`
`730`	`741`	`if err := verifyListenerUpdate(ctx, lw.updateCh, listenerUpdateErrTuple{err: wantErr}); err != nil {`
`731`	`742`	`t.Fatal(err)`
`732`	`743`	`}`
`@@ -900,7 +911,7 @@ func (s) TestLDSWatch_ResourceRemoved(t *testing.T) {`
`900`	`911`	`// The first watcher should receive a resource removed error, while the`
`901`	`912`	`// second watcher should not see an update.`
`902`	`913`	`if err := verifyListenerUpdate(ctx, lw1.updateCh, listenerUpdateErrTuple{`
`903`		`- err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, ""),`
	`914`	`+ err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, ""),`
`904`	`915`	`}); err != nil {`
`905`	`916`	`t.Fatal(err)`
`906`	`917`	`}`
`@@ -1002,7 +1013,7 @@ func (s) TestLDSWatch_NewWatcherForRemovedResource(t *testing.T) {`
`1002`	`1013`	`}`
`1003`	`1014`
`1004`	`1015`	`// The existing watcher should receive a resource removed error.`
`1005`		`- updateError := listenerUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")}`
	`1016`	`+ updateError := listenerUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")}`
`1006`	`1017`	`if err := verifyListenerUpdate(ctx, lw1.updateCh, updateError); err != nil {`
`1007`	`1018`	`t.Fatal(err)`
`1008`	`1019`	`}`
`@@ -1061,15 +1072,15 @@ func (s) TestLDSWatch_NACKError(t *testing.T) {`
`1061`	`1072`	`}`
`1062`	`1073`
`1063`	`1074`	`// Verify that the expected error is propagated to the existing watcher.`
`1064`		`- if err := verifyListenerError(ctx, lw.updateCh, wantListenerNACKErr, nodeID); err != nil {`
	`1075`	`+ if err := verifyErrorType(ctx, lw.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {`
`1065`	`1076`	`t.Fatal(err)`
`1066`	`1077`	`}`
`1067`	`1078`
`1068`	`1079`	`// Verify that the expected error is propagated to the new watcher as well.`
`1069`	`1080`	`lw2 := newListenerWatcher()`
`1070`	`1081`	`ldsCancel2 := xdsresource.WatchListener(client, ldsName, lw2)`
`1071`	`1082`	`defer ldsCancel2()`
`1072`		`- if err := verifyListenerError(ctx, lw2.updateCh, wantListenerNACKErr, nodeID); err != nil {`
	`1083`	`+ if err := verifyErrorType(ctx, lw2.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {`
`1073`	`1084`	`t.Fatal(err)`
`1074`	`1085`	`}`
`1075`	`1086`	`}`
`@@ -1141,7 +1152,7 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {`
`1141`	`1152`	`}`
`1142`	`1153`
`1143`	`1154`	`// Verify that the expected error is propagated to the existing watcher.`
`1144`		`- if err := verifyListenerError(ctx, lw1.updateCh, wantListenerNACKErr, nodeID); err != nil {`
	`1155`	`+ if err := verifyErrorType(ctx, lw1.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {`
`1145`	`1156`	`t.Fatal(err)`
`1146`	`1157`	`}`
`1147`	`1158`
`@@ -1154,7 +1165,7 @@ func (s) TestLDSWatch_ResourceCaching_NACKError(t *testing.T) {`
`1154`	`1165`	`t.Fatal(err)`
`1155`	`1166`	`}`
`1156`	`1167`	`// Verify that the expected error is propagated to the existing watcher.`
`1157`		`- if err := verifyListenerError(ctx, lw2.updateCh, wantListenerNACKErr, nodeID); err != nil {`
	`1168`	`+ if err := verifyErrorType(ctx, lw2.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {`
`1158`	`1169`	`t.Fatal(err)`
`1159`	`1170`	`}`
`1160`	`1171`	`}`
`@@ -1232,7 +1243,7 @@ func (s) TestLDSWatch_PartialValid(t *testing.T) {`
`1232`	`1243`	`// Verify that the expected error is propagated to the watcher which`
`1233`	`1244`	`// requested for the bad resource.`
`1234`	`1245`	`// Verify that the expected error is propagated to the existing watcher.`
`1235`		`- if err := verifyListenerError(ctx, lw1.updateCh, wantListenerNACKErr, nodeID); err != nil {`
	`1246`	`+ if err := verifyErrorType(ctx, lw1.updateCh, xdsresource.ErrorTypeNACKed, nodeID); err != nil {`
`1236`	`1247`	`t.Fatal(err)`
`1237`	`1248`	`}`
`1238`	`1249`
Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ func (rw *testRouteConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFu`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`func (rw *testRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {`
`93`		`- rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})`
	`93`	`+ rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})`
`94`	`94`	`onDone()`
`95`	`95`	`}`
`96`	`96`
Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ func (rw *routeConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc)`
`81`	`81`	`}`
`82`	`82`
`83`	`83`	`func (rw *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {`
`84`		`- rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})`
	`84`	`+ rw.updateCh.Replace(routeConfigUpdateErrTuple{err: xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "RouteConfiguration not found in received response")})`
`85`	`85`	`onDone()`
`86`	`86`	`}`
`87`	`87`
`@@ -759,7 +759,7 @@ func (s) TestRDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {`
`759`	`759`	`// Verify that an empty update with the expected error is received.`
`760`	`760`	`ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)`
`761`	`761`	`defer cancel()`
`762`		`- wantErr := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "")`
	`762`	`+ wantErr := xdsresource.NewError(xdsresource.ErrorTypeResourceNotFound, "")`
`763`	`763`	`if err := verifyRouteConfigUpdate(ctx, rw.updateCh, routeConfigUpdateErrTuple{err: wantErr}); err != nil {`
`764`	`764`	`t.Fatal(err)`
`765`	`765`	`}`