🌱 inmemory: fix watch to continue serving based on resourceVersion parameter

test/infrastructure/inmemory/pkg/runtime/cache/cache.go

@@ -89,7 +89,8 @@ type resourceGroupTracker struct {
 	lock    sync.RWMutex
 	objects map[schema.GroupVersionKind]map[types.NamespacedName]client.Object
 	// ownedObjects tracks ownership. Key is the owner, values are the owned objects.
-	ownedObjects map[ownReference]map[ownReference]struct{}
+	ownedObjects        map[ownReference]map[ownReference]struct{}
+	nextResourceVersion uint64
 }
 
 type ownReference struct {
@@ -159,8 +160,9 @@ func (c *cache) AddResourceGroup(name string) {
 		return
 	}
 	c.resourceGroups[name] = &resourceGroupTracker{
-		objects:      map[schema.GroupVersionKind]map[types.NamespacedName]client.Object{},
-		ownedObjects: map[ownReference]map[ownReference]struct{}{},
+		objects:             map[schema.GroupVersionKind]map[types.NamespacedName]client.Object{},
+		ownedObjects:        map[ownReference]map[ownReference]struct{}{},
+		nextResourceVersion: 1,
 	}
 }
 

test/infrastructure/inmemory/pkg/runtime/cache/client.go

@@ -212,7 +212,7 @@ func (c *cache) store(resourceGroup string, obj client.Object, replaceExisting b
 				return apierrors.NewConflict(unsafeGuessGroupVersionResource(objGVK).GroupResource(), objKey.String(), fmt.Errorf("object has been modified"))
 			}
 
-			c.beforeUpdate(resourceGroup, trackedObj, obj)
+			c.beforeUpdate(resourceGroup, trackedObj, obj, &tracker.nextResourceVersion)
 
 			tracker.objects[objGVK][objKey] = obj.DeepCopyObject().(client.Object)
 			updateTrackerOwnerReferences(tracker, trackedObj, obj, objRef)
@@ -226,7 +226,7 @@ func (c *cache) store(resourceGroup string, obj client.Object, replaceExisting b
 		return apierrors.NewNotFound(unsafeGuessGroupVersionResource(objGVK).GroupResource(), objKey.String())
 	}
 
-	c.beforeCreate(resourceGroup, obj)
+	c.beforeCreate(resourceGroup, obj, &tracker.nextResourceVersion)
 
 	tracker.objects[objGVK][objKey] = obj.DeepCopyObject().(client.Object)
 	updateTrackerOwnerReferences(tracker, nil, obj, objRef)
