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) }) }