diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/generic/policy_test_context.go b/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/generic/policy_test_context.go index b98c10e7050..5a989028642 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/generic/policy_test_context.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/generic/policy_test_context.go @@ -164,7 +164,7 @@ func NewPolicyTestContext[P, B runtime.Object, E Evaluator]( return policiesAndBindingsTracker.List(fakePolicyGVR, fakePolicyGVK, "") }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { - return policiesAndBindingsTracker.Watch(fakePolicyGVR, "") + return policiesAndBindingsTracker.Watch(fakePolicyGVR, "", options) }, }, policiesAndBindingsTracker), Pexample, @@ -177,7 +177,7 @@ func NewPolicyTestContext[P, B runtime.Object, E Evaluator]( return policiesAndBindingsTracker.List(fakeBindingGVR, fakeBindingGVK, "") }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { - return policiesAndBindingsTracker.Watch(fakeBindingGVR, "") + return policiesAndBindingsTracker.Watch(fakeBindingGVR, "", options) }, }, policiesAndBindingsTracker), Bexample, diff --git a/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/internal/generic/controller_test.go b/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/internal/generic/controller_test.go index 9ccaffad25d..e3372c31f22 100644 --- a/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/internal/generic/controller_test.go +++ b/staging/src/k8s.io/apiserver/pkg/admission/plugin/policy/internal/generic/controller_test.go @@ -117,7 +117,7 @@ func setupTest(ctx context.Context, customReconciler func(string, string, runtim return tracker.List(fakeGVR, fakeGVK, "") }, WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { - return tracker.Watch(fakeGVR, "") + return tracker.Watch(fakeGVR, "", options) }, }, tracker), &unstructured.Unstructured{}, 30*time.Second, nil)} diff --git a/staging/src/k8s.io/client-go/dynamic/fake/simple.go b/staging/src/k8s.io/client-go/dynamic/fake/simple.go index 06ef859425f..39b3b646001 100644 --- a/staging/src/k8s.io/client-go/dynamic/fake/simple.go +++ b/staging/src/k8s.io/client-go/dynamic/fake/simple.go @@ -114,9 +114,13 @@ func NewSimpleDynamicClientWithCustomListKinds(scheme *runtime.Scheme, gvrToList cs := &FakeDynamicClient{scheme: scheme, gvrToListKind: completeGVRToListKind, tracker: o} cs.AddReactor("*", "*", testing.ObjectReaction(o)) cs.AddWatchReactor("*", func(action testing.Action) (handled bool, ret watch.Interface, err error) { + var opts metav1.ListOptions + if watchAction, ok := action.(testing.WatchActionImpl); ok { + opts = watchAction.ListOptions + } gvr := action.GetResource() ns := action.GetNamespace() - watch, err := o.Watch(gvr, ns) + watch, err := o.Watch(gvr, ns, opts) if err != nil { return false, nil, err } diff --git a/staging/src/k8s.io/client-go/dynamic/fake/simple_test.go b/staging/src/k8s.io/client-go/dynamic/fake/simple_test.go index 50f22c49b9e..50c6bd87e3b 100644 --- a/staging/src/k8s.io/client-go/dynamic/fake/simple_test.go +++ b/staging/src/k8s.io/client-go/dynamic/fake/simple_test.go @@ -186,7 +186,7 @@ func Test_ListKind(t *testing.T) { "kind": "TheKindList", "metadata": map[string]interface{}{ "continue": "", - "resourceVersion": "", + "resourceVersion": "3", // Three objects created so far. }, }, Items: []unstructured.Unstructured{ @@ -340,7 +340,7 @@ func TestListWithUnstructuredObjectsAndTypedScheme(t *testing.T) { expectedList := &unstructured.UnstructuredList{} expectedList.SetGroupVersionKind(listGVK) - expectedList.SetResourceVersion("") // by product of the fake setting resource version + expectedList.SetResourceVersion("1") // One object created so far. expectedList.SetContinue("") expectedList.Items = append(expectedList.Items, u) @@ -402,7 +402,7 @@ func TestListWithNoScheme(t *testing.T) { expectedList := &unstructured.UnstructuredList{} expectedList.SetGroupVersionKind(listGVK) - expectedList.SetResourceVersion("") // by product of the fake setting resource version + expectedList.SetResourceVersion("1") // One object created so far. expectedList.SetContinue("") expectedList.Items = append(expectedList.Items, u) @@ -443,7 +443,7 @@ func TestListWithTypedFixtures(t *testing.T) { expectedList := &unstructured.UnstructuredList{} expectedList.SetGroupVersionKind(listGVK) - expectedList.SetResourceVersion("") // by product of the fake setting resource version + expectedList.SetResourceVersion("1") // One object created so far. expectedList.SetContinue("") expectedList.Items = []unstructured.Unstructured{u} diff --git a/staging/src/k8s.io/client-go/examples/fake-client/main_test.go b/staging/src/k8s.io/client-go/examples/fake-client/main_test.go index a2f0eeef0df..548ac86bde6 100644 --- a/staging/src/k8s.io/client-go/examples/fake-client/main_test.go +++ b/staging/src/k8s.io/client-go/examples/fake-client/main_test.go @@ -41,9 +41,13 @@ func TestFakeClient(t *testing.T) { client := fake.NewSimpleClientset() // A catch-all watch reactor that allows us to inject the watcherStarted channel. client.PrependWatchReactor("*", func(action clienttesting.Action) (handled bool, ret watch.Interface, err error) { + var opts metav1.ListOptions + if watchAction, ok := action.(clienttesting.WatchActionImpl); ok { + opts = watchAction.ListOptions + } gvr := action.GetResource() ns := action.GetNamespace() - watch, err := client.Tracker().Watch(gvr, ns) + watch, err := client.Tracker().Watch(gvr, ns, opts) if err != nil { return false, nil, err } diff --git a/staging/src/k8s.io/client-go/metadata/fake/simple.go b/staging/src/k8s.io/client-go/metadata/fake/simple.go index fc50113fca1..e2d1f7aa2b2 100644 --- a/staging/src/k8s.io/client-go/metadata/fake/simple.go +++ b/staging/src/k8s.io/client-go/metadata/fake/simple.go @@ -68,9 +68,13 @@ func NewSimpleMetadataClient(scheme *runtime.Scheme, objects ...runtime.Object) cs := &FakeMetadataClient{scheme: scheme, tracker: o} cs.AddReactor("*", "*", testing.ObjectReaction(o)) cs.AddWatchReactor("*", func(action testing.Action) (handled bool, ret watch.Interface, err error) { + var opts metav1.ListOptions + if watchAction, ok := action.(testing.WatchActionImpl); ok { + opts = watchAction.ListOptions + } gvr := action.GetResource() ns := action.GetNamespace() - watch, err := o.Watch(gvr, ns) + watch, err := o.Watch(gvr, ns, opts) if err != nil { return false, nil, err } diff --git a/staging/src/k8s.io/client-go/testing/fixture.go b/staging/src/k8s.io/client-go/testing/fixture.go index 152a5c1bacb..7b8b614d3b2 100644 --- a/staging/src/k8s.io/client-go/testing/fixture.go +++ b/staging/src/k8s.io/client-go/testing/fixture.go @@ -20,6 +20,7 @@ import ( "fmt" "reflect" "sort" + "strconv" "strings" "sync" @@ -288,13 +289,41 @@ type tracker struct { scheme ObjectScheme decoder runtime.Decoder lock sync.RWMutex - objects map[schema.GroupVersionResource]map[types.NamespacedName]runtime.Object + objects map[schema.GroupVersionResource]map[types.NamespacedName]versionedObject // The value type of watchers is a map of which the key is either a namespace or // all/non namespace aka "" and its value is list of fake watchers. // Manipulations on resources will broadcast the notification events into the // watchers' channel. Note that too many unhandled events (currently 100, // see apimachinery/pkg/watch.DefaultChanSize) will cause a panic. watchers map[schema.GroupVersionResource]map[string][]*watch.RaceFreeFakeWatcher + // resourceVersions is the highest resource version of any tracked object with + // a certain gvr. The resource version for that set of objects gets bumped before + // storing a new or modified object, so all entries are larger than 0. + // Object content does not get changed to preserve the traditional behavior + // (hence also the versionedObject type instead of storing a runtime.Object + // with modified ResourceVersion). + // + // Resource version support (https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions) + // is very limited. It only supports one particular use case: + // List (no resource version check, returned ListMeta has ResourceVersion set) + + // Watch (Exact match for the ResourceVersion returned by List). + // + // This is sufficient for Reflector.ListAndWatch (https://github.com/kubernetes/kubernetes/blob/b53b9fb5573323484af9a19cf3f5bfe80760abba/staging/src/k8s.io/client-go/tools/cache/reflector.go#L401) + // when setting up informers in an informer factory. + // + // Strictly speaking, this should be by GroupVersion. But objects are + // also tracked by GroupVersionResource instead of GroupVersion, so the + // same is done here to match how List is implemented. + resourceVersions map[schema.GroupVersionResource]int64 +} + +// versionedObject stores an object together with the resource version that was +// assigned to it by the tracker. The version could be stored inline in the object, +// but this is not how fake client-go has traditionally worked and starting to do +// that now might break tests. +type versionedObject struct { + resourceVersion int64 + runtime.Object } var _ ObjectTracker = &tracker{} @@ -303,10 +332,11 @@ var _ ObjectTracker = &tracker{} // of objects for the fake clientset. Mostly useful for unit tests. func NewObjectTracker(scheme ObjectScheme, decoder runtime.Decoder) ObjectTracker { return &tracker{ - scheme: scheme, - decoder: decoder, - objects: make(map[schema.GroupVersionResource]map[types.NamespacedName]runtime.Object), - watchers: make(map[schema.GroupVersionResource]map[string][]*watch.RaceFreeFakeWatcher), + scheme: scheme, + decoder: decoder, + objects: make(map[schema.GroupVersionResource]map[types.NamespacedName]versionedObject), + watchers: make(map[schema.GroupVersionResource]map[string][]*watch.RaceFreeFakeWatcher), + resourceVersions: make(map[schema.GroupVersionResource]int64), } } @@ -343,13 +373,20 @@ func (t *tracker) List(gvr schema.GroupVersionResource, gvk schema.GroupVersionK return list, nil } - matchingObjs, err := filterByNamespace(objs, ns) + matchingVersionedObjs, err := filterByNamespace(objs, ns) if err != nil { return nil, err } + matchingObjs := make([]runtime.Object, len(matchingVersionedObjs)) + for i, obj := range matchingVersionedObjs { + matchingObjs[i] = obj.Object + } if err := meta.SetList(list, matchingObjs); err != nil { return nil, err } + if listMeta, err := meta.ListAccessor(list); err == nil { + listMeta.SetResourceVersion(fmt.Sprintf("%d", t.resourceVersions[gvr])) + } return list.DeepCopyObject(), nil } @@ -358,6 +395,17 @@ func (t *tracker) Watch(gvr schema.GroupVersionResource, ns string, opts ...meta if err != nil { return nil, err } + // By default, emulate the traditional behavior of the tracker and don't deliver + // *any* existing objects. This ensures compatibility which users + // which don't pass options and then also don't expect the additional objects. + resourceVersion := int64(-1) + if len(opts) > 0 && opts[0].ResourceVersion != "" { + rv, err := strconv.ParseInt(opts[0].ResourceVersion, 10, 64) + if err != nil { + return nil, fmt.Errorf("invalid ResourceVersion %q in ListOptions, must be int64: %w", opts[0].ResourceVersion, err) + } + resourceVersion = rv + } t.lock.Lock() defer t.lock.Unlock() @@ -368,6 +416,22 @@ func (t *tracker) Watch(gvr schema.GroupVersionResource, ns string, opts ...meta t.watchers[gvr] = make(map[string][]*watch.RaceFreeFakeWatcher) } t.watchers[gvr][ns] = append(t.watchers[gvr][ns], fakewatcher) + + // Deliver all objects that match the list options, for example + // between the initial List and the following Watch. + if resourceVersion != -1 { + objs := t.objects[gvr] + matchingObjs, err := filterByNamespace(objs, ns) + if err != nil { + return nil, err + } + for _, obj := range matchingObjs { + if resourceVersion == 0 || resourceVersion < obj.resourceVersion { + fakewatcher.Add(obj.Object) + } + } + } + return fakewatcher, nil } @@ -565,17 +629,19 @@ func (t *tracker) add(gvr schema.GroupVersionResource, obj runtime.Object, ns st _, ok := t.objects[gvr] if !ok { - t.objects[gvr] = make(map[types.NamespacedName]runtime.Object) + t.objects[gvr] = make(map[types.NamespacedName]versionedObject) } namespacedName := types.NamespacedName{Namespace: newMeta.GetNamespace(), Name: newMeta.GetName()} if _, ok = t.objects[gvr][namespacedName]; ok { if replaceExisting { + resourceVersion := t.resourceVersions[gvr] + 1 + t.resourceVersions[gvr] = resourceVersion for _, w := range t.getWatches(gvr, ns) { // To avoid the object from being accidentally modified by watcher w.Modify(obj.DeepCopyObject()) } - t.objects[gvr][namespacedName] = obj + t.objects[gvr][namespacedName] = versionedObject{resourceVersion, obj} return nil } return apierrors.NewAlreadyExists(gr, newMeta.GetName()) @@ -586,7 +652,9 @@ func (t *tracker) add(gvr schema.GroupVersionResource, obj runtime.Object, ns st return apierrors.NewNotFound(gr, newMeta.GetName()) } - t.objects[gvr][namespacedName] = obj + resourceVersion := t.resourceVersions[gvr] + 1 + t.resourceVersions[gvr] = resourceVersion + t.objects[gvr][namespacedName] = versionedObject{resourceVersion, obj} for _, w := range t.getWatches(gvr, ns) { // To avoid the object from being accidentally modified by watcher @@ -841,11 +909,11 @@ func (d *objectDefaulter) Default(_ runtime.Object) {} // filterByNamespace returns all objects in the collection that // match provided namespace. Empty namespace matches // non-namespaced objects. -func filterByNamespace(objs map[types.NamespacedName]runtime.Object, ns string) ([]runtime.Object, error) { - var res []runtime.Object +func filterByNamespace(objs map[types.NamespacedName]versionedObject, ns string) ([]versionedObject, error) { + var res []versionedObject for _, obj := range objs { - acc, err := meta.Accessor(obj) + acc, err := meta.Accessor(obj.Object) if err != nil { return nil, err } @@ -857,8 +925,8 @@ func filterByNamespace(objs map[types.NamespacedName]runtime.Object, ns string) // Sort res to get deterministic order. sort.Slice(res, func(i, j int) bool { - acc1, _ := meta.Accessor(res[i]) - acc2, _ := meta.Accessor(res[j]) + acc1, _ := meta.Accessor(res[i].Object) + acc2, _ := meta.Accessor(res[j].Object) if acc1.GetNamespace() != acc2.GetNamespace() { return acc1.GetNamespace() < acc2.GetNamespace() } diff --git a/staging/src/k8s.io/client-go/testing/internal/listandwatch_test.go b/staging/src/k8s.io/client-go/testing/internal/listandwatch_test.go new file mode 100644 index 00000000000..e1996490dfe --- /dev/null +++ b/staging/src/k8s.io/client-go/testing/internal/listandwatch_test.go @@ -0,0 +1,117 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package internal + +import ( + "context" + "testing" + "testing/synctest" + "time" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/tools/cache" + "k8s.io/klog/v2/ktesting" + _ "k8s.io/klog/v2/ktesting/init" // for -testing.v +) + +// TestListAndWatch mirrors how fake client-go is often used with real +// informers. It enforces a timing such that List completes, a new +// object gets created because of the completed cache sync, and only +// then is the Watch call in the reflector's "ListAndWatch" allowed to +// continue. +// +// The fake Watch implementation then must use the ResourceVersion to +// detect that it must send some (but not all!) objects to the new watch. +// +// This runs in a synctest bubble, therefore time is virtual. +func TestListAndWatch(t *testing.T) { synctest.Test(t, testListAndWatch) } +func testListAndWatch(t *testing.T) { + logger, ctx := ktesting.NewTestContext(t) + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cm1", + Namespace: "default", + }, + } + client := fake.NewClientset(cm) + stopCh := make(chan struct{}) + defer close(stopCh) + createDone := make(chan struct{}) + + f := informers.NewSharedInformerFactory(client, 0) + configMapInformer := f.InformerFor(&v1.ConfigMap{}, func(client kubernetes.Interface, defaultEventHandlerResyncPeriod time.Duration) cache.SharedIndexInformer { + + return cache.NewSharedIndexInformer(cache.ToListWatcherWithWatchListSemantics(&cache.ListWatch{ + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { + objs, err := client.CoreV1().ConfigMaps("").List(context.Background(), options) + logger.Info("Listed", "configMaps", objs, "err", err) + if err != nil { + t.Errorf("Unexpected List error: %v", err) + } else if objs.ResourceVersion != "1" { + t.Errorf("Expected ListMeta ResourceVersion 1, got %q", objs.ResourceVersion) + } + return objs, err + }, + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { + if options.ResourceVersion != "1" { + t.Errorf("Expected ListOptions ResourceVersion 1, got %q", options.ResourceVersion) + } + logger.Info("Delaying Watch...") + <-createDone + logger.Info("Continuing Watch...") + return client.CoreV1().ConfigMaps("").Watch(context.Background(), options) + }, + }, client), &v1.ConfigMap{}, defaultEventHandlerResyncPeriod, nil) + }) + + configMapStore := configMapInformer.GetStore() + f.Start(stopCh) + f.WaitForCacheSync(stopCh) + logger.Info("Caches synced") + + objs := configMapStore.List() + if len(objs) != 1 { + t.Fatalf("Unexpected item(s) in informer cache, want 1, got %d = %v", len(objs), objs) + } + + cm = &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cm2", + Namespace: "default", + }, + } + _, err := client.CoreV1().ConfigMaps(cm.Namespace).Create(ctx, cm, metav1.CreateOptions{}) + if err != nil { + t.Fatalf("Unexpected error creating ConfigMap: %v", err) + } + logger.Info("Created second ConfigMap") + close(createDone) + + // Wait for watch setup and event processing. + synctest.Wait() + + objs = configMapStore.List() + if len(objs) != 2 { + t.Fatalf("Unexpected item(s) in informer cache, want 2, got %d = %v", len(objs), objs) + } +} diff --git a/staging/src/k8s.io/kubectl/pkg/polymorphichelpers/helpers_test.go b/staging/src/k8s.io/kubectl/pkg/polymorphichelpers/helpers_test.go index 76628cf6312..1ed27015e7b 100644 --- a/staging/src/k8s.io/kubectl/pkg/polymorphichelpers/helpers_test.go +++ b/staging/src/k8s.io/kubectl/pkg/polymorphichelpers/helpers_test.go @@ -50,6 +50,9 @@ func TestGetPodList(t *testing.T) { podList: newPodList(2, -1, -1, labelSet), sortBy: func(pods []*corev1.Pod) sort.Interface { return podutils.ByLogging(pods) }, expected: &corev1.PodList{ + ListMeta: metav1.ListMeta{ + ResourceVersion: "2", // Two objects created. + }, Items: []corev1.Pod{ { ObjectMeta: metav1.ObjectMeta{