Repository: puppetlabs/wash
Pull request #185 (2019-04-05 17:40:40)

Fix deadlock
Allow cached lookup of one category within a cached lookup of a different category of call to work without deadlock. Fixes an issue where CachedList would call cachedDescribeInstance in ec2Instance and sometimes deadlock trying to use the same mutex.

Also avoid caching MetadataOp on ec2InstanceMetadataJSON unnecessarily.


Patch:
diff --git a/datastore/cache.go b/datastore/cache.go
index 5f25ddc..6f1c5f2 100644
--- a/datastore/cache.go
+++ b/datastore/cache.go
@@ -3,6 +3,7 @@ package datastore
 
 import (
 	"regexp"
+	"sync"
 	"time"
 
 	// TODO: Once https://github.com/patrickmn/go-cache/pull/75
@@ -14,7 +15,7 @@ import (
 
 // Cache is an interface for a cache.
 type Cache interface {
-	GetOrUpdate(key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error)
+	GetOrUpdate(category, key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error)
 	Flush()
 	Delete(matcher *regexp.Regexp) []string
 }
@@ -23,7 +24,7 @@ type Cache interface {
 // to get-or-update cached data in a single transaction to avoid redundant update activity.
 type MemCache struct {
 	instance    *cache.Cache
-	locks       []*locksutil.LockEntry
+	locks       sync.Map
 	hasEviction bool
 }
 
@@ -34,7 +35,6 @@ func NewMemCache() *MemCache {
 	cache := cache.New(cache.NoExpiration, 1*time.Minute)
 	return &MemCache{
 		instance:    cache,
-		locks:       locksutil.CreateLocks(),
 		hasEviction: false,
 	}
 }
@@ -48,20 +48,28 @@ func NewMemCacheWithEvicted(f func(string, interface{})) *MemCache {
 	return cache
 }
 
-// LockForKey retrieve the lock used for a specific key.
-func (cache *MemCache) lockForKey(key string) *locksutil.LockEntry {
-	return locksutil.LockForKey(cache.locks, key)
+// LockForKey retrieve the lock used for a specific category/key pair.
+func (cache *MemCache) lockForKey(category, key string) *locksutil.LockEntry {
+	// If a lockset is present for the category, use it. Otherwise create one and add it.
+	obj, ok := cache.locks.Load(category)
+	if !ok {
+		obj, _ = cache.locks.LoadOrStore(category, locksutil.CreateLocks())
+	}
+	return locksutil.LockForKey(obj.([]*locksutil.LockEntry), key)
 }
 
 // GetOrUpdate attempts to retrieve the value stored at the given key.
 // If the value does not exist, then it generates the value using
 // the generateValue function and stores it with the specified ttl.
 // If resetTTLOnHit is true, will reset the cache expiration for the entry.
-func (cache *MemCache) GetOrUpdate(key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error) {
-	l := cache.lockForKey(key)
+func (cache *MemCache) GetOrUpdate(category, key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error) {
+	l := cache.lockForKey(category, key)
 	l.Lock()
 	defer l.Unlock()
 
+	// From here on key is a composition of category and key so we can maintain
+	// a single cache.
+	key = category + "::" + key
 	value, found := cache.instance.Get(key)
 	if found {
 		log.Tracef("Cache hit on %v", key)
diff --git a/journal/core.go b/journal/core.go
index 997ea62..41801bd 100644
--- a/journal/core.go
+++ b/journal/core.go
@@ -79,7 +79,8 @@ func Record(ctx context.Context, msg string, a ...interface{}) {
 
 	log.Debugf(msg, a...)
 
-	obj, err := journalCache.GetOrUpdate(id, expires, true, func() (interface{}, error) {
+	// This is a single-use cache, so pass in an empty category.
+	obj, err := journalCache.GetOrUpdate("", id, expires, true, func() (interface{}, error) {
 		jdir := Dir()
 		if err := os.MkdirAll(jdir, 0750); err != nil {
 			return nil, err
diff --git a/plugin/aws/ec2Instance.go b/plugin/aws/ec2Instance.go
index 36d3503..0e2ef5c 100644
--- a/plugin/aws/ec2Instance.go
+++ b/plugin/aws/ec2Instance.go
@@ -50,6 +50,7 @@ func newEC2Instance(ctx context.Context, inst *ec2Client.Instance, session *sess
 		cloudwatchClient: cloudwatchlogs.New(session),
 	}
 	ec2Instance.SetTTLOf(plugin.ListOp, 30*time.Second)
+	ec2Instance.DisableCachingFor(plugin.MetadataOp)
 
 	metaObj := newDescribeInstanceResult(inst)
 
diff --git a/plugin/cache.go b/plugin/cache.go
index 96affd3..b96f9b9 100644
--- a/plugin/cache.go
+++ b/plugin/cache.go
@@ -241,5 +241,5 @@ func cachedOp(ctx context.Context, opName string, entry Entry, ttl time.Duration
 		return op()
 	}
 
-	return cache.GetOrUpdate(opName+"::"+entry.id(), ttl, false, op)
+	return cache.GetOrUpdate(opName, entry.id(), ttl, false, op)
 }


Test patch:
diff --git a/api/cache_test.go b/api/cache_test.go
index 695801e..8131c16 100644
--- a/api/cache_test.go
+++ b/api/cache_test.go
@@ -27,7 +27,8 @@ func newMockCache() *mockCache {
 	return &mockCache{items: make(map[string]interface{})}
 }
 
-func (m *mockCache) GetOrUpdate(key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error) {
+func (m *mockCache) GetOrUpdate(cat, key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error) {
+	key = cat + "::" + key
 	if v, ok := m.items[key]; ok {
 		return v, nil
 	}
diff --git a/datastore/cache_test.go b/datastore/cache_test.go
index ec27d95..450b6f2 100644
--- a/datastore/cache_test.go
+++ b/datastore/cache_test.go
@@ -37,23 +37,23 @@ func (suite *MemCacheTestSuite) validate(item interface{}, err error) {
 func (suite *MemCacheTestSuite) TestGetOrUpdateNoReset() {
 	suite.thing.On("update").Return(anything, nil)
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Second, false, suite.update))
-	item, ok := suite.mem.instance.Get("an entry")
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Second, false, suite.update))
+	item, ok := suite.mem.instance.Get("cat::an entry")
 	if suite.True(ok) {
 		suite.Equal("anything", item)
 	}
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Nanosecond, false, suite.update))
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Nanosecond, false, suite.update))
 	time.Sleep(time.Nanosecond)
-	item, ok = suite.mem.instance.Get("an entry")
+	item, ok = suite.mem.instance.Get("cat::an entry")
 	if suite.True(ok) {
 		suite.Equal("anything", item)
 	}
 	suite.thing.AssertNumberOfCalls(suite.T(), "update", 1)
 
-	suite.mem.instance.Delete("an entry")
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Second, false, suite.update))
-	item, ok = suite.mem.instance.Get("an entry")
+	suite.mem.instance.Delete("cat::an entry")
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Second, false, suite.update))
+	item, ok = suite.mem.instance.Get("cat::an entry")
 	if suite.True(ok) {
 		suite.Equal("anything", item)
 	}
@@ -63,13 +63,13 @@ func (suite *MemCacheTestSuite) TestGetOrUpdateNoReset() {
 func (suite *MemCacheTestSuite) TestGetOrUpdateExpire() {
 	suite.thing.On("update").Return(anything, nil)
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Nanosecond, false, suite.update))
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Nanosecond, false, suite.update))
 	time.Sleep(time.Nanosecond)
-	_, ok := suite.mem.instance.Get("an entry")
+	_, ok := suite.mem.instance.Get("cat::an entry")
 	suite.False(ok)
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Second, false, suite.update))
-	item, ok := suite.mem.instance.Get("an entry")
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Second, false, suite.update))
+	item, ok := suite.mem.instance.Get("cat::an entry")
 	if suite.True(ok) {
 		suite.Equal("anything", item)
 	}
