Fix fialed to watch context canceled error when streaming logs

pkg/pipelinerun/tracker.go

@@ -16,6 +16,7 @@ package pipelinerun
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 
@@ -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)
@@ -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 *cache.Reflector, err error) {
+	if !errors.Is(err, context.Canceled) {
+		cache.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)
+		})
+	}
+}