Skip to content

Commit

Permalink
lru cache: Only send notifications to the new subscriber
Browse files Browse the repository at this point in the history
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>
  • Loading branch information
sorindumitru committed Jul 6, 2024
1 parent a62344a commit 1580ace
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 13 deletions.
17 changes: 9 additions & 8 deletions pkg/agent/manager/cache/lru_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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()
}
Expand All @@ -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()
}
Expand Down
14 changes: 9 additions & 5 deletions pkg/agent/manager/cache/lru_cache_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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) {
Expand Down

0 comments on commit 1580ace

Please sign in to comment.