fix(continuousscanning): resume watches from the last version and stop on error events

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,

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

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

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 = ""
 	}

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