From 3759d817bf4b49ceb41c22eacbfe458a6ea34928 Mon Sep 17 00:00:00 2001 From: itsvishalyadav Date: Tue, 10 Nov 2026 10:43:36 +0530 Subject: [PATCH] fix(continuousscanning): resume watches from the last version and stop on error events SelfHealingWatch reconnected with an empty resourceVersion, so after every watch timeout the API server replayed every object as a synthetic ADDED event and continuous scanning triggered a scan for each of them again. A watch error event was also forwarded, and its *metav1.Status object then failed the e.Object.(metav1.Object) assertion in the service and panicked. Keep the version of the last object received, forget it when the server says it has expired, and end the watch on an error event instead of forwarding it. This reuses the helpers the watcher package got in #439, exported as ResumeFrom and DropGoneVersion. Co-Authored-By: Claude Opus 5.5 Signed-off-by: itsvishalyadav --- continuousscanning/watchbuilder.go | 16 ++++- continuousscanning/watchbuilder_test.go | 91 +++++++++++++++++++++++++ watcher/commandswatcher.go | 4 +- watcher/watchretry.go | 14 ++-- watcher/watchretry_test.go | 4 +- 5 files changed, 115 insertions(+), 14 deletions(-) diff --git a/continuousscanning/watchbuilder.go b/continuousscanning/watchbuilder.go index b9b310ee..845830d0 100644 --- a/continuousscanning/watchbuilder.go +++ b/continuousscanning/watchbuilder.go @@ -7,6 +7,8 @@ import ( "github.com/kubescape/go-logger" "github.com/kubescape/go-logger/helpers" "github.com/kubescape/k8s-interface/k8sinterface" + "github.com/kubescape/operator/watcher" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/watch" @@ -53,11 +55,18 @@ func (w *SelfHealingWatch) RunUntilWatchCloses(ctx context.Context, out chan<- w watchEvents := w.currWatch.ResultChan() select { case event, ok := <-watchEvents: - if ok { - out <- event - } else { + if !ok { return nil } + // Without a version the next watch replays every object as ADDED. + watcher.ResumeFrom(&w.opts, event) + if event.Type == watch.Error { + // The event carries a Status, not an object, and the watch is over. + logger.L().Ctx(ctx).Warning("watch failed", helpers.String("gvr", w.gvr.String()), helpers.Error(apierrors.FromObject(event.Object))) + w.currWatch.Stop() + return nil + } + out <- event case <-ctx.Done(): return ctx.Err() } @@ -76,6 +85,7 @@ func (w *SelfHealingWatch) Run(ctx context.Context, readyWg *sync.WaitGroup, out logger.L().Debug("creating watch for GVR", gvr) watchFunc, err := w.makeWatchFunc(ctx, w.client, w.gvr, w.namespace, w.opts) if err != nil { + watcher.DropGoneVersion(&w.opts, err) logger.L().Ctx(ctx).Warning( "got error when creating a watch for gvr", gvr, diff --git a/continuousscanning/watchbuilder_test.go b/continuousscanning/watchbuilder_test.go index 359ac0f6..23bf7107 100644 --- a/continuousscanning/watchbuilder_test.go +++ b/continuousscanning/watchbuilder_test.go @@ -2,12 +2,18 @@ package continuousscanning import ( "context" + "errors" + "net/http" + "sync" "testing" "github.com/stretchr/testify/assert" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/dynamic" dynamicfake "k8s.io/client-go/dynamic/fake" ktest "k8s.io/client-go/testing" ) @@ -179,3 +185,88 @@ func TestNewWatchPoolClusterScopedGVRs(t *testing.T) { }) } } + +// TestSelfHealingWatchReconnect checks what an ended watch leaves for the next one. The +// next watch has to carry on from the last object: started without a version, the server +// replays every object as ADDED and each one is scanned again. +func TestSelfHealingWatchReconnect(t *testing.T) { + object := func(version string) *unstructured.Unstructured { + o := &unstructured.Unstructured{} + o.SetKind("Deployment") + o.SetNamespace("default") + o.SetName("app") + o.SetResourceVersion(version) + return o + } + failure := func(code int32, reason metav1.StatusReason) *metav1.Status { + return &metav1.Status{Status: metav1.StatusFailure, Code: code, Reason: reason} + } + + tt := []struct { + name string + events []watch.Event + wantResume string + wantForwarded int + }{ + { + name: "resumes from the last object", + events: []watch.Event{ + {Type: watch.Added, Object: object("7")}, + {Type: watch.Modified, Object: object("8")}, + }, + wantResume: "8", + wantForwarded: 2, + }, + { + name: "an error event is not forwarded and keeps the version", + events: []watch.Event{ + {Type: watch.Added, Object: object("7")}, + {Type: watch.Error, Object: failure(http.StatusInternalServerError, metav1.StatusReasonInternalError)}, + }, + wantResume: "7", + wantForwarded: 1, + }, + { + name: "an expired version is forgotten", + events: []watch.Event{ + {Type: watch.Added, Object: object("7")}, + {Type: watch.Error, Object: failure(http.StatusGone, metav1.StatusReasonExpired)}, + }, + wantResume: "", + wantForwarded: 1, + }, + } + + for _, tc := range tt { + t.Run(tc.name, func(t *testing.T) { + first := watch.NewFakeWithChanSize(len(tc.events), false) + for _, e := range tc.events { + first.Action(e.Type, e.Object) + } + first.Stop() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var calls []metav1.ListOptions + w := NewSelfHealingWatch(nil, schema.GroupVersionResource{}, "", metav1.ListOptions{}) + w.makeWatchFunc = func(_ context.Context, _ dynamic.Interface, _ schema.GroupVersionResource, _ string, opts metav1.ListOptions) (watch.Interface, error) { + calls = append(calls, opts) + if len(calls) == 1 { + return first, nil + } + cancel() // the reconnect is what this test is after + return nil, errors.New("stop") + } + + ready := &sync.WaitGroup{} + ready.Add(1) + out := make(chan watch.Event, len(tc.events)) + assert.ErrorIs(t, w.Run(ctx, ready, out), context.Canceled) + + assert.Len(t, calls, 2) + assert.Equal(t, tc.wantResume, calls[1].ResourceVersion) + assert.Len(t, out, tc.wantForwarded) + }) + } +} diff --git a/watcher/commandswatcher.go b/watcher/commandswatcher.go index c49249d7..80ccb050 100644 --- a/watcher/commandswatcher.go +++ b/watcher/commandswatcher.go @@ -83,12 +83,12 @@ func (cwh *CommandWatchHandler) watchRetry(ctx context.Context) { if err := backoff.RetryNotify(func() error { watcher, err := cwh.k8sAPI.GetDynamicClient().Resource(v1alpha1.SchemaGroupVersionResource).Namespace(cwh.config.Namespace()).Watch(context.Background(), watchOpts) if err != nil { - dropGoneVersion(&watchOpts, err) + DropGoneVersion(&watchOpts, err) return fmt.Errorf("failed to get commands watcher: %w", err) } for { event, chanActive := <-watcher.ResultChan() - resumeFrom(&watchOpts, event) + ResumeFrom(&watchOpts, event) if cwh.eventQueue.Closed() { watcher.Stop() return backoff.Permanent(errors.New("event queue closed")) diff --git a/watcher/watchretry.go b/watcher/watchretry.go index 315e33b2..4a685e5a 100644 --- a/watcher/watchretry.go +++ b/watcher/watchretry.go @@ -25,12 +25,12 @@ func (wh *WatchHandler) watchRetry(ctx context.Context, watchOpts metav1.ListOpt if err := backoff.RetryNotify(func() error { watcher, err := wh.k8sAPI.KubernetesClient.CoreV1().Pods("").Watch(context.Background(), watchOpts) if err != nil { - dropGoneVersion(&watchOpts, err) + DropGoneVersion(&watchOpts, err) return fmt.Errorf("client resource: %w", err) } for { event, chanActive := <-watcher.ResultChan() - resumeFrom(&watchOpts, event) + ResumeFrom(&watchOpts, event) if wh.eventQueue.Closed() { watcher.Stop() return backoff.Permanent(errors.New("event queue closed")) @@ -60,26 +60,26 @@ func (wh *WatchHandler) watchRetry(ctx context.Context, watchOpts metav1.ListOpt } } -// resumeFrom records where the next watch resumes, the way client-go's RetryWatcher does: +// ResumeFrom records where the next watch resumes, the way client-go's RetryWatcher does: // at the version of the last object received, so a reconnect carries on from there, or // from the current state when the server says that version is gone, since nothing else // can follow it. An error event carries a Status, not an object; its empty version is not // a place to resume, and taking it for one restarted the watch from scratch after any // error, replaying every object. -func resumeFrom(opts *metav1.ListOptions, event watch.Event) { +func ResumeFrom(opts *metav1.ListOptions, event watch.Event) { switch event.Type { case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark: if metaObject, ok := event.Object.(resourceVersionGetter); ok { opts.ResourceVersion = metaObject.GetResourceVersion() } case watch.Error: - dropGoneVersion(opts, apierrors.FromObject(event.Object)) + DropGoneVersion(opts, apierrors.FromObject(event.Object)) } } -// dropGoneVersion forgets the resume version when err says the server no longer has it, +// DropGoneVersion forgets the resume version when err says the server no longer has it, // whether the server said so in an event or by refusing the watch request itself. -func dropGoneVersion(opts *metav1.ListOptions, err error) { +func DropGoneVersion(opts *metav1.ListOptions, err error) { if apierrors.IsResourceExpired(err) || apierrors.IsGone(err) { opts.ResourceVersion = "" } diff --git a/watcher/watchretry_test.go b/watcher/watchretry_test.go index 611edb3f..409551bf 100644 --- a/watcher/watchretry_test.go +++ b/watcher/watchretry_test.go @@ -51,7 +51,7 @@ func TestResumeFromTakesTheVersionFromObjectsOnly(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { opts := metav1.ListOptions{ResourceVersion: "3"} - resumeFrom(&opts, tt.event) + ResumeFrom(&opts, tt.event) assert.Equal(t, tt.want, opts.ResourceVersion) }) } @@ -72,7 +72,7 @@ func TestDropGoneVersionOnlyForAGoneVersion(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { opts := metav1.ListOptions{ResourceVersion: "3"} - dropGoneVersion(&opts, tt.err) + DropGoneVersion(&opts, tt.err) assert.Equal(t, tt.want, opts.ResourceVersion) }) }