Add timeout for watch requests to member clusters to prevent request …

…hang

Signed-off-by: xigang <[email protected]>

pkg/search/proxy/store/multi_cluster_cache.go

@@ -327,25 +327,65 @@ func (c *MultiClusterCache) Watch(ctx context.Context, gvr schema.GroupVersionRe
 
 	mux := newWatchMux()
 	clusters := c.getClusterNames()
+
+	var wg sync.WaitGroup
+	errChan := make(chan error, len(clusters))
+
 	for i := range clusters {
 		cluster := clusters[i]
+		options := options.DeepCopy
 		options.ResourceVersion = resourceVersion.get(cluster)
 		cache := c.cacheForClusterResource(cluster, gvr)
 		if cache == nil {
 			continue
 		}
-		w, err := cache.Watch(ctx, options)
-		if err != nil {
-			return nil, err
-		}
 
-		mux.AddSource(w, func(e watch.Event) {
-			setObjectResourceVersionFunc(cluster, e.Object)
-			addCacheSourceAnnotation(e.Object, cluster)
-		})
+		wg.Add(1)
+		go func(cluster string, cache *resourceCache, options *metainternalversion.ListOptions) {
+			defer wg.Done()
+
+			watchChan := make(chan watch.Interface, 1)
+			watchErrChan := make(chan error, 1)
+
+			go func(cluster string, cache *resourceCache, options *metainternalversion.ListOptions) {
+				w, err := cache.Watch(ctx, options)
+				if err != nil {
+					watchErrChan <- fmt.Errorf("failed to start watch for resource %v in cluster %q: %v", gvr.String(), cluster, err)
+					return
+				}
+
+				select {
+					case watchChan <- w:
+					case <-ctx.Done():
+						w.Stop()
+				}
+			}(cluster, cache, options)
+
+			select {
+			case w := <-watchChan:
+				mux.AddSource(w, func(e watch.Event) {
+					setObjectResourceVersionFunc(cluster, e.Object)
+					addCacheSourceAnnotation(e.Object, cluster)
+				})
+			case err := <-watchErrChan:
+				errChan <- err
+			case <-time.After(30 * time.Second):
+				errChan <- fmt.Errorf("timeout waiting for watch for resource %v in cluster %q", gvr.String(), cluster)
+			case <-ctx.Done():
+				return
+			}
+		}(cluster, cache, options)
+	}
+
+	wg.Wait()
+
+	select {
+	case err := <-errChan:
+		return nil, err
+	default:
+		mux.Start()
+		return mux, nil
 	}
-	mux.Start()
-	return mux, nil
 }
 
 func (c *MultiClusterCache) getClusterNames() []string {