karmada-search: Fix lock race affects watch RestChan not close, causi…

…ng client watch api to hang

Signed-off-by: xigang <wangxiang2014@gmail.com>

pkg/search/proxy/store/util.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"sort"
 	"sync"
+	"time"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -166,22 +167,47 @@ func (c *multiClusterContinue) String() string {
 	return base64.RawURLEncoding.EncodeToString(buf)
 }
 
+// nothing will ever be sent down this channel
+var neverExitWatch <-chan time.Time = make(chan time.Time)
+
+// timeoutFactory abstracts watch timeout logic for testing
+type TimeoutFactory interface {
+	TimeoutCh() (<-chan time.Time, func() bool)
+}
+
+// realTimeoutFactory implements timeoutFactory
+type realTimeoutFactory struct {
+	timeout time.Duration
+}
+
+// TimeoutCh returns a channel which will receive something when the watch times out,
+// and a cleanup function to call when this happens.
+func (w *realTimeoutFactory) TimeoutCh() (<-chan time.Time, func() bool) {
+	if w.timeout == 0 {
+		return neverExitWatch, func() bool { return false }
+	}
+	t := time.NewTimer(w.timeout)
+	return t.C, t.Stop
+}
+
 type decoratedWatcher struct {
 	watcher   watch.Interface
 	decorator func(watch.Event)
 }
 
 type watchMux struct {
-	lock    sync.RWMutex
-	sources []decoratedWatcher
-	result  chan watch.Event
-	done    chan struct{}
+	lock           sync.RWMutex
+	sources        []decoratedWatcher
+	result         chan watch.Event
+	done           chan struct{}
+	timeoutFactory TimeoutFactory
 }
 
 func newWatchMux() *watchMux {
 	return &watchMux{
-		result: make(chan watch.Event),
-		done:   make(chan struct{}),
+		result:         make(chan watch.Event, 100),
+		done:           make(chan struct{}),
+		timeoutFactory: &realTimeoutFactory{timeout: time.Second * 10},
 	}
 }
 
@@ -225,8 +251,12 @@ func (w *watchMux) Stop() {
 }
 
 func (w *watchMux) startWatchSource(source watch.Interface, decorator func(watch.Event)) {
+	defer func() { _ = recover() }()
 	defer source.Stop()
 	defer w.Stop()
+
+	timeoutCh, cleanup := w.timeoutFactory.TimeoutCh()
+	defer cleanup()
 	for {
 		var copyEvent watch.Event
 		select {
@@ -255,8 +285,9 @@ func (w *watchMux) startWatchSource(source watch.Interface, decorator func(watch
 			select {
 			case <-w.done:
 				return
-			default:
-				w.result <- copyEvent
+			case <-timeoutCh:
+				return
+			case w.result <- copyEvent:
 			}
 		}()
 	}
@@ -386,4 +417,4 @@ func BuildMultiClusterResourceVersion(clusterResourceMap map[string]string) stri
 		m.set(cluster, rv)
 	}
 	return m.String()
-}
+}