Skip to content

Commit 57dbf23

Browse files
purnesh42Hpurnesh42H
committed
xdsclient: update watcher API as per gRFC A88
1 parent 724f450 commit 57dbf23

22 files changed

+380
-352
lines changed

‎xds/csds/csds_e2e_test.go

Lines changed: 10 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -70,49 +70,37 @@ func Test(t *testing.T) {
7070

7171
type nopListenerWatcher struct{}
7272

73-
func (nopListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
73+
func (nopListenerWatcher) OnResourceChanged(_ *xdsresource.ListenerResourceData, _ error, onDone xdsresource.OnDoneFunc) {
7474
onDone()
7575
}
76-
func (nopListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
77-
onDone()
78-
}
79-
func (nopListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
76+
func (nopListenerWatcher) OnAmbientError(_ error, onDone xdsresource.OnDoneFunc) {
8077
onDone()
8178
}
8279

8380
type nopRouteConfigWatcher struct{}
8481

85-
func (nopRouteConfigWatcher) OnUpdate(_ *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
82+
func (nopRouteConfigWatcher) OnResourceChanged(_ *xdsresource.RouteConfigResourceData, _ error, onDone xdsresource.OnDoneFunc) {
8683
onDone()
8784
}
88-
func (nopRouteConfigWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
89-
onDone()
90-
}
91-
func (nopRouteConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
85+
func (nopRouteConfigWatcher) OnAmbientError(_ error, onDone xdsresource.OnDoneFunc) {
9286
onDone()
9387
}
9488

9589
type nopClusterWatcher struct{}
9690

97-
func (nopClusterWatcher) OnUpdate(_ *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
98-
onDone()
99-
}
100-
func (nopClusterWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
91+
func (nopClusterWatcher) OnResourceChanged(_ *xdsresource.ClusterResourceData, _ error, onDone xdsresource.OnDoneFunc) {
10192
onDone()
10293
}
103-
func (nopClusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
94+
func (nopClusterWatcher) OnAmbientError(_ error, onDone xdsresource.OnDoneFunc) {
10495
onDone()
10596
}
10697

10798
type nopEndpointsWatcher struct{}
10899

109-
func (nopEndpointsWatcher) OnUpdate(_ *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
110-
onDone()
111-
}
112-
func (nopEndpointsWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
100+
func (nopEndpointsWatcher) OnResourceChanged(_ *xdsresource.EndpointsResourceData, _ error, onDone xdsresource.OnDoneFunc) {
113101
onDone()
114102
}
115-
func (nopEndpointsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
103+
func (nopEndpointsWatcher) OnAmbientError(_ error, onDone xdsresource.OnDoneFunc) {
116104
onDone()
117105
}
118106

@@ -137,13 +125,10 @@ func newBlockingListenerWatcher(testCtxDone <-chan struct{}) *blockingListenerWa
137125
}
138126
}
139127

140-
func (w *blockingListenerWatcher) OnUpdate(_ *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
141-
writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
142-
}
143-
func (w *blockingListenerWatcher) OnError(_ error, onDone xdsresource.OnDoneFunc) {
128+
func (w *blockingListenerWatcher) OnResourceChanged(_ *xdsresource.ListenerResourceData, _ error, onDone xdsresource.OnDoneFunc) {
144129
writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
145130
}
146-
func (w *blockingListenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
131+
func (w *blockingListenerWatcher) OnAmbientError(_ error, onDone xdsresource.OnDoneFunc) {
147132
writeOnDone(w.testCtxDone, w.onDoneCh, onDone)
148133
}
149134

‎xds/internal/balancer/cdsbalancer/cluster_watcher.go

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -32,21 +32,26 @@ type clusterWatcher struct {
3232
parent *cdsBalancer
3333
}
3434

35-
func (cw *clusterWatcher) OnUpdate(u *xdsresource.ClusterResourceData, onDone xdsresource.OnDoneFunc) {
35+
func (cw *clusterWatcher) OnResourceChanged(u *xdsresource.ClusterResourceData, err error, onDone xdsresource.OnDoneFunc) {
36+
if err != nil {
37+
if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound {
38+
handleNotFound := func(context.Context) { cw.parent.onClusterResourceNotFound(cw.name); onDone() }
39+
cw.parent.serializer.ScheduleOr(handleNotFound, onDone)
40+
} else {
41+
handleError := func(context.Context) { cw.parent.onClusterError(cw.name, err); onDone() }
42+
cw.parent.serializer.ScheduleOr(handleError, onDone)
43+
}
44+
return
45+
}
3646
handleUpdate := func(context.Context) { cw.parent.onClusterUpdate(cw.name, u.Resource); onDone() }
3747
cw.parent.serializer.ScheduleOr(handleUpdate, onDone)
3848
}
3949

40-
func (cw *clusterWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
50+
func (cw *clusterWatcher) OnAmbientError(err error, onDone xdsresource.OnDoneFunc) {
4151
handleError := func(context.Context) { cw.parent.onClusterError(cw.name, err); onDone() }
4252
cw.parent.serializer.ScheduleOr(handleError, onDone)
4353
}
4454

45-
func (cw *clusterWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
46-
handleNotFound := func(context.Context) { cw.parent.onClusterResourceNotFound(cw.name); onDone() }
47-
cw.parent.serializer.ScheduleOr(handleNotFound, onDone)
48-
}
49-
5055
// watcherState groups the state associated with a clusterWatcher.
5156
type watcherState struct {
5257
watcher *clusterWatcher // The underlying watcher.

‎xds/internal/balancer/clusterresolver/resource_resolver_eds.go

Lines changed: 24 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -76,20 +76,42 @@ func newEDSResolver(nameToWatch string, producer xdsresource.Producer, topLevelR
7676
}
7777

7878
// OnUpdate is invoked to report an update for the resource being watched.
79-
func (er *edsDiscoveryMechanism) OnUpdate(update *xdsresource.EndpointsResourceData, onDone xdsresource.OnDoneFunc) {
79+
func (er *edsDiscoveryMechanism) OnResourceChanged(update *xdsresource.EndpointsResourceData, err error, onDone xdsresource.OnDoneFunc) {
8080
if er.stopped.HasFired() {
8181
onDone()
8282
return
8383
}
8484

85+
if err != nil {
86+
if er.logger.V(2) {
87+
if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound {
88+
er.logger.Infof("EDS discovery mechanism for resource %q reported resource-does-not-exist error", er.nameToWatch)
89+
} else {
90+
er.logger.Infof("EDS discovery mechanism for resource %q reported error: %v", er.nameToWatch, err)
91+
}
92+
}
93+
// Report an empty update that would result in no priority child being
94+
// created for this discovery mechanism. This would result in the priority
95+
// LB policy reporting TRANSIENT_FAILURE (as there would be no priorities or
96+
// localities) if this was the only discovery mechanism, or would result in
97+
// the priority LB policy using a lower priority discovery mechanism when
98+
// that becomes available.
99+
er.mu.Lock()
100+
er.update = &xdsresource.EndpointsUpdate{}
101+
er.mu.Unlock()
102+
103+
er.topLevelResolver.onUpdate(onDone)
104+
return
105+
}
106+
85107
er.mu.Lock()
86108
er.update = &update.Resource
87109
er.mu.Unlock()
88110

89111
er.topLevelResolver.onUpdate(onDone)
90112
}
91113

92-
func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.OnDoneFunc) {
114+
func (er *edsDiscoveryMechanism) OnAmbientError(err error, onDone xdsresource.OnDoneFunc) {
93115
if er.stopped.HasFired() {
94116
onDone()
95117
return
@@ -119,26 +141,3 @@ func (er *edsDiscoveryMechanism) OnError(err error, onDone xdsresource.OnDoneFun
119141

120142
er.topLevelResolver.onUpdate(onDone)
121143
}
122-
123-
func (er *edsDiscoveryMechanism) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
124-
if er.stopped.HasFired() {
125-
onDone()
126-
return
127-
}
128-
129-
if er.logger.V(2) {
130-
er.logger.Infof("EDS discovery mechanism for resource %q reported resource-does-not-exist error", er.nameToWatch)
131-
}
132-
133-
// Report an empty update that would result in no priority child being
134-
// created for this discovery mechanism. This would result in the priority
135-
// LB policy reporting TRANSIENT_FAILURE (as there would be no priorities or
136-
// localities) if this was the only discovery mechanism, or would result in
137-
// the priority LB policy using a lower priority discovery mechanism when
138-
// that becomes available.
139-
er.mu.Lock()
140-
er.update = &xdsresource.EndpointsUpdate{}
141-
er.mu.Unlock()
142-
143-
er.topLevelResolver.onUpdate(onDone)
144-
}

‎xds/internal/resolver/watch_service.go

Lines changed: 20 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -36,21 +36,21 @@ func newListenerWatcher(resourceName string, parent *xdsResolver) *listenerWatch
3636
return lw
3737
}
3838

39-
func (l *listenerWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
39+
func (l *listenerWatcher) OnResourceChanged(update *xdsresource.ListenerResourceData, err error, onDone xdsresource.OnDoneFunc) {
40+
if err != nil {
41+
handleError := func(context.Context) { l.parent.onListenerResourceChangedError(err); onDone() }
42+
l.parent.serializer.ScheduleOr(handleError, onDone)
43+
return
44+
}
4045
handleUpdate := func(context.Context) { l.parent.onListenerResourceUpdate(update.Resource); onDone() }
4146
l.parent.serializer.ScheduleOr(handleUpdate, onDone)
4247
}
4348

44-
func (l *listenerWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
45-
handleError := func(context.Context) { l.parent.onListenerResourceError(err); onDone() }
49+
func (l *listenerWatcher) OnAmbientError(err error, onDone xdsresource.OnDoneFunc) {
50+
handleError := func(context.Context) { l.parent.onListenerResourceAmbientError(err); onDone() }
4651
l.parent.serializer.ScheduleOr(handleError, onDone)
4752
}
4853

49-
func (l *listenerWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
50-
handleNotFound := func(context.Context) { l.parent.onListenerResourceNotFound(); onDone() }
51-
l.parent.serializer.ScheduleOr(handleNotFound, onDone)
52-
}
53-
5454
func (l *listenerWatcher) stop() {
5555
l.cancel()
5656
l.parent.logger.Infof("Canceling watch on Listener resource %q", l.resourceName)
@@ -68,24 +68,29 @@ func newRouteConfigWatcher(resourceName string, parent *xdsResolver) *routeConfi
6868
return rw
6969
}
7070

71-
func (r *routeConfigWatcher) OnUpdate(u *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
71+
func (r *routeConfigWatcher) OnResourceChanged(u *xdsresource.RouteConfigResourceData, err error, onDone xdsresource.OnDoneFunc) {
72+
if err != nil {
73+
var handleError func(context.Context)
74+
if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound {
75+
handleError = func(context.Context) { r.parent.onRouteConfigResourceNotFound(r.resourceName); onDone() }
76+
} else {
77+
handleError = func(context.Context) { r.parent.onRouteConfigResourceError(r.resourceName, err); onDone() }
78+
}
79+
r.parent.serializer.ScheduleOr(handleError, onDone)
80+
return
81+
}
7282
handleUpdate := func(context.Context) {
7383
r.parent.onRouteConfigResourceUpdate(r.resourceName, u.Resource)
7484
onDone()
7585
}
7686
r.parent.serializer.ScheduleOr(handleUpdate, onDone)
7787
}
7888

79-
func (r *routeConfigWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
89+
func (r *routeConfigWatcher) OnAmbientError(err error, onDone xdsresource.OnDoneFunc) {
8090
handleError := func(context.Context) { r.parent.onRouteConfigResourceError(r.resourceName, err); onDone() }
8191
r.parent.serializer.ScheduleOr(handleError, onDone)
8292
}
8393

84-
func (r *routeConfigWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
85-
handleNotFound := func(context.Context) { r.parent.onRouteConfigResourceNotFound(r.resourceName); onDone() }
86-
r.parent.serializer.ScheduleOr(handleNotFound, onDone)
87-
}
88-
8994
func (r *routeConfigWatcher) stop() {
9095
r.cancel()
9196
r.parent.logger.Infof("Canceling watch on RouteConfiguration resource %q", r.resourceName)

‎xds/internal/resolver/xds_resolver.go

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -518,17 +518,21 @@ func (r *xdsResolver) onListenerResourceUpdate(update xdsresource.ListenerUpdate
518518
r.routeConfigWatcher = newRouteConfigWatcher(r.rdsResourceName, r)
519519
}
520520

521-
func (r *xdsResolver) onListenerResourceError(err error) {
521+
func (r *xdsResolver) onListenerResourceAmbientError(err error) {
522522
if r.logger.V(2) {
523523
r.logger.Infof("Received error for Listener resource %q: %v", r.ldsResourceName, err)
524524
}
525525
r.onError(err)
526526
}
527527

528528
// Only executed in the context of a serializer callback.
529-
func (r *xdsResolver) onListenerResourceNotFound() {
529+
func (r *xdsResolver) onListenerResourceChangedError(err error) {
530530
if r.logger.V(2) {
531-
r.logger.Infof("Received resource-not-found-error for Listener resource %q", r.ldsResourceName)
531+
if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound {
532+
r.logger.Infof("Received resource-not-found-error for Listener resource %q", r.ldsResourceName)
533+
} else {
534+
r.logger.Infof("Received on-resource-changed error for Listener resource %q: %v", r.ldsResourceName, err)
535+
}
532536
}
533537

534538
r.listenerUpdateRecvd = false

‎xds/internal/server/listener_wrapper.go

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -414,19 +414,33 @@ type ldsWatcher struct {
414414
name string
415415
}
416416

417-
func (lw *ldsWatcher) OnUpdate(update *xdsresource.ListenerResourceData, onDone xdsresource.OnDoneFunc) {
417+
func (lw *ldsWatcher) OnResourceChanged(update *xdsresource.ListenerResourceData, err error, onDone xdsresource.OnDoneFunc) {
418418
defer onDone()
419419
if lw.parent.closed.HasFired() {
420-
lw.logger.Warningf("Resource %q received update: %#v after listener was closed", lw.name, update)
420+
if err != nil {
421+
lw.logger.Warningf("Resource %q received err: %#v after listener was closed", lw.name, err)
422+
} else {
423+
lw.logger.Warningf("Resource %q received update: %#v after listener was closed", lw.name, update)
424+
}
421425
return
422426
}
423427
if lw.logger.V(2) {
424-
lw.logger.Infof("LDS watch for resource %q received update: %#v", lw.name, update.Resource)
428+
if err != nil {
429+
lw.logger.Infof("LDS watch for resource %q received error: %#v", lw.name, err)
430+
} else {
431+
lw.logger.Infof("LDS watch for resource %q received update: %#v", lw.name, update.Resource)
432+
}
433+
}
434+
if err != nil {
435+
if xdsresource.ErrType(err) == xdsresource.ErrorTypeResourceNotFound {
436+
lw.parent.onLDSResourceDoesNotExist(err)
437+
}
438+
return
425439
}
426440
lw.parent.handleLDSUpdate(update.Resource)
427441
}
428442

429-
func (lw *ldsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
443+
func (lw *ldsWatcher) OnAmbientError(err error, onDone xdsresource.OnDoneFunc) {
430444
defer onDone()
431445
if lw.parent.closed.HasFired() {
432446
lw.logger.Warningf("Resource %q received error: %v after listener was closed", lw.name, err)
@@ -438,17 +452,3 @@ func (lw *ldsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
438452
// For errors which are anything other than "resource-not-found", we
439453
// continue to use the old configuration.
440454
}
441-
442-
func (lw *ldsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
443-
defer onDone()
444-
if lw.parent.closed.HasFired() {
445-
lw.logger.Warningf("Resource %q received resource-does-not-exist error after listener was closed", lw.name)
446-
return
447-
}
448-
if lw.logger.V(2) {
449-
lw.logger.Infof("LDS watch for resource %q reported resource-does-not-exist error", lw.name)
450-
}
451-
452-
err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "resource name %q of type Listener not found in received response", lw.name)
453-
lw.parent.onLDSResourceDoesNotExist(err)
454-
}

‎xds/internal/server/rds_handler.go

Lines changed: 11 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,7 @@ type rdsWatcher struct {
147147
canceled bool // eats callbacks if true
148148
}
149149

150-
func (rw *rdsWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDone xdsresource.OnDoneFunc) {
150+
func (rw *rdsWatcher) OnResourceChanged(update *xdsresource.RouteConfigResourceData, err error, onDone xdsresource.OnDoneFunc) {
151151
defer onDone()
152152
rw.mu.Lock()
153153
if rw.canceled {
@@ -156,26 +156,20 @@ func (rw *rdsWatcher) OnUpdate(update *xdsresource.RouteConfigResourceData, onDo
156156
}
157157
rw.mu.Unlock()
158158
if rw.logger.V(2) {
159-
rw.logger.Infof("RDS watch for resource %q received update: %#v", rw.routeName, update.Resource)
159+
if err != nil {
160+
rw.logger.Infof("RDS watch for resource %q received error: %#v", rw.routeName, err)
161+
} else {
162+
rw.logger.Infof("RDS watch for resource %q received update: %#v", rw.routeName, update.Resource)
163+
}
160164
}
161-
rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{data: &update.Resource})
162-
}
163-
164-
func (rw *rdsWatcher) OnError(err error, onDone xdsresource.OnDoneFunc) {
165-
defer onDone()
166-
rw.mu.Lock()
167-
if rw.canceled {
168-
rw.mu.Unlock()
165+
if err != nil {
166+
rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{err: err})
169167
return
170168
}
171-
rw.mu.Unlock()
172-
if rw.logger.V(2) {
173-
rw.logger.Infof("RDS watch for resource %q reported error: %v", rw.routeName, err)
174-
}
175-
rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{err: err})
169+
rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{data: &update.Resource})
176170
}
177171

178-
func (rw *rdsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
172+
func (rw *rdsWatcher) OnAmbientError(err error, onDone xdsresource.OnDoneFunc) {
179173
defer onDone()
180174
rw.mu.Lock()
181175
if rw.canceled {
@@ -184,8 +178,7 @@ func (rw *rdsWatcher) OnResourceDoesNotExist(onDone xdsresource.OnDoneFunc) {
184178
}
185179
rw.mu.Unlock()
186180
if rw.logger.V(2) {
187-
rw.logger.Infof("RDS watch for resource %q reported resource-does-not-exist error: %v", rw.routeName)
181+
rw.logger.Infof("RDS watch for resource %q reported error: %v", rw.routeName, err)
188182
}
189-
err := xdsresource.NewErrorf(xdsresource.ErrorTypeResourceNotFound, "resource name %q of type RouteConfiguration not found in received response", rw.routeName)
190183
rw.parent.handleRouteUpdate(rw.routeName, rdsWatcherUpdate{err: err})
191184
}

0 commit comments

Comments
 (0)