Go clientSet Watch 运行后随机性失效

背景

List 和 Watch 机制是 kubernetes 中重要的机制之一。控制器通过 API Server 的 List API 来获取所有最新版本的 API 对象，通过 Watch API 来监听所有 API 对象的变化。
在程序设计过程中，往往也需要利用 List && Watch 机制，来观察 API 对象的状态，从而调用 EventHandler，做出响应。
基于此背景，Go 语言官方的 clientSet 中提供了相应的 API 接口供开发者使用。然而，笔者在使用 Watch 机制中踩到了不小坑。

问题

笔者在程序中创建 clientSet，并调用其 Watch 方法，监听某 Secret 资源变化，伪代码如下：

secretWatch, err := clientSet.CoreV1().Secrets("命名空间").Watch(context.TODO(), 
    metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("metadata.name", "API 对象名").String()})
for {
    for range secretWatch.ResultChan() {
        // 响应操作    
    }
}

笔者在启动后，经过几番调试确实可以监听到信息，安心提交。
然而，经过一段运行时间后，Watch 机制突然失灵，而且无法恢复。本地 DeBug 也始终找不到异常点。

解决方案

问题分析

clientSet 的 Watch 接口缺失能够监听了需要的信息，然而其难以处理各类异常。如下源码所示，因此当异常发生时，Watch 会自动关闭。

// Interface can be implemented by anything that knows how to watch and report changes.
type Interface interface {
    // Stops watching. Will close the channel returned by ResultChan(). Releases
    // any resources used by the watch.
    Stop()

    // Returns a chan which will receive all the events. If an error occurs
    // or Stop() is called, this channel will be closed, in which case the
    // watch should be completely cleaned up.  ！！！明确说了在出现错误或者被调用Stop时，通道会自动关闭的
    ResultChan() <-chan Event
}

解决方案

使用

使用 RetryWatcher 来替换传统 Watcher，简单来说，此类 RetryWatcher 能够保证 Watch 自动关闭后，重新拉起一个 Watcher，使得程序继续正常运行。
使用方法如下：

secretWatch, err := watchtools.NewRetryWatcher("资源版本", &cache.ListWatch{
    WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
        return clientSet.CoreV1().Secrets("命名空间").Watch(context.TODO(), metav1.ListOptions{
            FieldSelector: fields.OneTermEqualSelector("metadata.name", "API 对象名称").String(),
        })
    },
})
for {
    for range secretWatch.ResultChan() {
        // 响应操作    
    }
}

代码解读

了解使用后，笔者带着大家来深究一下源码：
首先来看看 newRetryWatcher 函数（NewRetryWatcher 接口会调用该内部函数），笔者发现其中启动了一个 rw.receive() 的协程。receive() 的官方解释是“reads the result from a watcher, restarting it if necessary.”，表示除了正常 Watch 之外，在必要时重启该 Watch。

func newRetryWatcher(initialResourceVersion string, watcherClient cache.Watcher, minRestartDelay time.Duration) (*RetryWatcher, error) {
    switch initialResourceVersion {
    case "", "0":
        // TODO: revisit this if we ever get WATCH v2 where it means start "now"
        //       without doing the synthetic list of objects at the beginning (see #74022)
        return nil, fmt.Errorf("initial RV %q is not supported due to issues with underlying WATCH", initialResourceVersion)
    default:
        break
    }

    rw := &RetryWatcher{
        lastResourceVersion: initialResourceVersion,
        watcherClient:       watcherClient,
        stopChan:            make(chan struct{}),
        doneChan:            make(chan struct{}),
        resultChan:          make(chan watch.Event, 0),
        minRestartDelay:     minRestartDelay,
    }

    go rw.receive()
    return rw, nil
}

更深层次的代码需要了解 doReceive() 函数（如下代码）。若正常结束，doReceive() 返回 true，上层 receive 函数会调用 cancel 退出程序。若不正常结束，则返回 false，receive 会调用 NonSlidingUntilWithContext 重建 Watcher 继续监听。

func (rw *RetryWatcher) receive() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go func() {
        select {
        case <-rw.stopChan:
            cancel()
            return
        case <-ctx.Done():
            return
        }
    }()

    // We use non sliding until so we don't introduce delays on happy path when WATCH call
    // timeouts or gets closed and we need to reestablish it while also avoiding hot loops.
    wait.NonSlidingUntilWithContext(ctx, func(ctx context.Context) {
        done, retryAfter := rw.doReceive()
        if done {
            cancel()
            return
        }

        time.Sleep(retryAfter)

        klog.V(4).Infof("Restarting RetryWatcher at RV=%q", rw.lastResourceVersion)
    }, rw.minRestartDelay)
}

func (rw *RetryWatcher) doReceive() (bool, time.Duration) {
    // ...
}

为了降低阅读难度，笔者将 doReceive() 重要的部分拆解。

• 首先：因为 RetryWatch 最终还是调用的传统 Watch 函数，因此，先捕捉获取 Watch 中出现的 error，若发现则返回 false；

watcher, err := rw.watcherClient.Watch(metav1.ListOptions{
        ResourceVersion:     rw.lastResourceVersion,
        AllowWatchBookmarks: true,
    })
    switch err {
    case nil:
        break

    case io.EOF:
        // ...
    // ...
    }

• 其次：在获取消息时，同样可能会出现异常，返回 false。

for {
        select {
        case <-rw.stopChan:
            // ...
        case event, ok := <-ch:
            if !ok {
                // ...
            }
            switch event.Type {
            case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark:
                // ...
            case watch.Error:
                // ...
            default:
                // ...
            }
        }
    }