Apiserver 问题排查: too old resource version

func (w *watchCache) GetAllEventsSinceThreadUnsafe(resourceVersion uint64) ([]*watchCacheEvent, error) {
	size := w.endIndex - w.startIndex
	// 寻找最旧的 rv，最旧的 rv 也就是最小的 rv。
	var oldest uint64
	switch {
	case size >= w.capacity:
		oldest = w.cache[w.startIndex%w.capacity].ResourceVersion
	case w.listResourceVersion > 0:
		oldest = w.listResourceVersion + 1
	case size > 0:
		oldest = w.cache[w.startIndex%w.capacity].ResourceVersion
	default:
		return nil, fmt.Errorf("watch cache isn't correctly initialized")
	}
	// 携带的 rv 比当前最小的 rv 还小，就报错。
	if resourceVersion < oldest-1 {
		return nil, errors.NewGone(fmt.Sprintf("too old resource version: %d (%d)", resourceVersion, oldest-1))
	}
	// 省去其他代码...
	return result, nil
}

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	var resourceVersion string
	// 初始的 list 请求，optoin 里只设置了 rv 选项
	options := metav1.ListOptions{ResourceVersion: "0"}
	// 省去不相关代码...
	if err := func() error {
		go func() {
			pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return r.listerWatcher.List(opts)
			}))
			if r.WatchListPageSize != 0 {
				pager.PageSize = r.WatchListPageSize
			}
			// 全量 list
			list, err = pager.List(context.Background(), options)
		}()
		// 将全量 list 的 resourceVersion 修改为最新的 rv
		resourceVersion = listMetaInterface.GetResourceVersion()
		// 省去不相关代码...
	}(); err != nil {
		return err
	}

	// 下面是 watch 请求，放在一个无限 for 循环里
	for {
		// 这里的 minWatchTimeout hardcode 定义为 5 * time.Minute
		timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
		options = metav1.ListOptions{
			ResourceVersion: resourceVersion,
			// 这里的时间是 [5, 10) 分钟的一个随机数
			TimeoutSeconds: &timeoutSeconds,
			// 开启 bookmark 机制
			AllowWatchBookmarks: true,
		}
		// 发送 watch 请求
		w, err := r.listerWatcher.Watch(options)
		if err != nil {
			switch err {
			case io.EOF:
			case io.ErrUnexpectedEOF:
				klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedType, err)
			default:
				utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedType, err))
			}
			// 如果是 ConnectRefused 错误，并不 return，使用 continue 继续下一轮循环，也就是重新发送 watch 请求
			// 注意这个是一个 watch 重新发送的 case 1，即不发送 list 请求，但是发送 watch 请求
			if utilnet.IsConnectionRefused(err) {
				time.Sleep(time.Second)
				continue
			}
			return nil
		}

		// 下面是处理 watch 请求了，我们看到 return 的情况是出现错误的情况
		// 这里有个陷阱是，这个 watchHandler 可能会无疾而终，不出现错误而退出
		if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {
			if err != errorStopRequested {
				switch {
				case apierrs.IsResourceExpired(err):
					klog.V(4).Infof("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
				default:
					klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
				}
			}
			return nil
		}
	}
}

// 监听事件
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case err := <-errc:
			return err
		case event, ok := <-w.ResultChan():
			// 这个 case 我们需要注意一下，因为这个 ok 可能确实不 ok，在不 ok 的情况下，这个 break 直接跳出了 for 循环
			// 在超时情况下，这个 ResultChan 返回给我们的，是一个已经关闭的 channel，这个时候就不 ok 了。
			if !ok {
				break loop
			}
			newResourceVersion := meta.GetResourceVersion()
			switch event.Type {
			// 下面是处理事件，不是我们考虑的重点，省去代码...
			case watch.Added:
			}
			// 更新 rv
			*resourceVersion = newResourceVersion
			r.setLastSyncResourceVersion(newResourceVersion)
		}
	}
	// break 跳出 for 循环后，就通过 return nil 的形式返回了。
	return nil
}

func NewEmptyWatch() Interface {
	ch := make(chan Event)
	close(ch)
	return emptyWatch(ch)
}

// 位于文件：staging/src/k8s.io/client-go/rest/request.go

// Timeout makes the request use the given duration as an overall timeout for the
// request. Additionally, if set passes the value as "timeout" parameter in URL.
func (r *Request) Timeout(d time.Duration) *Request {
	if r.err != nil {
		return r
	}
	r.timeout = d
	return r
}