@@ -79,20 +79,20 @@ func (suite *MemCacheTestSuite) TestGetOrUpdateExpire() {
 func (suite *MemCacheTestSuite) TestGetOrUpdateWithReset() {
 	suite.thing.On("update").Return(anything, nil)
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Second, true, suite.update))
-	item, ok := suite.mem.instance.Get("an entry")
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Second, true, suite.update))
+	item, ok := suite.mem.instance.Get("cat::an entry")
 	if suite.True(ok) {
 		suite.Equal("anything", item)
 	}
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Nanosecond, true, suite.update))
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Nanosecond, true, suite.update))
 	time.Sleep(time.Nanosecond)
-	_, ok = suite.mem.instance.Get("an entry")
+	_, ok = suite.mem.instance.Get("cat::an entry")
 	suite.False(ok)
 	suite.thing.AssertNumberOfCalls(suite.T(), "update", 1)
 
-	suite.validate(suite.mem.GetOrUpdate("an entry", time.Second, true, suite.update))
-	item, ok = suite.mem.instance.Get("an entry")
+	suite.validate(suite.mem.GetOrUpdate("cat", "an entry", time.Second, true, suite.update))
+	item, ok = suite.mem.instance.Get("cat::an entry")
 	if suite.True(ok) {
 		suite.Equal("anything", item)
 	}
