APIGOV-32372 cache rebuild optimizations (#1023)

* APIGOV-32372 rebuild only failed kinds

* APIGOV-32372 fix for listener/cache rebuild race condition

* APIGOV-32372 merge main

* APIGOV-32372 remove else statement

Co-authored-by: Copilot <copilot@github.com>

* APIGOV-32327 wait for clients to be ready

* APIGOV-32372 improvements

* APIGOV-32372 fix for unlock on unlocked mutex error

* APIGOV-32372 guard against listener potential data race

---------

Co-authored-by: Copilot <copilot@github.com>

Author: alrosca

pkg/agent/cache/cachevalidation.go

@@ -66,6 +66,17 @@ func (c *cacheManager) getCacheForKind(kind string) cache.Cache {
 	}
 }
 
+// FlushKind empties the dedicated cache for the given resource kind.
+// Kinds without a dedicated cache are silently ignored.
+func (c *cacheManager) FlushKind(kind string) {
+	c.ApplyResourceReadLock()
+	defer c.ReleaseResourceReadLock()
+
+	if resourceCache := c.getCacheForKind(kind); resourceCache != nil {
+		resourceCache.Flush()
+	}
+}
+
 // ResourceCacheKey builds a unique cache key from kind, scope name, and resource name.
 func ResourceCacheKey(kind, scopeName, name string) string {
 	return kind + "/" + scopeName + "/" + name

pkg/agent/cache/cachevalidation_test.go

@@ -219,3 +219,77 @@ func TestGetCachedResourcesByKind(t *testing.T) {
 		})
 	}
 }
+
+func TestFlushKind(t *testing.T) {
+	modTime := time.Date(2026, 3, 12, 10, 0, 0, 0, time.UTC)
+
+	tests := map[string]struct {
+		setup     func(Manager)
+		flushKind string
+		// after flush, verify these kinds still have data
+		expectNonEmpty []struct{ group, kind string }
+		// after flush, verify these kinds are empty
+		expectEmpty []struct{ group, kind string }
+	}{
+		"flush APIService leaves other caches intact": {
+			setup: func(cm Manager) {
+				cm.AddAPIService(makeAPIServiceRI("env1", "svc1", "ext-id-1", modTime))
+				cm.AddAPIServiceInstance(makeRI("management", management.APIServiceInstanceGVK().Kind, "Environment", "env1", "inst1", "id1", modTime))
+			},
+			flushKind: management.APIServiceGVK().Kind,
+			expectEmpty: []struct{ group, kind string }{
+				{"management", management.APIServiceGVK().Kind},
+			},
+			expectNonEmpty: []struct{ group, kind string }{
+				{"management", management.APIServiceInstanceGVK().Kind},
+			},
+		},
+		"flush APIServiceInstance": {
+			setup: func(cm Manager) {
+				cm.AddAPIServiceInstance(makeRI("management", management.APIServiceInstanceGVK().Kind, "Environment", "env1", "inst1", "id1", modTime))
+			},
+			flushKind: management.APIServiceInstanceGVK().Kind,
+			expectEmpty: []struct{ group, kind string }{
+				{"management", management.APIServiceInstanceGVK().Kind},
+			},
+		},
+		"flush ManagedApplication": {
+			setup: func(cm Manager) {
+				cm.AddManagedApplication(makeRI("management", management.ManagedApplicationGVK().Kind, "Environment", "env1", "app1", "id1", modTime))
+			},
+			flushKind: management.ManagedApplicationGVK().Kind,
+			expectEmpty: []struct{ group, kind string }{
+				{"management", management.ManagedApplicationGVK().Kind},
+			},
+		},
+		"flush unknown kind is a no-op": {
+			setup: func(cm Manager) {
+				cm.AddAPIService(makeAPIServiceRI("env1", "svc1", "ext-id-1", modTime))
+			},
+			flushKind: "UnknownKind",
+			expectNonEmpty: []struct{ group, kind string }{
+				{"management", management.APIServiceGVK().Kind},
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			cm := NewAgentCacheManager(&config.CentralConfiguration{}, false)
+			if tc.setup != nil {
+				tc.setup(cm)
+			}
+
+			cm.FlushKind(tc.flushKind)
+
+			for _, e := range tc.expectEmpty {
+				result := cm.GetCachedResourcesByKind(e.group, e.kind, "")
+				assert.Empty(t, result, "expected %s cache to be empty after flush", e.kind)
+			}
+			for _, e := range tc.expectNonEmpty {
+				result := cm.GetCachedResourcesByKind(e.group, e.kind, "")
+				assert.NotEmpty(t, result, "expected %s cache to still have data", e.kind)
+			}
+		})
+	}
+}

