#1008 - nil object in watch event in causes api-server to report errors (#399)

* Remove error watch event when obj is nil

* Add test coverage for watcher manager

Author: kushnaidu

pkg/engine/watchermanager.go

@@ -33,6 +33,19 @@ type ObjectWatcher interface {
 	OnPackageRevisionChange(eventType watch.EventType, obj repository.PackageRevision) bool
 }
 
+func isContextError(err error) bool {
+	return err == context.Canceled || err == context.DeadlineExceeded
+}
+
+func logWatcherAction(watcher *watcher, action string, err error) string {
+	if isContextError(err) {
+		klog.V(3).Infof("watcher %p %s: %v", watcher, action, err)
+		return "debug"
+	}
+	klog.Infof("watcher %p %s: %v", watcher, action, err)
+	return "info"
+}
+
 func NewWatcherManager() *watcherManager {
 	return &watcherManager{}
 }
@@ -81,10 +94,10 @@ func (r *watcherManager) WatchPackageRevisions(ctx context.Context, filter repos
 					r.watchers[i] = w
 					inserted = true
 					active += 1
-					klog.Infof("watcher %p finished with: %v and is replaced by watcher %p", watcher, err, w)
+					logWatcherAction(watcher, "finished and is replaced by watcher", err)
 				} else {
 					r.watchers[i] = nil
-					klog.Infof("watcher %p finished with: %v and is removed", watcher, err)
+					logWatcherAction(watcher, "finished and is removed", err)
 				}
 			} else {
 				active += 1
@@ -102,7 +115,7 @@ func (r *watcherManager) WatchPackageRevisions(ctx context.Context, filter repos
 		r.watchers = append(r.watchers, w)
 	}
 
-	klog.Infof("added watcher %p; there are now %d active watchers and %d slots", w, active, len(r.watchers))
+	klog.V(3).Infof("added watcher %p; there are now %d active watchers and %d slots", w, active, len(r.watchers))
 	return nil
 }
 
@@ -117,7 +130,7 @@ func (r *watcherManager) NotifyPackageRevisionChange(eventType watch.EventType,
 			continue
 		}
 		if err := watcher.isDoneFunction(); err != nil {
-			klog.Infof("stopping watcher in response to error %v", err)
+			logWatcherAction(watcher, "stopping in response to error", err)
 			r.watchers[i] = nil
 			continue
 		}

pkg/engine/watchermanager_test.go

@@ -2,6 +2,7 @@ package engine
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/nephio-project/porch/pkg/externalrepo/fake"
@@ -88,3 +89,42 @@ func countActiveWatchers(manager *watcherManager) (int, int) {
 	}
 	return active, len(manager.watchers)
 }
