From 3f8290045478eb4f9b27a29570ee34921e333917 Mon Sep 17 00:00:00 2001 From: Menghan Li Date: Thu, 9 Sep 2021 13:39:02 -0700 Subject: [PATCH] [xds_client_nack_part] remove Err from Update struct and add a tuple type --- xds/internal/xdsclient/callback.go | 56 +++--- xds/internal/xdsclient/cds_test.go | 36 ++-- xds/internal/xdsclient/client.go | 20 +-- xds/internal/xdsclient/client_test.go | 46 ++--- xds/internal/xdsclient/dump_test.go | 32 ++-- xds/internal/xdsclient/eds_test.go | 16 +- xds/internal/xdsclient/lds_test.go | 160 +++++++++--------- xds/internal/xdsclient/rds_test.go | 34 ++-- xds/internal/xdsclient/v2/cds_test.go | 10 +- xds/internal/xdsclient/v2/client_test.go | 26 +-- xds/internal/xdsclient/v2/eds_test.go | 14 +- xds/internal/xdsclient/v2/lds_test.go | 22 +-- xds/internal/xdsclient/v2/rds_test.go | 20 +-- .../xdsclient/watchers_cluster_test.go | 79 ++++----- .../xdsclient/watchers_endpoints_test.go | 53 +++--- .../xdsclient/watchers_listener_test.go | 63 ++++--- xds/internal/xdsclient/watchers_route_test.go | 49 +++--- xds/internal/xdsclient/xds.go | 64 ++++--- 18 files changed, 396 insertions(+), 404 deletions(-) diff --git a/xds/internal/xdsclient/callback.go b/xds/internal/xdsclient/callback.go index b765af231d0f..c1e73ed19a31 100644 --- a/xds/internal/xdsclient/callback.go +++ b/xds/internal/xdsclient/callback.go @@ -76,7 +76,7 @@ func (c *clientImpl) callCallback(wiu *watcherInfoWithUpdate) { // // A response can contain multiple resources. They will be parsed and put in a // map from resource name to the resource content. -func (c *clientImpl) NewListeners(updates map[string]ListenerUpdate, metadata UpdateMetadata) { +func (c *clientImpl) NewListeners(updates map[string]ListenerUpdateErr, metadata UpdateMetadata) { c.mu.Lock() defer c.mu.Unlock() @@ -84,9 +84,9 @@ func (c *clientImpl) NewListeners(updates map[string]ListenerUpdate, metadata Up if metadata.ErrState != nil { c.ldsVersion = metadata.ErrState.Version } - for name, update := range updates { + for name, uErr := range updates { if s, ok := c.ldsWatchers[name]; ok { - if update.Err != nil { + if uErr.Err != nil { // On error, keep previous version for each resource. But update // status and error. mdCopy := c.ldsMD[name] @@ -94,17 +94,17 @@ func (c *clientImpl) NewListeners(updates map[string]ListenerUpdate, metadata Up mdCopy.Status = metadata.Status c.ldsMD[name] = mdCopy for wi := range s { - wi.newError(update.Err) + wi.newError(uErr.Err) } continue } // If the resource is valid, send the update. for wi := range s { - wi.newUpdate(update) + wi.newUpdate(uErr.Update) } // Sync cache. - c.logger.Debugf("LDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(update)) - c.ldsCache[name] = update + c.logger.Debugf("LDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(uErr)) + c.ldsCache[name] = uErr.Update // Set status to ACK, and clear error state. The metadata might be a // NACK metadata because some other resources in the same response // are invalid. @@ -141,7 +141,7 @@ func (c *clientImpl) NewListeners(updates map[string]ListenerUpdate, metadata Up // // A response can contain multiple resources. They will be parsed and put in a // map from resource name to the resource content. -func (c *clientImpl) NewRouteConfigs(updates map[string]RouteConfigUpdate, metadata UpdateMetadata) { +func (c *clientImpl) NewRouteConfigs(updates map[string]RouteConfigUpdateErr, metadata UpdateMetadata) { c.mu.Lock() defer c.mu.Unlock() @@ -150,9 +150,9 @@ func (c *clientImpl) NewRouteConfigs(updates map[string]RouteConfigUpdate, metad if metadata.ErrState != nil { c.rdsVersion = metadata.ErrState.Version } - for name, update := range updates { + for name, uErr := range updates { if s, ok := c.rdsWatchers[name]; ok { - if update.Err != nil { + if uErr.Err != nil { // On error, keep previous version for each resource. But update // status and error. mdCopy := c.rdsMD[name] @@ -160,17 +160,17 @@ func (c *clientImpl) NewRouteConfigs(updates map[string]RouteConfigUpdate, metad mdCopy.Status = metadata.Status c.rdsMD[name] = mdCopy for wi := range s { - wi.newError(update.Err) + wi.newError(uErr.Err) } continue } // If the resource is valid, send the update. for wi := range s { - wi.newUpdate(update) + wi.newUpdate(uErr.Update) } // Sync cache. - c.logger.Debugf("RDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(update)) - c.rdsCache[name] = update + c.logger.Debugf("RDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(uErr)) + c.rdsCache[name] = uErr.Update // Set status to ACK, and clear error state. The metadata might be a // NACK metadata because some other resources in the same response // are invalid. @@ -190,7 +190,7 @@ func (c *clientImpl) NewRouteConfigs(updates map[string]RouteConfigUpdate, metad // // A response can contain multiple resources. They will be parsed and put in a // map from resource name to the resource content. -func (c *clientImpl) NewClusters(updates map[string]ClusterUpdate, metadata UpdateMetadata) { +func (c *clientImpl) NewClusters(updates map[string]ClusterUpdateErr, metadata UpdateMetadata) { c.mu.Lock() defer c.mu.Unlock() @@ -198,9 +198,9 @@ func (c *clientImpl) NewClusters(updates map[string]ClusterUpdate, metadata Upda if metadata.ErrState != nil { c.cdsVersion = metadata.ErrState.Version } - for name, update := range updates { + for name, uErr := range updates { if s, ok := c.cdsWatchers[name]; ok { - if update.Err != nil { + if uErr.Err != nil { // On error, keep previous version for each resource. But update // status and error. mdCopy := c.cdsMD[name] @@ -210,17 +210,17 @@ func (c *clientImpl) NewClusters(updates map[string]ClusterUpdate, metadata Upda for wi := range s { // Send the watcher the individual error, instead of the // overall combined error from the metadata.ErrState. - wi.newError(update.Err) + wi.newError(uErr.Err) } continue } // If the resource is valid, send the update. for wi := range s { - wi.newUpdate(update) + wi.newUpdate(uErr.Update) } // Sync cache. - c.logger.Debugf("CDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(update)) - c.cdsCache[name] = update + c.logger.Debugf("CDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(uErr)) + c.cdsCache[name] = uErr.Update // Set status to ACK, and clear error state. The metadata might be a // NACK metadata because some other resources in the same response // are invalid. @@ -257,7 +257,7 @@ func (c *clientImpl) NewClusters(updates map[string]ClusterUpdate, metadata Upda // // A response can contain multiple resources. They will be parsed and put in a // map from resource name to the resource content. -func (c *clientImpl) NewEndpoints(updates map[string]EndpointsUpdate, metadata UpdateMetadata) { +func (c *clientImpl) NewEndpoints(updates map[string]EndpointsUpdateErr, metadata UpdateMetadata) { c.mu.Lock() defer c.mu.Unlock() @@ -265,9 +265,9 @@ func (c *clientImpl) NewEndpoints(updates map[string]EndpointsUpdate, metadata U if metadata.ErrState != nil { c.edsVersion = metadata.ErrState.Version } - for name, update := range updates { + for name, uErr := range updates { if s, ok := c.edsWatchers[name]; ok { - if update.Err != nil { + if uErr.Err != nil { // On error, keep previous version for each resource. But update // status and error. mdCopy := c.edsMD[name] @@ -277,17 +277,17 @@ func (c *clientImpl) NewEndpoints(updates map[string]EndpointsUpdate, metadata U for wi := range s { // Send the watcher the individual error, instead of the // overall combined error from the metadata.ErrState. - wi.newError(update.Err) + wi.newError(uErr.Err) } continue } // If the resource is valid, send the update. for wi := range s { - wi.newUpdate(update) + wi.newUpdate(uErr.Update) } // Sync cache. - c.logger.Debugf("EDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(update)) - c.edsCache[name] = update + c.logger.Debugf("EDS resource with name %v, value %+v added to cache", name, pretty.ToJSON(uErr)) + c.edsCache[name] = uErr.Update // Set status to ACK, and clear error state. The metadata might be a // NACK metadata because some other resources in the same response // are invalid. diff --git a/xds/internal/xdsclient/cds_test.go b/xds/internal/xdsclient/cds_test.go index c43a285727fb..b0b8febf0712 100644 --- a/xds/internal/xdsclient/cds_test.go +++ b/xds/internal/xdsclient/cds_test.go @@ -945,7 +945,7 @@ func (s) TestUnmarshalCluster(t *testing.T) { tests := []struct { name string resources []*anypb.Any - wantUpdate map[string]ClusterUpdate + wantUpdate map[string]ClusterUpdateErr wantMD UpdateMetadata wantErr bool }{ @@ -988,7 +988,7 @@ func (s) TestUnmarshalCluster(t *testing.T) { ClusterDiscoveryType: &v3clusterpb.Cluster_Type{Type: v3clusterpb.Cluster_STATIC}, }), }, - wantUpdate: map[string]ClusterUpdate{ + wantUpdate: map[string]ClusterUpdateErr{ "test": {Err: cmpopts.AnyError}, }, wantMD: UpdateMetadata{ @@ -1004,12 +1004,12 @@ func (s) TestUnmarshalCluster(t *testing.T) { { name: "v2 cluster", resources: []*anypb.Any{v2ClusterAny}, - wantUpdate: map[string]ClusterUpdate{ - v2ClusterName: { + wantUpdate: map[string]ClusterUpdateErr{ + v2ClusterName: {Update: ClusterUpdate{ ClusterName: v2ClusterName, EDSServiceName: v2Service, EnableLRS: true, Raw: v2ClusterAny, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -1019,12 +1019,12 @@ func (s) TestUnmarshalCluster(t *testing.T) { { name: "v3 cluster", resources: []*anypb.Any{v3ClusterAny}, - wantUpdate: map[string]ClusterUpdate{ - v3ClusterName: { + wantUpdate: map[string]ClusterUpdateErr{ + v3ClusterName: {Update: ClusterUpdate{ ClusterName: v3ClusterName, EDSServiceName: v3Service, EnableLRS: true, Raw: v3ClusterAny, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -1034,17 +1034,17 @@ func (s) TestUnmarshalCluster(t *testing.T) { { name: "multiple clusters", resources: []*anypb.Any{v2ClusterAny, v3ClusterAny}, - wantUpdate: map[string]ClusterUpdate{ - v2ClusterName: { + wantUpdate: map[string]ClusterUpdateErr{ + v2ClusterName: {Update: ClusterUpdate{ ClusterName: v2ClusterName, EDSServiceName: v2Service, EnableLRS: true, Raw: v2ClusterAny, - }, - v3ClusterName: { + }}, + v3ClusterName: {Update: ClusterUpdate{ ClusterName: v3ClusterName, EDSServiceName: v3Service, EnableLRS: true, Raw: v3ClusterAny, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -1063,17 +1063,17 @@ func (s) TestUnmarshalCluster(t *testing.T) { }), v3ClusterAny, }, - wantUpdate: map[string]ClusterUpdate{ - v2ClusterName: { + wantUpdate: map[string]ClusterUpdateErr{ + v2ClusterName: {Update: ClusterUpdate{ ClusterName: v2ClusterName, EDSServiceName: v2Service, EnableLRS: true, Raw: v2ClusterAny, - }, - v3ClusterName: { + }}, + v3ClusterName: {Update: ClusterUpdate{ ClusterName: v3ClusterName, EDSServiceName: v3Service, EnableLRS: true, Raw: v3ClusterAny, - }, + }}, "bad": {Err: cmpopts.AnyError}, }, wantMD: UpdateMetadata{ diff --git a/xds/internal/xdsclient/client.go b/xds/internal/xdsclient/client.go index 32dbf89b694f..85a9097401ca 100644 --- a/xds/internal/xdsclient/client.go +++ b/xds/internal/xdsclient/client.go @@ -134,14 +134,14 @@ type loadReportingOptions struct { // resource updates from an APIClient for a specific version. type UpdateHandler interface { // NewListeners handles updates to xDS listener resources. - NewListeners(map[string]ListenerUpdate, UpdateMetadata) + NewListeners(map[string]ListenerUpdateErr, UpdateMetadata) // NewRouteConfigs handles updates to xDS RouteConfiguration resources. - NewRouteConfigs(map[string]RouteConfigUpdate, UpdateMetadata) + NewRouteConfigs(map[string]RouteConfigUpdateErr, UpdateMetadata) // NewClusters handles updates to xDS Cluster resources. - NewClusters(map[string]ClusterUpdate, UpdateMetadata) + NewClusters(map[string]ClusterUpdateErr, UpdateMetadata) // NewEndpoints handles updates to xDS ClusterLoadAssignment (or tersely // referred to as Endpoints) resources. - NewEndpoints(map[string]EndpointsUpdate, UpdateMetadata) + NewEndpoints(map[string]EndpointsUpdateErr, UpdateMetadata) // NewConnectionError handles connection errors from the xDS stream. The // error will be reported to all the resource watchers. NewConnectionError(err error) @@ -220,9 +220,6 @@ type ListenerUpdate struct { // Raw is the resource from the xds response. Raw *anypb.Any - // Err is non-nil if the resource in the response is invalid. It contains - // the unmarshal error message. - Err error } // HTTPFilter represents one HTTP filter from an LDS response's HTTP connection @@ -257,9 +254,6 @@ type RouteConfigUpdate struct { VirtualHosts []*VirtualHost // Raw is the resource from the xds response. Raw *anypb.Any - // Err is non-nil if the resource in the response is invalid. It contains - // the unmarshal error message. - Err error } // VirtualHost contains the routes for a list of Domains. @@ -488,9 +482,6 @@ type ClusterUpdate struct { // Raw is the resource from the xds response. Raw *anypb.Any - // Err is non-nil if the resource in the response is invalid. It contains - // the unmarshal error message. - Err error } // OverloadDropConfig contains the config to drop overloads. @@ -540,9 +531,6 @@ type EndpointsUpdate struct { // Raw is the resource from the xds response. Raw *anypb.Any - // Err is non-nil if the resource in the response is invalid. It contains - // the unmarshal error message. - Err error } // Function to be overridden in tests. diff --git a/xds/internal/xdsclient/client_test.go b/xds/internal/xdsclient/client_test.go index c89c526debbe..b224e7839fb0 100644 --- a/xds/internal/xdsclient/client_test.go +++ b/xds/internal/xdsclient/client_test.go @@ -162,7 +162,7 @@ func (s) TestWatchCallAnotherWatch(t *testing.T) { clusterUpdateCh := testutils.NewChannel() firstTime := true client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) // Calls another watch inline, to ensure there's deadlock. client.WatchCluster("another-random-name", func(ClusterUpdate, error) {}) @@ -176,13 +176,13 @@ func (s) TestWatchCallAnotherWatch(t *testing.T) { } wantUpdate := ClusterUpdate{ClusterName: testEDSName} - client.NewClusters(map[string]ClusterUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } wantUpdate2 := ClusterUpdate{ClusterName: testEDSName + "2"} - client.NewClusters(map[string]ClusterUpdate{testCDSName: wantUpdate2}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: {Update: wantUpdate2}}, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate2, nil); err != nil { t.Fatal(err) } @@ -193,15 +193,15 @@ func verifyListenerUpdate(ctx context.Context, updateCh *testutils.Channel, want if err != nil { return fmt.Errorf("timeout when waiting for listener update: %v", err) } - gotUpdate := u.(ldsUpdateErr) + gotUpdate := u.(ListenerUpdateErr) if wantErr != nil { - if gotUpdate.err != wantErr { - return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.err, wantErr) + if gotUpdate.Err != wantErr { + return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.Err, wantErr) } return nil } - if gotUpdate.err != nil || !cmp.Equal(gotUpdate.u, wantUpdate) { - return fmt.Errorf("unexpected endpointsUpdate: (%v, %v), want: (%v, nil)", gotUpdate.u, gotUpdate.err, wantUpdate) + if gotUpdate.Err != nil || !cmp.Equal(gotUpdate.Update, wantUpdate) { + return fmt.Errorf("unexpected endpointsUpdate: (%v, %v), want: (%v, nil)", gotUpdate.Update, gotUpdate.Err, wantUpdate) } return nil } @@ -211,15 +211,15 @@ func verifyRouteConfigUpdate(ctx context.Context, updateCh *testutils.Channel, w if err != nil { return fmt.Errorf("timeout when waiting for route configuration update: %v", err) } - gotUpdate := u.(rdsUpdateErr) + gotUpdate := u.(RouteConfigUpdateErr) if wantErr != nil { - if gotUpdate.err != wantErr { - return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.err, wantErr) + if gotUpdate.Err != wantErr { + return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.Err, wantErr) } return nil } - if gotUpdate.err != nil || !cmp.Equal(gotUpdate.u, wantUpdate) { - return fmt.Errorf("unexpected route config update: (%v, %v), want: (%v, nil)", gotUpdate.u, gotUpdate.err, wantUpdate) + if gotUpdate.Err != nil || !cmp.Equal(gotUpdate.Update, wantUpdate) { + return fmt.Errorf("unexpected route config update: (%v, %v), want: (%v, nil)", gotUpdate.Update, gotUpdate.Err, wantUpdate) } return nil } @@ -229,15 +229,15 @@ func verifyClusterUpdate(ctx context.Context, updateCh *testutils.Channel, wantU if err != nil { return fmt.Errorf("timeout when waiting for cluster update: %v", err) } - gotUpdate := u.(clusterUpdateErr) + gotUpdate := u.(ClusterUpdateErr) if wantErr != nil { - if gotUpdate.err != wantErr { - return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.err, wantErr) + if gotUpdate.Err != wantErr { + return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.Err, wantErr) } return nil } - if !cmp.Equal(gotUpdate.u, wantUpdate) { - return fmt.Errorf("unexpected clusterUpdate: (%v, %v), want: (%v, nil)", gotUpdate.u, gotUpdate.err, wantUpdate) + if !cmp.Equal(gotUpdate.Update, wantUpdate) { + return fmt.Errorf("unexpected clusterUpdate: (%v, %v), want: (%v, nil)", gotUpdate.Update, gotUpdate.Err, wantUpdate) } return nil } @@ -247,15 +247,15 @@ func verifyEndpointsUpdate(ctx context.Context, updateCh *testutils.Channel, wan if err != nil { return fmt.Errorf("timeout when waiting for endpoints update: %v", err) } - gotUpdate := u.(endpointsUpdateErr) + gotUpdate := u.(EndpointsUpdateErr) if wantErr != nil { - if gotUpdate.err != wantErr { - return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.err, wantErr) + if gotUpdate.Err != wantErr { + return fmt.Errorf("unexpected error: %v, want %v", gotUpdate.Err, wantErr) } return nil } - if gotUpdate.err != nil || !cmp.Equal(gotUpdate.u, wantUpdate, cmpopts.EquateEmpty()) { - return fmt.Errorf("unexpected endpointsUpdate: (%v, %v), want: (%v, nil)", gotUpdate.u, gotUpdate.err, wantUpdate) + if gotUpdate.Err != nil || !cmp.Equal(gotUpdate.Update, wantUpdate, cmpopts.EquateEmpty()) { + return fmt.Errorf("unexpected endpointsUpdate: (%v, %v), want: (%v, nil)", gotUpdate.Update, gotUpdate.Err, wantUpdate) } return nil } diff --git a/xds/internal/xdsclient/dump_test.go b/xds/internal/xdsclient/dump_test.go index 6670da581654..125c65e6bc7a 100644 --- a/xds/internal/xdsclient/dump_test.go +++ b/xds/internal/xdsclient/dump_test.go @@ -101,10 +101,10 @@ func (s) TestLDSConfigDump(t *testing.T) { t.Fatalf(err.Error()) } - update0 := make(map[string]xdsclient.ListenerUpdate) + update0 := make(map[string]xdsclient.ListenerUpdateErr) want0 := make(map[string]xdsclient.UpdateWithMD) for n, r := range listenerRaws { - update0[n] = xdsclient.ListenerUpdate{Raw: r} + update0[n] = xdsclient.ListenerUpdateErr{Update: xdsclient.ListenerUpdate{Raw: r}} want0[n] = xdsclient.UpdateWithMD{ MD: xdsclient.UpdateMetadata{Status: xdsclient.ServiceStatusACKed, Version: testVersion}, Raw: r, @@ -120,9 +120,9 @@ func (s) TestLDSConfigDump(t *testing.T) { const nackVersion = "lds-version-nack" var nackErr = fmt.Errorf("lds nack error") updateHandler.NewListeners( - map[string]xdsclient.ListenerUpdate{ + map[string]xdsclient.ListenerUpdateErr{ ldsTargets[0]: {Err: nackErr}, - ldsTargets[1]: {Raw: listenerRaws[ldsTargets[1]]}, + ldsTargets[1]: {Update: xdsclient.ListenerUpdate{Raw: listenerRaws[ldsTargets[1]]}}, }, xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusNACKed, @@ -215,10 +215,10 @@ func (s) TestRDSConfigDump(t *testing.T) { t.Fatalf(err.Error()) } - update0 := make(map[string]xdsclient.RouteConfigUpdate) + update0 := make(map[string]xdsclient.RouteConfigUpdateErr) want0 := make(map[string]xdsclient.UpdateWithMD) for n, r := range routeRaws { - update0[n] = xdsclient.RouteConfigUpdate{Raw: r} + update0[n] = xdsclient.RouteConfigUpdateErr{Update: xdsclient.RouteConfigUpdate{Raw: r}} want0[n] = xdsclient.UpdateWithMD{ MD: xdsclient.UpdateMetadata{Status: xdsclient.ServiceStatusACKed, Version: testVersion}, Raw: r, @@ -234,9 +234,9 @@ func (s) TestRDSConfigDump(t *testing.T) { const nackVersion = "rds-version-nack" var nackErr = fmt.Errorf("rds nack error") updateHandler.NewRouteConfigs( - map[string]xdsclient.RouteConfigUpdate{ + map[string]xdsclient.RouteConfigUpdateErr{ rdsTargets[0]: {Err: nackErr}, - rdsTargets[1]: {Raw: routeRaws[rdsTargets[1]]}, + rdsTargets[1]: {Update: xdsclient.RouteConfigUpdate{Raw: routeRaws[rdsTargets[1]]}}, }, xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusNACKed, @@ -329,10 +329,10 @@ func (s) TestCDSConfigDump(t *testing.T) { t.Fatalf(err.Error()) } - update0 := make(map[string]xdsclient.ClusterUpdate) + update0 := make(map[string]xdsclient.ClusterUpdateErr) want0 := make(map[string]xdsclient.UpdateWithMD) for n, r := range clusterRaws { - update0[n] = xdsclient.ClusterUpdate{Raw: r} + update0[n] = xdsclient.ClusterUpdateErr{Update: xdsclient.ClusterUpdate{Raw: r}} want0[n] = xdsclient.UpdateWithMD{ MD: xdsclient.UpdateMetadata{Status: xdsclient.ServiceStatusACKed, Version: testVersion}, Raw: r, @@ -348,9 +348,9 @@ func (s) TestCDSConfigDump(t *testing.T) { const nackVersion = "cds-version-nack" var nackErr = fmt.Errorf("cds nack error") updateHandler.NewClusters( - map[string]xdsclient.ClusterUpdate{ + map[string]xdsclient.ClusterUpdateErr{ cdsTargets[0]: {Err: nackErr}, - cdsTargets[1]: {Raw: clusterRaws[cdsTargets[1]]}, + cdsTargets[1]: {Update: xdsclient.ClusterUpdate{Raw: clusterRaws[cdsTargets[1]]}}, }, xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusNACKed, @@ -429,10 +429,10 @@ func (s) TestEDSConfigDump(t *testing.T) { t.Fatalf(err.Error()) } - update0 := make(map[string]xdsclient.EndpointsUpdate) + update0 := make(map[string]xdsclient.EndpointsUpdateErr) want0 := make(map[string]xdsclient.UpdateWithMD) for n, r := range endpointRaws { - update0[n] = xdsclient.EndpointsUpdate{Raw: r} + update0[n] = xdsclient.EndpointsUpdateErr{Update: xdsclient.EndpointsUpdate{Raw: r}} want0[n] = xdsclient.UpdateWithMD{ MD: xdsclient.UpdateMetadata{Status: xdsclient.ServiceStatusACKed, Version: testVersion}, Raw: r, @@ -448,9 +448,9 @@ func (s) TestEDSConfigDump(t *testing.T) { const nackVersion = "eds-version-nack" var nackErr = fmt.Errorf("eds nack error") updateHandler.NewEndpoints( - map[string]xdsclient.EndpointsUpdate{ + map[string]xdsclient.EndpointsUpdateErr{ edsTargets[0]: {Err: nackErr}, - edsTargets[1]: {Raw: endpointRaws[edsTargets[1]]}, + edsTargets[1]: {Update: xdsclient.EndpointsUpdate{Raw: endpointRaws[edsTargets[1]]}}, }, xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusNACKed, diff --git a/xds/internal/xdsclient/eds_test.go b/xds/internal/xdsclient/eds_test.go index ae1499e09e0e..1442c6bc7bd3 100644 --- a/xds/internal/xdsclient/eds_test.go +++ b/xds/internal/xdsclient/eds_test.go @@ -138,7 +138,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { tests := []struct { name string resources []*anypb.Any - wantUpdate map[string]EndpointsUpdate + wantUpdate map[string]EndpointsUpdateErr wantMD UpdateMetadata wantErr bool }{ @@ -181,7 +181,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { clab0.addLocality("locality-2", 1, 2, []string{"addr2:159"}, nil) return clab0.Build() }())}, - wantUpdate: map[string]EndpointsUpdate{"test": {Err: cmpopts.AnyError}}, + wantUpdate: map[string]EndpointsUpdateErr{"test": {Err: cmpopts.AnyError}}, wantMD: UpdateMetadata{ Status: ServiceStatusNACKed, Version: testVersion, @@ -195,8 +195,8 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { { name: "v3 endpoints", resources: []*anypb.Any{v3EndpointsAny}, - wantUpdate: map[string]EndpointsUpdate{ - "test": { + wantUpdate: map[string]EndpointsUpdateErr{ + "test": {Update: EndpointsUpdate{ Drops: nil, Localities: []Locality{ { @@ -221,7 +221,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { }, }, Raw: v3EndpointsAny, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -240,8 +240,8 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { return clab0.Build() }()), }, - wantUpdate: map[string]EndpointsUpdate{ - "test": { + wantUpdate: map[string]EndpointsUpdateErr{ + "test": {Update: EndpointsUpdate{ Drops: nil, Localities: []Locality{ { @@ -266,7 +266,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { }, }, Raw: v3EndpointsAny, - }, + }}, "bad": {Err: cmpopts.AnyError}, }, wantMD: UpdateMetadata{ diff --git a/xds/internal/xdsclient/lds_test.go b/xds/internal/xdsclient/lds_test.go index 5f3efc94a843..aad1019683d2 100644 --- a/xds/internal/xdsclient/lds_test.go +++ b/xds/internal/xdsclient/lds_test.go @@ -185,7 +185,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { tests := []struct { name string resources []*anypb.Any - wantUpdate map[string]ListenerUpdate + wantUpdate map[string]ListenerUpdateErr wantMD UpdateMetadata wantErr bool }{ @@ -215,7 +215,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }(), }, }, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, @@ -227,7 +227,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { ApiListener: testutils.MarshalAny(&v2xdspb.Listener{}), }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, @@ -243,7 +243,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }), }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, @@ -257,7 +257,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }), }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, @@ -280,7 +280,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }), }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, @@ -294,8 +294,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with no filters", resources: []*anypb.Any{v3LisWithFilters()}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: {RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: routerFilterList, Raw: v3LisWithFilters()}, + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: routerFilterList, Raw: v3LisWithFilters()}}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -323,15 +323,15 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }), }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, { name: "v3 with custom filter", resources: []*anypb.Any{v3LisWithFilters(customFilter)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: []HTTPFilter{ { @@ -342,7 +342,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { routerFilter, }, Raw: v3LisWithFilters(customFilter), - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -352,8 +352,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with custom filter in typed struct", resources: []*anypb.Any{v3LisWithFilters(typedStructFilter)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: []HTTPFilter{ { @@ -364,7 +364,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { routerFilter, }, Raw: v3LisWithFilters(typedStructFilter), - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -374,8 +374,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with optional custom filter", resources: []*anypb.Any{v3LisWithFilters(customOptionalFilter)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: []HTTPFilter{ { @@ -386,7 +386,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { routerFilter, }, Raw: v3LisWithFilters(customOptionalFilter), - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -396,15 +396,15 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with two filters with same name", resources: []*anypb.Any{v3LisWithFilters(customFilter, customFilter)}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, { name: "v3 with two filters - same type different name", resources: []*anypb.Any{v3LisWithFilters(customFilter, customFilter2)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: []HTTPFilter{{ Name: "customFilter", @@ -418,7 +418,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { routerFilter, }, Raw: v3LisWithFilters(customFilter, customFilter2), - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -428,20 +428,20 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with server-only filter", resources: []*anypb.Any{v3LisWithFilters(serverOnlyCustomFilter)}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, { name: "v3 with optional server-only filter", resources: []*anypb.Any{v3LisWithFilters(serverOnlyOptionalCustomFilter)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, Raw: v3LisWithFilters(serverOnlyOptionalCustomFilter), HTTPFilters: routerFilterList, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -451,8 +451,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with client-only filter", resources: []*anypb.Any{v3LisWithFilters(clientOnlyCustomFilter)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: []HTTPFilter{ { @@ -462,7 +462,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }, routerFilter}, Raw: v3LisWithFilters(clientOnlyCustomFilter), - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -472,34 +472,34 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 with err filter", resources: []*anypb.Any{v3LisWithFilters(errFilter)}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, { name: "v3 with optional err filter", resources: []*anypb.Any{v3LisWithFilters(errOptionalFilter)}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, { name: "v3 with unknown filter", resources: []*anypb.Any{v3LisWithFilters(unknownFilter)}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: true, }, { name: "v3 with unknown filter (optional)", resources: []*anypb.Any{v3LisWithFilters(unknownOptionalFilter)}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: routerFilterList, Raw: v3LisWithFilters(unknownOptionalFilter), - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -509,8 +509,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v2 listener resource", resources: []*anypb.Any{v2Lis}, - wantUpdate: map[string]ListenerUpdate{ - v2LDSTarget: {RouteConfigName: v2RouteConfigName, Raw: v2Lis}, + wantUpdate: map[string]ListenerUpdateErr{ + v2LDSTarget: {Update: ListenerUpdate{RouteConfigName: v2RouteConfigName, Raw: v2Lis}}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -520,8 +520,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 listener resource", resources: []*anypb.Any{v3LisWithFilters()}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: {RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: routerFilterList, Raw: v3LisWithFilters()}, + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, HTTPFilters: routerFilterList, Raw: v3LisWithFilters()}}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -531,8 +531,8 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "v3 listener with inline route configuration", resources: []*anypb.Any{v3LisWithInlineRoute}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ InlineRouteConfig: &RouteConfigUpdate{ VirtualHosts: []*VirtualHost{{ Domains: []string{v3LDSTarget}, @@ -541,7 +541,7 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { MaxStreamDuration: time.Second, Raw: v3LisWithInlineRoute, HTTPFilters: routerFilterList, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -551,9 +551,9 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { { name: "multiple listener resources", resources: []*anypb.Any{v2Lis, v3LisWithFilters()}, - wantUpdate: map[string]ListenerUpdate{ - v2LDSTarget: {RouteConfigName: v2RouteConfigName, Raw: v2Lis}, - v3LDSTarget: {RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, Raw: v3LisWithFilters(), HTTPFilters: routerFilterList}, + wantUpdate: map[string]ListenerUpdateErr{ + v2LDSTarget: {Update: ListenerUpdate{RouteConfigName: v2RouteConfigName, Raw: v2Lis}}, + v3LDSTarget: {Update: ListenerUpdate{RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, Raw: v3LisWithFilters(), HTTPFilters: routerFilterList}}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -574,9 +574,9 @@ func (s) TestUnmarshalListener_ClientSide(t *testing.T) { }}), v3LisWithFilters(), }, - wantUpdate: map[string]ListenerUpdate{ - v2LDSTarget: {RouteConfigName: v2RouteConfigName, Raw: v2Lis}, - v3LDSTarget: {RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, Raw: v3LisWithFilters(), HTTPFilters: routerFilterList}, + wantUpdate: map[string]ListenerUpdateErr{ + v2LDSTarget: {Update: ListenerUpdate{RouteConfigName: v2RouteConfigName, Raw: v2Lis}}, + v3LDSTarget: {Update: ListenerUpdate{RouteConfigName: v3RouteConfigName, MaxStreamDuration: time.Second, Raw: v3LisWithFilters(), HTTPFilters: routerFilterList}}, "bad": {Err: cmpopts.AnyError}, }, wantMD: errMD, @@ -766,7 +766,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { tests := []struct { name string resources []*anypb.Any - wantUpdate map[string]ListenerUpdate + wantUpdate map[string]ListenerUpdateErr wantMD UpdateMetadata wantErr string }{ @@ -778,7 +778,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { {Name: "listener-filter-1"}, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "unsupported field 'listener_filters'", }, @@ -788,14 +788,14 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { Name: v3LDSTarget, UseOriginalDst: &wrapperspb.BoolValue{Value: true}, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "unsupported field 'use_original_dst'", }, { name: "no address field", resources: []*anypb.Any{testutils.MarshalAny(&v3listenerpb.Listener{Name: v3LDSTarget})}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "no address field in LDS response", }, @@ -805,7 +805,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { Name: v3LDSTarget, Address: &v3corepb.Address{}, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "no socket_address field in LDS response", }, @@ -821,7 +821,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "no supported filter chains and no default filter chain", }, @@ -836,7 +836,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "missing HttpConnectionManager filter", }, @@ -858,7 +858,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "missing name field in filter", }, @@ -897,7 +897,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "duplicate filter name", }, @@ -924,7 +924,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "http filters list is empty", }, @@ -952,7 +952,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "is a terminal filter but it is not last in the filter chain", }, @@ -980,7 +980,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "is not a terminal filter", }, @@ -1001,7 +1001,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "unsupported config_type", }, @@ -1025,7 +1025,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "multiple filter chains with overlapping matching rules are defined", }, @@ -1048,7 +1048,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "unsupported network filter", }, @@ -1074,7 +1074,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "failed unmarshaling of network filter", }, @@ -1093,7 +1093,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "transport_socket field has unexpected name", }, @@ -1115,7 +1115,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "transport_socket field has unexpected typeURL", }, @@ -1140,7 +1140,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "failed to unmarshal DownstreamTlsContext in LDS response", }, @@ -1162,7 +1162,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "DownstreamTlsContext in LDS response does not contain a CommonTlsContext", }, @@ -1192,15 +1192,15 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "validation context contains unexpected type", }, { name: "empty transport socket", resources: []*anypb.Any{listenerEmptyTransportSocket}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ InboundListenerCfg: &InboundListenerConfig{ Address: "0.0.0.0", Port: "9999", @@ -1226,7 +1226,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, Raw: listenerEmptyTransportSocket, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -1259,7 +1259,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "security configuration on the server-side does not contain root certificate provider instance name, but require_client_cert field is set", }, @@ -1283,15 +1283,15 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, })}, - wantUpdate: map[string]ListenerUpdate{v3LDSTarget: {Err: cmpopts.AnyError}}, + wantUpdate: map[string]ListenerUpdateErr{v3LDSTarget: {Err: cmpopts.AnyError}}, wantMD: errMD, wantErr: "security configuration on the server-side does not contain identity certificate provider instance name", }, { name: "happy case with no validation context", resources: []*anypb.Any{listenerNoValidationContext}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ InboundListenerCfg: &InboundListenerConfig{ Address: "0.0.0.0", Port: "9999", @@ -1329,7 +1329,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, Raw: listenerNoValidationContext, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -1339,8 +1339,8 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { { name: "happy case with validation context provider instance", resources: []*anypb.Any{listenerWithValidationContext}, - wantUpdate: map[string]ListenerUpdate{ - v3LDSTarget: { + wantUpdate: map[string]ListenerUpdateErr{ + v3LDSTarget: {Update: ListenerUpdate{ InboundListenerCfg: &InboundListenerConfig{ Address: "0.0.0.0", Port: "9999", @@ -1384,7 +1384,7 @@ func (s) TestUnmarshalListener_ServerSide(t *testing.T) { }, }, Raw: listenerWithValidationContext, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, diff --git a/xds/internal/xdsclient/rds_test.go b/xds/internal/xdsclient/rds_test.go index a570e43ebd2c..58263628613e 100644 --- a/xds/internal/xdsclient/rds_test.go +++ b/xds/internal/xdsclient/rds_test.go @@ -668,7 +668,7 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { tests := []struct { name string resources []*anypb.Any - wantUpdate map[string]RouteConfigUpdate + wantUpdate map[string]RouteConfigUpdateErr wantMD UpdateMetadata wantErr bool }{ @@ -713,8 +713,8 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { { name: "v2 routeConfig resource", resources: []*anypb.Any{v2RouteConfig}, - wantUpdate: map[string]RouteConfigUpdate{ - v2RouteConfigName: { + wantUpdate: map[string]RouteConfigUpdateErr{ + v2RouteConfigName: {Update: RouteConfigUpdate{ VirtualHosts: []*VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -730,7 +730,7 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }, }, Raw: v2RouteConfig, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -740,8 +740,8 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { { name: "v3 routeConfig resource", resources: []*anypb.Any{v3RouteConfig}, - wantUpdate: map[string]RouteConfigUpdate{ - v3RouteConfigName: { + wantUpdate: map[string]RouteConfigUpdateErr{ + v3RouteConfigName: {Update: RouteConfigUpdate{ VirtualHosts: []*VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -757,7 +757,7 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }, }, Raw: v3RouteConfig, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -767,8 +767,8 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { { name: "multiple routeConfig resources", resources: []*anypb.Any{v2RouteConfig, v3RouteConfig}, - wantUpdate: map[string]RouteConfigUpdate{ - v3RouteConfigName: { + wantUpdate: map[string]RouteConfigUpdateErr{ + v3RouteConfigName: {Update: RouteConfigUpdate{ VirtualHosts: []*VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -784,8 +784,8 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }, }, Raw: v3RouteConfig, - }, - v2RouteConfigName: { + }}, + v2RouteConfigName: {Update: RouteConfigUpdate{ VirtualHosts: []*VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -801,7 +801,7 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }, }, Raw: v2RouteConfig, - }, + }}, }, wantMD: UpdateMetadata{ Status: ServiceStatusACKed, @@ -822,8 +822,8 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }}}}}), v3RouteConfig, }, - wantUpdate: map[string]RouteConfigUpdate{ - v3RouteConfigName: { + wantUpdate: map[string]RouteConfigUpdateErr{ + v3RouteConfigName: {Update: RouteConfigUpdate{ VirtualHosts: []*VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -839,8 +839,8 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }, }, Raw: v3RouteConfig, - }, - v2RouteConfigName: { + }}, + v2RouteConfigName: {Update: RouteConfigUpdate{ VirtualHosts: []*VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -856,7 +856,7 @@ func (s) TestUnmarshalRouteConfig(t *testing.T) { }, }, Raw: v2RouteConfig, - }, + }}, "bad": {Err: cmpopts.AnyError}, }, wantMD: UpdateMetadata{ diff --git a/xds/internal/xdsclient/v2/cds_test.go b/xds/internal/xdsclient/v2/cds_test.go index b1683db17ad4..b81e72b4f580 100644 --- a/xds/internal/xdsclient/v2/cds_test.go +++ b/xds/internal/xdsclient/v2/cds_test.go @@ -101,7 +101,7 @@ func (s) TestCDSHandleResponse(t *testing.T) { name string cdsResponse *xdspb.DiscoveryResponse wantErr bool - wantUpdate map[string]xdsclient.ClusterUpdate + wantUpdate map[string]xdsclient.ClusterUpdateErr wantUpdateMD xdsclient.UpdateMetadata wantUpdateErr bool }{ @@ -149,8 +149,8 @@ func (s) TestCDSHandleResponse(t *testing.T) { name: "one-uninteresting-cluster", cdsResponse: goodCDSResponse2, wantErr: false, - wantUpdate: map[string]xdsclient.ClusterUpdate{ - goodClusterName2: {ClusterName: goodClusterName2, EDSServiceName: serviceName2, Raw: marshaledCluster2}, + wantUpdate: map[string]xdsclient.ClusterUpdateErr{ + goodClusterName2: {Update: xdsclient.ClusterUpdate{ClusterName: goodClusterName2, EDSServiceName: serviceName2, Raw: marshaledCluster2}}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, @@ -162,8 +162,8 @@ func (s) TestCDSHandleResponse(t *testing.T) { name: "one-good-cluster", cdsResponse: goodCDSResponse1, wantErr: false, - wantUpdate: map[string]xdsclient.ClusterUpdate{ - goodClusterName1: {ClusterName: goodClusterName1, EDSServiceName: serviceName1, EnableLRS: true, Raw: marshaledCluster1}, + wantUpdate: map[string]xdsclient.ClusterUpdateErr{ + goodClusterName1: {Update: xdsclient.ClusterUpdate{ClusterName: goodClusterName1, EDSServiceName: serviceName1, EnableLRS: true, Raw: marshaledCluster1}}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, diff --git a/xds/internal/xdsclient/v2/client_test.go b/xds/internal/xdsclient/v2/client_test.go index b9e5c4d6a7ba..f04aa8fbcc47 100644 --- a/xds/internal/xdsclient/v2/client_test.go +++ b/xds/internal/xdsclient/v2/client_test.go @@ -301,7 +301,7 @@ type testUpdateReceiver struct { f func(rType xdsclient.ResourceType, d map[string]interface{}, md xdsclient.UpdateMetadata) } -func (t *testUpdateReceiver) NewListeners(d map[string]xdsclient.ListenerUpdate, metadata xdsclient.UpdateMetadata) { +func (t *testUpdateReceiver) NewListeners(d map[string]xdsclient.ListenerUpdateErr, metadata xdsclient.UpdateMetadata) { dd := make(map[string]interface{}) for k, v := range d { dd[k] = v @@ -309,7 +309,7 @@ func (t *testUpdateReceiver) NewListeners(d map[string]xdsclient.ListenerUpdate, t.newUpdate(xdsclient.ListenerResource, dd, metadata) } -func (t *testUpdateReceiver) NewRouteConfigs(d map[string]xdsclient.RouteConfigUpdate, metadata xdsclient.UpdateMetadata) { +func (t *testUpdateReceiver) NewRouteConfigs(d map[string]xdsclient.RouteConfigUpdateErr, metadata xdsclient.UpdateMetadata) { dd := make(map[string]interface{}) for k, v := range d { dd[k] = v @@ -317,7 +317,7 @@ func (t *testUpdateReceiver) NewRouteConfigs(d map[string]xdsclient.RouteConfigU t.newUpdate(xdsclient.RouteConfigResource, dd, metadata) } -func (t *testUpdateReceiver) NewClusters(d map[string]xdsclient.ClusterUpdate, metadata xdsclient.UpdateMetadata) { +func (t *testUpdateReceiver) NewClusters(d map[string]xdsclient.ClusterUpdateErr, metadata xdsclient.UpdateMetadata) { dd := make(map[string]interface{}) for k, v := range d { dd[k] = v @@ -325,7 +325,7 @@ func (t *testUpdateReceiver) NewClusters(d map[string]xdsclient.ClusterUpdate, m t.newUpdate(xdsclient.ClusterResource, dd, metadata) } -func (t *testUpdateReceiver) NewEndpoints(d map[string]xdsclient.EndpointsUpdate, metadata xdsclient.UpdateMetadata) { +func (t *testUpdateReceiver) NewEndpoints(d map[string]xdsclient.EndpointsUpdateErr, metadata xdsclient.UpdateMetadata) { dd := make(map[string]interface{}) for k, v := range d { dd[k] = v @@ -363,27 +363,27 @@ func testWatchHandle(t *testing.T, test *watchHandleTestcase) { if rType == test.rType { switch test.rType { case xdsclient.ListenerResource: - dd := make(map[string]xdsclient.ListenerUpdate) + dd := make(map[string]xdsclient.ListenerUpdateErr) for n, u := range d { - dd[n] = u.(xdsclient.ListenerUpdate) + dd[n] = u.(xdsclient.ListenerUpdateErr) } gotUpdateCh.Send(updateErr{dd, md, nil}) case xdsclient.RouteConfigResource: - dd := make(map[string]xdsclient.RouteConfigUpdate) + dd := make(map[string]xdsclient.RouteConfigUpdateErr) for n, u := range d { - dd[n] = u.(xdsclient.RouteConfigUpdate) + dd[n] = u.(xdsclient.RouteConfigUpdateErr) } gotUpdateCh.Send(updateErr{dd, md, nil}) case xdsclient.ClusterResource: - dd := make(map[string]xdsclient.ClusterUpdate) + dd := make(map[string]xdsclient.ClusterUpdateErr) for n, u := range d { - dd[n] = u.(xdsclient.ClusterUpdate) + dd[n] = u.(xdsclient.ClusterUpdateErr) } gotUpdateCh.Send(updateErr{dd, md, nil}) case xdsclient.EndpointsResource: - dd := make(map[string]xdsclient.EndpointsUpdate) + dd := make(map[string]xdsclient.EndpointsUpdateErr) for n, u := range d { - dd[n] = u.(xdsclient.EndpointsUpdate) + dd[n] = u.(xdsclient.EndpointsUpdateErr) } gotUpdateCh.Send(updateErr{dd, md, nil}) } @@ -664,7 +664,7 @@ func (s) TestV2ClientWatchWithoutStream(t *testing.T) { if v, err := callbackCh.Receive(ctx); err != nil { t.Fatal("Timeout when expecting LDS update") - } else if _, ok := v.(xdsclient.ListenerUpdate); !ok { + } else if _, ok := v.(xdsclient.ListenerUpdateErr); !ok { t.Fatalf("Expect an LDS update from watcher, got %v", v) } } diff --git a/xds/internal/xdsclient/v2/eds_test.go b/xds/internal/xdsclient/v2/eds_test.go index 5d26d0a75b8c..04bddc12db7c 100644 --- a/xds/internal/xdsclient/v2/eds_test.go +++ b/xds/internal/xdsclient/v2/eds_test.go @@ -76,7 +76,7 @@ func (s) TestEDSHandleResponse(t *testing.T) { name string edsResponse *v2xdspb.DiscoveryResponse wantErr bool - wantUpdate map[string]xdsclient.EndpointsUpdate + wantUpdate map[string]xdsclient.EndpointsUpdateErr wantUpdateMD xdsclient.UpdateMetadata wantUpdateErr bool }{ @@ -113,8 +113,8 @@ func (s) TestEDSHandleResponse(t *testing.T) { name: "one-uninterestring-assignment", edsResponse: goodEDSResponse2, wantErr: false, - wantUpdate: map[string]xdsclient.EndpointsUpdate{ - "not-goodEDSName": { + wantUpdate: map[string]xdsclient.EndpointsUpdateErr{ + "not-goodEDSName": {Update: xdsclient.EndpointsUpdate{ Localities: []xdsclient.Locality{ { Endpoints: []xdsclient.Endpoint{{Address: "addr1:314"}}, @@ -124,7 +124,7 @@ func (s) TestEDSHandleResponse(t *testing.T) { }, }, Raw: marshaledGoodCLA2, - }, + }}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, @@ -136,8 +136,8 @@ func (s) TestEDSHandleResponse(t *testing.T) { name: "one-good-assignment", edsResponse: goodEDSResponse1, wantErr: false, - wantUpdate: map[string]xdsclient.EndpointsUpdate{ - goodEDSName: { + wantUpdate: map[string]xdsclient.EndpointsUpdateErr{ + goodEDSName: {Update: xdsclient.EndpointsUpdate{ Localities: []xdsclient.Locality{ { Endpoints: []xdsclient.Endpoint{{Address: "addr1:314"}}, @@ -153,7 +153,7 @@ func (s) TestEDSHandleResponse(t *testing.T) { }, }, Raw: marshaledGoodCLA1, - }, + }}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, diff --git a/xds/internal/xdsclient/v2/lds_test.go b/xds/internal/xdsclient/v2/lds_test.go index 3d2199a150d3..3e9fff0f580a 100644 --- a/xds/internal/xdsclient/v2/lds_test.go +++ b/xds/internal/xdsclient/v2/lds_test.go @@ -36,7 +36,7 @@ func (s) TestLDSHandleResponse(t *testing.T) { name string ldsResponse *v2xdspb.DiscoveryResponse wantErr bool - wantUpdate map[string]xdsclient.ListenerUpdate + wantUpdate map[string]xdsclient.ListenerUpdateErr wantUpdateMD xdsclient.UpdateMetadata wantUpdateErr bool }{ @@ -75,7 +75,7 @@ func (s) TestLDSHandleResponse(t *testing.T) { name: "no-apiListener-in-response", ldsResponse: noAPIListenerLDSResponse, wantErr: true, - wantUpdate: map[string]xdsclient.ListenerUpdate{ + wantUpdate: map[string]xdsclient.ListenerUpdateErr{ goodLDSTarget1: {Err: cmpopts.AnyError}, }, wantUpdateMD: xdsclient.UpdateMetadata{ @@ -91,8 +91,8 @@ func (s) TestLDSHandleResponse(t *testing.T) { name: "one-good-listener", ldsResponse: goodLDSResponse1, wantErr: false, - wantUpdate: map[string]xdsclient.ListenerUpdate{ - goodLDSTarget1: {RouteConfigName: goodRouteName1, Raw: marshaledListener1}, + wantUpdate: map[string]xdsclient.ListenerUpdateErr{ + goodLDSTarget1: {Update: xdsclient.ListenerUpdate{RouteConfigName: goodRouteName1, Raw: marshaledListener1}}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, @@ -105,9 +105,9 @@ func (s) TestLDSHandleResponse(t *testing.T) { name: "multiple-good-listener", ldsResponse: ldsResponseWithMultipleResources, wantErr: false, - wantUpdate: map[string]xdsclient.ListenerUpdate{ - goodLDSTarget1: {RouteConfigName: goodRouteName1, Raw: marshaledListener1}, - goodLDSTarget2: {RouteConfigName: goodRouteName1, Raw: marshaledListener2}, + wantUpdate: map[string]xdsclient.ListenerUpdateErr{ + goodLDSTarget1: {Update: xdsclient.ListenerUpdate{RouteConfigName: goodRouteName1, Raw: marshaledListener1}}, + goodLDSTarget2: {Update: xdsclient.ListenerUpdate{RouteConfigName: goodRouteName1, Raw: marshaledListener2}}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, @@ -121,8 +121,8 @@ func (s) TestLDSHandleResponse(t *testing.T) { name: "good-bad-ugly-listeners", ldsResponse: goodBadUglyLDSResponse, wantErr: true, - wantUpdate: map[string]xdsclient.ListenerUpdate{ - goodLDSTarget1: {RouteConfigName: goodRouteName1, Raw: marshaledListener1}, + wantUpdate: map[string]xdsclient.ListenerUpdateErr{ + goodLDSTarget1: {Update: xdsclient.ListenerUpdate{RouteConfigName: goodRouteName1, Raw: marshaledListener1}}, goodLDSTarget2: {Err: cmpopts.AnyError}, }, wantUpdateMD: xdsclient.UpdateMetadata{ @@ -138,8 +138,8 @@ func (s) TestLDSHandleResponse(t *testing.T) { name: "one-uninteresting-listener", ldsResponse: goodLDSResponse2, wantErr: false, - wantUpdate: map[string]xdsclient.ListenerUpdate{ - goodLDSTarget2: {RouteConfigName: goodRouteName1, Raw: marshaledListener2}, + wantUpdate: map[string]xdsclient.ListenerUpdateErr{ + goodLDSTarget2: {Update: xdsclient.ListenerUpdate{RouteConfigName: goodRouteName1, Raw: marshaledListener2}}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, diff --git a/xds/internal/xdsclient/v2/rds_test.go b/xds/internal/xdsclient/v2/rds_test.go index 2b15958d6f2a..32b721387bed 100644 --- a/xds/internal/xdsclient/v2/rds_test.go +++ b/xds/internal/xdsclient/v2/rds_test.go @@ -50,7 +50,7 @@ func (s) TestRDSHandleResponseWithRouting(t *testing.T) { name string rdsResponse *xdspb.DiscoveryResponse wantErr bool - wantUpdate map[string]xdsclient.RouteConfigUpdate + wantUpdate map[string]xdsclient.RouteConfigUpdateErr wantUpdateMD xdsclient.UpdateMetadata wantUpdateErr bool }{ @@ -89,11 +89,11 @@ func (s) TestRDSHandleResponseWithRouting(t *testing.T) { name: "no-virtual-hosts-in-response", rdsResponse: noVirtualHostsInRDSResponse, wantErr: false, - wantUpdate: map[string]xdsclient.RouteConfigUpdate{ - goodRouteName1: { + wantUpdate: map[string]xdsclient.RouteConfigUpdateErr{ + goodRouteName1: {Update: xdsclient.RouteConfigUpdate{ VirtualHosts: nil, Raw: marshaledNoVirtualHostsRouteConfig, - }, + }}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, @@ -105,8 +105,8 @@ func (s) TestRDSHandleResponseWithRouting(t *testing.T) { name: "one-uninteresting-route-config", rdsResponse: goodRDSResponse2, wantErr: false, - wantUpdate: map[string]xdsclient.RouteConfigUpdate{ - goodRouteName2: { + wantUpdate: map[string]xdsclient.RouteConfigUpdateErr{ + goodRouteName2: {Update: xdsclient.RouteConfigUpdate{ VirtualHosts: []*xdsclient.VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -123,7 +123,7 @@ func (s) TestRDSHandleResponseWithRouting(t *testing.T) { }, }, Raw: marshaledGoodRouteConfig2, - }, + }}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, @@ -135,8 +135,8 @@ func (s) TestRDSHandleResponseWithRouting(t *testing.T) { name: "one-good-route-config", rdsResponse: goodRDSResponse1, wantErr: false, - wantUpdate: map[string]xdsclient.RouteConfigUpdate{ - goodRouteName1: { + wantUpdate: map[string]xdsclient.RouteConfigUpdateErr{ + goodRouteName1: {Update: xdsclient.RouteConfigUpdate{ VirtualHosts: []*xdsclient.VirtualHost{ { Domains: []string{uninterestingDomain}, @@ -153,7 +153,7 @@ func (s) TestRDSHandleResponseWithRouting(t *testing.T) { }, }, Raw: marshaledGoodRouteConfig1, - }, + }}, }, wantUpdateMD: xdsclient.UpdateMetadata{ Status: xdsclient.ServiceStatusACKed, diff --git a/xds/internal/xdsclient/watchers_cluster_test.go b/xds/internal/xdsclient/watchers_cluster_test.go index 805f479c9f9c..0dd1dcec65d0 100644 --- a/xds/internal/xdsclient/watchers_cluster_test.go +++ b/xds/internal/xdsclient/watchers_cluster_test.go @@ -28,11 +28,6 @@ import ( "google.golang.org/grpc/internal/testutils" ) -type clusterUpdateErr struct { - u ClusterUpdate - err error -} - // TestClusterWatch covers the cases: // - an update is received after a watch() // - an update for another resource name @@ -57,21 +52,21 @@ func (s) TestClusterWatch(t *testing.T) { clusterUpdateCh := testutils.NewChannel() cancelWatch := client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := ClusterUpdate{ClusterName: testEDSName} - client.NewClusters(map[string]ClusterUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } // Another update, with an extra resource for a different resource name. - client.NewClusters(map[string]ClusterUpdate{ - testCDSName: wantUpdate, + client.NewClusters(map[string]ClusterUpdateErr{ + testCDSName: {Update: wantUpdate}, "randomName": {}, }, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil { @@ -80,7 +75,7 @@ func (s) TestClusterWatch(t *testing.T) { // Cancel watch, and send update again. cancelWatch() - client.NewClusters(map[string]ClusterUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() if u, err := clusterUpdateCh.Receive(sCtx); err != context.DeadlineExceeded { @@ -115,7 +110,7 @@ func (s) TestClusterTwoWatchSameResourceName(t *testing.T) { clusterUpdateCh := testutils.NewChannel() clusterUpdateChs = append(clusterUpdateChs, clusterUpdateCh) cancelLastWatch = client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -128,7 +123,7 @@ func (s) TestClusterTwoWatchSameResourceName(t *testing.T) { } wantUpdate := ClusterUpdate{ClusterName: testEDSName} - client.NewClusters(map[string]ClusterUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count; i++ { if err := verifyClusterUpdate(ctx, clusterUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -137,7 +132,7 @@ func (s) TestClusterTwoWatchSameResourceName(t *testing.T) { // Cancel the last watch, and send update again. cancelLastWatch() - client.NewClusters(map[string]ClusterUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count-1; i++ { if err := verifyClusterUpdate(ctx, clusterUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -178,7 +173,7 @@ func (s) TestClusterThreeWatchDifferentResourceName(t *testing.T) { clusterUpdateCh := testutils.NewChannel() clusterUpdateChs = append(clusterUpdateChs, clusterUpdateCh) client.WatchCluster(testCDSName+"1", func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -193,7 +188,7 @@ func (s) TestClusterThreeWatchDifferentResourceName(t *testing.T) { // Third watch for a different name. clusterUpdateCh2 := testutils.NewChannel() client.WatchCluster(testCDSName+"2", func(update ClusterUpdate, err error) { - clusterUpdateCh2.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh2.Send(ClusterUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -201,9 +196,9 @@ func (s) TestClusterThreeWatchDifferentResourceName(t *testing.T) { wantUpdate1 := ClusterUpdate{ClusterName: testEDSName + "1"} wantUpdate2 := ClusterUpdate{ClusterName: testEDSName + "2"} - client.NewClusters(map[string]ClusterUpdate{ - testCDSName + "1": wantUpdate1, - testCDSName + "2": wantUpdate2, + client.NewClusters(map[string]ClusterUpdateErr{ + testCDSName + "1": {Update: wantUpdate1}, + testCDSName + "2": {Update: wantUpdate2}, }, UpdateMetadata{}) for i := 0; i < count; i++ { @@ -238,15 +233,15 @@ func (s) TestClusterWatchAfterCache(t *testing.T) { clusterUpdateCh := testutils.NewChannel() client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := ClusterUpdate{ClusterName: testEDSName} - client.NewClusters(map[string]ClusterUpdate{ - testCDSName: wantUpdate, + client.NewClusters(map[string]ClusterUpdateErr{ + testCDSName: {Update: wantUpdate}, }, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) @@ -255,7 +250,7 @@ func (s) TestClusterWatchAfterCache(t *testing.T) { // Another watch for the resource in cache. clusterUpdateCh2 := testutils.NewChannel() client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) { - clusterUpdateCh2.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh2.Send(ClusterUpdateErr{Update: update, Err: err}) }) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() @@ -299,7 +294,7 @@ func (s) TestClusterWatchExpiryTimer(t *testing.T) { clusterUpdateCh := testutils.NewChannel() client.WatchCluster(testCDSName, func(u ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: u, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: u, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -309,9 +304,9 @@ func (s) TestClusterWatchExpiryTimer(t *testing.T) { if err != nil { t.Fatalf("timeout when waiting for cluster update: %v", err) } - gotUpdate := u.(clusterUpdateErr) - if gotUpdate.err == nil || !cmp.Equal(gotUpdate.u, ClusterUpdate{}) { - t.Fatalf("unexpected clusterUpdate: (%v, %v), want: (ClusterUpdate{}, nil)", gotUpdate.u, gotUpdate.err) + gotUpdate := u.(ClusterUpdateErr) + if gotUpdate.Err == nil || !cmp.Equal(gotUpdate.Update, ClusterUpdate{}) { + t.Fatalf("unexpected clusterUpdate: (%v, %v), want: (ClusterUpdate{}, nil)", gotUpdate.Update, gotUpdate.Err) } } @@ -338,15 +333,15 @@ func (s) TestClusterWatchExpiryTimerStop(t *testing.T) { clusterUpdateCh := testutils.NewChannel() client.WatchCluster(testCDSName, func(u ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: u, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: u, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := ClusterUpdate{ClusterName: testEDSName} - client.NewClusters(map[string]ClusterUpdate{ - testCDSName: wantUpdate, + client.NewClusters(map[string]ClusterUpdateErr{ + testCDSName: {Update: wantUpdate}, }, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) @@ -386,7 +381,7 @@ func (s) TestClusterResourceRemoved(t *testing.T) { clusterUpdateCh1 := testutils.NewChannel() client.WatchCluster(testCDSName+"1", func(update ClusterUpdate, err error) { - clusterUpdateCh1.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh1.Send(ClusterUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -395,7 +390,7 @@ func (s) TestClusterResourceRemoved(t *testing.T) { // Another watch for a different name. clusterUpdateCh2 := testutils.NewChannel() client.WatchCluster(testCDSName+"2", func(update ClusterUpdate, err error) { - clusterUpdateCh2.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh2.Send(ClusterUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -403,9 +398,9 @@ func (s) TestClusterResourceRemoved(t *testing.T) { wantUpdate1 := ClusterUpdate{ClusterName: testEDSName + "1"} wantUpdate2 := ClusterUpdate{ClusterName: testEDSName + "2"} - client.NewClusters(map[string]ClusterUpdate{ - testCDSName + "1": wantUpdate1, - testCDSName + "2": wantUpdate2, + client.NewClusters(map[string]ClusterUpdateErr{ + testCDSName + "1": {Update: wantUpdate1}, + testCDSName + "2": {Update: wantUpdate2}, }, UpdateMetadata{}) if err := verifyClusterUpdate(ctx, clusterUpdateCh1, wantUpdate1, nil); err != nil { t.Fatal(err) @@ -415,10 +410,10 @@ func (s) TestClusterResourceRemoved(t *testing.T) { } // Send another update to remove resource 1. - client.NewClusters(map[string]ClusterUpdate{testCDSName + "2": wantUpdate2}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName + "2": {Update: wantUpdate2}}, UpdateMetadata{}) // Watcher 1 should get an error. - if u, err := clusterUpdateCh1.Receive(ctx); err != nil || ErrType(u.(clusterUpdateErr).err) != ErrorTypeResourceNotFound { + if u, err := clusterUpdateCh1.Receive(ctx); err != nil || ErrType(u.(ClusterUpdateErr).Err) != ErrorTypeResourceNotFound { t.Errorf("unexpected clusterUpdate: %v, error receiving from channel: %v, want update with error resource not found", u, err) } @@ -428,7 +423,7 @@ func (s) TestClusterResourceRemoved(t *testing.T) { } // Send one more update without resource 1. - client.NewClusters(map[string]ClusterUpdate{testCDSName + "2": wantUpdate2}, UpdateMetadata{}) + client.NewClusters(map[string]ClusterUpdateErr{testCDSName + "2": {Update: wantUpdate2}}, UpdateMetadata{}) // Watcher 1 should not see an update. sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) @@ -465,7 +460,7 @@ func (s) TestClusterWatchNACKError(t *testing.T) { clusterUpdateCh := testutils.NewChannel() cancelWatch := client.WatchCluster(testCDSName, func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) }) defer cancelWatch() if _, err := apiClient.addWatches[ClusterResource].Receive(ctx); err != nil { @@ -473,7 +468,7 @@ func (s) TestClusterWatchNACKError(t *testing.T) { } wantError := fmt.Errorf("testing error") - client.NewClusters(map[string]ClusterUpdate{testCDSName: { + client.NewClusters(map[string]ClusterUpdateErr{testCDSName: { Err: wantError, }}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) if err := verifyClusterUpdate(ctx, clusterUpdateCh, ClusterUpdate{}, wantError); err != nil { @@ -509,7 +504,7 @@ func (s) TestClusterWatchPartialValid(t *testing.T) { for _, name := range []string{testCDSName, badResourceName} { clusterUpdateCh := testutils.NewChannel() cancelWatch := client.WatchCluster(name, func(update ClusterUpdate, err error) { - clusterUpdateCh.Send(clusterUpdateErr{u: update, err: err}) + clusterUpdateCh.Send(ClusterUpdateErr{Update: update, Err: err}) }) defer func() { cancelWatch() @@ -525,8 +520,8 @@ func (s) TestClusterWatchPartialValid(t *testing.T) { wantError := fmt.Errorf("testing error") wantError2 := fmt.Errorf("individual error") - client.NewClusters(map[string]ClusterUpdate{ - testCDSName: {ClusterName: testEDSName}, + client.NewClusters(map[string]ClusterUpdateErr{ + testCDSName: {Update: ClusterUpdate{ClusterName: testEDSName}}, badResourceName: {Err: wantError2}, }, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) diff --git a/xds/internal/xdsclient/watchers_endpoints_test.go b/xds/internal/xdsclient/watchers_endpoints_test.go index d0f680a7ec1f..f25cff06b69d 100644 --- a/xds/internal/xdsclient/watchers_endpoints_test.go +++ b/xds/internal/xdsclient/watchers_endpoints_test.go @@ -46,11 +46,6 @@ var ( } ) -type endpointsUpdateErr struct { - u EndpointsUpdate - err error -} - // TestEndpointsWatch covers the cases: // - an update is received after a watch() // - an update for another resource name (which doesn't trigger callback) @@ -75,20 +70,20 @@ func (s) TestEndpointsWatch(t *testing.T) { endpointsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchEndpoints(testCDSName, func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[EndpointsResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := EndpointsUpdate{Localities: []Locality{testLocalities[0]}} - client.NewEndpoints(map[string]EndpointsUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewEndpoints(map[string]EndpointsUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyEndpointsUpdate(ctx, endpointsUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } // Another update for a different resource name. - client.NewEndpoints(map[string]EndpointsUpdate{"randomName": {}}, UpdateMetadata{}) + client.NewEndpoints(map[string]EndpointsUpdateErr{"randomName": {}}, UpdateMetadata{}) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() if u, err := endpointsUpdateCh.Receive(sCtx); err != context.DeadlineExceeded { @@ -97,7 +92,7 @@ func (s) TestEndpointsWatch(t *testing.T) { // Cancel watch, and send update again. cancelWatch() - client.NewEndpoints(map[string]EndpointsUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewEndpoints(map[string]EndpointsUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() if u, err := endpointsUpdateCh.Receive(sCtx); err != context.DeadlineExceeded { @@ -134,7 +129,7 @@ func (s) TestEndpointsTwoWatchSameResourceName(t *testing.T) { endpointsUpdateCh := testutils.NewChannel() endpointsUpdateChs = append(endpointsUpdateChs, endpointsUpdateCh) cancelLastWatch = client.WatchEndpoints(testCDSName, func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -147,7 +142,7 @@ func (s) TestEndpointsTwoWatchSameResourceName(t *testing.T) { } wantUpdate := EndpointsUpdate{Localities: []Locality{testLocalities[0]}} - client.NewEndpoints(map[string]EndpointsUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewEndpoints(map[string]EndpointsUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count; i++ { if err := verifyEndpointsUpdate(ctx, endpointsUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -156,7 +151,7 @@ func (s) TestEndpointsTwoWatchSameResourceName(t *testing.T) { // Cancel the last watch, and send update again. cancelLastWatch() - client.NewEndpoints(map[string]EndpointsUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewEndpoints(map[string]EndpointsUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count-1; i++ { if err := verifyEndpointsUpdate(ctx, endpointsUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -197,7 +192,7 @@ func (s) TestEndpointsThreeWatchDifferentResourceName(t *testing.T) { endpointsUpdateCh := testutils.NewChannel() endpointsUpdateChs = append(endpointsUpdateChs, endpointsUpdateCh) client.WatchEndpoints(testCDSName+"1", func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -212,7 +207,7 @@ func (s) TestEndpointsThreeWatchDifferentResourceName(t *testing.T) { // Third watch for a different name. endpointsUpdateCh2 := testutils.NewChannel() client.WatchEndpoints(testCDSName+"2", func(update EndpointsUpdate, err error) { - endpointsUpdateCh2.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh2.Send(EndpointsUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[EndpointsResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -220,9 +215,9 @@ func (s) TestEndpointsThreeWatchDifferentResourceName(t *testing.T) { wantUpdate1 := EndpointsUpdate{Localities: []Locality{testLocalities[0]}} wantUpdate2 := EndpointsUpdate{Localities: []Locality{testLocalities[1]}} - client.NewEndpoints(map[string]EndpointsUpdate{ - testCDSName + "1": wantUpdate1, - testCDSName + "2": wantUpdate2, + client.NewEndpoints(map[string]EndpointsUpdateErr{ + testCDSName + "1": {Update: wantUpdate1}, + testCDSName + "2": {Update: wantUpdate2}, }, UpdateMetadata{}) for i := 0; i < count; i++ { @@ -257,14 +252,14 @@ func (s) TestEndpointsWatchAfterCache(t *testing.T) { endpointsUpdateCh := testutils.NewChannel() client.WatchEndpoints(testCDSName, func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[EndpointsResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := EndpointsUpdate{Localities: []Locality{testLocalities[0]}} - client.NewEndpoints(map[string]EndpointsUpdate{testCDSName: wantUpdate}, UpdateMetadata{}) + client.NewEndpoints(map[string]EndpointsUpdateErr{testCDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyEndpointsUpdate(ctx, endpointsUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } @@ -272,7 +267,7 @@ func (s) TestEndpointsWatchAfterCache(t *testing.T) { // Another watch for the resource in cache. endpointsUpdateCh2 := testutils.NewChannel() client.WatchEndpoints(testCDSName, func(update EndpointsUpdate, err error) { - endpointsUpdateCh2.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh2.Send(EndpointsUpdateErr{Update: update, Err: err}) }) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() @@ -316,7 +311,7 @@ func (s) TestEndpointsWatchExpiryTimer(t *testing.T) { endpointsUpdateCh := testutils.NewChannel() client.WatchEndpoints(testCDSName, func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[EndpointsResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -326,9 +321,9 @@ func (s) TestEndpointsWatchExpiryTimer(t *testing.T) { if err != nil { t.Fatalf("timeout when waiting for endpoints update: %v", err) } - gotUpdate := u.(endpointsUpdateErr) - if gotUpdate.err == nil || !cmp.Equal(gotUpdate.u, EndpointsUpdate{}) { - t.Fatalf("unexpected endpointsUpdate: (%v, %v), want: (EndpointsUpdate{}, nil)", gotUpdate.u, gotUpdate.err) + gotUpdate := u.(EndpointsUpdateErr) + if gotUpdate.Err == nil || !cmp.Equal(gotUpdate.Update, EndpointsUpdate{}) { + t.Fatalf("unexpected endpointsUpdate: (%v, %v), want: (EndpointsUpdate{}, nil)", gotUpdate.Update, gotUpdate.Err) } } @@ -354,7 +349,7 @@ func (s) TestEndpointsWatchNACKError(t *testing.T) { endpointsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchEndpoints(testCDSName, func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) defer cancelWatch() if _, err := apiClient.addWatches[EndpointsResource].Receive(ctx); err != nil { @@ -362,7 +357,7 @@ func (s) TestEndpointsWatchNACKError(t *testing.T) { } wantError := fmt.Errorf("testing error") - client.NewEndpoints(map[string]EndpointsUpdate{testCDSName: {Err: wantError}}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) + client.NewEndpoints(map[string]EndpointsUpdateErr{testCDSName: {Err: wantError}}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) if err := verifyEndpointsUpdate(ctx, endpointsUpdateCh, EndpointsUpdate{}, wantError); err != nil { t.Fatal(err) } @@ -396,7 +391,7 @@ func (s) TestEndpointsWatchPartialValid(t *testing.T) { for _, name := range []string{testCDSName, badResourceName} { endpointsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchEndpoints(name, func(update EndpointsUpdate, err error) { - endpointsUpdateCh.Send(endpointsUpdateErr{u: update, err: err}) + endpointsUpdateCh.Send(EndpointsUpdateErr{Update: update, Err: err}) }) defer func() { cancelWatch() @@ -412,8 +407,8 @@ func (s) TestEndpointsWatchPartialValid(t *testing.T) { wantError := fmt.Errorf("testing error") wantError2 := fmt.Errorf("individual error") - client.NewEndpoints(map[string]EndpointsUpdate{ - testCDSName: {Localities: []Locality{testLocalities[0]}}, + client.NewEndpoints(map[string]EndpointsUpdateErr{ + testCDSName: {Update: EndpointsUpdate{Localities: []Locality{testLocalities[0]}}}, badResourceName: {Err: wantError2}, }, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) diff --git a/xds/internal/xdsclient/watchers_listener_test.go b/xds/internal/xdsclient/watchers_listener_test.go index 9b555368f350..be7bda4e2b77 100644 --- a/xds/internal/xdsclient/watchers_listener_test.go +++ b/xds/internal/xdsclient/watchers_listener_test.go @@ -26,11 +26,6 @@ import ( "google.golang.org/grpc/internal/testutils" ) -type ldsUpdateErr struct { - u ListenerUpdate - err error -} - // TestLDSWatch covers the cases: // - an update is received after a watch() // - an update for another resource name @@ -55,21 +50,21 @@ func (s) TestLDSWatch(t *testing.T) { ldsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchListener(testLDSName, func(update ListenerUpdate, err error) { - ldsUpdateCh.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh.Send(ListenerUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ListenerResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := ListenerUpdate{RouteConfigName: testRDSName} - client.NewListeners(map[string]ListenerUpdate{testLDSName: wantUpdate}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyListenerUpdate(ctx, ldsUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } // Another update, with an extra resource for a different resource name. - client.NewListeners(map[string]ListenerUpdate{ - testLDSName: wantUpdate, + client.NewListeners(map[string]ListenerUpdateErr{ + testLDSName: {Update: wantUpdate}, "randomName": {}, }, UpdateMetadata{}) if err := verifyListenerUpdate(ctx, ldsUpdateCh, wantUpdate, nil); err != nil { @@ -78,7 +73,7 @@ func (s) TestLDSWatch(t *testing.T) { // Cancel watch, and send update again. cancelWatch() - client.NewListeners(map[string]ListenerUpdate{testLDSName: wantUpdate}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName: {Update: wantUpdate}}, UpdateMetadata{}) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() if u, err := ldsUpdateCh.Receive(sCtx); err != context.DeadlineExceeded { @@ -116,7 +111,7 @@ func (s) TestLDSTwoWatchSameResourceName(t *testing.T) { ldsUpdateCh := testutils.NewChannel() ldsUpdateChs = append(ldsUpdateChs, ldsUpdateCh) cancelLastWatch = client.WatchListener(testLDSName, func(update ListenerUpdate, err error) { - ldsUpdateCh.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh.Send(ListenerUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -129,7 +124,7 @@ func (s) TestLDSTwoWatchSameResourceName(t *testing.T) { } wantUpdate := ListenerUpdate{RouteConfigName: testRDSName} - client.NewListeners(map[string]ListenerUpdate{testLDSName: wantUpdate}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count; i++ { if err := verifyListenerUpdate(ctx, ldsUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -138,7 +133,7 @@ func (s) TestLDSTwoWatchSameResourceName(t *testing.T) { // Cancel the last watch, and send update again. cancelLastWatch() - client.NewListeners(map[string]ListenerUpdate{testLDSName: wantUpdate}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count-1; i++ { if err := verifyListenerUpdate(ctx, ldsUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -180,7 +175,7 @@ func (s) TestLDSThreeWatchDifferentResourceName(t *testing.T) { ldsUpdateCh := testutils.NewChannel() ldsUpdateChs = append(ldsUpdateChs, ldsUpdateCh) client.WatchListener(testLDSName+"1", func(update ListenerUpdate, err error) { - ldsUpdateCh.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh.Send(ListenerUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -195,7 +190,7 @@ func (s) TestLDSThreeWatchDifferentResourceName(t *testing.T) { // Third watch for a different name. ldsUpdateCh2 := testutils.NewChannel() client.WatchListener(testLDSName+"2", func(update ListenerUpdate, err error) { - ldsUpdateCh2.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh2.Send(ListenerUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ListenerResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -203,9 +198,9 @@ func (s) TestLDSThreeWatchDifferentResourceName(t *testing.T) { wantUpdate1 := ListenerUpdate{RouteConfigName: testRDSName + "1"} wantUpdate2 := ListenerUpdate{RouteConfigName: testRDSName + "2"} - client.NewListeners(map[string]ListenerUpdate{ - testLDSName + "1": wantUpdate1, - testLDSName + "2": wantUpdate2, + client.NewListeners(map[string]ListenerUpdateErr{ + testLDSName + "1": {Update: wantUpdate1}, + testLDSName + "2": {Update: wantUpdate2}, }, UpdateMetadata{}) for i := 0; i < count; i++ { @@ -240,14 +235,14 @@ func (s) TestLDSWatchAfterCache(t *testing.T) { ldsUpdateCh := testutils.NewChannel() client.WatchListener(testLDSName, func(update ListenerUpdate, err error) { - ldsUpdateCh.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh.Send(ListenerUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ListenerResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) } wantUpdate := ListenerUpdate{RouteConfigName: testRDSName} - client.NewListeners(map[string]ListenerUpdate{testLDSName: wantUpdate}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyListenerUpdate(ctx, ldsUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } @@ -255,7 +250,7 @@ func (s) TestLDSWatchAfterCache(t *testing.T) { // Another watch for the resource in cache. ldsUpdateCh2 := testutils.NewChannel() client.WatchListener(testLDSName, func(update ListenerUpdate, err error) { - ldsUpdateCh2.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh2.Send(ListenerUpdateErr{Update: update, Err: err}) }) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() @@ -302,7 +297,7 @@ func (s) TestLDSResourceRemoved(t *testing.T) { ldsUpdateCh1 := testutils.NewChannel() client.WatchListener(testLDSName+"1", func(update ListenerUpdate, err error) { - ldsUpdateCh1.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh1.Send(ListenerUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ListenerResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -310,7 +305,7 @@ func (s) TestLDSResourceRemoved(t *testing.T) { // Another watch for a different name. ldsUpdateCh2 := testutils.NewChannel() client.WatchListener(testLDSName+"2", func(update ListenerUpdate, err error) { - ldsUpdateCh2.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh2.Send(ListenerUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[ListenerResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -318,9 +313,9 @@ func (s) TestLDSResourceRemoved(t *testing.T) { wantUpdate1 := ListenerUpdate{RouteConfigName: testEDSName + "1"} wantUpdate2 := ListenerUpdate{RouteConfigName: testEDSName + "2"} - client.NewListeners(map[string]ListenerUpdate{ - testLDSName + "1": wantUpdate1, - testLDSName + "2": wantUpdate2, + client.NewListeners(map[string]ListenerUpdateErr{ + testLDSName + "1": {Update: wantUpdate1}, + testLDSName + "2": {Update: wantUpdate2}, }, UpdateMetadata{}) if err := verifyListenerUpdate(ctx, ldsUpdateCh1, wantUpdate1, nil); err != nil { t.Fatal(err) @@ -330,10 +325,10 @@ func (s) TestLDSResourceRemoved(t *testing.T) { } // Send another update to remove resource 1. - client.NewListeners(map[string]ListenerUpdate{testLDSName + "2": wantUpdate2}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName + "2": {Update: wantUpdate2}}, UpdateMetadata{}) // Watcher 1 should get an error. - if u, err := ldsUpdateCh1.Receive(ctx); err != nil || ErrType(u.(ldsUpdateErr).err) != ErrorTypeResourceNotFound { + if u, err := ldsUpdateCh1.Receive(ctx); err != nil || ErrType(u.(ListenerUpdateErr).Err) != ErrorTypeResourceNotFound { t.Errorf("unexpected ListenerUpdate: %v, error receiving from channel: %v, want update with error resource not found", u, err) } @@ -343,7 +338,7 @@ func (s) TestLDSResourceRemoved(t *testing.T) { } // Send one more update without resource 1. - client.NewListeners(map[string]ListenerUpdate{testLDSName + "2": wantUpdate2}, UpdateMetadata{}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName + "2": {Update: wantUpdate2}}, UpdateMetadata{}) // Watcher 1 should not see an update. sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) @@ -380,7 +375,7 @@ func (s) TestListenerWatchNACKError(t *testing.T) { ldsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchListener(testLDSName, func(update ListenerUpdate, err error) { - ldsUpdateCh.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh.Send(ListenerUpdateErr{Update: update, Err: err}) }) defer cancelWatch() if _, err := apiClient.addWatches[ListenerResource].Receive(ctx); err != nil { @@ -388,7 +383,7 @@ func (s) TestListenerWatchNACKError(t *testing.T) { } wantError := fmt.Errorf("testing error") - client.NewListeners(map[string]ListenerUpdate{testLDSName: {Err: wantError}}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) + client.NewListeners(map[string]ListenerUpdateErr{testLDSName: {Err: wantError}}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) if err := verifyListenerUpdate(ctx, ldsUpdateCh, ListenerUpdate{}, wantError); err != nil { t.Fatal(err) } @@ -422,7 +417,7 @@ func (s) TestListenerWatchPartialValid(t *testing.T) { for _, name := range []string{testLDSName, badResourceName} { ldsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchListener(name, func(update ListenerUpdate, err error) { - ldsUpdateCh.Send(ldsUpdateErr{u: update, err: err}) + ldsUpdateCh.Send(ListenerUpdateErr{Update: update, Err: err}) }) defer func() { cancelWatch() @@ -438,8 +433,8 @@ func (s) TestListenerWatchPartialValid(t *testing.T) { wantError := fmt.Errorf("testing error") wantError2 := fmt.Errorf("individual error") - client.NewListeners(map[string]ListenerUpdate{ - testLDSName: {RouteConfigName: testEDSName}, + client.NewListeners(map[string]ListenerUpdateErr{ + testLDSName: {Update: ListenerUpdate{RouteConfigName: testEDSName}}, badResourceName: {Err: wantError2}, }, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) diff --git a/xds/internal/xdsclient/watchers_route_test.go b/xds/internal/xdsclient/watchers_route_test.go index 8065e48330f1..63f10ab13196 100644 --- a/xds/internal/xdsclient/watchers_route_test.go +++ b/xds/internal/xdsclient/watchers_route_test.go @@ -28,11 +28,6 @@ import ( "google.golang.org/grpc/internal/testutils" ) -type rdsUpdateErr struct { - u RouteConfigUpdate - err error -} - // TestRDSWatch covers the cases: // - an update is received after a watch() // - an update for another resource name (which doesn't trigger callback) @@ -57,7 +52,7 @@ func (s) TestRDSWatch(t *testing.T) { rdsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchRouteConfig(testRDSName, func(update RouteConfigUpdate, err error) { - rdsUpdateCh.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[RouteConfigResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -71,13 +66,13 @@ func (s) TestRDSWatch(t *testing.T) { }, }, } - client.NewRouteConfigs(map[string]RouteConfigUpdate{testRDSName: wantUpdate}, UpdateMetadata{}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{testRDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyRouteConfigUpdate(ctx, rdsUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } // Another update for a different resource name. - client.NewRouteConfigs(map[string]RouteConfigUpdate{"randomName": {}}, UpdateMetadata{}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{"randomName": {}}, UpdateMetadata{}) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() if u, err := rdsUpdateCh.Receive(sCtx); err != context.DeadlineExceeded { @@ -86,7 +81,7 @@ func (s) TestRDSWatch(t *testing.T) { // Cancel watch, and send update again. cancelWatch() - client.NewRouteConfigs(map[string]RouteConfigUpdate{testRDSName: wantUpdate}, UpdateMetadata{}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{testRDSName: {Update: wantUpdate}}, UpdateMetadata{}) sCtx, sCancel = context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() if u, err := rdsUpdateCh.Receive(sCtx); err != context.DeadlineExceeded { @@ -123,7 +118,7 @@ func (s) TestRDSTwoWatchSameResourceName(t *testing.T) { rdsUpdateCh := testutils.NewChannel() rdsUpdateChs = append(rdsUpdateChs, rdsUpdateCh) cancelLastWatch = client.WatchRouteConfig(testRDSName, func(update RouteConfigUpdate, err error) { - rdsUpdateCh.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -143,7 +138,7 @@ func (s) TestRDSTwoWatchSameResourceName(t *testing.T) { }, }, } - client.NewRouteConfigs(map[string]RouteConfigUpdate{testRDSName: wantUpdate}, UpdateMetadata{}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{testRDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count; i++ { if err := verifyRouteConfigUpdate(ctx, rdsUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -152,7 +147,7 @@ func (s) TestRDSTwoWatchSameResourceName(t *testing.T) { // Cancel the last watch, and send update again. cancelLastWatch() - client.NewRouteConfigs(map[string]RouteConfigUpdate{testRDSName: wantUpdate}, UpdateMetadata{}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{testRDSName: {Update: wantUpdate}}, UpdateMetadata{}) for i := 0; i < count-1; i++ { if err := verifyRouteConfigUpdate(ctx, rdsUpdateChs[i], wantUpdate, nil); err != nil { t.Fatal(err) @@ -193,7 +188,7 @@ func (s) TestRDSThreeWatchDifferentResourceName(t *testing.T) { rdsUpdateCh := testutils.NewChannel() rdsUpdateChs = append(rdsUpdateChs, rdsUpdateCh) client.WatchRouteConfig(testRDSName+"1", func(update RouteConfigUpdate, err error) { - rdsUpdateCh.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) if i == 0 { @@ -208,7 +203,7 @@ func (s) TestRDSThreeWatchDifferentResourceName(t *testing.T) { // Third watch for a different name. rdsUpdateCh2 := testutils.NewChannel() client.WatchRouteConfig(testRDSName+"2", func(update RouteConfigUpdate, err error) { - rdsUpdateCh2.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh2.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[RouteConfigResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -230,9 +225,9 @@ func (s) TestRDSThreeWatchDifferentResourceName(t *testing.T) { }, }, } - client.NewRouteConfigs(map[string]RouteConfigUpdate{ - testRDSName + "1": wantUpdate1, - testRDSName + "2": wantUpdate2, + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{ + testRDSName + "1": {Update: wantUpdate1}, + testRDSName + "2": {Update: wantUpdate2}, }, UpdateMetadata{}) for i := 0; i < count; i++ { @@ -267,7 +262,7 @@ func (s) TestRDSWatchAfterCache(t *testing.T) { rdsUpdateCh := testutils.NewChannel() client.WatchRouteConfig(testRDSName, func(update RouteConfigUpdate, err error) { - rdsUpdateCh.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) if _, err := apiClient.addWatches[RouteConfigResource].Receive(ctx); err != nil { t.Fatalf("want new watch to start, got error %v", err) @@ -281,7 +276,7 @@ func (s) TestRDSWatchAfterCache(t *testing.T) { }, }, } - client.NewRouteConfigs(map[string]RouteConfigUpdate{testRDSName: wantUpdate}, UpdateMetadata{}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{testRDSName: {Update: wantUpdate}}, UpdateMetadata{}) if err := verifyRouteConfigUpdate(ctx, rdsUpdateCh, wantUpdate, nil); err != nil { t.Fatal(err) } @@ -289,7 +284,7 @@ func (s) TestRDSWatchAfterCache(t *testing.T) { // Another watch for the resource in cache. rdsUpdateCh2 := testutils.NewChannel() client.WatchRouteConfig(testRDSName, func(update RouteConfigUpdate, err error) { - rdsUpdateCh2.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh2.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) sCtx, sCancel := context.WithTimeout(ctx, defaultTestShortTimeout) defer sCancel() @@ -298,7 +293,7 @@ func (s) TestRDSWatchAfterCache(t *testing.T) { } // New watch should receives the update. - if u, err := rdsUpdateCh2.Receive(ctx); err != nil || !cmp.Equal(u, rdsUpdateErr{wantUpdate, nil}, cmp.AllowUnexported(rdsUpdateErr{})) { + if u, err := rdsUpdateCh2.Receive(ctx); err != nil || !cmp.Equal(u, RouteConfigUpdateErr{wantUpdate, nil}, cmp.AllowUnexported(RouteConfigUpdateErr{})) { t.Errorf("unexpected RouteConfigUpdate: %v, error receiving from channel: %v", u, err) } @@ -332,7 +327,7 @@ func (s) TestRouteWatchNACKError(t *testing.T) { rdsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchRouteConfig(testCDSName, func(update RouteConfigUpdate, err error) { - rdsUpdateCh.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) defer cancelWatch() if _, err := apiClient.addWatches[RouteConfigResource].Receive(ctx); err != nil { @@ -340,7 +335,7 @@ func (s) TestRouteWatchNACKError(t *testing.T) { } wantError := fmt.Errorf("testing error") - client.NewRouteConfigs(map[string]RouteConfigUpdate{testCDSName: {Err: wantError}}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{testCDSName: {Err: wantError}}, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) if err := verifyRouteConfigUpdate(ctx, rdsUpdateCh, RouteConfigUpdate{}, wantError); err != nil { t.Fatal(err) } @@ -374,7 +369,7 @@ func (s) TestRouteWatchPartialValid(t *testing.T) { for _, name := range []string{testRDSName, badResourceName} { rdsUpdateCh := testutils.NewChannel() cancelWatch := client.WatchRouteConfig(name, func(update RouteConfigUpdate, err error) { - rdsUpdateCh.Send(rdsUpdateErr{u: update, err: err}) + rdsUpdateCh.Send(RouteConfigUpdateErr{Update: update, Err: err}) }) defer func() { cancelWatch() @@ -390,11 +385,11 @@ func (s) TestRouteWatchPartialValid(t *testing.T) { wantError := fmt.Errorf("testing error") wantError2 := fmt.Errorf("individual error") - client.NewRouteConfigs(map[string]RouteConfigUpdate{ - testRDSName: {VirtualHosts: []*VirtualHost{{ + client.NewRouteConfigs(map[string]RouteConfigUpdateErr{ + testRDSName: {Update: RouteConfigUpdate{VirtualHosts: []*VirtualHost{{ Domains: []string{testLDSName}, Routes: []*Route{{Prefix: newStringP(""), WeightedClusters: map[string]WeightedCluster{testCDSName: {Weight: 1}}}}, - }}}, + }}}}, badResourceName: {Err: wantError2}, }, UpdateMetadata{ErrState: &UpdateErrorMetadata{Err: wantError}}) diff --git a/xds/internal/xdsclient/xds.go b/xds/internal/xdsclient/xds.go index 71985daa6174..53677d431e9c 100644 --- a/xds/internal/xdsclient/xds.go +++ b/xds/internal/xdsclient/xds.go @@ -58,8 +58,8 @@ const transportSocketName = "envoy.transport_sockets.tls" // UnmarshalListener processes resources received in an LDS response, validates // them, and transforms them into a native struct which contains only fields we // are interested in. -func UnmarshalListener(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]ListenerUpdate, UpdateMetadata, error) { - update := make(map[string]ListenerUpdate) +func UnmarshalListener(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]ListenerUpdateErr, UpdateMetadata, error) { + update := make(map[string]ListenerUpdateErr) md, err := processAllResources(version, resources, logger, update) return update, md, err } @@ -296,8 +296,8 @@ func processServerSideListener(lis *v3listenerpb.Listener) (*ListenerUpdate, err // validates them, and transforms them into a native struct which contains only // fields we are interested in. The provided hostname determines the route // configuration resources of interest. -func UnmarshalRouteConfig(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]RouteConfigUpdate, UpdateMetadata, error) { - update := make(map[string]RouteConfigUpdate) +func UnmarshalRouteConfig(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]RouteConfigUpdateErr, UpdateMetadata, error) { + update := make(map[string]RouteConfigUpdateErr) md, err := processAllResources(version, resources, logger, update) return update, md, err } @@ -631,8 +631,8 @@ func hashPoliciesProtoToSlice(policies []*v3routepb.RouteAction_HashPolicy, logg // UnmarshalCluster processes resources received in an CDS response, validates // them, and transforms them into a native struct which contains only fields we // are interested in. -func UnmarshalCluster(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]ClusterUpdate, UpdateMetadata, error) { - update := make(map[string]ClusterUpdate) +func UnmarshalCluster(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]ClusterUpdateErr, UpdateMetadata, error) { + update := make(map[string]ClusterUpdateErr) md, err := processAllResources(version, resources, logger, update) return update, md, err } @@ -894,8 +894,8 @@ func circuitBreakersFromCluster(cluster *v3clusterpb.Cluster) *uint32 { // UnmarshalEndpoints processes resources received in an EDS response, // validates them, and transforms them into a native struct which contains only // fields we are interested in. -func UnmarshalEndpoints(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]EndpointsUpdate, UpdateMetadata, error) { - update := make(map[string]EndpointsUpdate) +func UnmarshalEndpoints(version string, resources []*anypb.Any, logger *grpclog.PrefixLogger) (map[string]EndpointsUpdateErr, UpdateMetadata, error) { + update := make(map[string]EndpointsUpdateErr) md, err := processAllResources(version, resources, logger, update) return update, md, err } @@ -989,6 +989,30 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate, return ret, nil } +// ListenerUpdateErr is a tuple with the update and error. +type ListenerUpdateErr struct { + Update ListenerUpdate + Err error +} + +// RouteConfigUpdateErr is a tuple with the update and error. +type RouteConfigUpdateErr struct { + Update RouteConfigUpdate + Err error +} + +// ClusterUpdateErr is a tuple with the update and error. +type ClusterUpdateErr struct { + Update ClusterUpdate + Err error +} + +// EndpointsUpdateErr is a tuple with the update and error. +type EndpointsUpdateErr struct { + Update EndpointsUpdate + Err error +} + // processAllResources unmarshals and validates the resources, populates the // provided ret (a map), and returns metadata and error. // @@ -1009,10 +1033,10 @@ func processAllResources(version string, resources []*anypb.Any, logger *grpclog for _, r := range resources { switch ret2 := ret.(type) { - case map[string]ListenerUpdate: + case map[string]ListenerUpdateErr: name, update, err := unmarshalListenerResource(r, logger) if err == nil { - ret2[name] = update + ret2[name] = ListenerUpdateErr{Update: update} continue } if name == "" { @@ -1022,11 +1046,11 @@ func processAllResources(version string, resources []*anypb.Any, logger *grpclog perResourceErrors[name] = err // Add place holder in the map so we know this resource name was in // the response. - ret2[name] = ListenerUpdate{Err: err} - case map[string]RouteConfigUpdate: + ret2[name] = ListenerUpdateErr{Err: err} + case map[string]RouteConfigUpdateErr: name, update, err := unmarshalRouteConfigResource(r, logger) if err == nil { - ret2[name] = update + ret2[name] = RouteConfigUpdateErr{Update: update} continue } if name == "" { @@ -1036,11 +1060,11 @@ func processAllResources(version string, resources []*anypb.Any, logger *grpclog perResourceErrors[name] = err // Add place holder in the map so we know this resource name was in // the response. - ret2[name] = RouteConfigUpdate{Err: err} - case map[string]ClusterUpdate: + ret2[name] = RouteConfigUpdateErr{Err: err} + case map[string]ClusterUpdateErr: name, update, err := unmarshalClusterResource(r, logger) if err == nil { - ret2[name] = update + ret2[name] = ClusterUpdateErr{Update: update} continue } if name == "" { @@ -1050,11 +1074,11 @@ func processAllResources(version string, resources []*anypb.Any, logger *grpclog perResourceErrors[name] = err // Add place holder in the map so we know this resource name was in // the response. - ret2[name] = ClusterUpdate{Err: err} - case map[string]EndpointsUpdate: + ret2[name] = ClusterUpdateErr{Err: err} + case map[string]EndpointsUpdateErr: name, update, err := unmarshalEndpointsResource(r, logger) if err == nil { - ret2[name] = update + ret2[name] = EndpointsUpdateErr{Update: update} continue } if name == "" { @@ -1064,7 +1088,7 @@ func processAllResources(version string, resources []*anypb.Any, logger *grpclog perResourceErrors[name] = err // Add place holder in the map so we know this resource name was in // the response. - ret2[name] = EndpointsUpdate{Err: err} + ret2[name] = EndpointsUpdateErr{Err: err} } }