Adds graceful shutdown to concurrent core

Author: reederc42

core/concurrent_cache/backend_test.go

@@ -31,7 +31,6 @@ func TestUpsertBackend_Metrics(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Reset metrics before each test
 			metrics.BackendsGauge.Reset()
@@ -132,7 +131,6 @@ func TestDeleteBackend_Metrics(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Reset metrics before each test
 			metrics.BackendsGauge.Reset()
@@ -294,7 +292,6 @@ func TestReadBackend(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Set up initial state
 			backends.lock()
@@ -354,7 +351,6 @@ func TestInconsistentReadBackend(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Set up initial state
 			backends.lock()
@@ -417,7 +413,6 @@ func TestReadBackendByName(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Set up initial state
 			backends.lock()
@@ -485,7 +480,6 @@ func TestInconsistentReadBackendByName(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Set up initial state
 			backends.lock()

core/concurrent_cache/concurrent_cache.go

@@ -28,6 +28,10 @@ func init() {
 
 // Initialize Need this as public for unit tests
 func Initialize() {
+	rootCache = cache{
+		data:          make(map[string]SmartCopier),
+		resourceLocks: locks.NewGCNamedMutex(),
+	}
 	nodes = cache{
 		data:          make(map[string]SmartCopier),
 		resourceLocks: locks.NewGCNamedMutex(),
@@ -49,7 +53,6 @@ func Initialize() {
 	volumes = cache{
 		data:          make(map[string]SmartCopier),
 		resourceLocks: locks.NewGCNamedMutex(),
-
 		// volume internal name
 		key: &uniqueKey{
 			data:     make(map[string]string),
@@ -84,6 +87,8 @@ func Initialize() {
 	for r := range schema {
 		resourceRanks[r] = rankForResource(r)
 	}
+
+	rootCache.data["."] = SmartCopier(nil)
 }
 
 // SmartCopier is an interface for objects that can be copied for the cache. If the object implements interior
@@ -143,6 +148,7 @@ func (c *cache) runlock() {
 }
 
 var (
+	rootCache          cache
 	nodes              cache
 	storageClasses     cache
 	backends           cache
@@ -153,6 +159,7 @@ var (
 	autogrowPolicies   cache
 
 	caches = map[resource]*cache{
+		root:              &rootCache,
 		node:              &nodes,
 		storageClass:      &storageClasses,
 		backend:           &backends,
@@ -244,6 +251,7 @@ func assembleQueries(queries [][]Subquery, roots [][]int, cachesPresent map[reso
 		}
 		queries[i], roots[i] = buildTrees(ri, q)
 	}
+
 	return nil
 }
 
@@ -363,7 +371,7 @@ func dedupe(query []Subquery) (map[resource]int, error) {
 	return resourceIndices, nil
 }
 
-// buildTrees takes subqueries and finds the roots, and fills in any missing parents
+// buildTrees takes subqueries and finds the roots, and fills in any missing dependencies
 func buildTrees(resourceIndices map[resource]int, query []Subquery) ([]Subquery, []int) {
 	roots := make([]int, 0)
 	l := len(query)
@@ -377,11 +385,11 @@ func buildTrees(resourceIndices map[resource]int, query []Subquery) ([]Subquery,
 			continue
 		}
 
-		// if the current subquery has ownable resources in the query it is not a root
+		// if the current subquery has dependent resources in the query it is not a root
 		root := true
-		ownables := inverseSchema[query[i].res]
-		for j := 0; j < len(ownables) && root; j++ {
-			if _, ok := resourceIndices[ownables[j]]; ok {
+		dependents := inverseSchema[query[i].res]
+		for j := 0; j < len(dependents) && root; j++ {
+			if _, ok := resourceIndices[dependents[j]]; ok {
 				root = false
 			}
 		}
@@ -479,11 +487,13 @@ func fillInIDs(root int, query []Subquery) error {
 }
 
 func mergeQueries(queries [][]Subquery) []Subquery {
-	length := 0
+	length := 1
 	for _, q := range queries {
 		length += len(q)
 	}
 	merged := make([]Subquery, 0, length)
+	// always add implied read root, if write root is present it will take precedence
+	merged = append(merged, Subquery{res: root, op: read, id: "."})
 	for i, q := range queries {
 		for j := range q {
 			q[j].result = i
@@ -505,9 +515,9 @@ func mergeQueries(queries [][]Subquery) []Subquery {
 		}
 
 		switch {
-		case resourceRanks[i.res] > resourceRanks[j.res]:
-			return -1
 		case resourceRanks[i.res] < resourceRanks[j.res]:
+			return -1
+		case resourceRanks[i.res] > resourceRanks[j.res]:
 			return 1
 		}
 
@@ -688,7 +698,7 @@ func checkDependency(s []Subquery, i int, r resource) error {
 
 func rankForResource(r resource) int {
 	rank := 0
-	for _, c := range inverseSchema[r] {
+	for _, c := range schema[r] {
 		cRank := 1 + rankForResource(c)
 		if cRank > rank {
 			rank = cRank
@@ -786,7 +796,8 @@ var resourceNames = map[resource]string{
 }
 
 const (
-	node = resource(iota)
+	root = resource(iota)
+	node
 	storageClass
 	backend
 	volume
@@ -825,6 +836,7 @@ const (
 // schema is the authoritative source for relationships between resources. For example,
 // a snapshot depends on a volume, and a volume depends on a backend.
 var schema = map[resource][]resource{
+	root:              nil, // root is an implied dependency for all resources
 	node:              nil,
 	storageClass:      nil,
 	backend:           nil,

core/concurrent_cache/concurrent_cache_test.go

@@ -164,7 +164,6 @@ func TestFillInIDs(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			initCaches()
-			defer cleanCaches()
 			m, err := dedupe(tt.query)
 			assert.NoError(t, err)
 			query, roots := buildTrees(m, tt.query)
@@ -214,7 +213,6 @@ func TestMergeQueries(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			initCaches()
 			nodes.data["node2"] = &models.Node{Name: "node2"}
-			defer cleanCaches()
 
 			queries := make([][]Subquery, 0, len(tt.queries))
 			for _, q := range tt.queries {
@@ -229,7 +227,7 @@ func TestMergeQueries(t *testing.T) {
 			}
 
 			merged := mergeQueries(queries)
-			assert.Len(t, merged, 13)
+			assert.Len(t, merged, 14)
 		})
 	}
 }
@@ -238,7 +236,6 @@ func TestLock(t *testing.T) {
 	// start two goroutines that try to update the same resources using multiple queries
 	// passes if there is no deadlock
 	initCaches()
-	defer cleanCaches()
 	makeQueries := func() [][]Subquery {
 		return [][]Subquery{
 			{
@@ -295,14 +292,11 @@ func TestLock(t *testing.T) {
 }
 
 func initCaches() {
-	nodes.data = make(map[string]SmartCopier)
-	backends.data = make(map[string]SmartCopier)
-	volumes.data = make(map[string]SmartCopier)
-	volumePublications.data = make(map[string]SmartCopier)
-	snapshots.data = make(map[string]SmartCopier)
-	autogrowPolicies.data = make(map[string]SmartCopier)
+	Initialize()
 
-	nodes.data["node1"] = &models.Node{}
+	nodes.data["node1"] = &models.Node{
+		Name: "node1",
+	}
 	backends.data["backend1"] = storage.NewTestStorageBackend()
 	volumes.data["volume1"] = &storage.Volume{
 		BackendUUID: "backend1",
@@ -323,15 +317,6 @@ func initCaches() {
 	autogrowPolicies.data["policy1"] = storage.NewAutogrowPolicy("policy1", "80", "20", "1000Gi", storage.AutogrowPolicyStateSuccess)
 }
 
-func cleanCaches() {
-	nodes.data = make(map[string]SmartCopier)
-	backends.data = make(map[string]SmartCopier)
-	volumes.data = make(map[string]SmartCopier)
-	volumePublications.data = make(map[string]SmartCopier)
-	snapshots.data = make(map[string]SmartCopier)
-	autogrowPolicies.data = make(map[string]SmartCopier)
-}
-
 func createQueryGenerators(numIds int) [][]queryGenerator {
 	generators := make([][]queryGenerator, 0, numIds)
 

core/concurrent_cache/metrics_test.go

@@ -91,7 +91,6 @@ func TestAddAndDeleteBackendMetrics(t *testing.T) {
 			metrics.TridentBackendInfo.Reset()
 
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Create mock backend
 			mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{
@@ -150,7 +149,6 @@ func TestAddAndDeleteBackendMetrics_MultipleBackends(t *testing.T) {
 	// Add all backends to metrics
 	for i, b := range backends {
 		mockCtrl := gomock.NewController(t)
-		defer mockCtrl.Finish()
 
 		mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{
 			"driverName": b.driverName,
@@ -243,7 +241,6 @@ func TestAddAndDeleteVolumeMetrics(t *testing.T) {
 			metrics.VolumeAllocatedBytesGauge.Reset()
 
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Create mock backend
 			mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{
@@ -348,7 +345,6 @@ func TestAddAndDeleteVolumeMetrics_InvalidSize(t *testing.T) {
 			metrics.VolumeAllocatedBytesGauge.Reset()
 
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Create mock backend
 			mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{
@@ -457,7 +453,6 @@ func TestAddAndDeleteSnapshotMetrics(t *testing.T) {
 			metrics.SnapshotAllocatedBytesGauge.Reset()
 
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Create mock backend
 			mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{

core/concurrent_cache/root.go

@@ -0,0 +1,13 @@
+package concurrent_cache
+
+// LockCache takes a write lock on the cache root, which blocks all other operations.
+func LockCache() Subquery {
+	return Subquery{
+		res: root,
+		op:  upsert,
+		id:  ".",
+		setResults: func(s *Subquery, r *Result) error {
+			return nil
+		},
+	}
+}

core/concurrent_cache/root_test.go

@@ -0,0 +1,68 @@
+package concurrent_cache
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestLockCache tests that a Lock call with the LockCache subquery holds the root resource
+// and blocks any other Lock call until it is released.
+func TestLockCache(t *testing.T) {
+	initCaches()
+
+	// Goroutine 1: acquire Lock with LockCache() and hold it
+	lockHeld := make(chan struct{})
+	releaseRequested := make(chan struct{})
+	var unlockFirst func()
+	var firstErr error
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_, unlock, err := Lock(context.Background(), Query(LockCache()))
+		firstErr = err
+		unlockFirst = unlock
+		close(lockHeld)
+		<-releaseRequested
+		if unlockFirst != nil {
+			unlockFirst()
+		}
+	}()
+
+	// Wait until the first Lock has been acquired
+	<-lockHeld
+	assert.NoError(t, firstErr, "first Lock(LockCache()) should succeed")
+
+	// Goroutine 2: attempt Lock with another query; it must block while the first holds LockCache
+	secondDone := make(chan struct{})
+	var secondErr error
+	go func() {
+		_, _, secondErr = Lock(context.Background(), Query(ReadBackend("backend1"), ReadVolume("volume1")))
+		close(secondDone)
+	}()
+
+	// Give the second goroutine a moment to reach the Lock call and block
+	time.Sleep(50 * time.Millisecond)
+	select {
+	case <-secondDone:
+		assert.FailNow(t, "second Lock completed while first still holds LockCache; LockCache should block other Lock calls")
+	default:
+		// Second Lock is still blocked, as expected
+	}
+
+	// Release the first lock; second Lock should now complete
+	close(releaseRequested)
+	wg.Wait()
+
+	// Wait for second Lock to finish (with a timeout)
+	select {
+	case <-secondDone:
+		assert.NoError(t, secondErr, "second Lock should succeed after first releases")
+	case <-time.After(2 * time.Second):
+		assert.FailNow(t, "second Lock did not complete after first released LockCache")
+	}
+}

core/concurrent_cache/snapshot_test.go

@@ -50,7 +50,6 @@ func TestUpsertSnapshot_Metrics(t *testing.T) {
 			metrics.SnapshotAllocatedBytesGauge.Reset()
 
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Create mock backend and volume
 			mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{
@@ -203,7 +202,6 @@ func TestDeleteSnapshot_Metrics(t *testing.T) {
 			metrics.SnapshotAllocatedBytesGauge.Reset()
 
 			mockCtrl := gomock.NewController(t)
-			defer mockCtrl.Finish()
 
 			// Create mock backend
 			mockBackend := getMockBackendWithMap(mockCtrl, map[string]string{