DO_NOT_MERGE: enable WatchListClient FG

hack/check-everything.sh

@@ -41,6 +41,42 @@ tmp_bin=/tmp/cr-tests-bin
 )
 export KUBEBUILDER_ASSETS="$(${tmp_bin}/setup-envtest use --use-env -p path "${ENVTEST_K8S_VERSION}")"
 
+# HACK
+k8s_clone_dir=$tmp_root/kubernetes
+(
+  k8s_repo_url=https://github.com/kubernetes/kubernetes.git
+
+  echo "Cloning Kube repository from $k8s_repo_url..."
+  git clone $k8s_repo_url $k8s_clone_dir
+
+  cd $k8s_clone_dir
+
+  pr_number="132704"
+  echo "Fetching pull request #$pr_number..."
+  git fetch origin pull/$pr_number/head:pr-$pr_number
+
+  echo "Checking out pull request #$pr_number..."
+  git checkout pr-$pr_number
+
+  echo "Building Kube from source code..."
+  make
+)
+k8s_bin_dir=$(
+  k8s_output_dir=${k8s_clone_dir}/_output/local/go/bin
+  if [ -d "${k8s_output_dir}" ]; then
+    cd ${k8s_output_dir}
+    pwd
+  else
+    echo "Directory ${k8s_output_dir} does not exist."
+    exit 1
+  fi
+)
+echo "Replacing kube-apiserver binary from ${k8s_bin_dir} to ${KUBEBUILDER_ASSETS}"
+cp -f "${k8s_bin_dir}/kube-apiserver" "${KUBEBUILDER_ASSETS}/kube-apiserver"
+
+echo "Enabling WatchListClient feature"
+export KUBE_FEATURE_WatchListClient=true
+
 # Run tests.
 ${hack_dir}/test-all.sh
 

pkg/client/watch_list_test.go

@@ -0,0 +1,78 @@
+package client_test
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	clientfeatures "k8s.io/client-go/features"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/utils/ptr"
+)
+
+var _ = Describe("WatchList", func() {
+	It("should work against the kube-apiserver", func() {
+
+		Expect(clientfeatures.FeatureGates().Enabled(clientfeatures.WatchListClient)).To(BeTrue())
+
+		testenv = &envtest.Environment{}
+		cfg, err := testenv.Start()
+		Expect(err).NotTo(HaveOccurred())
+		clientset, err = kubernetes.NewForConfig(cfg)
+		Expect(err).NotTo(HaveOccurred())
+
+		ctx := context.Background()
+		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+		defer cancel()
+		opts := metav1.ListOptions{}
+		opts.AllowWatchBookmarks = true
+		opts.SendInitialEvents = ptr.To(true)
+		opts.ResourceVersionMatch = metav1.ResourceVersionMatchNotOlderThan
+		w, err := clientset.CoreV1().Secrets("kube-system").Watch(ctx, opts)
+		Expect(err).NotTo(HaveOccurred())
+		defer w.Stop()
+
+		receivedWatchListStreamFromTheServer := false
+		func() {
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case event, ok := <-w.ResultChan():
+					if !ok {
+						panic("unexpected watch close")
+
+					}
+					if event.Type == watch.Error {
+						panic(fmt.Sprintf("unexpected watch event: %v", apierrors.FromObject(event.Object)))
+					}
+					meta, err := meta.Accessor(event.Object)
+					Expect(err).NotTo(HaveOccurred())
+
+					switch event.Type {
+					case watch.Bookmark:
+						if meta.GetAnnotations()[metav1.InitialEventsAnnotationKey] != "true" {
+							continue
+						}
+						receivedWatchListStreamFromTheServer = true
+						return
+					case watch.Added, watch.Modified, watch.Deleted, watch.Error:
+					default:
+						panic(fmt.Sprintf("unexpected watch event: %v", event.Object))
+					}
+				}
+			}
+		}()
+
+		Expect(receivedWatchListStreamFromTheServer).To(BeTrue())
+	})
+})