Skip to content

xds/internal: remove Generic resource Decoder and add concrete functions #8652

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open

xds/internal: remove Generic resource Decoder and add concrete functions #8652

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
127107b
Refactor the resource types
vinothkumarr227 Oct 15, 2025
e8d0567
small tweaks
vinothkumarr227 Oct 15, 2025
a5d8b85
Fixed the review changes
vinothkumarr227 Oct 16, 2025
1b12d28
small tweaks
vinothkumarr227 Oct 16, 2025
d4221a6
Fixed the review changes
vinothkumarr227 Oct 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions internal/xds/balancer/cdsbalancer/cdsbalancer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -448,7 +448,7 @@ func (b *cdsBalancer) annotateErrorWithNodeID(err error) error {
// graph is resolved, generates child policy config and pushes it down.
//
// Only executed in the context of a serializer callback.
func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpdate) {
func (b *cdsBalancer) onClusterUpdate(name string, update *xdsresource.ClusterUpdate) {
state := b.watchers[name]
if state == nil {
// We are currently not watching this cluster anymore. Return early.
Expand All @@ -458,7 +458,7 @@ func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpd
b.logger.Infof("Received Cluster resource: %s", pretty.ToJSON(update))

// Update the watchers map with the update for the cluster.
state.lastUpdate = &update
state.lastUpdate = update

// For an aggregate cluster, always use the security configuration on the
// root cluster.
Expand Down
5 changes: 3 additions & 2 deletions internal/xds/balancer/cdsbalancer/cluster_watcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,9 @@ type clusterWatcher struct {
parent *cdsBalancer
}

func (cw *clusterWatcher) ResourceChanged(u *xdsresource.ClusterResourceData, onDone func()) {
handleUpdate := func(context.Context) { cw.parent.onClusterUpdate(cw.name, u.Resource); onDone() }
func (cw *clusterWatcher) ResourceChanged(u *xdsresource.ClusterUpdate, onDone func()) {
handleUpdate := func(context.Context) { cw.parent.onClusterUpdate(cw.name, u); onDone() }

cw.parent.serializer.ScheduleOr(handleUpdate, onDone)
}

Expand Down
4 changes: 2 additions & 2 deletions internal/xds/balancer/clusterresolver/resource_resolver_eds.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,14 +76,14 @@ func newEDSResolver(nameToWatch string, producer xdsresource.Producer, topLevelR
}

// ResourceChanged is invoked to report an update for the resource being watched.
func (er *edsDiscoveryMechanism) ResourceChanged(update *xdsresource.EndpointsResourceData, onDone func()) {
func (er *edsDiscoveryMechanism) ResourceChanged(update *xdsresource.EndpointsUpdate, onDone func()) {
if er.stopped.HasFired() {
onDone()
return
}

er.mu.Lock()
er.update = &update.Resource
er.update = update
er.mu.Unlock()

er.topLevelResolver.onUpdate(onDone)
Expand Down
4 changes: 2 additions & 2 deletions internal/xds/resolver/watch_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,9 +68,9 @@ func newRouteConfigWatcher(resourceName string, parent *xdsResolver) *routeConfi
return rw
}

func (r *routeConfigWatcher) ResourceChanged(u *xdsresource.RouteConfigResourceData, onDone func()) {
func (r *routeConfigWatcher) ResourceChanged(u *xdsresource.RouteConfigUpdate, onDone func()) {
handleUpdate := func(context.Context) {
r.parent.onRouteConfigResourceUpdate(r.resourceName, u.Resource)
r.parent.onRouteConfigResourceUpdate(r.resourceName, u)
onDone()
}
r.parent.serializer.ScheduleOr(handleUpdate, onDone)
Expand Down
8 changes: 4 additions & 4 deletions internal/xds/resolver/xds_resolver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,7 @@ type xdsResolver struct {
rdsResourceName string
routeConfigWatcher *routeConfigWatcher
routeConfigUpdateRecvd bool
currentRouteConfig xdsresource.RouteConfigUpdate
currentRouteConfig *xdsresource.RouteConfigUpdate
currentVirtualHost *xdsresource.VirtualHost // Matched virtual host for quick access.

// activeClusters is a map from cluster name to information about the
Expand Down Expand Up @@ -461,7 +461,7 @@ func (r *xdsResolver) onResolutionComplete() {
r.curConfigSelector = cs
}

func (r *xdsResolver) applyRouteConfigUpdate(update xdsresource.RouteConfigUpdate) {
func (r *xdsResolver) applyRouteConfigUpdate(update *xdsresource.RouteConfigUpdate) {
matchVh := xdsresource.FindBestMatchingVirtualHost(r.dataplaneAuthority, update.VirtualHosts)
if matchVh == nil {
// TODO(purnesh42h): Should this be a resource or ambient error? Note
Expand Down Expand Up @@ -527,7 +527,7 @@ func (r *xdsResolver) onListenerResourceUpdate(update *xdsresource.ListenerUpdat
r.routeConfigWatcher = nil
}

r.applyRouteConfigUpdate(*update.InlineRouteConfig)
r.applyRouteConfigUpdate(update.InlineRouteConfig)
return
}

Expand Down Expand Up @@ -580,7 +580,7 @@ func (r *xdsResolver) onListenerResourceError(err error) {
}

// Only executed in the context of a serializer callback.
func (r *xdsResolver) onRouteConfigResourceUpdate(name string, update xdsresource.RouteConfigUpdate) {
func (r *xdsResolver) onRouteConfigResourceUpdate(name string, update *xdsresource.RouteConfigUpdate) {
if r.logger.V(2) {
r.logger.Infof("Received update for RouteConfiguration resource %q: %v", name, pretty.ToJSON(update))
}
Expand Down
3 changes: 1 addition & 2 deletions internal/xds/server/listener_wrapper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,7 @@ type ServingModeCallback func(addr net.Addr, mode connectivity.ServingMode, err
// XDSClient wraps the methods on the XDSClient which are required by
// the listenerWrapper.
type XDSClient interface {
WatchResource(rType xdsresource.Type, resourceName string, watcher xdsresource.ResourceWatcher) (cancel func())
WatchResourceV2(typeURL, resourceName string, watcher xdsclient.ResourceWatcher) (cancel func())
WatchResource(typeURL, resourceName string, watcher xdsclient.ResourceWatcher) (cancel func())
BootstrapConfig() *bootstrap.Config
}

Expand Down
6 changes: 3 additions & 3 deletions internal/xds/server/rds_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ type rdsWatcher struct {
canceled bool // eats callbacks if true
}

func (rw *rdsWatcher) ResourceChanged(update *xdsresource.RouteConfigResourceData, onDone func()) {
func (rw *rdsWatcher) ResourceChanged(update *xdsresource.RouteConfigUpdate, onDone func()) {
defer onDone()
rw.mu.Lock()
if rw.canceled {
Expand All @@ -144,11 +144,11 @@ func (rw *rdsWatcher) ResourceChanged(update *xdsresource.RouteConfigResourceDat
}
rw.mu.Unlock()
if rw.logger.V(2) {
rw.logger.Infof("RDS watch for resource %q received update: %#v", rw.routeName, update.Resource)
rw.logger.Infof("RDS watch for resource %q received update: %#v", rw.routeName, update)
}

routeName := rw.routeName
rwu := rdsWatcherUpdate{data: &update.Resource}
rwu := rdsWatcherUpdate{data: update}
rw.parent.updates[routeName] = rwu
rw.parent.callback(routeName, rwu)
}
Expand Down
8 changes: 1 addition & 7 deletions internal/xds/xdsclient/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import (
"google.golang.org/grpc/internal/xds/bootstrap"
"google.golang.org/grpc/internal/xds/clients/lrsclient"
"google.golang.org/grpc/internal/xds/clients/xdsclient"
"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"

v3statuspb "github.com/envoyproxy/go-control-plane/envoy/service/status/v3"
)
Expand All @@ -49,12 +48,7 @@ type XDSClient interface {
// During a race (e.g. an xDS response is received while the user is calling
// cancel()), there's a small window where the callback can be called after
// the watcher is canceled. Callers need to handle this case.
WatchResource(rType xdsresource.Type, resourceName string, watcher xdsresource.ResourceWatcher) (cancel func())

// WatchResourceV2 matches the API of the external xdsclient interface.
// Once all users of xdsclient have been moved to this watch API, we can
// remove the WatchResource API above, and rename this to WatchResource.
WatchResourceV2(typeURL, resourceName string, watcher xdsclient.ResourceWatcher) (cancel func())
WatchResource(typeURL, resourceName string, watcher xdsclient.ResourceWatcher) (cancel func())

ReportLoad(*bootstrap.ServerConfig) (*lrsclient.LoadStore, func(context.Context))

Expand Down
26 changes: 13 additions & 13 deletions internal/xds/xdsclient/clientimpl_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,9 +82,9 @@ func (s) TestBuildXDSClientConfig_Success(t *testing.T) {
Authorities: map[string]xdsclient.Authority{},
ResourceTypes: map[string]xdsclient.ResourceType{
version.V3ListenerURL: {TypeURL: version.V3ListenerURL, TypeName: xdsresource.ListenerResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewListenerResourceTypeDecoder(c)},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericRouteConfigResourceTypeDecoder()},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewGenericClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericEndpointsResourceTypeDecoder()},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewRouteConfigResourceTypeDecoder(c)},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewEndpointsResourceTypeDecoder(c)},
},
MetricsReporter: &metricsReporter{recorder: stats.NewTestMetricsRecorder(), target: testTargetName},
TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{
Expand Down Expand Up @@ -113,16 +113,16 @@ func (s) TestBuildXDSClientConfig_Success(t *testing.T) {
topLevelSCfg, auth2SCfg := c.XDSServers()[0], c.Authorities()["auth2"].XDSServers[0]
expTopLevelS := xdsclient.ServerConfig{ServerIdentifier: clients.ServerIdentifier{ServerURI: topLevelSCfg.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}}
expAuth2S := xdsclient.ServerConfig{ServerIdentifier: clients.ServerIdentifier{ServerURI: auth2SCfg.ServerURI(), Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"}}}
gSCfgMap := map[xdsclient.ServerConfig]*bootstrap.ServerConfig{expTopLevelS: topLevelSCfg, expAuth2S: auth2SCfg}
gServerCfgMap := map[xdsclient.ServerConfig]*bootstrap.ServerConfig{expTopLevelS: topLevelSCfg, expAuth2S: auth2SCfg}
return xdsclient.Config{
Servers: []xdsclient.ServerConfig{expTopLevelS},
Node: clients.Node{ID: node.GetId(), Cluster: node.GetCluster(), Metadata: node.Metadata, UserAgentName: node.UserAgentName, UserAgentVersion: node.GetUserAgentVersion()},
Authorities: map[string]xdsclient.Authority{"auth1": {XDSServers: []xdsclient.ServerConfig{expTopLevelS}}, "auth2": {XDSServers: []xdsclient.ServerConfig{expAuth2S}}},
ResourceTypes: map[string]xdsclient.ResourceType{
version.V3ListenerURL: {TypeURL: version.V3ListenerURL, TypeName: xdsresource.ListenerResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewListenerResourceTypeDecoder(c)},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericRouteConfigResourceTypeDecoder()},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewGenericClusterResourceTypeDecoder(c, gSCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericEndpointsResourceTypeDecoder()},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewRouteConfigResourceTypeDecoder(c)},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewEndpointsResourceTypeDecoder(c)},
},
MetricsReporter: &metricsReporter{recorder: stats.NewTestMetricsRecorder(), target: testTargetName},
TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{
Expand Down Expand Up @@ -152,9 +152,9 @@ func (s) TestBuildXDSClientConfig_Success(t *testing.T) {
Authorities: map[string]xdsclient.Authority{},
ResourceTypes: map[string]xdsclient.ResourceType{
version.V3ListenerURL: {TypeURL: version.V3ListenerURL, TypeName: xdsresource.ListenerResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewListenerResourceTypeDecoder(c)},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericRouteConfigResourceTypeDecoder()},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewGenericClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericEndpointsResourceTypeDecoder()},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewRouteConfigResourceTypeDecoder(c)},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewEndpointsResourceTypeDecoder(c)},
},
MetricsReporter: &metricsReporter{recorder: stats.NewTestMetricsRecorder(), target: testTargetName},
TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{
Expand Down Expand Up @@ -184,9 +184,9 @@ func (s) TestBuildXDSClientConfig_Success(t *testing.T) {
Authorities: map[string]xdsclient.Authority{},
ResourceTypes: map[string]xdsclient.ResourceType{
version.V3ListenerURL: {TypeURL: version.V3ListenerURL, TypeName: xdsresource.ListenerResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewListenerResourceTypeDecoder(c)},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericRouteConfigResourceTypeDecoder()},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewGenericClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewGenericEndpointsResourceTypeDecoder()},
version.V3RouteConfigURL: {TypeURL: version.V3RouteConfigURL, TypeName: xdsresource.RouteConfigTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewRouteConfigResourceTypeDecoder(c)},
version.V3ClusterURL: {TypeURL: version.V3ClusterURL, TypeName: xdsresource.ClusterResourceTypeName, AllResourcesRequiredInSotW: true, Decoder: xdsresource.NewClusterResourceTypeDecoder(c, gServerCfgMap)},
version.V3EndpointsURL: {TypeURL: version.V3EndpointsURL, TypeName: xdsresource.EndpointsResourceTypeName, AllResourcesRequiredInSotW: false, Decoder: xdsresource.NewEndpointsResourceTypeDecoder(c)},
},
MetricsReporter: &metricsReporter{recorder: stats.NewTestMetricsRecorder(), target: testTargetName},
TransportBuilder: grpctransport.NewBuilder(map[string]grpctransport.Config{
Expand Down
7 changes: 1 addition & 6 deletions internal/xds/xdsclient/clientimpl_watchers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,18 +19,13 @@ package xdsclient

import (
"google.golang.org/grpc/internal/xds/clients/xdsclient"
"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
)

// WatchResource uses xDS to discover the resource associated with the provided
// resource name. The resource type implementation determines how xDS responses
// are are deserialized and validated, as received from the xDS management
// server. Upon receipt of a response from the management server, an
// appropriate callback on the watcher is invoked.
func (c *clientImpl) WatchResource(rType xdsresource.Type, resourceName string, watcher xdsresource.ResourceWatcher) (cancel func()) {
return c.XDSClient.WatchResource(rType.TypeURL(), resourceName, xdsresource.GenericResourceWatcher(watcher))
}

func (c *clientImpl) WatchResourceV2(typeURL, resourceName string, watcher xdsclient.ResourceWatcher) (cancel func()) {
func (c *clientImpl) WatchResource(typeURL, resourceName string, watcher xdsclient.ResourceWatcher) (cancel func()) {
return c.XDSClient.WatchResource(typeURL, resourceName, watcher)
}
6 changes: 3 additions & 3 deletions internal/xds/xdsclient/resource_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,19 +36,19 @@ func supportedResourceTypes(config *bootstrap.Config, gServerCfgMap map[xdsclien
TypeURL: version.V3RouteConfigURL,
TypeName: xdsresource.RouteConfigTypeName,
AllResourcesRequiredInSotW: false,
Decoder: xdsresource.NewGenericRouteConfigResourceTypeDecoder(),
Decoder: xdsresource.NewRouteConfigResourceTypeDecoder(config),
},
version.V3ClusterURL: {
TypeURL: version.V3ClusterURL,
TypeName: xdsresource.ClusterResourceTypeName,
AllResourcesRequiredInSotW: true,
Decoder: xdsresource.NewGenericClusterResourceTypeDecoder(config, gServerCfgMap),
Decoder: xdsresource.NewClusterResourceTypeDecoder(config, gServerCfgMap),
},
version.V3EndpointsURL: {
TypeURL: version.V3EndpointsURL,
TypeName: xdsresource.EndpointsResourceTypeName,
AllResourcesRequiredInSotW: false,
Decoder: xdsresource.NewGenericEndpointsResourceTypeDecoder(),
Decoder: xdsresource.NewEndpointsResourceTypeDecoder(config),
},
}
}
8 changes: 4 additions & 4 deletions internal/xds/xdsclient/tests/authority_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,13 +316,13 @@ func (s) TestAuthority_Fallback(t *testing.T) {
if err != nil {
t.Fatalf("Error when waiting for a resource update callback: %v", err)
}
gotUpdate := v.(xdsresource.ClusterUpdate)
gotUpdate := v.(*xdsresource.ClusterUpdate)
wantUpdate := xdsresource.ClusterUpdate{
ClusterName: clusterName,
EDSServiceName: edsSecondaryName,
}
cmpOpts := []cmp.Option{cmpopts.EquateEmpty(), cmpopts.IgnoreFields(xdsresource.ClusterUpdate{}, "Raw", "LBPolicy", "TelemetryLabels")}
if diff := cmp.Diff(wantUpdate, gotUpdate, cmpOpts...); diff != "" {
if diff := cmp.Diff(wantUpdate, *gotUpdate, cmpOpts...); diff != "" {
t.Fatalf("Diff in the cluster resource update: (-want, got):\n%s", diff)
}

Expand Down Expand Up @@ -351,8 +351,8 @@ func newClusterWatcherV2() *clusterWatcherV2 {
}
}

func (cw *clusterWatcherV2) ResourceChanged(update *xdsresource.ClusterResourceData, onDone func()) {
cw.updateCh.Send(update.Resource)
func (cw *clusterWatcherV2) ResourceChanged(update *xdsresource.ClusterUpdate, onDone func()) {
cw.updateCh.Send(update)
onDone()
}

Expand Down
6 changes: 3 additions & 3 deletions internal/xds/xdsclient/tests/cds_watchers_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ import (

type noopClusterWatcher struct{}

func (noopClusterWatcher) ResourceChanged(_ *xdsresource.ClusterResourceData, onDone func()) {
func (noopClusterWatcher) ResourceChanged(_ *xdsresource.ClusterUpdate, onDone func()) {
onDone()
}
func (noopClusterWatcher) ResourceError(_ error, onDone func()) {
Expand All @@ -67,8 +67,8 @@ func newClusterWatcher() *clusterWatcher {
return &clusterWatcher{updateCh: testutils.NewChannel()}
}

func (cw *clusterWatcher) ResourceChanged(update *xdsresource.ClusterResourceData, onDone func()) {
cw.updateCh.Send(clusterUpdateErrTuple{update: update.Resource})
func (cw *clusterWatcher) ResourceChanged(update *xdsresource.ClusterUpdate, onDone func()) {
cw.updateCh.Send(clusterUpdateErrTuple{update: *update})
onDone()
}

Expand Down
Loading