Merge pull request #13229 from fabriziopandini/fix-inmemory-watch

🌱 Fix inMemory watch

Author: k8s-ci-robot

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

@@ -48,6 +48,8 @@ type Cache interface {
 	Update(resourceGroup string, obj client.Object) error
 	Patch(resourceGroup string, obj client.Object, patch client.Patch) error
 
+	GetBookmarkResourceVersion(resourceGroup string) (string, error)
+
 	GetInformer(ctx context.Context, obj client.Object) (Informer, error)
 	GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (Informer, error)
 }

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

@@ -79,6 +79,22 @@ func (c *cache) Get(resourceGroup string, objKey client.ObjectKey, obj client.Ob
 	return nil
 }
 
+func (c *cache) GetBookmarkResourceVersion(resourceGroup string) (string, error) {
+	if resourceGroup == "" {
+		return "", apierrors.NewBadRequest("resourceGroup must not be empty")
+	}
+
+	tracker := c.resourceGroupTracker(resourceGroup)
+	if tracker == nil {
+		return "", apierrors.NewBadRequest(fmt.Sprintf("resourceGroup %s does not exist", resourceGroup))
+	}
+
+	tracker.lock.RLock()
+	defer tracker.lock.RUnlock()
+
+	return fmt.Sprintf("%d", tracker.lastResourceVersion), nil
+}
+
 func (c *cache) List(resourceGroup string, list client.ObjectList, opts ...client.ListOption) error {
 	if resourceGroup == "" {
 		return apierrors.NewBadRequest("resourceGroup must not be empty")

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

@@ -277,6 +277,49 @@ func Test_cache_client(t *testing.T) {
 		})
 	})
 
+	t.Run("Get bookmark resourceVersion", func(t *testing.T) {
+		c := NewCache(scheme).(*cache)
+		c.AddResourceGroup("foo")
+
+		t.Run("fails if resourceGroup is empty", func(t *testing.T) {
+			g := NewWithT(t)
+
+			_, err := c.GetBookmarkResourceVersion("")
+			g.Expect(err).To(HaveOccurred())
+			g.Expect(apierrors.IsBadRequest(err)).To(BeTrue())
+		})
+
+		t.Run("fails if resourceGroup doesn't exist", func(t *testing.T) {
+			g := NewWithT(t)
+
+			_, err := c.GetBookmarkResourceVersion("bar")
+			g.Expect(err).To(HaveOccurred())
+			g.Expect(apierrors.IsBadRequest(err)).To(BeTrue())
+		})
+
+		t.Run("get when no objects exists", func(t *testing.T) {
+			g := NewWithT(t)
+
+			v, err := c.GetBookmarkResourceVersion("foo")
+			g.Expect(err).ToNot(HaveOccurred())
+
+			// Check all the computed fields are as expected.
+			g.Expect(v).To(Equal("0"), "resourceVersion must be set")
+		})
+
+		t.Run("get when objects exists", func(t *testing.T) {
+			g := NewWithT(t)
+
+			createMachine(t, c, "foo", "bar")
+
+			v, err := c.GetBookmarkResourceVersion("foo")
+			g.Expect(err).ToNot(HaveOccurred())
+
+			// Check all the computed fields are as expected.
+			g.Expect(v).To(Equal("1"), "resourceVersion must be set")
+		})
+	})
+
 	t.Run("list objects", func(t *testing.T) {
 		c := NewCache(scheme).(*cache)
 		c.AddResourceGroup("foo")

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

@@ -25,6 +25,7 @@ import (
 
 	"github.com/emicklei/go-restful/v3"
 	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/watch"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -169,6 +170,20 @@ func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful.
 		initialEvents = append(initialEvents, Event{Type: eventType, Object: &obj})
 	}
 
+	// Add the bookmark event.
+	bookmarkResourceVersion, err := c.GetBookmarkResourceVersion(resourceGroup)
+	if err != nil {
+		return err
+	}
+
+	obj := &unstructured.Unstructured{}
+	obj.SetAPIVersion(gvk.GroupVersion().String())
+	obj.SetKind(gvk.Kind)
+	obj.SetResourceVersion(bookmarkResourceVersion)
+	obj.SetAnnotations(map[string]string{"k8s.io/initial-events-end": "true"})
+
+	initialEvents = append(initialEvents, Event{Type: watch.Bookmark, Object: obj})
+
 	return watcher.Run(ctx, queryTimeout, initialEvents, list.GetResourceVersion(), resp)
 }
 
@@ -184,6 +199,7 @@ func (m *WatchEventDispatcher) Run(ctx context.Context, timeout string, initialE
 		return errors.New("can't start Watch: can't get restful.Response")
 	}
 	w.Header().Set("Transfer-Encoding", "chunked")
+	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(http.StatusOK)
 
 	// Write all initial events.

test/infrastructure/inmemory/pkg/server/listener.go

@@ -28,8 +28,6 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 
 	"sigs.k8s.io/cluster-api/util/certs"
 )
@@ -119,28 +117,3 @@ func (s *WorkloadClusterListener) RESTConfig() (*rest.Config, error) {
 
 	return restConfig, nil
 }
-
-// GetClient returns a client for a WorkloadClusterListener.
-func (s *WorkloadClusterListener) GetClient() (client.WithWatch, error) {
-	restConfig, err := s.RESTConfig()
-	if err != nil {
-		return nil, err
-	}
-
-	httpClient, err := rest.HTTPClientFor(restConfig)
-	if err != nil {
-		return nil, err
-	}
-
-	mapper, err := apiutil.NewDynamicRESTMapper(restConfig, httpClient)
-	if err != nil {
-		return nil, err
-	}
-
-	c, err := client.NewWithWatch(restConfig, client.Options{Scheme: s.scheme, Mapper: mapper})
-	if err != nil {
-		return nil, err
-	}
-
-	return c, nil
-}