pkg/agent/cache/manager.go

@@ -42,6 +42,7 @@ type Manager interface {
 	HasLoadedPersistedCache() bool
 	SaveCache()
 	Flush()
+	FlushKind(kind string)
 
 	// API Service cache related methods
 	AddAPIService(resource *v1.ResourceInstance) error

pkg/agent/cachevalidationjob.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	agentcache "github.com/Axway/agent-sdk/pkg/agent/cache"
@@ -41,17 +42,39 @@ func newCacheValidator(
 	}
 }
 
-func (cv *cacheValidator) Execute() error {
+// Execute validates each filter in the watch topic against the API server.
+// It returns the slice of filters whose cache is out of sync, plus errCacheOutOfSync
+// if any failed, or a nil slice and nil error if all filters are in sync.
+func (cv *cacheValidator) Execute() ([]management.WatchTopicSpecFilters, error) {
 	cv.logger.Debug("executing cache validation")
 
-	for _, filter := range cv.watchTopic.Spec.Filters {
-		if !cv.validateKind(filter) {
-			return errCacheOutOfSync
-		}
+	filters := cv.watchTopic.Spec.Filters
+	ch := make(chan management.WatchTopicSpecFilters, len(filters))
+
+	var wg sync.WaitGroup
+	for _, filter := range filters {
+		wg.Add(1)
+		go func(f management.WatchTopicSpecFilters) {
+			defer wg.Done()
+			if !cv.validateKind(f) {
+				ch <- f
+			}
+		}(filter)
+	}
+	wg.Wait()
+	close(ch)
+
+	var failed []management.WatchTopicSpecFilters
+	for f := range ch {
+		failed = append(failed, f)
+	}
+
+	if len(failed) > 0 {
+		return failed, errCacheOutOfSync
 	}
 
 	cv.logger.Debug("cache validation passed")
-	return nil
+	return nil, nil
 }
 
 func (cv *cacheValidator) validateKind(filter management.WatchTopicSpecFilters) bool {

pkg/agent/cachevalidationjob_test.go

@@ -10,6 +10,7 @@ import (
 	apiv1 "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/api/v1"
 	management "github.com/Axway/agent-sdk/pkg/apic/apiserver/models/management/v1alpha1"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // mockCacheGetter implements cacheGetter for tests
@@ -123,9 +124,10 @@ func TestCacheValidator_Execute(t *testing.T) {
 	}
 
 	type testCase struct {
-		setup      func(*mockResourceClient, *mockCacheGetter)
-		watchTopic *management.WatchTopic
-		expectErr  error
+		setup               func(*mockResourceClient, *mockCacheGetter)
+		watchTopic          *management.WatchTopic
+		expectErr           error
+		expectedFailedKinds []string
 	}
 
 	tests := map[string]testCase{
@@ -137,7 +139,8 @@ func TestCacheValidator_Execute(t *testing.T) {
 					},
 				},
 			},
-			expectErr: nil,
+			expectErr:           nil,
+			expectedFailedKinds: nil,
 		},
 		"in sync": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -146,8 +149,9 @@ func TestCacheValidator_Execute(t *testing.T) {
 					agentcache.ResourceCacheKey(svcGVK.Kind, scopeName, "svc1"): modTime,
 				})
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  nil,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           nil,
+			expectedFailedKinds: nil,
 		},
 		"count mismatch": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -156,8 +160,9 @@ func TestCacheValidator_Execute(t *testing.T) {
 					agentcache.ResourceCacheKey(svcGVK.Kind, scopeName, "svc1"): modTime,
 				})
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  errCacheOutOfSync,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{svcGVK.Kind},
 		},
 		"resource missing from cache": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -166,8 +171,9 @@ func TestCacheValidator_Execute(t *testing.T) {
 					agentcache.ResourceCacheKey(svcGVK.Kind, scopeName, "svc-other"): modTime,
 				})
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  errCacheOutOfSync,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{svcGVK.Kind},
 		},
 		"modify time mismatch": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -177,8 +183,9 @@ func TestCacheValidator_Execute(t *testing.T) {
 					agentcache.ResourceCacheKey(svcGVK.Kind, scopeName, "svc1"): modTime,
 				})
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  errCacheOutOfSync,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{svcGVK.Kind},
 		},
 		"zero timestamps are ignored": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -188,8 +195,9 @@ func TestCacheValidator_Execute(t *testing.T) {
 					agentcache.ResourceCacheKey(svcGVK.Kind, scopeName, "svc1"): time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC),
 				})
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  nil,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           nil,
+			expectedFailedKinds: nil,
 		},
 		"extra resource in cache": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -199,15 +207,17 @@ func TestCacheValidator_Execute(t *testing.T) {
 					agentcache.ResourceCacheKey(svcGVK.Kind, scopeName, "svc3"): modTime,
 				})
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  errCacheOutOfSync,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{svcGVK.Kind},
 		},
 		"fetch error": {
 			setup: func(client *mockResourceClient, _ *mockCacheGetter) {
 				client.err = fmt.Errorf("network error")
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  errCacheOutOfSync,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{svcGVK.Kind},
 		},
 		"APIServiceInstances from multiple scopes are each validated independently": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -228,7 +238,8 @@ func TestCacheValidator_Execute(t *testing.T) {
 					},
 				},
 			},