+
+func TestIsContextError(t *testing.T) {
+	tests := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{"context canceled", context.Canceled, true},
+		{"context deadline", context.DeadlineExceeded, true},
+		{"regular error", errors.New("test error"), false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := isContextError(tt.err)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestLogWatcherAction(t *testing.T) {
+	tests := []struct {
+		name     string
+		err      error
+		expected string
+	}{
+		{"context canceled", context.Canceled, "debug"},
+		{"context deadline", context.DeadlineExceeded, "debug"},
+		{"regular error", errors.New("test error"), "info"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			w := &watcher{}
+			result := logWatcherAction(w, "test", tt.err)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

pkg/registry/porch/background.go

@@ -140,7 +140,7 @@ loop:
 			} else if repository, ok := event.Object.(*configapi.Repository); ok {
 				if event.Type == watch.Bookmark {
 					bookmark = repository.ResourceVersion
-					klog.Infof("Bookmark: %q", bookmark)
+					klog.V(2).Infof("Bookmark: %q", bookmark)
 				} else {
 					if err := b.updateCache(ctx, event.Type, repository); err != nil {
 						klog.Warningf("error updating cache: %v", err)
@@ -151,7 +151,7 @@ loop:
 			}
 
 		case t := <-ticker.C:
-			klog.Infof("Background task %s", t)
+			klog.V(2).Infof("Background task %s", t)
 			if err := b.runOnce(ctx); err != nil {
 				klog.Errorf("Periodic repository refresh failed: %v", err)
 			}
@@ -227,7 +227,7 @@ func (b *background) handleRepositoryEvent(ctx context.Context, repo *configapi.
 }
 
 func (b *background) runOnce(ctx context.Context) error {
-	klog.Infof("background-refreshing repositories")
+	klog.V(2).Infof("background-refreshing repositories")
 	repositories := &configapi.RepositoryList{}
 	if err := b.coreClient.List(ctx, repositories); err != nil {
 		return fmt.Errorf("error listing repository objects: %w", err)

pkg/registry/porch/watch.go

@@ -122,11 +122,12 @@ func (w *watcher) listAndWatch(ctx context.Context, r packageReader, filter repo
 
 	if err := w.listAndWatchInner(ctx, r, filter); err != nil {
 		// TODO: We need to populate the object on this error
-		klog.Warningf("sending error to watch stream: %v", err)
-		ev := watch.Event{
-			Type: watch.Error,
+		if err == context.Canceled || err == context.DeadlineExceeded {
+			klog.V(3).Infof("sending error to watch stream: %v", err)
+		} else {
+			klog.Warningf("sending error to watch stream: %v", err)
 		}
-		w.resultChan <- ev
+		// Don't send error events with nil objects
 	}
 	w.cancel()
 	close(w.resultChan)
@@ -149,6 +150,9 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
 			errorResult <- err
 			return false
 		}
+		if obj == nil {
+			return true
+		}
 
 		backlog = append(backlog, watch.Event{
 			Type:   eventType,
@@ -173,6 +177,9 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
 			w.mutex.Unlock()
 			return err
 		}
+		if obj == nil {
+			return nil
+		}
 		// TODO: Check resource version?
 		ev := watch.Event{
 			Type:   watch.Added,
@@ -217,7 +224,7 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
 		w.sendWatchEvent(ev)
 	}
 
-	klog.Infof("watch %p: moving watch into streaming mode after sentAdd %d, sentBacklog %d, sentNewBacklog %d", w, sentAdd, sentBacklog, sentNewBacklog)
+	klog.V(3).Infof("watch %p: moving watch into streaming mode after sentAdd %d, sentBacklog %d, sentNewBacklog %d", w, sentAdd, sentBacklog, sentNewBacklog)
 	w.eventCallback = func(eventType watch.EventType, pr repository.PackageRevision) bool {
 		if w.done {
 			return false
@@ -228,6 +235,9 @@ func (w *watcher) listAndWatchInner(ctx context.Context, r packageReader, filter
 			errorResult <- err
 			return false
 		}
+		if obj == nil {
+			return true
+		}
 		// TODO: Check resource version?
 		ev := watch.Event{
 			Type:   eventType,

pkg/registry/porch/watch_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/nephio-project/porch/pkg/externalrepo/fake"
 	"github.com/nephio-project/porch/pkg/repository"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
 )
 
@@ -97,3 +98,112 @@ func (f *fakePackageReader) watchPackages(ctx context.Context, filter repository
 func (f *fakePackageReader) listPackageRevisions(ctx context.Context, filter repository.ListPackageRevisionFilter, callback func(ctx context.Context, p repository.PackageRevision) error) error {
 	return nil
 }
+func TestWatcherNilObject(t *testing.T) {
+	tests := []struct {
+		name             string
+		packages         []repository.PackageRevision
+		waitForStreaming bool
+		sendEvent        bool
+	}{
+		{
+			name:      "backlog phase",
+			packages:  nil,
+			sendEvent: false,
+		},
+		{
+			name:             "streaming phase",
+			packages:         nil,
+			waitForStreaming: true,
+			sendEvent:        true,
+		},
+		{
+			name: "list phase",
+			packages: []repository.PackageRevision{
+				&fake.FakePackageRevision{
+					PackageRevision: &porchapi.PackageRevision{
+						ObjectMeta: metav1.ObjectMeta{
+							Labels: make(map[string]string),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancelFunc := context.WithCancel(context.Background())
+			defer cancelFunc()
+
+			w := &watcher{
+				cancel:     cancelFunc,
+				resultChan: make(chan watch.Event, 64),
+				extractor: func(ctx context.Context, pr repository.PackageRevision) (runtime.Object, error) {
+					return nil, nil
+				},
+			}
+
+			r := &nilCheckFakeReader{
+				packages:           tt.packages,
+				sendEventInBacklog: tt.name == "backlog phase",
+			}
+			r.Add(1)
+			var filter repository.ListPackageRevisionFilter
+			go w.listAndWatch(ctx, r, filter)
+			r.Wait()
+
+			if tt.waitForStreaming {
+				time.Sleep(100 * time.Millisecond)
+			}
+
+			if tt.sendEvent {
+				pkgRev := &fake.FakePackageRevision{
+					PackageRevision: &porchapi.PackageRevision{
+						ObjectMeta: metav1.ObjectMeta{
+							Labels: make(map[string]string),
+						},
+					},
+				}
+				cont := r.callback.OnPackageRevisionChange(watch.Modified, pkgRev)
+				if !cont {
+					t.Error("Expected callback to return true for nil object")
+				}
+			} else {
+				time.Sleep(10 * time.Millisecond)
+			}
+		})
+	}
+}
+
+type nilCheckFakeReader struct {
+	sync.WaitGroup
+	callback           engine.ObjectWatcher
+	packages           []repository.PackageRevision
+	sendEventInBacklog bool
+}
+
+func (f *nilCheckFakeReader) watchPackages(ctx context.Context, filter repository.ListPackageRevisionFilter, callback engine.ObjectWatcher) error {
+	f.callback = callback
+	if f.sendEventInBacklog {
+		// Send event synchronously in backlog phase
+		pkgRev := &fake.FakePackageRevision{
+			PackageRevision: &porchapi.PackageRevision{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: make(map[string]string),
+				},
+			},
+		}
+		callback.OnPackageRevisionChange(watch.Modified, pkgRev)
+	}
+	f.Done()
+	return nil
+}
+
+func (f *nilCheckFakeReader) listPackageRevisions(ctx context.Context, filter repository.ListPackageRevisionFilter, callback func(ctx context.Context, p repository.PackageRevision) error) error {
+	for _, pkg := range f.packages {
+		if err := callback(ctx, pkg); err != nil {
+			return err
+		}
+	}
+	return nil
+}