@@ -422,7 +422,7 @@ func (c *cache) doTryDeleteLocked(resourceGroup string, tracker *resourceGroupTr
 		oldObj := obj.DeepCopyObject().(client.Object)
 		now := metav1.Time{Time: time.Now().UTC()}
 		obj.SetDeletionTimestamp(&now)
-		c.beforeUpdate(resourceGroup, oldObj, obj)
+		c.beforeUpdate(resourceGroup, oldObj, obj, &tracker.nextResourceVersion)
 
 		objects[objKey] = obj
 		c.afterUpdate(resourceGroup, oldObj, obj)

test/infrastructure/inmemory/pkg/runtime/cache/client_test.go

@@ -104,7 +104,7 @@ func Test_cache_client(t *testing.T) {
 			r := c.resourceGroups["foo"].objects[cloudv1.GroupVersion.WithKind(cloudv1.CloudMachineKind)][key]
 			g.Expect(r.GetObjectKind().GroupVersionKind()).To(BeComparableTo(cloudv1.GroupVersion.WithKind(cloudv1.CloudMachineKind)), "gvk must be set")
 			g.Expect(r.GetName()).To(Equal("bar"), "name must be equal to object tracker key")
-			g.Expect(r.GetResourceVersion()).To(Equal("v1"), "resourceVersion must be set")
+			g.Expect(r.GetResourceVersion()).To(Equal("0"), "resourceVersion must be set")
 			g.Expect(r.GetCreationTimestamp()).ToNot(BeZero(), "creation timestamp must be set")
 			g.Expect(r.GetAnnotations()).To(HaveKey(lastSyncTimeAnnotation), "last sync annotation must exists")
 
@@ -271,7 +271,7 @@ func Test_cache_client(t *testing.T) {
 			// Check all the computed fields are as expected.
 			g.Expect(obj.GetObjectKind().GroupVersionKind()).To(BeComparableTo(cloudv1.GroupVersion.WithKind(cloudv1.CloudMachineKind)), "gvk must be set")
 			g.Expect(obj.GetName()).To(Equal("bar"), "name must be equal to object tracker key")
-			g.Expect(obj.GetResourceVersion()).To(Equal("v1"), "resourceVersion must be set")
+			g.Expect(obj.GetResourceVersion()).To(Equal("1"), "resourceVersion must be set")
 			g.Expect(obj.GetCreationTimestamp()).ToNot(BeZero(), "creation timestamp must be set")
 			g.Expect(obj.GetAnnotations()).To(HaveKey(lastSyncTimeAnnotation), "last sync annotation must be set")
 		})
@@ -435,6 +435,8 @@ func Test_cache_client(t *testing.T) {
 			g.Expect(objBefore.GetResourceVersion()).ToNot(Equal(objUpdate.GetResourceVersion()), "Object version must be changed")
 			objBefore.SetResourceVersion(objUpdate.GetResourceVersion())
 			objBefore.Labels = objUpdate.Labels
+			g.Expect(objUpdate.GetGeneration()).To(Equal(objBefore.GetGeneration()+1), "Object Generation must increment")
+			objBefore.Generation = objUpdate.GetGeneration()
 			g.Expect(objBefore).To(BeComparableTo(objUpdate), "everything else must be the same")
 
 			g.Expect(h.Events()).To(ContainElement("foo, CloudMachine=baz, Updated"))
@@ -670,6 +672,8 @@ func Test_cache_client(t *testing.T) {
 			g.Expect(objBefore.GetResourceVersion()).ToNot(Equal(objAfterUpdate.GetResourceVersion()), "Object version must be changed")
 			objBefore.SetResourceVersion(objAfterUpdate.GetResourceVersion())
 			objBefore.Labels = objAfterUpdate.Labels
+			g.Expect(objAfterUpdate.GetGeneration()).To(Equal(objBefore.GetGeneration()+1), "Object Generation must increment")
+			objBefore.Generation = objAfterUpdate.GetGeneration()
 			g.Expect(objBefore).To(BeComparableTo(objAfterUpdate), "everything else must be the same")
 
 			g.Expect(h.Events()).To(ContainElement("foo, CloudMachine=baz, Deleted"))

test/infrastructure/inmemory/pkg/runtime/cache/hooks.go

@@ -19,29 +19,30 @@ package cache
 import (
 	"fmt"
 	"reflect"
-	"strconv"
-	"strings"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func (c *cache) beforeCreate(_ string, obj client.Object) {
+func (c *cache) beforeCreate(_ string, obj client.Object, resourceVersion *uint64) {
 	now := time.Now().UTC()
 	obj.SetCreationTimestamp(metav1.Time{Time: now})
 	// TODO: UID
 	obj.SetAnnotations(appendAnnotations(obj, lastSyncTimeAnnotation, now.Format(time.RFC3339)))
-	obj.SetResourceVersion(fmt.Sprintf("v%d", 1))
+	obj.SetResourceVersion(fmt.Sprintf("%d", *resourceVersion))
+	obj.SetGeneration(1)
+	*resourceVersion++
 }
 
 func (c *cache) afterCreate(resourceGroup string, obj client.Object) {
 	c.informCreate(resourceGroup, obj)
 }
 
-func (c *cache) beforeUpdate(_ string, oldObj, newObj client.Object) {
+func (c *cache) beforeUpdate(_ string, oldObj, newObj client.Object, resourceVersion *uint64) {
 	newObj.SetCreationTimestamp(oldObj.GetCreationTimestamp())
 	newObj.SetResourceVersion(oldObj.GetResourceVersion())
+	newObj.SetGeneration(oldObj.GetGeneration())
 	// TODO: UID
 	newObj.SetAnnotations(appendAnnotations(newObj, lastSyncTimeAnnotation, oldObj.GetAnnotations()[lastSyncTimeAnnotation]))
 	if !oldObj.GetDeletionTimestamp().IsZero() {
@@ -51,8 +52,9 @@ func (c *cache) beforeUpdate(_ string, oldObj, newObj client.Object) {
 		now := time.Now().UTC()
 		newObj.SetAnnotations(appendAnnotations(newObj, lastSyncTimeAnnotation, now.Format(time.RFC3339)))
 
-		oldResourceVersion, _ := strconv.Atoi(strings.TrimPrefix(oldObj.GetResourceVersion(), "v"))
-		newObj.SetResourceVersion(fmt.Sprintf("v%d", oldResourceVersion+1))
+		newObj.SetResourceVersion(fmt.Sprintf("%d", *resourceVersion))
+		newObj.SetGeneration(oldObj.GetGeneration() + 1)
+		*resourceVersion++
 	}
 }
 

test/infrastructure/inmemory/pkg/server/api/handler.go

@@ -50,6 +50,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	inmemoryruntime "sigs.k8s.io/cluster-api/test/infrastructure/inmemory/pkg/runtime"
+	inmemoryclient "sigs.k8s.io/cluster-api/test/infrastructure/inmemory/pkg/runtime/client"
 	inmemoryportforward "sigs.k8s.io/cluster-api/test/infrastructure/inmemory/pkg/server/api/portforward"
 )
 
@@ -315,6 +316,25 @@ func (h *apiServerHandler) apiV1List(req *restful.Request, resp *restful.Respons
 		return
 	}
 
+	h.log.V(3).Info(fmt.Sprintf("Serving List for %v", req.Request.URL), "resourceGroup", resourceGroup)
+
+	list, err := h.apiV1list(ctx, req, *gvk, inmemoryClient)
+	if err != nil {
+		if status, ok := err.(apierrors.APIStatus); ok || errors.As(err, &status) {
+			_ = resp.WriteHeaderAndEntity(int(status.Status().Code), status)
+			return
+		}
+		_ = resp.WriteErrorString(http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	if err := resp.WriteEntity(list); err != nil {
+		_ = resp.WriteErrorString(http.StatusInternalServerError, err.Error())
+		return
+	}
+}
+
+func (h *apiServerHandler) apiV1list(ctx context.Context, req *restful.Request, gvk schema.GroupVersionKind, inmemoryClient inmemoryclient.Client) (*unstructured.UnstructuredList, error) {
 	// Reads and returns the requested data.
 	list := &unstructured.UnstructuredList{}
 	list.SetAPIVersion(gvk.GroupVersion().String())
@@ -328,33 +348,23 @@ func (h *apiServerHandler) apiV1List(req *restful.Request, resp *restful.Respons
 	// TODO: The only field Selector which works is for `spec.nodeName` on pods.
 	fieldSelector, err := fields.ParseSelector(req.QueryParameter("fieldSelector"))
 	if err != nil {
-		_ = resp.WriteErrorString(http.StatusInternalServerError, err.Error())
-		return
+		return nil, err
 	}
 	if fieldSelector != nil {
 		listOpts = append(listOpts, client.MatchingFieldsSelector{Selector: fieldSelector})
 	}
 
 	labelSelector, err := labels.Parse(req.QueryParameter("labelSelector"))
 	if err != nil {
-		_ = resp.WriteErrorString(http.StatusInternalServerError, err.Error())
-		return
+		return nil, err
 	}
 	if labelSelector != nil {
 		listOpts = append(listOpts, client.MatchingLabelsSelector{Selector: labelSelector})
 	}
 	if err := inmemoryClient.List(ctx, list, listOpts...); err != nil {
-		if status, ok := err.(apierrors.APIStatus); ok || errors.As(err, &status) {
-			_ = resp.WriteHeaderAndEntity(int(status.Status().Code), status)
-			return
-		}
-		_ = resp.WriteHeaderAndEntity(http.StatusInternalServerError, err.Error())
-		return
-	}
-	if err := resp.WriteEntity(list); err != nil {
-		_ = resp.WriteErrorString(http.StatusInternalServerError, err.Error())
-		return
+		return nil, err
 	}
+	return list, nil
 }
 
 func (h *apiServerHandler) apiV1Watch(req *restful.Request, resp *restful.Response) {
@@ -372,6 +382,8 @@ func (h *apiServerHandler) apiV1Watch(req *restful.Request, resp *restful.Respon
 		return
 	}
 
+	h.log.V(3).Info(fmt.Sprintf("Serving Watch for %v", req.Request.URL), "resourceGroup", resourceGroup)
+
 	// If the request is a Watch handle it using watchForResource.
 	err = h.watchForResource(req, resp, resourceGroup, *gvk)
 	if err != nil {