Fix fialed to watch context canceled error when streaming logs

When streaming pipelineRun or taskRun logs, closing the stop channel
to terminate the informer was causing "Failed to watch" error messages
to be logged with "context canceled".

This patch add custom watchErrorHandler function that filter out context.Canceled
errors while passing other errors to the default handler. This prevents the error
log messages when the CLI intentionally stops watching after a run completes.

Signed-off-by: Shiv Verma <[email protected]>

Author: pratap0007

pkg/pipelinerun/tracker.go

@@ -16,6 +16,7 @@ package pipelinerun
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
@@ -28,7 +29,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/client-go/tools/cache"
+	k8scache "k8s.io/client-go/tools/cache"
 )
 
 // Tracker tracks the progress of a PipelineRun
@@ -68,6 +69,11 @@ func (t *Tracker) Monitor(allowed []string) <-chan []taskrunpkg.Run {
 
 	genericInformer, _ := factory.ForResource(*gvr)
 	informer := genericInformer.Informer()
+
+	// Set a custom watch error handler that ignores context.Canceled errors
+	// to prevent "Failed to watch" log messages when the informer is stopped intentionally
+	_ = informer.SetWatchErrorHandlerWithContext(watchErrorHandler)
+
 	mu := &sync.Mutex{}
 	stopC := make(chan struct{})
 	trC := make(chan []taskrunpkg.Run)
@@ -105,7 +111,7 @@ func (t *Tracker) Monitor(allowed []string) <-chan []taskrunpkg.Run {
 	}
 
 	_, err := informer.AddEventHandler(
-		cache.ResourceEventHandlerFuncs{
+		k8scache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				// To ensure synchonization and checks is the stopC channel has received a signal to stop
 				// If it receives a signal then return and does nothing
@@ -154,7 +160,15 @@ func pipelinerunOpts(name string) func(opts *metav1.ListOptions) {
 	return func(opts *metav1.ListOptions) {
 		opts.FieldSelector = fields.OneTermEqualSelector("metadata.name", name).String()
 	}
+}
 
+// watchErrorHandler is a custom watch error handler that filters out context.Canceled errors
+// to prevent "Failed to watch" log messages when the informer is stopped intentionally.
+// Other errors are passed to the default handler.
+func watchErrorHandler(ctx context.Context, r *k8scache.Reflector, err error) {
+	if !errors.Is(err, context.Canceled) {
+		k8scache.DefaultWatchErrorHandler(ctx, r, err)
+	}
 }
 
 // handles changes to pipelinerun and pushes the Run information to the

pkg/pipelinerun/tracker_test.go

@@ -15,6 +15,8 @@
 package pipelinerun
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -239,3 +241,28 @@ func startPipelineRun(t *testing.T, data pipelinetest.Data, prStatus ...v1.Pipel
 		Dynamic: dynamic,
 	}
 }
+
+func TestTracker_watchErrorHandler(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{
+			name: "context.Canceled should be filtered",
+			err:  context.Canceled,
+		},
+		{
+			name: "wrapped context.Canceled should be filtered",
+			err:  errors.Join(errors.New("watch failed"), context.Canceled),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(_ *testing.T) {
+			// Call watchErrorHandler with context.Canceled errors
+			// These should be filtered (not passed to DefaultWatchErrorHandler)
+			// so passing nil reflector is safe
+			watchErrorHandler(context.Background(), nil, tt.err)
+		})
+	}
+}

pkg/pods/pod.go

@@ -16,6 +16,7 @@ package pods
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"sync"
@@ -115,7 +116,12 @@ func (p *Pod) watcher(stopC chan struct{}, result *podResult, mu *sync.Mutex) {
 		}
 	}
 
-	_, err := factory.Core().V1().Pods().Informer().AddEventHandler(
+	informer := factory.Core().V1().Pods().Informer()
+	// Set a custom watch error handler that ignores context.Canceled errors
+	// to prevent "Failed to watch" log messages when the informer is stopped intentionally
+	_ = informer.SetWatchErrorHandlerWithContext(watchErrorHandler)
+
+	_, err := informer.AddEventHandler(
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				select {
@@ -156,6 +162,15 @@ func podOpts(name string) func(opts *metav1.ListOptions) {
 	}
 }
 
+// watchErrorHandler is a custom watch error handler that filters out context.Canceled errors
+// to prevent "Failed to watch" log messages when the informer is stopped intentionally.
+// Other errors are passed to the default handler.
+func watchErrorHandler(ctx context.Context, r *cache.Reflector, err error) {
+	if !errors.Is(err, context.Canceled) {
+		cache.DefaultWatchErrorHandler(ctx, r, err)
+	}
+}
+
 func checkPodStatus(obj interface{}) (*corev1.Pod, error) {
 	pod, ok := obj.(*corev1.Pod)
 	if !ok {

pkg/pods/pod_test.go

@@ -15,6 +15,8 @@
 package pods
 
 import (
+	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -212,3 +214,28 @@ func simulateDeleteWatch(t *testing.T, initial *corev1.Pod, later *corev1.Pod) k
 
 	return clients.Kube
 }
+
+func Test_watchErrorHandler(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+	}{
+		{
+			name: "context.Canceled should be filtered",
+			err:  context.Canceled,
+		},
+		{
+			name: "wrapped context.Canceled should be filtered",
+			err:  errors.Join(errors.New("watch failed"), context.Canceled),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(_ *testing.T) {
+			// Call watchErrorHandler with context.Canceled errors
+			// These should be filtered (not passed to DefaultWatchErrorHandler)
+			// so passing nil reflector is safe
+			watchErrorHandler(context.Background(), nil, tt.err)
+		})
+	}
+}