lru cache: Only send notifications to the new subscriber

When there's a new subscription to the cache we shouldn't notify all existing
subscriptions. That causes them to receive updates even if there was no
change to the SVIDs or bundles.

This can cause issues with users that trigger side effects, such as
restarting a service, when they receive updates on the X509 stream.

Signed-off-by: Sorin Dumitru <sdumitru@bloomberg.net>

pkg/agent/manager/cache/lru_cache.go

@@ -514,16 +514,16 @@ func (c *LRUCache) SyncSVIDsWithSubscribers() {
 	c.syncSVIDsWithSubscribers()
 }
 
-// Notify subscribers of selector set only if all SVIDs for corresponding selector set are cached
+// Notify subscriber of selector set only if all SVIDs for corresponding selector set are cached
 // It returns whether all SVIDs are cached or not.
 // This method should be retried with backoff to avoid lock contention.
-func (c *LRUCache) Notify(selectors []*common.Selector) bool {
+func (c *LRUCache) notifySubscriberIfSVIDAvailable(selectors []*common.Selector, subscriber *lruCacheSubscriber) bool {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	set, setFree := allocSelectorSet(selectors...)
 	defer setFree()
 	if !c.missingSVIDRecords(set) {
-		c.notifyBySelectorSet(set)
+		c.notify(subscriber)
 		return true
 	}
 	return false
@@ -537,11 +537,12 @@ func (c *LRUCache) subscribeToWorkloadUpdates(ctx context.Context, selectors Sel
 	subscriber := c.NewSubscriber(selectors)
 	bo := c.subscribeBackoffFn()
 
+	sub, ok := subscriber.(*lruCacheSubscriber)
+	if !ok {
+		return nil, fmt.Errorf("unexpected subscriber type %T", sub)
+	}
+
 	if len(selectors) == 0 {
-		sub, ok := subscriber.(*lruCacheSubscriber)
-		if !ok {
-			return nil, fmt.Errorf("unexpected subscriber type %T", sub)
-		}
 		if notifyCallbackFn != nil {
 			notifyCallbackFn()
 		}
@@ -552,7 +553,7 @@ func (c *LRUCache) subscribeToWorkloadUpdates(ctx context.Context, selectors Sel
 	// block until all svids are cached and subscriber is notified
 	for {
 		// notifyCallbackFn is used for testing
-		if c.Notify(selectors) {
+		if c.notifySubscriberIfSVIDAvailable(selectors, sub) {
 			if notifyCallbackFn != nil {
 				notifyCallbackFn()
 			}

pkg/agent/manager/cache/lru_cache_test.go

@@ -711,9 +711,9 @@ func TestLRUCacheSVIDCacheExpiry(t *testing.T) {
 	assert.Equal(t, 10, cache.CountSVIDs())
 
 	// foo SVID should be removed from cache as it does not have active subscriber
-	assert.False(t, cache.Notify(makeSelectors("A")))
+	assert.False(t, cache.notifySubscriberIfSVIDAvailable(makeSelectors("A"), subA.(*lruCacheSubscriber)))
 	// bar SVID should be cached as it has active subscriber
-	assert.True(t, cache.Notify(makeSelectors("B")))
+	assert.True(t, cache.notifySubscriberIfSVIDAvailable(makeSelectors("B"), subB.(*lruCacheSubscriber)))
 
 	subA = cache.NewSubscriber(makeSelectors("A"))
 	defer subA.Finish()
@@ -791,20 +791,24 @@ func TestSyncSVIDsWithSubscribers(t *testing.T) {
 	assert.Equal(t, 5, cache.CountSVIDs())
 }
 
-func TestNotify(t *testing.T) {
+func TestNotifySubscriberWhenSVIDIsAvailable(t *testing.T) {
 	cache := newTestLRUCache(t)
 
+	subscriber := cache.NewSubscriber(makeSelectors("A"))
+	sub, ok := subscriber.(*lruCacheSubscriber)
+	require.True(t, ok)
+
 	foo := makeRegistrationEntry("FOO", "A")
 	cache.UpdateEntries(&UpdateEntries{
 		Bundles:             makeBundles(bundleV1),
 		RegistrationEntries: makeRegistrationEntries(foo),
 	}, nil)
 
-	assert.False(t, cache.Notify(makeSelectors("A")))
+	assert.False(t, cache.notifySubscriberIfSVIDAvailable(makeSelectors("A"), sub))
 	cache.UpdateSVIDs(&UpdateSVIDs{
 		X509SVIDs: makeX509SVIDs(foo),
 	})
-	assert.True(t, cache.Notify(makeSelectors("A")))
+	assert.True(t, cache.notifySubscriberIfSVIDAvailable(makeSelectors("A"), sub))
 }
 
 func TestSubscribeToWorkloadUpdatesLRUNoSelectors(t *testing.T) {