-			expectErr: nil,
+			expectErr:           nil,
+			expectedFailedKinds: nil,
 		},
 		"multiple kinds in sync": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -249,7 +260,8 @@ func TestCacheValidator_Execute(t *testing.T) {
 					},
 				},
 			},
-			expectErr: nil,
+			expectErr:           nil,
+			expectedFailedKinds: nil,
 		},
 		"second kind out of sync": {
 			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
@@ -268,7 +280,25 @@ func TestCacheValidator_Execute(t *testing.T) {
 					},
 				},
 			},
-			expectErr: errCacheOutOfSync,
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{instGVK.Kind},
+		},
+		"both kinds out of sync": {
+			setup: func(client *mockResourceClient, cacheMan *mockCacheGetter) {
+				client.resources[svc1.GetKindLink()] = []*apiv1.ResourceInstance{svc1}
+				client.resources[inst1.GetKindLink()] = []*apiv1.ResourceInstance{inst1}
+				// both empty in cache
+			},
+			watchTopic: &management.WatchTopic{
+				Spec: management.WatchTopicSpec{
+					Filters: []management.WatchTopicSpecFilters{
+						{Group: svcGVK.Group, Kind: svcGVK.Kind, Name: "*", Scope: &management.WatchTopicSpecScope{Kind: "Environment", Name: scopeName}},
+						{Group: instGVK.Group, Kind: instGVK.Kind, Name: "*", Scope: &management.WatchTopicSpecScope{Kind: "Environment", Name: scopeName}},
+					},
+				},
+			},
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{svcGVK.Kind, instGVK.Kind},
 		},
 		"empty server and cache": {
 			setup: func(client *mockResourceClient, _ *mockCacheGetter) {
@@ -285,15 +315,17 @@ func TestCacheValidator_Execute(t *testing.T) {
 				}
 				client.resources[ri.GetKindLink()] = []*apiv1.ResourceInstance{}
 			},
-			watchTopic: singleScopeWatchTopic(svcGVK, scopeName),
-			expectErr:  nil,
+			watchTopic:          singleScopeWatchTopic(svcGVK, scopeName),
+			expectErr:           nil,
+			expectedFailedKinds: nil,
 		},
 		"ComplianceRuntimeResult out of sync is detected": {
 			setup: func(client *mockResourceClient, _ *mockCacheGetter) {
 				client.resources[crr1.GetKindLink()] = []*apiv1.ResourceInstance{crr1}
 			},
-			watchTopic: singleScopeWatchTopic(crrGVK, scopeName),
-			expectErr:  errCacheOutOfSync,
+			watchTopic:          singleScopeWatchTopic(crrGVK, scopeName),
+			expectErr:           errCacheOutOfSync,
+			expectedFailedKinds: []string{crrGVK.Kind},
 		},
 	}
 
@@ -305,8 +337,19 @@ func TestCacheValidator_Execute(t *testing.T) {
 				tc.setup(client, cacheMan)
 			}
 			cv := newCacheValidator(client, tc.watchTopic, cacheMan)
-			err := cv.Execute()
+			failedFilters, err := cv.Execute()
 			assert.Equal(t, tc.expectErr, err)
+
+			if len(tc.expectedFailedKinds) == 0 {
+				assert.Empty(t, failedFilters)
+			} else {
+				require.Len(t, failedFilters, len(tc.expectedFailedKinds))
+				actualKinds := make([]string, len(failedFilters))
+				for i, f := range failedFilters {
+					actualKinds[i] = f.Kind
+				}
+				assert.ElementsMatch(t, tc.expectedFailedKinds, actualKinds)
+			}
 		})
 	}
 }