client-go watch接口隔一段时间自动退出怎么办？

1pod, err := mycluster.Clusterclientset.CoreV1().Pods(appNamespace).Watch(context.TODO(), metav1.ListOptions{LabelSelector: label})
2if err != nil {
3    log.Error(err)
4}
5...
6event, ok := <-pod.ResultChan()
7if !ok {
8    log.Error(err)
9}

 1type StreamWatcher struct {
 2	sync.Mutex
 3	source   Decoder
 4	reporter Reporter
 5	result   chan Event
 6	stopped  bool
 7}
 8...
 9func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
10	sw := &StreamWatcher{
11		source:   d,
12		reporter: r,
13		// It's easy for a consumer to add buffering via an extra
14		// goroutine/channel, but impossible for them to remove it,
15		// so nonbuffered is better.
16		result: make(chan Event),
17	}
18	go sw.receive()
19	return sw
20}
21...
22func (sw *StreamWatcher) receive() {
23	defer close(sw.result)
24	defer sw.Stop()
25	defer utilruntime.HandleCrash()
26	for {
27		action, obj, err := sw.source.Decode()
28		if err != nil {
29			// Ignore expected error.
30			if sw.stopping() {
31				return
32			}
33			switch err {
34			case io.EOF:
35				// watch closed normally
36			case io.ErrUnexpectedEOF:
37				klog.V(1).Infof("Unexpected EOF during watch stream event decoding: %v", err)
38			default:
39				if net.IsProbableEOF(err) || net.IsTimeout(err) {
40					klog.V(5).Infof("Unable to decode an event from the watch stream: %v", err)
41				} else {
42					sw.result <- Event{
43						Type:   Error,
44						Object: sw.reporter.AsObject(fmt.Errorf("unable to decode an event from the watch stream: %v", err)),
45					}
46				}
47			}
48			return
49		}
50		sw.result <- Event{
51			Type:   action,
52			Object: obj,
53		}
54	}
55}

 1for {
 2	pod, err := mycluster.Clusterclientset.CoreV1().Pods(appNamespace).Watch(context.TODO(), metav1.ListOptions{LabelSelector: label})
 3	if err != nil {
 4		log.Error(err)
 5	}
 6loopier:
 7	for {
 8		event, ok := <-pod.ResultChan()
 9		if !ok {
10			time.Sleep(time.Second * 5)
11			log.Info("Restarting watcher...")
12			break loopier
13		}
14		// your process logic
15	}
16}

 1type RetryWatcher struct {
 2	lastResourceVersion string
 3	watcherClient       cache.Watcher
 4	resultChan          chan watch.Event
 5	stopChan            chan struct{}
 6	doneChan            chan struct{}
 7	minRestartDelay     time.Duration
 8}
 9...
