diff --git a/meshsync/discovery.go b/meshsync/discovery.go index 853fea6b..213dd7be 100644 --- a/meshsync/discovery.go +++ b/meshsync/discovery.go @@ -45,5 +45,28 @@ func (h *Handler) startDiscovery(pipelineCh chan struct{}) { return } + h.replaceStores(data) +} + +// replaceStores swaps in the per-GVR informer store set produced by a discovery +// run. stores is read concurrently by handleInformerStoreRequest, so the swap is +// guarded by storesMu. +func (h *Handler) replaceStores(data map[string]cache.Store) { + h.storesMu.Lock() + defer h.storesMu.Unlock() h.stores = data } + +// snapshotStores returns the current per-GVR informer stores. storesMu is held +// only while copying the map values, never across the caller's subsequent +// store reads, so a store List() never blocks the discovery goroutine's next +// replaceStores. +func (h *Handler) snapshotStores() []cache.Store { + h.storesMu.RLock() + defer h.storesMu.RUnlock() + stores := make([]cache.Store, 0, len(h.stores)) + for _, s := range h.stores { + stores = append(stores, s) + } + return stores +} diff --git a/meshsync/handlers.go b/meshsync/handlers.go index 6a4781cf..b042d953 100644 --- a/meshsync/handlers.go +++ b/meshsync/handlers.go @@ -293,7 +293,7 @@ func (h *Handler) publishMeshSyncMeta() { func (h *Handler) listStoreObjects() []model.KubernetesResource { objects := make([]interface{}, 0) - for _, v := range h.stores { + for _, v := range h.snapshotStores() { objects = append(objects, v.List()...) } parsedObjects := make([]model.KubernetesResource, 0) diff --git a/meshsync/meshsync.go b/meshsync/meshsync.go index ed63accd..82fa4b22 100644 --- a/meshsync/meshsync.go +++ b/meshsync/meshsync.go @@ -30,10 +30,15 @@ type Handler struct { // channelPool holds the fixed system channels (Stop/OS/ReSync) and is // read-only after construction. Dynamic exec/log-stream sessions live in // sessions (guarded by sessionsMu), not here, so the two never race. - channelPool map[string]channels.GenericChannel - sessions map[string]channels.StructChannel - sessionsMu sync.Mutex + channelPool map[string]channels.GenericChannel + sessions map[string]channels.StructChannel + sessionsMu sync.Mutex + // stores holds the per-GVR informer store set from the latest discovery. + // startDiscovery replaces it wholesale on the discovery goroutine on every + // (re)discovery while handleInformerStoreRequest reads it on the request + // listener goroutine; storesMu guards that concurrent field access. stores map[string]cache.Store + storesMu sync.RWMutex outputWriter output.Writer outputFiltration internalconfig.OutputFiltrationContainer } diff --git a/meshsync/stores_test.go b/meshsync/stores_test.go new file mode 100644 index 00000000..38362398 --- /dev/null +++ b/meshsync/stores_test.go @@ -0,0 +1,67 @@ +package meshsync + +import ( + "fmt" + "sync" + "testing" + + "k8s.io/client-go/tools/cache" +) + +func newStoresHandler() *Handler { + return &Handler{stores: make(map[string]cache.Store)} +} + +// TestStoresConcurrentAccess must pass under -race: the discovery goroutine +// replaces the stores map wholesale on every (re)discovery while the request +// listener goroutine snapshots it to answer informer-store requests. Without +// storesMu that is an unsynchronized read/write of the map field. +func TestStoresConcurrentAccess(t *testing.T) { + h := newStoresHandler() + + const workers = 16 + const iterations = 500 + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < iterations; i++ { + if w%2 == 0 { + h.replaceStores(map[string]cache.Store{ + fmt.Sprintf("gvr-%d", w): cache.NewStore(cache.MetaNamespaceKeyFunc), + }) + } else { + for _, s := range h.snapshotStores() { + _ = s.List() + } + } + } + }(w) + } + wg.Wait() +} + +// TestSnapshotStoresReturnsAllStores verifies snapshotStores returns every +// current store so informer-store replies stay complete after the guard. +func TestSnapshotStoresReturnsAllStores(t *testing.T) { + h := newStoresHandler() + h.replaceStores(map[string]cache.Store{ + "a": cache.NewStore(cache.MetaNamespaceKeyFunc), + "b": cache.NewStore(cache.MetaNamespaceKeyFunc), + "c": cache.NewStore(cache.MetaNamespaceKeyFunc), + }) + if got := len(h.snapshotStores()); got != 3 { + t.Fatalf("snapshotStores returned %d stores, want 3", got) + } +} + +// TestSnapshotStoresNilMap confirms a Handler whose stores map was never +// initialized (nil, as before the first discovery) snapshots to empty rather +// than panicking. +func TestSnapshotStoresNilMap(t *testing.T) { + h := &Handler{} + if got := len(h.snapshotStores()); got != 0 { + t.Fatalf("snapshotStores on a nil map returned %d, want 0", got) + } +}