Skip to content

Commit caa3989

Browse files
lexfreilexfrei
committed
fix(api): send the initial-events-end bookmark only to WatchList clients
The apiserver defaults SendInitialEvents on for every watch whose resourceVersion is empty or "0", and leaves AllowWatchBookmarks off. The aggregated storages read SendInitialEvents alone as a WatchList request, so a plain `kubectl get -w` on applications, tenant modules, namespaces, secrets or security groups got a bookmark annotated as the end of initial events. The k8s.io/apiserver v0.35 watch handler lets that through. From v0.36 it treats the annotated bookmark as the end of a WatchList and calls the WatchList completion hook, which it only sets up when SendInitialEvents and AllowWatchBookmarks are both on, so the handler dereferences nil and the watch dies. The handler is the one compiled into cozystack-api, so bumping the library is what breaks it, whatever the cluster version. The marker now requires both flags, the same condition the watch cache uses, and the backing watches ask for bookmarks only when the client did, since every backing bookmark is forwarded. Security groups still forward SendInitialEvents to their backing watch so it replays existing objects as ADDED. Assisted-by: LLM Signed-off-by: Aleksei Sviridkin <[email protected]>
1 parent ab0c9f6 commit caa3989

14 files changed

Lines changed: 822 additions & 42 deletions

File tree

‎pkg/cmd/server/watch_initial_events_test.go‎