10func newRetryWatcher(initialResourceVersion string, watcherClient cache.Watcher, minRestartDelay time.Duration) (*RetryWatcher, error) {
11	switch initialResourceVersion {
12	case "", "0":
13		// TODO: revisit this if we ever get WATCH v2 where it means start "now"
14		//       without doing the synthetic list of objects at the beginning (see #74022)
15		return nil, fmt.Errorf("initial RV %q is not supported due to issues with underlying WATCH", initialResourceVersion)
16	default:
17		break
18	}
19
20	rw := &RetryWatcher{
21		lastResourceVersion: initialResourceVersion,
22		watcherClient:       watcherClient,
23		stopChan:            make(chan struct{}),
24		doneChan:            make(chan struct{}),
25		resultChan:          make(chan watch.Event, 0),
26		minRestartDelay:     minRestartDelay,
27	}
28
29	go rw.receive()
30	return rw, nil
31}

 1func (rw *RetryWatcher) receive() {
 2	defer close(rw.doneChan)
 3	defer close(rw.resultChan)
 4
 5	klog.V(4).Info("Starting RetryWatcher.")
 6	defer klog.V(4).Info("Stopping RetryWatcher.")
 7
 8	ctx, cancel := context.WithCancel(context.Background())
 9	defer cancel()
10	go func() {
11		select {
12		case <-rw.stopChan:
13			cancel()
14			return
15		case <-ctx.Done():
16			return
17		}
18	}()
19
20	// We use non sliding until so we don't introduce delays on happy path when WATCH call
21	// timeouts or gets closed and we need to reestablish it while also avoiding hot loops.
22	wait.NonSlidingUntilWithContext(ctx, func(ctx context.Context) {
23		done, retryAfter := rw.doReceive()
24		if done {
25			cancel()
26			return
27		}
28
29		time.Sleep(retryAfter)
30
31		klog.V(4).Infof("Restarting RetryWatcher at RV=%q", rw.lastResourceVersion)
32	}, rw.minRestartDelay)
33}
34...
35func (rw *RetryWatcher) Stop() {
36	close(rw.stopChan)
37}

 1func (rw *RetryWatcher) doReceive() (bool, time.Duration) {
 2	watcher, err := rw.watcherClient.Watch(metav1.ListOptions{
 3		ResourceVersion:     rw.lastResourceVersion,
 4		AllowWatchBookmarks: true,
 5	})
 6	// We are very unlikely to hit EOF here since we are just establishing the call,
 7	// but it may happen that the apiserver is just shutting down (e.g. being restarted)
 8	// This is consistent with how it is handled for informers
 9	switch err {
10...
11// 省略watch的一些错误处理，都会返回false，也就是继续wait重试
12...
13	}
14
15	if watcher == nil {
16		klog.Error("Watch returned nil watcher")
17		// Retry
18		return false, 0
19	}
20
21	ch := watcher.ResultChan()
22	defer watcher.Stop()
23
24	for {
25		select {
26		case <-rw.stopChan:
27			klog.V(4).Info("Stopping RetryWatcher.")
28			return true, 0
29		case event, ok := <-ch:
30			if !ok {
31				klog.V(4).Infof("Failed to get event! Re-creating the watcher. Last RV: %s", rw.lastResourceVersion)
32				return false, 0
33			}
34
35			// We need to inspect the event and get ResourceVersion out of it
36			switch event.Type {
37			case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark:
38				metaObject, ok := event.Object.(resourceVersionGetter)
39				...
40				resourceVersion := metaObject.GetResourceVersion()
41				...
42        // All is fine; send the non-bookmark events and update resource version.
43				if event.Type != watch.Bookmark {
44					ok = rw.send(event)
45					if !ok {
46						return true, 0
47					}
48				}
49				rw.lastResourceVersion = resourceVersion
50
51				continue
52
53			case watch.Error:
54			...
55			}
56		}
57	}
58}
59...
60func (rw *RetryWatcher) send(event watch.Event) bool {
61	// Writing to an unbuffered channel is blocking operation
62	// and we need to check if stop wasn't requested while doing so.
63	select {
64	case rw.resultChan <- event:
65		return true
66	case <-rw.stopChan:
67		return false
68	}
69}

1type Watcher interface {
2	// Watch should begin a watch at the specified version.
3	Watch(options metav1.ListOptions) (watch.Interface, error)
4}

1type PodInterface interface {
2	...
3	Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)
4	...
5}

1type PodListWatch struct {
2	MyCluster    *initk8s.MyCluster
3	AppNamespace string
4}
5// watch指定命名空间下的Pod例子 
6func (p PodListWatch) Watch(options metav1.ListOptions) (watch.Interface, error) {
7	return p.MyCluster.Clusterclientset.CoreV1().Pods(p.AppNamespace).Watch(context.TODO(), options)
8}

 1func ResourceWatcher(mycluster *initk8s.MyCluster, appNamespace string) {
 2	podWatcher := PodListWatch{MyCluster: mycluster, AppNamespace: appNamespace}
 3	// generate new retrywatcher, use podRetryWatcher.ResultChan() to keep a chronic watch operation
 4	podRetryWatcher, err := retrywatcher.NewRetryWatcher(label, podWatcher)
 5	if err != nil {
 6		log.Error(err)
 7	}
 8FORWATCHER:
 9	for {
10		select {
11		case <-podRetryWatcher.Done():
12			log.Info("Retrywatcher is exiting, please check...")
13			break FORWATCHER
14		case event, ok := <-podRetryWatcher.ResultChan():
15			if !ok {
16				log.Warn("Retrywatcher is not open, please check...")
17				continue
18			}
19			...