Race condition / data race in watch cache indexer mutation

Go PoC (demonstrates a classic race between concurrent mutation of an in-memory indexer and a read path that enumerates the indexer without proper synchronization). Notes: - This PoC uses a simplified UnsafeIndexer to mimic an indexer mutated by one goroutine and read by another without locks, illustrating the race condition the fix addresses. In Kubernetes, the vulnerable path would be similar to mutating the watch cache indexer while another goroutine enumerates or snapshots it. Code (save as poc.go and run with: go run -race poc.go): package main import ( "fmt" "sync" "time" ) type UnsafeIndexer struct { items []string } func (u *UnsafeIndexer) Add(v string) { // Intentionally no lock: simulates mutation outside a global lock u.items = append(u.items, v) } func (u *UnsafeIndexer) ListPrefix(prefix string) []string { out := []string{} for _, s := range u.items { if len(prefix) == 0 || (len(s) >= len(prefix) && s[:len(prefix)] == prefix) { out = append(out, s) } } return out } func main() { idx := &UnsafeIndexer{} var wg sync.WaitGroup wg.Add(2) // Writer: mutates the indexer rapidly without any lock go func() { defer wg.Done() for i := 0; i < 10000; i++ { idx.Add(fmt.Sprintf("obj-%d", i)) time.Sleep(1 * time.Microsecond) } }() // Reader: enumerates the indexer concurrently without synchronization go func() { defer wg.Done() for i := 0; i < 10000; i++ { _ = idx.ListPrefix("obj-") time.Sleep(1 * time.Microsecond) } }() wg.Wait() fmt.Println("done (race detector may report a race if run with -race)") } Expected outcome (with -race): the race detector should report concurrent write/read on the shared slice, illustrating the vulnerability that the fix guards against by introducing proper locking or snapshotting. In Kubernetes, the watch cache fix prevents such races during snapshot generation and reads."

diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go index 566166d69bdf2..8c70259e70da3 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go @@ -668,13 +668,42 @@ func (w *watchCache) waitAndGetLatestSnapshot(ctx context.Context, minResourceVe return listSnapshot{Items: result}, w.resourceVersion, matchValue.IndexName, nil } } - if w.storage.snapshots != nil { + return w.getLatestSnapshotLocked(key, continueKey) +} + +func (w *watchCache) getLatestSnapshotLocked(key, continueKey string) (store.Snapshot, uint64, string, error) { + if w.storage.snapshots != nil && w.storage.snapshottingEnabled.Load() { snap, ok := w.storage.snapshots.Latest() if ok { + // Snapshots are added in order as we update store, so the + // latest snapshot match latest store state and latest revision. return snap, w.resourceVersion, "", nil } } - return w.storage.store, w.resourceVersion, "", nil + // TODO: Consider using Indexer Clone() after benchmarking. + snap, err := orderedSnapshotResponseFromIndexer(w.storage.store, key, continueKey) + if err != nil { + return nil, 0, "", err + } + return snap, w.resourceVersion, "", nil +} + +func orderedSnapshotResponseFromIndexer(indexer store.Indexer, key, continueKey string) (store.Snapshot, error) { + items, err := indexer.OrderedListPrefix(key, continueKey) + if err != nil { + return nil, err + } + return orderedListSnapshot{Items: items}, nil +} + +type orderedListSnapshot struct { + Items []interface{} +} + +var _ store.Snapshot = (*orderedListSnapshot)(nil) + +func (o orderedListSnapshot) OrderedListPrefix(prefix, continueKey string) ([]interface{}, error) { + return o.Items, nil } type listSnapshot struct { diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_test.go index aadd7c6f1cb48..c7dae3dc52a92 100644 --- a/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_test.go +++ b/staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache_test.go @@ -133,6 +133,8 @@ func newTestWatchCache(capacity int, eventFreshDuration time.Duration, indexers pr := progress.NewConditionalProgressRequester(wc.RequestWatchProgress, &immediateTickerFactory{}, nil) go pr.Run(wc.stopCh) getCurrentRV := func(context.Context) (uint64, error) { + wc.RLock() + defer wc.RUnlock() return wc.resourceVersion, nil } wc.watchCache = newWatchCache(keyFunc, mockHandler, getAttrsFunc, versioner, indexers, testingclock.NewFakeClock(time.Now()), eventFreshDuration, schema.GroupResource{Resource: "pods"}, pr, getCurrentRV) @@ -1413,3 +1415,110 @@ func TestCacheSnapshots(t *testing.T) { assert.Len(t, elements, 1) assert.Equal(t, makeTestPod("foo", 600), elements[0].(*store.Element).Object) } + +func TestWatchCacheSnapshotConcurrency(t *testing.T) { + testCases := []struct { + name string + listFromSnapshot bool + resourceVersionMatch metav1.ResourceVersionMatch + expectItemRVLessOrEqualListRV bool + expectExactResourceVersion bool + }{ + { + name: "latest list with snapshotting disabled", + listFromSnapshot: false, + resourceVersionMatch: "", + expectItemRVLessOrEqualListRV: true, + }, + { + name: "latest list with snapshotting enabled", + listFromSnapshot: true, + resourceVersionMatch: "", + expectItemRVLessOrEqualListRV: true, + }, + { + name: "exact match with snapshotting enabled", + listFromSnapshot: true, + resourceVersionMatch: metav1.ResourceVersionMatchExact, + expectItemRVLessOrEqualListRV: true, + expectExactResourceVersion: true, + }, + } + + for i := range testCases { + tc := &testCases[i] + t.Run(tc.name, func(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ListFromCacheSnapshot, tc.listFromSnapshot) + + s := newTestWatchCache(50000, DefaultEventFreshDuration, &cache.Indexers{}) + defer s.Stop() + + require.NoError(t, s.Add(makeTestPod("initial", 1))) + + testWatchCacheSnapshotConcurrency(t, s, tc.resourceVersionMatch, tc.expectItemRVLessOrEqualListRV, tc.expectExactResourceVersion) + }) + } +} + +func testWatchCacheSnapshotConcurrency(t *testing.T, s *testWatchCache, resourceVersionMatch metav1.ResourceVersionMatch, expectItemRVLessOrEqualListRV, expectExactResourceVersion bool) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + stopUpdates := make(chan struct{}) + var wg wait.Group + wg.Start(func() { + var rv uint64 = 2 + for { + select { + case <-stopUpdates: + return + default: + err := s.Update(makeTestPod("pod-test", rv)) + if err != nil { + t.Errorf("failed to update: %v", err) + return + } + rv++ + } + } + }) + + for j := range 1000 { + opts := storage.ListOptions{ + Recursive: true, + ResourceVersionMatch: resourceVersionMatch, + } + var targetRV uint64 + if resourceVersionMatch == metav1.ResourceVersionMatchExact { + targetRV = uint64(j + 2) + opts.ResourceVersion = strconv.FormatUint(targetRV, 10) + } + + resp, _, err := s.WaitUntilFreshAndGetList(ctx, "/prefix/", opts) + require.NoError(t, err) + + if expectExactResourceVersion && resp.ResourceVersion != targetRV { + t.Errorf("Expected list ResourceVersion %d, got %d", targetRV, resp.ResourceVersion) + } + if expectItemRVLessOrEqualListRV { + maxItemRV := getMaxItemRV(t, s.versioner, resp.Items) + if maxItemRV > resp.ResourceVersion { + t.Errorf("Violated consistency: max item resource version %d is greater than list resource version %d", maxItemRV, resp.ResourceVersion) + } + } + } + + close(stopUpdates) + wg.Wait() +} + +func getMaxItemRV(t *testing.T, versioner storage.Versioner, items []interface{}) uint64 { + var maxRV uint64 + for _, item := range items { + elem := item.(*store.Element) + itemRV, err := versioner.ObjectResourceVersion(elem.Object) + require.NoError(t, err) + maxRV = max(maxRV, itemRV) + } + return maxRV +}