Lines changed: 170 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,170 @@
1+
/*
2+
Copyright 2026 The Cozystack Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package server_test
18+
19+
import (
20+
"bufio"
21+
"context"
22+
"net/http"
23+
"net/http/httptest"
24+
"strings"
25+
"testing"
26+
"time"
27+
28+
helmv2 "github.com/fluxcd/helm-controller/api/v2"
29+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
30+
"k8s.io/apimachinery/pkg/runtime"
31+
"k8s.io/apiserver/pkg/registry/rest"
32+
genericapiserver "k8s.io/apiserver/pkg/server"
33+
utilfeature "k8s.io/apiserver/pkg/util/feature"
34+
restclient "k8s.io/client-go/rest"
35+
basecompatibility "k8s.io/component-base/compatibility"
36+
baseversion "k8s.io/component-base/version"
37+
"sigs.k8s.io/controller-runtime/pkg/client"
38+
"sigs.k8s.io/controller-runtime/pkg/client/fake"
39+
40+
appsv1alpha1 "github.com/cozystack/cozystack/pkg/apis/apps/v1alpha1"
41+
"github.com/cozystack/cozystack/pkg/apiserver"
42+
cozyserver "github.com/cozystack/cozystack/pkg/cmd/server"
43+
"github.com/cozystack/cozystack/pkg/config"
44+
applicationstorage "github.com/cozystack/cozystack/pkg/registry/apps/application"
45+
)
46+
47+
// startAppsServer serves the apps group through the real apiserver handler
48+
// chain, with the application storage backed by a fake client.
49+
func startAppsServer(t *testing.T) (*httptest.Server, client.Client) {
50+
t.Helper()
51+
rc := &config.ResourceConfig{Resources: []config.Resource{{
52+
Application: config.ApplicationConfig{Kind: "Redis", Singular: "redis", Plural: "redises"},
53+
Release: config.ReleaseConfig{Prefix: "redis-"},
54+
}}}
55+
if err := appsv1alpha1.RegisterDynamicTypes(apiserver.Scheme, rc); err != nil {
56+
t.Fatal(err)
57+
}
58+
59+
backing := runtime.NewScheme()
60+
if err := helmv2.AddToScheme(backing); err != nil {
61+
t.Fatal(err)
62+
}
63+
cli := fake.NewClientBuilder().WithScheme(backing).Build()
64+
65+
cfg := genericapiserver.NewConfig(apiserver.Codecs)
66+
cfg.ExternalAddress = "localhost:443"
67+
cfg.LoopbackClientConfig = &restclient.Config{}
68+
cfg.FeatureGate = utilfeature.DefaultMutableFeatureGate
69+
if baseversion.DefaultKubeBinaryVersion != "" {
70+
cfg.EffectiveVersion = basecompatibility.NewEffectiveVersionFromString(baseversion.DefaultKubeBinaryVersion, "", "")
71+
}
72+
cozyserver.ConfigureOpenAPI(cfg, cozyserver.KindSchemasFromConfig(rc), "test", "test")
73+
srv, err := cfg.Complete(nil).New("watch-initial-events-test", genericapiserver.NewEmptyDelegate())
74+
if err != nil {
75+
t.Fatal(err)
76+
}
77+
storage := map[string]rest.Storage{
78+
"redises": applicationstorage.NewREST(cli, cli, &rc.Resources[0]),
79+
}
80+
if err := apiserver.InstallAppsAPIGroup(srv, storage); err != nil {
81+
t.Fatal(err)
82+
}
83+
srv.PrepareRun()
84+
ts := httptest.NewServer(srv.Handler)
85+
t.Cleanup(ts.Close)
86+
return ts, cli
87+
}
88+
89+
// watchStream opens a watch with the given query, then creates and updates a
90+
// release behind it, and returns the stream lines received within the timeout.
91+
func watchStream(t *testing.T, ts *httptest.Server, cli client.Client, query string) []string {
92+
t.Helper()
93+
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
94+
defer cancel()
95+
req, err := http.NewRequestWithContext(ctx, http.MethodGet,
96+
ts.URL+"/apis/apps.cozystack.io/v1alpha1/namespaces/tenant-a/redises?"+query, nil)
97+
if err != nil {
98+
t.Fatal(err)
99+
}
100+
resp, err := http.DefaultClient.Do(req)
101+
if err != nil {
102+
t.Fatalf("watch: %v", err)
103+
}
104+
defer resp.Body.Close()
105+
if resp.StatusCode != http.StatusOK {
106+
t.Fatalf("watch: status %d", resp.StatusCode)
107+
}
108+
109+
lines := make(chan string)
110+
go func() {
111+
defer close(lines)
112+
sc := bufio.NewScanner(resp.Body)
113+
for sc.Scan() {
114+
lines <- sc.Text()
115+
}
116+
}()
117+
118+
hr := &helmv2.HelmRelease{ObjectMeta: metav1.ObjectMeta{
119+
Name: "redis-cache",
120+
Namespace: "tenant-a",
121+
Labels: map[string]string{
122+
applicationstorage.ApplicationKindLabel: "Redis",
123+
applicationstorage.ApplicationGroupLabel: appsv1alpha1.GroupName,
124+
},
125+
}}
126+
if err := cli.Create(context.Background(), hr); err != nil {
127+
t.Fatalf("create HelmRelease: %v", err)
128+
}
129+
// The storage sends a pending initial-events-end bookmark ahead of the
130+
// first live event, so drive one.
131+
hr.Annotations = map[string]string{"touched": "1"}
132+
if err := cli.Update(context.Background(), hr); err != nil {
133+
t.Fatalf("update HelmRelease: %v", err)
134+
}
135+
136+
var out []string
137+
for line := range lines {
138+
out = append(out, line)
139+
}
140+
return out
141+
}
142+
143+
// A plain watch has SendInitialEvents defaulted on by the handler but no
144+
// bookmarks; from k8s.io/apiserver v0.36 the handler panics if the storage
145+
// answers it with the initial-events-end bookmark, so none may appear.
146+
func TestWatchHandler_PlainWatchGetsNoInitialEventsEndBookmark(t *testing.T) {
147+
ts, cli := startAppsServer(t)
148+
lines := watchStream(t, ts, cli, "watch=true")
149+
sawModified := false
150+
for _, l := range lines {
151+
if strings.Contains(l, metav1.InitialEventsAnnotationKey) {
152+
t.Fatalf("plain watch received the initial-events-end bookmark: %s", l)
153+
}
154+
sawModified = sawModified || strings.Contains(l, `"type":"MODIFIED"`)
155+
}
156+
if !sawModified {
157+
t.Fatalf("plain watch saw no MODIFIED event, stream: %v", lines)
158+
}
159+
}
160+
161+
func TestWatchHandler_WatchListGetsInitialEventsEndBookmark(t *testing.T) {
162+
ts, cli := startAppsServer(t)
163+
lines := watchStream(t, ts, cli, "watch=true&sendInitialEvents=true&allowWatchBookmarks=true&resourceVersionMatch=NotOlderThan")
164+
for _, l := range lines {
165+
if strings.Contains(l, `"type":"BOOKMARK"`) && strings.Contains(l, metav1.InitialEventsAnnotationKey) {
166+
return
167+
}
168+
}
169+
t.Fatalf("watch list received no initial-events-end bookmark, stream: %v", lines)
170+
}

‎pkg/registry/apps/application/rest.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -794,12 +794,12 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
794794
}
795795
helmLabelSelector = labels.NewSelector().Add(labelRequirements...)
796796

797-
// Honor SendInitialEvents (WatchList): the client expects all existing
797+
// Honor a WatchList request: the client expects all existing
798798
// objects as ADDED events, then a Bookmark annotated with
799799
// metav1.InitialEventsAnnotationKey. controller-runtime's cache already
800800
// replays the ADDED events, so we just emit the terminating bookmark.
801-
sendInitialEvents := options.SendInitialEvents != nil && *options.SendInitialEvents
802-
bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, options.ResourceVersion, func() runtime.Object {
801+
initialEventsEnd := registry.InitialEventsEndBookmarkRequested(options)
802+
bookmarker := registry.NewInitialEventsBookmarker(initialEventsEnd, options.ResourceVersion, func() runtime.Object {
803803
app := &appsv1alpha1.Application{}
804804
app.TypeMeta = metav1.TypeMeta{
805805
APIVersion: appsv1alpha1.SchemeGroupVersion.String(),
@@ -824,7 +824,7 @@ func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptio
824824
// Ask the backing watch for bookmarks on a WatchList request; the
825825
// apiserver omits them by default, leaving the terminating
826826
// initial-events-end bookmark with no reliable trigger.
827-
Raw: &metav1.ListOptions{AllowWatchBookmarks: sendInitialEvents},
827+
Raw: &metav1.ListOptions{AllowWatchBookmarks: initialEventsEnd},
828828
})
829829
if err != nil {
830830
klog.Errorf("Error setting up watch for HelmReleases: %v", err)

‎pkg/registry/apps/application/rest_watch_initial_events_test.go‎

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
/*
2+
Copyright 2026 The Cozystack Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package application
18+
19+
import (
20+
"context"
21+
"testing"
22+
"time"
23+
24+
helmv2 "github.com/fluxcd/helm-controller/api/v2"
25+
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
26+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
27+
"k8s.io/apimachinery/pkg/runtime"
28+
"k8s.io/apimachinery/pkg/watch"
29+
"k8s.io/apiserver/pkg/endpoints/request"
30+
"sigs.k8s.io/controller-runtime/pkg/client/fake"
31+
32+
appsv1alpha1 "github.com/cozystack/cozystack/pkg/apis/apps/v1alpha1"
33+
"github.com/cozystack/cozystack/pkg/config"
34+
"github.com/cozystack/cozystack/pkg/registry/registrytest"
35+
)
36+
37+
func newWatchTestREST(t *testing.T) *REST {
38+
t.Helper()
39+
scheme := runtime.NewScheme()
40+
if err := helmv2.AddToScheme(scheme); err != nil {
41+
t.Fatalf("register helmv2 scheme: %v", err)
42+
}
43+
fc := fake.NewClientBuilder().WithScheme(scheme).Build()
44+
return NewREST(fc, fc, &config.Resource{
45+
Application: config.ApplicationConfig{Kind: "Redis", Singular: "redis", Plural: "redises"},
46+
Release: config.ReleaseConfig{Prefix: "redis-"},
47+
})
48+
}
49+
50+
func redisRelease(name string) *helmv2.HelmRelease {
51+
return &helmv2.HelmRelease{
52+
ObjectMeta: metav1.ObjectMeta{
53+
Name: "redis-" + name,
54+
Namespace: "tenant-a",
55+
Labels: map[string]string{
56+
ApplicationKindLabel: "Redis",
57+
ApplicationGroupLabel: appsv1alpha1.GroupName,
58+
},
59+
},
60+
}
61+
}
62+
63+
func collectWatchEvents(w watch.Interface, n int, timeout time.Duration) []watch.Event {
64+
var out []watch.Event
65+
deadline := time.After(timeout)
66+
for len(out) < n {
67+
select {
68+
case ev, ok := <-w.ResultChan():
69+
if !ok {
70+
return out
71+
}
72+
out = append(out, ev)
73+
case <-deadline:
74+
return out
75+
}
76+
}
77+
return out
78+
}
79+
80+
// driveCreateAndUpdate creates a release behind the watch and then touches it,
81+
// because fake.Client.Watch doesn't replay existing objects as ADDED.
82+
func driveCreateAndUpdate(ctx context.Context, t *testing.T, r *REST) {
83+
t.Helper()
84+
hr := redisRelease("cache")
85+
if err := r.c.Create(ctx, hr); err != nil {
86+
t.Fatalf("create HelmRelease: %v", err)
87+
}
88+
hr.Annotations = map[string]string{"touched": "1"}
89+
if err := r.c.Update(ctx, hr); err != nil {
90+
t.Fatalf("update HelmRelease: %v", err)
91+
}
92+
}
93+
94+
func TestWatch_SendInitialEvents_EmitsInitialEventsEndBookmark(t *testing.T) {
95+
r := newWatchTestREST(t)
96+
ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), "tenant-a"))
97+
defer cancel()
98+
99+
sendInitialEvents := true
100+
w, err := r.Watch(ctx, &metainternalversion.ListOptions{SendInitialEvents: &sendInitialEvents, AllowWatchBookmarks: true})
101+
if err != nil {
102+
t.Fatalf("Watch returned error: %v", err)
103+
}
104+
defer w.Stop()
105+
driveCreateAndUpdate(ctx, t, r)
106+
107+
evs := collectWatchEvents(w, 3, 2*time.Second)
108+
if len(evs) < 3 {
109+
t.Fatalf("expected at least 3 events (Added, Bookmark, Modified), got %d: %+v", len(evs), evs)
110+
}
111+
if evs[0].Type != watch.Added {
112+
t.Fatalf("event[0]: expected Added, got %s", evs[0].Type)
113+
}
114+
if evs[1].Type != watch.Bookmark {
115+
t.Fatalf("event[1]: expected Bookmark, got %s", evs[1].Type)
116+
}
117+
bookmark, ok := evs[1].Object.(*appsv1alpha1.Application)
118+
if !ok {
119+
t.Fatalf("event[1]: expected *Application, got %T", evs[1].Object)
120+
}
121+
if got := bookmark.Annotations[metav1.InitialEventsAnnotationKey]; got != "true" {
122+
t.Fatalf("event[1]: expected annotation %s=true, got %q", metav1.InitialEventsAnnotationKey, got)
123+
}
124+
if evs[2].Type != watch.Modified {
125+
t.Fatalf("event[2]: expected Modified after bookmark, got %s", evs[2].Type)
126+
}
127+
}
128+
129+
// TestWatch_BackingBookmarksFollowTheClient pins that the backing watch asks for
130+
// bookmarks only for a WatchList request: every backing bookmark is forwarded,
131+
// and the fake client never sends one, so the event stream alone cannot show it.
132+
func TestWatch_BackingBookmarksFollowTheClient(t *testing.T) {
133+
yes := true
134+
for _, tc := range []struct {
135+
name string
136+
opts metainternalversion.ListOptions
137+
want bool
138+
}{
139+
{"plain watch defaulted by the apiserver", metainternalversion.ListOptions{SendInitialEvents: &yes}, false},
140+
{"watch list", metainternalversion.ListOptions{SendInitialEvents: &yes, AllowWatchBookmarks: true}, true},
141+
} {
142+
t.Run(tc.name, func(t *testing.T) {
143+
r := newWatchTestREST(t)
144+
spy := &registrytest.RawRecordingWatch{WithWatch: r.w}
145+
r.w = spy
146+
ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), "tenant-a"))
147+
defer cancel()
148+
149+
w, err := r.Watch(ctx, &tc.opts)
150+
if err != nil {
151+
t.Fatalf("Watch returned error: %v", err)
152+
}
153+
defer w.Stop()
154+
155+
raw := spy.RawFor(&helmv2.HelmReleaseList{})
156+
if raw == nil {
157+
t.Fatal("backing watch was called without raw options")
158+
}
159+
if raw.AllowWatchBookmarks != tc.want {
160+
t.Fatalf("backing AllowWatchBookmarks = %v, want %v", raw.AllowWatchBookmarks, tc.want)
161+
}
162+
})
163+
}
164+
}
165+
166+
// TestWatch_PlainWatch_EmitsNoInitialEventsEndBookmark covers a watch with no
167+
// resourceVersion: the apiserver defaults SendInitialEvents on for it but leaves
168+
// bookmarks off, so the client never asked for the terminating bookmark.
169+
func TestWatch_PlainWatch_EmitsNoInitialEventsEndBookmark(t *testing.T) {
170+
r := newWatchTestREST(t)
171+
ctx, cancel := context.WithCancel(request.WithNamespace(context.Background(), "tenant-a"))
172+
defer cancel()
173+
174+
sendInitialEvents := true
175+
w, err := r.Watch(ctx, &metainternalversion.ListOptions{SendInitialEvents: &sendInitialEvents})
176+
if err != nil {
177+
t.Fatalf("Watch returned error: %v", err)
178+
}
179+
defer w.Stop()
180+
driveCreateAndUpdate(ctx, t, r)
181+
182+
evs := collectWatchEvents(w, 3, 2*time.Second)
183+
var types []watch.EventType
184+
for _, ev := range evs {
185+
types = append(types, ev.Type)
186+
if ev.Type == watch.Bookmark {
187+
t.Fatalf("unrequested bookmark on a plain watch: %+v", ev.Object)
188+
}
189+
}
190+
if len(types) != 2 || types[0] != watch.Added || types[1] != watch.Modified {
191+
t.Fatalf("expected [ADDED MODIFIED], got %v", types)
192+
}
193+
}

0 commit comments

Comments
 (0)