easwars review #1

Author: purnesh42H

xds/internal/clients/xdsclient/ads_stream.go

@@ -68,10 +68,10 @@ type dataAndErrTuple struct {
 // occur on the ADS stream. Methods on this interface may be invoked
 // concurrently and implementations need to handle them in a thread-safe manner.
 type adsStreamEventHandler interface {
-	onStreamError(error)                                       // Called when the ADS stream breaks.
-	onWatchExpiry(ResourceType, string)                        // Called when the watch timer expires for a resource.
-	onResponse(response, func()) ([]string, error)             // Called when a response is received on the ADS stream.
-	onRequiredToRemoveUnsubscribedCacheEntries(typeURL string) // Called when it is needed to remove unsubscribed cache entries.
+	onStreamError(error)                           // Called when the ADS stream breaks.
+	onWatchExpiry(ResourceType, string)            // Called when the watch timer expires for a resource.
+	onResponse(response, func()) ([]string, error) // Called when a response is received on the ADS stream.
+	onRequest(typeURL string)                      // Called when a request is about to be sent on the ADS stream.
 }
 
 // state corresponding to a resource type.
@@ -446,7 +446,7 @@ func (s *adsStreamImpl) sendMessageLocked(stream clients.Stream, names []string,
 	}
 
 	// Call the event handler to remove unsubscribed cache entries.
-	s.eventHandler.onRequiredToRemoveUnsubscribedCacheEntries(url)
+	s.eventHandler.onRequest(url)
 
 	msg, err := proto.Marshal(req)
 	if err != nil {

xds/internal/clients/xdsclient/authority.go

@@ -293,16 +293,18 @@ func (a *authority) fallbackToServer(xc *xdsChannelWithConfig) bool {
 	// Subscribe to all existing resources from the new management server.
 	for typ, resources := range a.resources {
 		for name, state := range resources {
-			if a.logger.V(2) {
-				a.logger.Infof("Resubscribing to resource of type %q and name %q", typ.TypeName, name)
-			}
-			xc.channel.subscribe(typ, name)
+			if len(state.watchers) > 0 {
+				if a.logger.V(2) {
+					a.logger.Infof("Resubscribing to resource of type %q and name %q", typ.TypeName, name)
+				}
+				xc.channel.subscribe(typ, name)
 
-			// Add the new channel to the list of xdsChannels from which this
-			// resource has been requested from. Retain the cached resource and
-			// the set of existing watchers (and other metadata fields) in the
-			// resource state.
-			state.xdsChannelConfigs[xc] = true
+				// Add the new channel to the list of xdsChannels from which this
+				// resource has been requested from. Retain the cached resource and
+				// the set of existing watchers (and other metadata fields) in the
+				// resource state.
+				state.xdsChannelConfigs[xc] = true
+			}
 		}
 	}
 	return true
@@ -731,10 +733,6 @@ func (a *authority) unwatchResource(rType ResourceType, resourceName string, wat
 			// there when the watch was registered.
 			resources := a.resources[rType]
 			state := resources[resourceName]
-			if state == nil {
-				a.logger.Warningf("Attempting to unwatch resource %q of type %q which is not currently watched", resourceName, rType.TypeName)
-				return
-			}
 
 			// Delete this particular watcher from the list of watchers, so that its
 			// callback will not be invoked in the future.
@@ -808,7 +806,7 @@ func (a *authority) closeXDSChannels() {
 func (a *authority) watcherExistsForUncachedResource() bool {
 	for _, resourceStates := range a.resources {
 		for _, state := range resourceStates {
-			if state.md.Status == xdsresource.ServiceStatusRequested {
+			if len(state.watchers) > 0 && state.md.Status == xdsresource.ServiceStatusRequested {
 				return true
 			}
 		}
@@ -840,6 +838,9 @@ func (a *authority) resourceConfig() []*v3statuspb.ClientConfig_GenericXdsConfig
 	for rType, resourceStates := range a.resources {
 		typeURL := rType.TypeURL
 		for name, state := range resourceStates {
+			if len(state.watchers) == 0 {
+				continue
+			}
 			var raw *anypb.Any
 			if state.cache != nil {
 				raw = &anypb.Any{TypeUrl: typeURL, Value: state.cache.Bytes()}
@@ -878,6 +879,8 @@ func (a *authority) close() {
 // after sending a discovery request to ensure that resources that were
 // unsubscribed (and thus have no watchers) are eventually removed from the
 // authority's cache.
+//
+// This method is only executed in the context of a serializer callback.
 func (a *authority) removeUnsubscribedCacheEntries(rType ResourceType) {
 	resources := a.resources[rType]
 	if resources == nil {
@@ -887,7 +890,7 @@ func (a *authority) removeUnsubscribedCacheEntries(rType ResourceType) {
 	for name, state := range resources {
 		if len(state.watchers) == 0 {
 			if a.logger.V(2) {
-				a.logger.Infof("Removing resource state for %q of type %q as it has no watchers after an update cycle", name, rType.TypeName)
+				a.logger.Infof("Removing resource state for %q of type %q as it has no watchers", name, rType.TypeName)
 			}
 			delete(resources, name)
 		}

xds/internal/clients/xdsclient/channel.go

@@ -60,9 +60,9 @@ type xdsChannelEventHandler interface {
 	// requested ADS resource does not exist.
 	adsResourceDoesNotExist(ResourceType, string)
 
-	// requiredToRemoveUnsubscribedCacheEntries is called when the xdsChannel
+	// adsResourceRemoveUnsubscribedCacheEntries is called when the xdsChannel
 	// needs to remove unsubscribed cache entries.
-	requiredToRemoveUnsubscribedCacheEntries(ResourceType)
+	adsResourceRemoveUnsubscribedCacheEntries(ResourceType)
 }
 
 // xdsChannelOpts holds the options for creating a new xdsChannel.
@@ -160,7 +160,7 @@ func (xc *xdsChannel) close() {
 	// on all authorities that were interested in this channel.
 	if _, ok := xc.eventHandler.(*channelState); ok {
 		for _, typ := range typesHandledByStream {
-			xc.eventHandler.requiredToRemoveUnsubscribedCacheEntries(typ)
+			xc.eventHandler.adsResourceRemoveUnsubscribedCacheEntries(typ)
 		}
 	}
 
@@ -254,7 +254,9 @@ func (xc *xdsChannel) onResponse(resp response, onDone func()) ([]string, error)
 	return names, err
 }
 
-func (xc *xdsChannel) onRequiredToRemoveUnsubscribedCacheEntries(typeURL string) {
+// onRequest invoked when a request is about to be sent on the ADS stream. It
+// removes the cache entries for the resource type that are no longer subscribed to.
+func (xc *xdsChannel) onRequest(typeURL string) {
 	if xc.closed.HasFired() {
 		if xc.logger.V(2) {
 			xc.logger.Infof("Received an update from the ADS stream on closed ADS stream")
@@ -269,7 +271,7 @@ func (xc *xdsChannel) onRequiredToRemoveUnsubscribedCacheEntries(typeURL string)
 		return
 	}
 
-	xc.eventHandler.requiredToRemoveUnsubscribedCacheEntries(rType)
+	xc.eventHandler.adsResourceRemoveUnsubscribedCacheEntries(rType)
 }
 
 // decodeResponse decodes the resources in the given ADS response.

xds/internal/clients/xdsclient/channel_test.go

@@ -773,5 +773,5 @@ func (ta *testEventHandler) waitForResourceDoesNotExist(ctx context.Context) (Re
 	return typ, name, nil
 }
 
-func (*testEventHandler) requiredToRemoveUnsubscribedCacheEntries(ResourceType) {
+func (*testEventHandler) adsResourceRemoveUnsubscribedCacheEntries(ResourceType) {
 }

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

@@ -575,10 +575,10 @@ func (s) TestRaceUnsubscribeResubscribe(t *testing.T) {
 	}
 	defer client.Close()
 
-	ldsResourceName1 := "test-listener-resource1"
-	ldsResourceName2 := "test-route-configuration-resource1"
-	rdsName1 := "test-listener-resource2"
-	rdsName2 := "test-route-configuration-resource2"
+	const ldsResourceName1 = "test-listener-resource1"
+	const ldsResourceName2 = "test-listener-resource2"
+	const rdsName1 = "test-route-configuration-resource1"
+	const rdsName2 = "test-route-configuration-resource2"
 	listenerResource1 := e2e.DefaultClientListener(ldsResourceName1, rdsName1)
 	listenerResource2 := e2e.DefaultClientListener(ldsResourceName2, rdsName2)
 

xds/internal/clients/xdsclient/xdsclient.go

@@ -439,7 +439,7 @@ func (cs *channelState) adsResourceDoesNotExist(typ ResourceType, resourceName s
 	}
 }
 
-func (cs *channelState) requiredToRemoveUnsubscribedCacheEntries(rType ResourceType) {
+func (cs *channelState) adsResourceRemoveUnsubscribedCacheEntries(rType ResourceType) {
 	if cs.parent.done.HasFired() {
 		return
 	}