@@ -158,11 +158,11 @@ func (suite *MemCacheEvictedTestSuite) TestExpired() {
 	suite.mem.instance.Set("an entry", struct{}{}, time.Nanosecond)
 	time.Sleep(time.Nanosecond)
 
-	_, err := suite.mem.GetOrUpdate("an entry", time.Second, false, func() (interface{}, error) {
+	_, err := suite.mem.GetOrUpdate("cat", "an entry", time.Second, false, func() (interface{}, error) {
 		return nil, errors.New("nope")
 	})
 	suite.Equal(errors.New("nope"), err)
-	if val, ok := suite.mem.instance.Get("an entry"); suite.True(ok) {
+	if val, ok := suite.mem.instance.Get("cat::an entry"); suite.True(ok) {
 		suite.Equal(errors.New("nope"), val)
 	}
 }
diff --git a/plugin/cache_test.go b/plugin/cache_test.go
index 3fb0757..1979d56 100644
--- a/plugin/cache_test.go
+++ b/plugin/cache_test.go
@@ -16,8 +16,8 @@ type cacheTestsMockCache struct {
 	mock.Mock
 }
 
-func (m *cacheTestsMockCache) GetOrUpdate(key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error) {
-	args := m.Called(key, ttl, resetTTLOnHit, generateValue)
+func (m *cacheTestsMockCache) GetOrUpdate(cat, key string, ttl time.Duration, resetTTLOnHit bool, generateValue func() (interface{}, error)) (interface{}, error) {
+	args := m.Called(cat, key, ttl, resetTTLOnHit, generateValue)
 	return args.Get(0), args.Error(1)
 }
 
@@ -175,14 +175,13 @@ func (suite *CacheTestSuite) TestCachedOp() {
 	opName := "Op"
 	opTTL := 5 * time.Second
 	op := func() (interface{}, error) { return "result", nil }
-	opKey := "Op::id"
 	generateValueMatcher := suite.makeGenerateValueMatcher("result")
-	suite.cache.On("GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher)).Return("result", nil).Once()
+	suite.cache.On("GetOrUpdate", opName, entry.id(), opTTL, false, mock.MatchedBy(generateValueMatcher)).Return("result", nil).Once()
 	v, err := CachedOp(context.Background(), opName, entry, opTTL, op)
 	if suite.NoError(err) {
 		suite.Equal("result", v)
 	}
-	suite.cache.AssertCalled(suite.T(), "GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher))
+	suite.cache.AssertCalled(suite.T(), "GetOrUpdate", opName, entry.id(), opTTL, false, mock.MatchedBy(generateValueMatcher))
 }
 
 func (suite *CacheTestSuite) TestDuplicateCNameErr() {
@@ -247,14 +246,13 @@ func (suite *CacheTestSuite) testCachedDefaultOp(
 	opTTL := 5 * time.Second
 	entry.SetTTLOf(op, opTTL)
 	entry.On(opName, ctx).Return(opValue, nil)
-	opKey := opName + "::" + "id"
 	generateValueMatcher := suite.makeGenerateValueMatcher(mungedOpValue)
-	suite.cache.On("GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher)).Return(mungedOpValue, nil).Once()
+	suite.cache.On("GetOrUpdate", opName, entry.id(), opTTL, false, mock.MatchedBy(generateValueMatcher)).Return(mungedOpValue, nil).Once()
 	v, err = cachedDefaultOp(ctx, entry)
 	if suite.NoError(err) {
 		suite.Equal(mungedOpValue, v)
 	}
-	suite.cache.AssertCalled(suite.T(), "GetOrUpdate", opKey, opTTL, false, mock.MatchedBy(generateValueMatcher))
+	suite.cache.AssertCalled(suite.T(), "GetOrUpdate", opName, entry.id(), opTTL, false, mock.MatchedBy(generateValueMatcher))
 }
 
 func toMap(children []Entry) map[string]Entry {