From 3f7c796d78e6c04911dd9f2db994cf02731adf2b Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 2 Jun 2026 15:18:38 +0000 Subject: [PATCH] fix(api): emit initial-events-end bookmark for core.cozystack.io watches The aggregated apiserver resources in group core.cozystack.io (TenantSecret, TenantModule, TenantNamespace) ignored the SendInitialEvents field of ListOptions. Clients using the WatchList / streaming-list protocol (client-go's WatchListClient feature, on by default since v1.35) never received the required k8s.io/initial-events-end bookmark, so their reflectors stayed stuck in the initial-stream phase and logged "hasn't received required bookmark event marking the end of initial events stream" every ~10s, never reaching HasSynced. apps.cozystack.io/Application already implemented this contract inline. Extract that logic into a shared registry.InitialEventsBookmarker helper (pkg/registry/utils.go) and wire it into all four translating Watch implementations: after the initial ADDED events, the first backing bookmark is annotated as the initial-events-end marker, with fallbacks that emit it before the first live event and on watcher close. Application is refactored onto the shared helper with its WorkloadMonitor pending-event draining preserved. Adds unit tests for the bookmarker and watch tests for each of the three core resources asserting ADDED -> annotated Bookmark -> live event. ```release-note [apiserver] core.cozystack.io resources (TenantSecret, TenantModule, TenantNamespace) now emit the k8s.io/initial-events-end bookmark for SendInitialEvents=true (WatchList) watches, so client-go informers reach HasSynced instead of looping on the missing-bookmark warning. ``` Co-Authored-By: Claude Opus 4.8 Signed-off-by: Simon Koudijs (cherry picked from commit a28e77c9a24580c1ec54dfc7317a09b0eaa0d2c6) --- pkg/registry/apps/application/rest.go | 160 ++++++----------- pkg/registry/core/tenantmodule/rest.go | 68 ++++--- .../core/tenantmodule/rest_watch_test.go | 118 ++++++++++++ pkg/registry/core/tenantnamespace/rest.go | 63 +++++-- .../core/tenantnamespace/rest_watch_test.go | 121 +++++++++++++ pkg/registry/core/tenantsecret/rest.go | 64 +++++-- .../core/tenantsecret/rest_watch_test.go | 100 +++++++++++ pkg/registry/utils.go | 88 +++++++++ pkg/registry/utils_test.go | 169 ++++++++++++++++++ 9 files changed, 798 insertions(+), 153 deletions(-) create mode 100644 pkg/registry/core/tenantmodule/rest_watch_test.go create mode 100644 pkg/registry/core/tenantnamespace/rest_watch_test.go create mode 100644 pkg/registry/core/tenantsecret/rest_watch_test.go create mode 100644 pkg/registry/utils_test.go diff --git a/pkg/registry/apps/application/rest.go b/pkg/registry/apps/application/rest.go index 0f0aae97c..6f457127b 100644 --- a/pkg/registry/apps/application/rest.go +++ b/pkg/registry/apps/application/rest.go @@ -681,13 +681,19 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } helmLabelSelector = labels.NewSelector().Add(labelRequirements...) - // Handle SendInitialEvents for WatchList feature (Kubernetes 1.27+) - // When sendInitialEvents=true, the client expects: - // 1. All existing resources as ADDED events - // 2. A Bookmark event with "k8s.io/initial-events-end": "true" annotation - // controller-runtime cache already sends ADDED events for all cached objects, - // so we just need to send the bookmark after those initial events + // Honor SendInitialEvents (WatchList): the client expects all existing + // objects as ADDED events, then a Bookmark annotated with + // metav1.InitialEventsAnnotationKey. controller-runtime's cache already + // replays the ADDED events, so we just emit the terminating bookmark. sendInitialEvents := options.SendInitialEvents != nil && *options.SendInitialEvents + bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, options.ResourceVersion, func() runtime.Object { + app := &appsv1alpha1.Application{} + app.TypeMeta = metav1.TypeMeta{ + APIVersion: appsv1alpha1.SchemeGroupVersion.String(), + Kind: r.kindName, + } + return app + }) // Create a custom watcher to transform events customW := &customWatcher{ @@ -702,6 +708,10 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio helmWatcher, err := r.w.Watch(ctx, hrList, &client.ListOptions{ Namespace: namespace, LabelSelector: helmLabelSelector, + // Ask the backing watch for bookmarks on a WatchList request; the + // apiserver omits them by default, leaving the terminating + // initial-events-end bookmark with no reliable trigger. + Raw: &metav1.ListOptions{AllowWatchBookmarks: sendInitialEvents}, }) if err != nil { klog.Errorf("Error setting up watch for HelmReleases: %v", err) @@ -738,10 +748,6 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio defer wmWatcherForCleanup.Stop() } - // Track whether we've sent the initial-events-end bookmark - initialEventsEndSent := !sendInitialEvents // If not sendInitialEvents, consider it already sent - var lastResourceVersion string - // Buffer of WorkloadMonitor events that arrived before the initial // snapshot finished. The watch-list contract requires the stream to // deliver all ADDED events followed by the initial-events-end bookmark @@ -762,48 +768,26 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } } - drainPendingWMEvents := func() { + // send forwards an event, returning false if the watch or context ended. + send := func(ev watch.Event) bool { + select { + case customW.resultChan <- ev: + return true + case <-customW.stopChan: + return false + case <-ctx.Done(): + return false + } + } + + drainPendingWMEvents := func() bool { for _, ev := range pendingWMEvents { - select { - case customW.resultChan <- ev: - case <-customW.stopChan: - return - case <-ctx.Done(): - return + if !send(ev) { + return false } } pendingWMEvents = nil - } - - // Helper function to send initial-events-end bookmark - sendInitialEventsEndBookmark := func() { - if initialEventsEndSent { - return - } - initialEventsEndSent = true - - bookmarkApp := &appsv1alpha1.Application{} - bookmarkApp.SetResourceVersion(lastResourceVersion) - bookmarkApp.TypeMeta = metav1.TypeMeta{ - APIVersion: appsv1alpha1.SchemeGroupVersion.String(), - Kind: r.kindName, - } - bookmarkApp.SetAnnotations(map[string]string{ - "k8s.io/initial-events-end": "true", - }) - bookmarkEvent := watch.Event{ - Type: watch.Bookmark, - Object: bookmarkApp, - } - klog.V(6).Infof("Sending initial-events-end bookmark with RV=%s", lastResourceVersion) - select { - case customW.resultChan <- bookmarkEvent: - case <-customW.stopChan: - return - case <-ctx.Done(): - return - } - drainPendingWMEvents() + return true } // Process watch events @@ -814,45 +798,25 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio // The watcher has been closed klog.Warning("HelmRelease watcher closed") // Send initial-events-end bookmark before closing if not yet sent - sendInitialEventsEndBookmark() + if bookmark, ok := bookmarker.OnClose(); ok { + if send(bookmark) { + drainPendingWMEvents() + } + } return } // Handle bookmark events - these are critical for informer sync if event.Type == watch.Bookmark { if hr, ok := event.Object.(*helmv2.HelmRelease); ok { - lastResourceVersion = hr.GetResourceVersion() - - // If sendInitialEvents and we haven't sent initial-events-end yet, - // add the annotation to this bookmark - bookmarkApp := &appsv1alpha1.Application{} - bookmarkApp.SetResourceVersion(lastResourceVersion) - bookmarkApp.TypeMeta = metav1.TypeMeta{ - APIVersion: appsv1alpha1.SchemeGroupVersion.String(), - Kind: r.kindName, - } - justFlipped := false - if !initialEventsEndSent { - initialEventsEndSent = true - justFlipped = true - bookmarkApp.SetAnnotations(map[string]string{ - "k8s.io/initial-events-end": "true", - }) - klog.V(6).Infof("Sending initial-events-end bookmark with RV=%s", lastResourceVersion) - } - bookmarkEvent := watch.Event{ - Type: watch.Bookmark, - Object: bookmarkApp, - } - select { - case customW.resultChan <- bookmarkEvent: - case <-customW.stopChan: - return - case <-ctx.Done(): + // controller-runtime bookmarks after the initial list + // replay; the first one becomes the terminating marker. + bookmark, initialEventsEnd := bookmarker.OnBackingBookmark(hr.GetResourceVersion()) + if !send(bookmark) { return } - if justFlipped { - drainPendingWMEvents() + if initialEventsEnd && !drainPendingWMEvents() { + return } } continue @@ -871,8 +835,8 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio continue } - // Update lastResourceVersion for bookmark - lastResourceVersion = hr.GetResourceVersion() + // Track resourceVersion for the eventual bookmark + bookmarker.Observe(hr.GetResourceVersion()) // Apply manual field selector filtering (metadata.name and metadata.namespace) // controller-runtime cache doesn't support field selectors @@ -909,9 +873,15 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } } - // If this is not an ADDED event and we haven't sent initial-events-end, send it now - if event.Type != watch.Added && !initialEventsEndSent { - sendInitialEventsEndBookmark() + // Emit the terminating bookmark before the first live event, then + // replay any buffered WorkloadMonitor events. + if bookmark, ok := bookmarker.BeforeLiveEvent(event.Type); ok { + if !send(bookmark) { + return + } + if !drainPendingWMEvents() { + return + } } // Skip ADDED events based on resourceVersion comparison @@ -926,18 +896,8 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } // When startingRV == 0, always send ADDED events (client wants full state) - // Create watch event with Application object - appEvent := watch.Event{ - Type: event.Type, - Object: &app, - } - // Send event to custom watcher - select { - case customW.resultChan <- appEvent: - case <-customW.stopChan: - return - case <-ctx.Done(): + if !send(watch.Event{Type: event.Type, Object: &app}) { return } @@ -1026,16 +986,12 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio // Buffer WM-triggered events that arrive before the // initial-events-end bookmark. They will be replayed in order // immediately after the bookmark is emitted. - if !initialEventsEndSent { + if !bookmarker.Sent() { pendingWMEvents = append(pendingWMEvents, outEvent) continue } - lastResourceVersion = wm.GetResourceVersion() - select { - case customW.resultChan <- outEvent: - case <-customW.stopChan: - return - case <-ctx.Done(): + bookmarker.Observe(wm.GetResourceVersion()) + if !send(outEvent) { return } diff --git a/pkg/registry/core/tenantmodule/rest.go b/pkg/registry/core/tenantmodule/rest.go index 888d44144..09fb288d5 100644 --- a/pkg/registry/core/tenantmodule/rest.go +++ b/pkg/registry/core/tenantmodule/rest.go @@ -366,6 +366,11 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } } + // For a SendInitialEvents (WatchList) request, ask the backing watch for + // bookmarks — the apiserver omits them by default, which would leave the + // terminating initial-events-end bookmark with no reliable trigger. + sendInitialEvents := options.SendInitialEvents != nil && *options.SendInitialEvents + // Start watch on HelmRelease with label selector only // Field selectors are not supported by controller-runtime cache // See: https://github.com/kubernetes-sigs/controller-runtime/issues/612 @@ -373,12 +378,24 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio helmWatcher, err := r.w.Watch(ctx, hrList, &client.ListOptions{ Namespace: namespace, LabelSelector: helmLabelSelector, + Raw: &metav1.ListOptions{AllowWatchBookmarks: sendInitialEvents}, }) if err != nil { klog.Errorf("Error setting up watch for HelmReleases: %v", err) return nil, err } + // Emit the initial-events-end bookmark after the initial ADDED events so + // client-go reflectors reach HasSynced. + bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, options.ResourceVersion, func() runtime.Object { + module := &corev1alpha1.TenantModule{} + module.TypeMeta = metav1.TypeMeta{ + APIVersion: corev1alpha1.SchemeGroupVersion.String(), + Kind: r.kindName, + } + return module + }) + // Create a custom watcher to transform events customW := &customWatcher{ resultChan: make(chan watch.Event), @@ -390,32 +407,35 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio defer close(customW.resultChan) defer customW.underlying.Stop() + // send forwards an event, returning false if the watch or context ended. + send := func(ev watch.Event) bool { + select { + case customW.resultChan <- ev: + return true + case <-customW.stopChan: + return false + case <-ctx.Done(): + return false + } + } + for { select { case event, ok := <-customW.underlying.ResultChan(): if !ok { klog.Warning("HelmRelease watcher closed") + // Flush the terminating bookmark if still pending. + if bookmark, ok := bookmarker.OnClose(); ok { + send(bookmark) + } return } // Handle bookmark events if event.Type == watch.Bookmark { if hr, ok := event.Object.(*helmv2.HelmRelease); ok { - bookmarkModule := &corev1alpha1.TenantModule{} - bookmarkModule.SetResourceVersion(hr.GetResourceVersion()) - bookmarkModule.TypeMeta = metav1.TypeMeta{ - APIVersion: corev1alpha1.SchemeGroupVersion.String(), - Kind: r.kindName, - } - bookmarkEvent := watch.Event{ - Type: watch.Bookmark, - Object: bookmarkModule, - } - select { - case customW.resultChan <- bookmarkEvent: - case <-customW.stopChan: - return - case <-ctx.Done(): + bookmark, _ := bookmarker.OnBackingBookmark(hr.GetResourceVersion()) + if !send(bookmark) { return } } @@ -434,6 +454,7 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio klog.V(4).Infof("Expected HelmRelease object, got %T", event.Object) continue } + bookmarker.Observe(hr.GetResourceVersion()) // Apply manual field selector filtering if !fieldFilter.MatchesName(hr.Name) { @@ -482,18 +503,15 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio } } - // Create watch event with TenantModule object - moduleEvent := watch.Event{ - Type: event.Type, - Object: &module, + // Emit the initial-events-end bookmark before the first live event. + if bookmark, ok := bookmarker.BeforeLiveEvent(event.Type); ok { + if !send(bookmark) { + return + } } - // Send event to custom watcher - select { - case customW.resultChan <- moduleEvent: - case <-customW.stopChan: - return - case <-ctx.Done(): + // Send the translated TenantModule event. + if !send(watch.Event{Type: event.Type, Object: &module}) { return } diff --git a/pkg/registry/core/tenantmodule/rest_watch_test.go b/pkg/registry/core/tenantmodule/rest_watch_test.go new file mode 100644 index 000000000..acb29b349 --- /dev/null +++ b/pkg/registry/core/tenantmodule/rest_watch_test.go @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: Apache-2.0 + +package tenantmodule + +import ( + "context" + "testing" + "time" + + helmv2 "github.com/fluxcd/helm-controller/api/v2" + metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/endpoints/request" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + corev1alpha1 "github.com/cozystack/cozystack/pkg/apis/core/v1alpha1" +) + +const testNamespace = "tenant-foo" + +// collectEvents drains up to n events from the watch, or returns early if the +// channel closes or the timeout fires. +func collectEvents(t *testing.T, w watch.Interface, n int, timeout time.Duration) []watch.Event { + t.Helper() + out := make([]watch.Event, 0, n) + deadline := time.NewTimer(timeout) + defer deadline.Stop() + for len(out) < n { + select { + case ev, ok := <-w.ResultChan(): + if !ok { + return out + } + out = append(out, ev) + case <-deadline.C: + return out + } + } + return out +} + +// TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark asserts the +// WatchList contract for TenantModule: ADDED events, then a bookmark annotated +// with k8s.io/initial-events-end, then live events. +// +// fake.Client.Watch doesn't replay existing objects as ADDED, so the HelmRelease +// is created after the watch starts and then mutated to drive a live event. +func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) { + scheme := runtime.NewScheme() + if err := helmv2.AddToScheme(scheme); err != nil { + t.Fatalf("add helmv2 to scheme: %v", err) + } + fc := fake.NewClientBuilder().WithScheme(scheme).Build() + r := NewREST(fc, fc) + + ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), testNamespace)) + defer cancel() + + sendInitialEvents := true + w, err := r.Watch(ctx, &metainternalversion.ListOptions{SendInitialEvents: &sendInitialEvents}) + if err != nil { + t.Fatalf("Watch returned error: %v", err) + } + defer w.Stop() + + hr := &helmv2.HelmRelease{ + ObjectMeta: metav1.ObjectMeta{ + Name: "module-a", + Namespace: testNamespace, + Labels: map[string]string{ + TenantModuleLabelKey: TenantModuleLabelValue, + }, + }, + } + if err := r.c.Create(ctx, hr); err != nil { + t.Fatalf("create HelmRelease: %v", err) + } + hr.Annotations = map[string]string{"touched": "1"} + if err := r.c.Update(ctx, hr); err != nil { + t.Fatalf("update HelmRelease: %v", err) + } + + evs := collectEvents(t, w, 3, 2*time.Second) + if len(evs) < 3 { + t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs) + } + + if evs[0].Type != watch.Added { + t.Fatalf("event[0]: expected Added, got %s", evs[0].Type) + } + added, ok := evs[0].Object.(*corev1alpha1.TenantModule) + if !ok { + t.Fatalf("event[0]: expected *TenantModule, got %T", evs[0].Object) + } + if added.Name != "module-a" { + t.Fatalf("event[0]: expected name module-a, got %q", added.Name) + } + + if evs[1].Type != watch.Bookmark { + t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type) + } + bookmark, ok := evs[1].Object.(*corev1alpha1.TenantModule) + if !ok { + t.Fatalf("event[1]: expected *TenantModule, got %T", evs[1].Object) + } + if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" { + t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got) + } + if bookmark.ResourceVersion == "" { + t.Fatal("event[1]: expected non-empty resourceVersion on bookmark") + } + + if evs[2].Type != watch.Modified { + t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type) + } +} diff --git a/pkg/registry/core/tenantnamespace/rest.go b/pkg/registry/core/tenantnamespace/rest.go index 4ee97eb08..1962caaff 100644 --- a/pkg/registry/core/tenantnamespace/rest.go +++ b/pkg/registry/core/tenantnamespace/rest.go @@ -170,10 +170,16 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch nsList := &corev1.NamespaceList{} + // For a SendInitialEvents (WatchList) request, ask the backing watch for + // bookmarks — the apiserver omits them by default, which would leave the + // terminating initial-events-end bookmark with no reliable trigger. + sendInitialEvents := opts.SendInitialEvents != nil && *opts.SendInitialEvents + // Build upstream watch options with field and label selectors rawOpts := &metav1.ListOptions{ - Watch: true, - ResourceVersion: opts.ResourceVersion, + Watch: true, + ResourceVersion: opts.ResourceVersion, + AllowWatchBookmarks: sendInitialEvents, } if opts.FieldSelector != nil { rawOpts.FieldSelector = opts.FieldSelector.String() @@ -195,26 +201,44 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch } } + // Emit the initial-events-end bookmark after the initial ADDED events so + // client-go reflectors reach HasSynced. + bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, opts.ResourceVersion, func() runtime.Object { + return &corev1alpha1.TenantNamespace{ + TypeMeta: metav1.TypeMeta{ + APIVersion: corev1alpha1.SchemeGroupVersion.String(), + Kind: "TenantNamespace", + }, + } + }) + events := make(chan watch.Event) pw := watch.NewProxyWatcher(events) go func() { defer pw.Stop() + defer nsWatch.Stop() + + // send forwards an event, returning false if the watch or context ended. + send := func(ev watch.Event) bool { + select { + case events <- ev: + return true + case <-pw.StopChan(): + return false + case <-ctx.Done(): + return false + } + } for ev := range nsWatch.ResultChan() { // Handle bookmark events if ev.Type == watch.Bookmark { if ns, ok := ev.Object.(*corev1.Namespace); ok { - out := &corev1alpha1.TenantNamespace{ - TypeMeta: metav1.TypeMeta{ - APIVersion: corev1alpha1.SchemeGroupVersion.String(), - Kind: "TenantNamespace", - }, - ObjectMeta: metav1.ObjectMeta{ - ResourceVersion: ns.ResourceVersion, - }, + bookmark, _ := bookmarker.OnBackingBookmark(ns.ResourceVersion) + if !send(bookmark) { + return } - events <- watch.Event{Type: watch.Bookmark, Object: out} } continue } @@ -223,6 +247,7 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch if !ok || !strings.HasPrefix(ns.Name, prefix) { continue } + bookmarker.Observe(ns.ResourceVersion) // Apply defensive filtering for field and label selectors if opts.FieldSelector != nil { @@ -274,7 +299,21 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch } // When startingRV == 0, always send ADDED events (client wants full state) - events <- watch.Event{Type: ev.Type, Object: out} + // Emit the initial-events-end bookmark before the first live event. + if bookmark, ok := bookmarker.BeforeLiveEvent(ev.Type); ok { + if !send(bookmark) { + return + } + } + + if !send(watch.Event{Type: ev.Type, Object: out}) { + return + } + } + + // Backing watcher closed: flush the terminating bookmark if still pending. + if bookmark, ok := bookmarker.OnClose(); ok { + send(bookmark) } }() diff --git a/pkg/registry/core/tenantnamespace/rest_watch_test.go b/pkg/registry/core/tenantnamespace/rest_watch_test.go new file mode 100644 index 000000000..baae91f34 --- /dev/null +++ b/pkg/registry/core/tenantnamespace/rest_watch_test.go @@ -0,0 +1,121 @@ +// SPDX-License-Identifier: Apache-2.0 + +package tenantnamespace + +import ( + "context" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/authentication/user" + "k8s.io/apiserver/pkg/endpoints/request" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + corev1alpha1 "github.com/cozystack/cozystack/pkg/apis/core/v1alpha1" +) + +// collectEvents drains up to n events from the watch, or returns early if the +// channel closes or the timeout fires. +func collectEvents(t *testing.T, w watch.Interface, n int, timeout time.Duration) []watch.Event { + t.Helper() + out := make([]watch.Event, 0, n) + deadline := time.NewTimer(timeout) + defer deadline.Stop() + for len(out) < n { + select { + case ev, ok := <-w.ResultChan(): + if !ok { + return out + } + out = append(out, ev) + case <-deadline.C: + return out + } + } + return out +} + +// TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark asserts the +// WatchList contract for TenantNamespace: ADDED events, then a bookmark +// annotated with k8s.io/initial-events-end, then live events. +// +// The user is in system:masters so the access check passes without RBAC +// fixtures. fake.Client.Watch doesn't replay existing objects as ADDED, so the +// namespace is created after the watch starts and then mutated to drive a live +// event. +func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) { + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1 to scheme: %v", err) + } + fc := fake.NewClientBuilder().WithScheme(scheme).Build() + r := &REST{ + c: fc, + w: fc, + gvr: schema.GroupVersionResource{ + Group: corev1alpha1.GroupName, + Version: "v1alpha1", + Resource: "tenantnamespaces", + }, + } + + u := &user.DefaultInfo{Name: "admin", Groups: []string{"system:masters"}} + ctx, cancel := context.WithCancel(request.WithUser(context.Background(), u)) + defer cancel() + + sendInitialEvents := true + w, err := r.Watch(ctx, &metainternal.ListOptions{SendInitialEvents: &sendInitialEvents}) + if err != nil { + t.Fatalf("Watch returned error: %v", err) + } + defer w.Stop() + + ns := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "tenant-foo"}} + if err := r.c.Create(ctx, ns); err != nil { + t.Fatalf("create namespace: %v", err) + } + ns.Labels = map[string]string{"touched": "1"} + if err := r.c.Update(ctx, ns); err != nil { + t.Fatalf("update namespace: %v", err) + } + + evs := collectEvents(t, w, 3, 2*time.Second) + if len(evs) < 3 { + t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs) + } + + if evs[0].Type != watch.Added { + t.Fatalf("event[0]: expected Added, got %s", evs[0].Type) + } + added, ok := evs[0].Object.(*corev1alpha1.TenantNamespace) + if !ok { + t.Fatalf("event[0]: expected *TenantNamespace, got %T", evs[0].Object) + } + if added.Name != "tenant-foo" { + t.Fatalf("event[0]: expected name tenant-foo, got %q", added.Name) + } + + if evs[1].Type != watch.Bookmark { + t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type) + } + bookmark, ok := evs[1].Object.(*corev1alpha1.TenantNamespace) + if !ok { + t.Fatalf("event[1]: expected *TenantNamespace, got %T", evs[1].Object) + } + if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" { + t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got) + } + if bookmark.ResourceVersion == "" { + t.Fatal("event[1]: expected non-empty resourceVersion on bookmark") + } + + if evs[2].Type != watch.Modified { + t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type) + } +} diff --git a/pkg/registry/core/tenantsecret/rest.go b/pkg/registry/core/tenantsecret/rest.go index c75255ed8..38bf6fc67 100644 --- a/pkg/registry/core/tenantsecret/rest.go +++ b/pkg/registry/core/tenantsecret/rest.go @@ -465,13 +465,19 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch return watch.NewProxyWatcher(ch), nil } + // For a SendInitialEvents (WatchList) request, ask the backing watch for + // bookmarks — the apiserver omits them by default, which would leave the + // terminating initial-events-end bookmark with no reliable trigger. + sendInitialEvents := opts.SendInitialEvents != nil && *opts.SendInitialEvents + secList := &corev1.SecretList{} base, err := r.w.Watch(ctx, secList, &client.ListOptions{ Namespace: ns, LabelSelector: ls, Raw: &metav1.ListOptions{ - Watch: true, - ResourceVersion: opts.ResourceVersion, + Watch: true, + ResourceVersion: opts.ResourceVersion, + AllowWatchBookmarks: sendInitialEvents, }, }) if err != nil { @@ -486,26 +492,44 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch } } + // Emit the initial-events-end bookmark after the initial ADDED events so + // client-go reflectors reach HasSynced. + bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, opts.ResourceVersion, func() runtime.Object { + return &corev1alpha1.TenantSecret{ + TypeMeta: metav1.TypeMeta{ + APIVersion: corev1alpha1.SchemeGroupVersion.String(), + Kind: kindTenantSecret, + }, + } + }) + ch := make(chan watch.Event) proxy := watch.NewProxyWatcher(ch) go func() { defer proxy.Stop() + defer base.Stop() + + // send forwards an event, returning false if the watch or context ended. + send := func(ev watch.Event) bool { + select { + case ch <- ev: + return true + case <-proxy.StopChan(): + return false + case <-ctx.Done(): + return false + } + } for ev := range base.ResultChan() { // Handle bookmark events if ev.Type == watch.Bookmark { if sec, ok := ev.Object.(*corev1.Secret); ok { - out := &corev1alpha1.TenantSecret{ - TypeMeta: metav1.TypeMeta{ - APIVersion: corev1alpha1.SchemeGroupVersion.String(), - Kind: kindTenantSecret, - }, - ObjectMeta: metav1.ObjectMeta{ - ResourceVersion: sec.ResourceVersion, - }, + bookmark, _ := bookmarker.OnBackingBookmark(sec.ResourceVersion) + if !send(bookmark) { + return } - ch <- watch.Event{Type: watch.Bookmark, Object: out} } continue } @@ -514,6 +538,7 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch if !ok || sec == nil { continue } + bookmarker.Observe(sec.ResourceVersion) // Defensive: post-filter against the merged selector. The underlying // watch already filters by label, but this guards against any client @@ -539,10 +564,21 @@ func (r *REST) Watch(ctx context.Context, opts *metainternal.ListOptions) (watch } // When startingRV == 0, always send ADDED events (client wants full state) - ch <- watch.Event{ - Type: ev.Type, - Object: tenant, + // Emit the initial-events-end bookmark before the first live event. + if bookmark, ok := bookmarker.BeforeLiveEvent(ev.Type); ok { + if !send(bookmark) { + return + } } + + if !send(watch.Event{Type: ev.Type, Object: tenant}) { + return + } + } + + // Backing watcher closed: flush the terminating bookmark if still pending. + if bookmark, ok := bookmarker.OnClose(); ok { + send(bookmark) } }() diff --git a/pkg/registry/core/tenantsecret/rest_watch_test.go b/pkg/registry/core/tenantsecret/rest_watch_test.go new file mode 100644 index 000000000..3d6ab713a --- /dev/null +++ b/pkg/registry/core/tenantsecret/rest_watch_test.go @@ -0,0 +1,100 @@ +// SPDX-License-Identifier: Apache-2.0 + +package tenantsecret + +import ( + "context" + "testing" + "time" + + metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/apiserver/pkg/endpoints/request" + + corev1alpha1 "github.com/cozystack/cozystack/pkg/apis/core/v1alpha1" +) + +// collectEvents drains up to n events from the watch, or returns early if the +// channel closes or the timeout fires. +func collectEvents(t *testing.T, w watch.Interface, n int, timeout time.Duration) []watch.Event { + t.Helper() + out := make([]watch.Event, 0, n) + deadline := time.NewTimer(timeout) + defer deadline.Stop() + for len(out) < n { + select { + case ev, ok := <-w.ResultChan(): + if !ok { + return out + } + out = append(out, ev) + case <-deadline.C: + return out + } + } + return out +} + +// TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark asserts the +// WatchList contract for TenantSecret: ADDED events, then a bookmark annotated +// with k8s.io/initial-events-end, then live events. +// +// fake.Client.Watch doesn't replay existing objects as ADDED, so the secret is +// created after the watch starts and then mutated to drive a live event. +func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) { + r := newTestREST(t) + + ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), testNamespace)) + defer cancel() + + sendInitialEvents := true + w, err := r.Watch(ctx, &metainternal.ListOptions{SendInitialEvents: &sendInitialEvents}) + if err != nil { + t.Fatalf("Watch returned error: %v", err) + } + defer w.Stop() + + sec := makeTenantSecret("harbor-test-credentials", nil) + if err := r.c.Create(ctx, sec); err != nil { + t.Fatalf("create secret: %v", err) + } + sec.Annotations = map[string]string{"touched": "1"} + if err := r.c.Update(ctx, sec); err != nil { + t.Fatalf("update secret: %v", err) + } + + evs := collectEvents(t, w, 3, 2*time.Second) + if len(evs) < 3 { + t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs) + } + + if evs[0].Type != watch.Added { + t.Fatalf("event[0]: expected Added, got %s", evs[0].Type) + } + added, ok := evs[0].Object.(*corev1alpha1.TenantSecret) + if !ok { + t.Fatalf("event[0]: expected *TenantSecret, got %T", evs[0].Object) + } + if added.Name != "harbor-test-credentials" { + t.Fatalf("event[0]: expected name harbor-test-credentials, got %q", added.Name) + } + + if evs[1].Type != watch.Bookmark { + t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type) + } + bookmark, ok := evs[1].Object.(*corev1alpha1.TenantSecret) + if !ok { + t.Fatalf("event[1]: expected *TenantSecret, got %T", evs[1].Object) + } + if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" { + t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got) + } + if bookmark.ResourceVersion == "" { + t.Fatal("event[1]: expected non-empty resourceVersion on bookmark") + } + + if evs[2].Type != watch.Modified { + t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type) + } +} diff --git a/pkg/registry/utils.go b/pkg/registry/utils.go index c5deb422e..e2951c56d 100644 --- a/pkg/registry/utils.go +++ b/pkg/registry/utils.go @@ -20,7 +20,9 @@ import ( "strconv" "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" ) // MaxResourceVersion returns the maximum resourceVersion from all items in a list. @@ -55,3 +57,89 @@ func MaxResourceVersion(list runtime.Object) (string, error) { return strconv.FormatUint(max, 10), nil } + +// InitialEventsBookmarker implements the WatchList initial-events-end contract +// for a Watch that translates events from a backing controller-runtime watcher. +// It annotates the first backing Bookmark as the terminating marker, falling +// back to emitting one before the first live event or when the watcher closes. +// +// Construct with NewInitialEventsBookmarker. Methods are not safe for concurrent +// use; call them from a single watch goroutine. +type InitialEventsBookmarker struct { + // newBookmark returns an empty, TypeMeta-populated object of the translated + // resource type; the bookmarker fills in the resourceVersion and annotation. + newBookmark func() runtime.Object + sent bool + lastRV string +} + +// NewInitialEventsBookmarker returns a bookmarker for a watch. When +// sendInitialEvents is false every method is a no-op. startingRV seeds the +// resourceVersion (the client's requested ResourceVersion) so the terminating +// bookmark carries a valid version even if the watcher closes before any +// backing event is observed. +func NewInitialEventsBookmarker(sendInitialEvents bool, startingRV string, newBookmark func() runtime.Object) *InitialEventsBookmarker { + return &InitialEventsBookmarker{ + newBookmark: newBookmark, + lastRV: startingRV, + sent: !sendInitialEvents, + } +} + +// Sent reports whether the initial-events-end bookmark has been emitted (or was +// never required). Callers buffering live events use it to decide when to start +// delivering them. +func (b *InitialEventsBookmarker) Sent() bool { return b.sent } + +// Observe records the resourceVersion of the latest backing event so the +// bookmark carries an up-to-date version. Empty versions are ignored. +func (b *InitialEventsBookmarker) Observe(resourceVersion string) { + if resourceVersion != "" { + b.lastRV = resourceVersion + } +} + +// bookmarkEvent builds a Bookmark at the tracked resourceVersion, annotated as +// the terminating marker when end is true. +func (b *InitialEventsBookmarker) bookmarkEvent(end bool) watch.Event { + obj := b.newBookmark() + if accessor, err := meta.Accessor(obj); err == nil { + accessor.SetResourceVersion(b.lastRV) + if end { + accessor.SetAnnotations(map[string]string{metav1.InitialEventsAnnotationKey: "true"}) + } + } + return watch.Event{Type: watch.Bookmark, Object: obj} +} + +// OnBackingBookmark forwards the backing watcher's Bookmark at rv. The first one +// after the initial events becomes the terminating marker; the returned bool +// reports whether this call emitted it, so callers can flush buffered events. +func (b *InitialEventsBookmarker) OnBackingBookmark(rv string) (event watch.Event, initialEventsEnd bool) { + b.Observe(rv) + if b.sent { + return b.bookmarkEvent(false), false + } + b.sent = true + return b.bookmarkEvent(true), true +} + +// BeforeLiveEvent returns the terminating bookmark and true when a live +// (non-ADDED) event is about to be delivered and the marker is still pending. +func (b *InitialEventsBookmarker) BeforeLiveEvent(eventType watch.EventType) (event watch.Event, needed bool) { + if b.sent || eventType == watch.Added { + return watch.Event{}, false + } + b.sent = true + return b.bookmarkEvent(true), true +} + +// OnClose returns the terminating bookmark and true if the backing watcher +// closed before the marker was emitted. +func (b *InitialEventsBookmarker) OnClose() (event watch.Event, needed bool) { + if b.sent { + return watch.Event{}, false + } + b.sent = true + return b.bookmarkEvent(true), true +} diff --git a/pkg/registry/utils_test.go b/pkg/registry/utils_test.go new file mode 100644 index 000000000..b3a916f28 --- /dev/null +++ b/pkg/registry/utils_test.go @@ -0,0 +1,169 @@ +/* +Copyright 2024 The Cozystack 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 registry + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/watch" +) + +// newTestBookmarkObject returns a minimal object for the bookmarker to stamp. +func newTestBookmarkObject() runtime.Object { + return &metav1.PartialObjectMetadata{ + TypeMeta: metav1.TypeMeta{APIVersion: "core.cozystack.io/v1alpha1", Kind: "TenantSecret"}, + } +} + +// assertInitialEventsEnd verifies ev is a Bookmark carrying +// metav1.InitialEventsAnnotationKey=true and a non-empty resourceVersion. +func assertInitialEventsEnd(t *testing.T, ev watch.Event, wantRV string) { + t.Helper() + if ev.Type != watch.Bookmark { + t.Fatalf("expected Bookmark event, got %s", ev.Type) + } + obj, ok := ev.Object.(*metav1.PartialObjectMetadata) + if !ok { + t.Fatalf("expected *PartialObjectMetadata, got %T", ev.Object) + } + if got := obj.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" { + t.Fatalf("expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got) + } + if obj.ResourceVersion == "" { + t.Fatal("expected non-empty resourceVersion on terminating bookmark") + } + if wantRV != "" && obj.ResourceVersion != wantRV { + t.Fatalf("expected resourceVersion %q, got %q", wantRV, obj.ResourceVersion) + } +} + +func TestInitialEventsBookmarker_Disabled(t *testing.T) { + b := NewInitialEventsBookmarker(false, "", newTestBookmarkObject) + + if !b.Sent() { + t.Fatal("disabled bookmarker should report Sent()==true") + } + if _, ok := b.BeforeLiveEvent(watch.Modified); ok { + t.Fatal("disabled bookmarker should never need a before-live bookmark") + } + if _, ok := b.OnClose(); ok { + t.Fatal("disabled bookmarker should never need an on-close bookmark") + } + // A backing bookmark is still forwarded, but never carries the marker. + ev, end := b.OnBackingBookmark("7") + if end { + t.Fatal("disabled bookmarker should not mark a backing bookmark as initial-events-end") + } + if ev.Type != watch.Bookmark { + t.Fatalf("expected forwarded Bookmark, got %s", ev.Type) + } + if obj := ev.Object.(*metav1.PartialObjectMetadata); obj.Annotations[metav1.InitialEventsAnnotationKey] != "" { + t.Fatal("disabled bookmarker must not annotate forwarded bookmarks") + } +} + +func TestInitialEventsBookmarker_BackingBookmarkTerminatesStream(t *testing.T) { + b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject) + b.Observe("12") + + ev, end := b.OnBackingBookmark("15") + if !end { + t.Fatal("first backing bookmark should terminate the initial stream") + } + assertInitialEventsEnd(t, ev, "15") + if !b.Sent() { + t.Fatal("bookmarker should report Sent()==true after terminating bookmark") + } + + // Subsequent backing bookmarks are forwarded without the marker. + ev2, end2 := b.OnBackingBookmark("20") + if end2 { + t.Fatal("second backing bookmark must not re-terminate the stream") + } + if obj := ev2.Object.(*metav1.PartialObjectMetadata); obj.Annotations[metav1.InitialEventsAnnotationKey] != "" { + t.Fatal("subsequent bookmark must not carry the initial-events-end annotation") + } +} + +func TestInitialEventsBookmarker_BeforeLiveEvent(t *testing.T) { + b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject) + b.Observe("9") + + // ADDED events are part of the initial stream and never trigger the marker. + if _, ok := b.BeforeLiveEvent(watch.Added); ok { + t.Fatal("ADDED event should not trigger the initial-events-end bookmark") + } + + ev, ok := b.BeforeLiveEvent(watch.Modified) + if !ok { + t.Fatal("first non-ADDED event should trigger the initial-events-end bookmark") + } + assertInitialEventsEnd(t, ev, "9") + + // Once sent, later live events do not re-trigger it. + if _, ok := b.BeforeLiveEvent(watch.Deleted); ok { + t.Fatal("initial-events-end bookmark should only be emitted once") + } +} + +func TestInitialEventsBookmarker_OnClose(t *testing.T) { + b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject) + b.Observe("42") + + ev, ok := b.OnClose() + if !ok { + t.Fatal("closing before the marker was sent should flush a terminating bookmark") + } + assertInitialEventsEnd(t, ev, "42") + + if _, ok := b.OnClose(); ok { + t.Fatal("on-close bookmark must not be emitted twice") + } +} + +func TestInitialEventsBookmarker_ObserveIgnoresEmpty(t *testing.T) { + b := NewInitialEventsBookmarker(true, "", newTestBookmarkObject) + b.Observe("5") + b.Observe("") // must not clobber the tracked version + + ev, _ := b.OnClose() + assertInitialEventsEnd(t, ev, "5") +} + +func TestInitialEventsBookmarker_SeedsStartingResourceVersion(t *testing.T) { + // Watcher closes before any backing event is observed: the terminating + // bookmark falls back to the client's requested resourceVersion rather than + // an empty one. + b := NewInitialEventsBookmarker(true, "100", newTestBookmarkObject) + + ev, ok := b.OnClose() + if !ok { + t.Fatal("expected a terminating bookmark on close") + } + assertInitialEventsEnd(t, ev, "100") +} + +func TestInitialEventsBookmarker_ObserveOverridesStartingResourceVersion(t *testing.T) { + // Once a backing event is observed, its resourceVersion supersedes the seed. + b := NewInitialEventsBookmarker(true, "100", newTestBookmarkObject) + b.Observe("150") + + ev, _ := b.OnClose() + assertInitialEventsEnd(t, ev, "150") +}