K8s-kube-apiserver(List-Watch实现原理)

基于1.25

List-Watch是K8s的核心机制，基于List-Watch实现了资源变化的感知

• List调用ListAPI列出所有资源，基于HTTP短链实现
• Watch调用WatchAPI监听资源变更事件，基于HTTP长连接事件

长连接通信协议

支持Watch操作碧玺实现客户端和服务端的长连接。

kube-apiserver的Watch支持3种类型的长连接

• HTTP/1.1 Chunked Transfer Encoding
• HTTP/2
• WebSocket

HTTP/1.1 Chunked Transfer Encoding

HTTP/1.1 Chunked Transfer Encoding（分块传输编码）是HTTP中数据传输的一种机制，允许吧服务端发送给客户端的数据分为多个部分，分批次发送

# 验证支持HTTP/1.1 Chunked
curl -ik --http1.1 \
--cert /etc/kubernetes/pki/apiserver-kubelet-client.crt \
--key /etc/kubernetes/pki/apiserver-kubelet-cliet.key \
https://127.0.0.1:6443/api/v1/pods?watch=true

HTTP/2

HTTP/2,主要基于SPDY协议

# 验证支持HTTP2
curl -ik --http \
--cert /etc/kubernetes/pki/apiserver-kubelet-client.crt \
--key /etc/kubernetes/pki/apiserver-kubelet-cliet.key \
https://127.0.0.1:6443/api/v1/pods?watch=true

WebSocket

WebSocket是一种网络传输协议，可以在单个TCP连接上进行双向同时通信

WebSocket可以通过Upgrade Header从HTTP1.1升级到WebSocket协议

• WebSocket使用ws和wss作为统一资源标识符

# 验证支持WebSocket
curl -ik --http/1.1 \
--header "Connection:Upgrade" \
--header "Upgrade: websocket" \
--header "Sec-WebSocket-Key: xxx" \
--header "Sec-WebSocket-Version: 13" \
--header "Host: 127.0.0.1:8001 " \
--header "Origin: http://127.0.0.1:8001" \
--cert /etc/kubernetes/pki/apiserver-kubelet-client.crt \
--key /etc/kubernetes/pki/apiserver-kubelet-cliet.key \
https://127.0.0.1:6443/api/v1/pods?watch=true

List-Watch核心原理

kube-apiserver的List-Watch是在etcd List-Watch基础上的封装，类似于etcd的代理层

Informer启动之后，内部Controller运行Reflector,Relector通过Run可以通过BackoffUntil的方式重复执行r.ListAndWatch func（确保在Watch连接断开之后，能自动重新执行List-Wacth）

• Ref:https://github.com/kubernetes/client-go/blob/4f1c2e7b9711846cf31255fd709616c66cb3cfea/tools/cache/reflector.go#L220

  // Run repeatedly uses the reflector's ListAndWatch to fetch all the
  // objects and subsequent deltas.
  // Run will exit when stopCh is closed.
  func (r *Reflector) Run(stopCh <-chan struct{}) {
  	klog.V(3).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
  	wait.BackoffUntil(func() {
  		if err := r.ListAndWatch(stopCh); err != nil {
  			r.watchErrorHandler(r, err)
  		}
  	}, r.backoffManager, true, stopCh)
  	klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
  }

ListAndWatch func 首先执行List，把全量数据更新到DeltaFIFO队列中，然后从List中获取到最新的ResourceVersion开始Watch的后续的资源变化事件，通过Add/Update/Delete 方式传递到DeltaFIFO处理

• Ref:https://github.com/kubernetes/client-go/blob/4f1c2e7b9711846cf31255fd709616c66cb3cfea/tools/cache/reflector.go#L256

  // ListAndWatch first lists all items and get the resource version at the moment of call,
  // and then use the resource version to watch.
  // It returns error if ListAndWatch didn't even try to initialize watch.
  func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
  	klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name)
  
  	err := r.list(stopCh)
  	if err != nil {
  		return err
  	}
  
  	resyncerrc := make(chan error, 1)
  	cancelCh := make(chan struct{})
  	defer close(cancelCh)
  	go func() {
  		resyncCh, cleanup := r.resyncChan()
  		defer func() {
  			cleanup() // Call the last one written into cleanup
  		}()
  		for {
  			select {
  			case <-resyncCh:
  			case <-stopCh:
  				return
  			case <-cancelCh:
  				return
  			}
  			if r.ShouldResync == nil || r.ShouldResync() {
  				klog.V(4).Infof("%s: forcing resync", r.name)
  				if err := r.store.Resync(); err != nil {
  					resyncerrc <- err
  					return
  				}
  			}
  			cleanup()
  			resyncCh, cleanup = r.resyncChan()
  		}
  	}()
  
  	retry := NewRetryWithDeadline(r.MaxInternalErrorRetryDuration, time.Minute, apierrors.IsInternalError, r.clock)
  	for {
  		// give the stopCh a chance to stop the loop, even in case of continue statements further down on errors
  		select {
  		case <-stopCh:
  			return nil
  		default:
  		}
  
  		timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
  		options := metav1.ListOptions{
  			ResourceVersion: r.LastSyncResourceVersion(),
  			// We want to avoid situations of hanging watchers. Stop any watchers that do not
  			// receive any events within the timeout window.
  			TimeoutSeconds: &timeoutSeconds,
  			// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
  			// Reflector doesn't assume bookmarks are returned at all (if the server do not support
  			// watch bookmarks, it will ignore this field).
  			AllowWatchBookmarks: true,
  		}
  
  		// start the clock before sending the request, since some proxies won't flush headers until after the first watch event is sent
  		start := r.clock.Now()
  		w, err := r.listerWatcher.Watch(options)
  		if err != nil {
  			// If this is "connection refused" error, it means that most likely apiserver is not responsive.
  			// It doesn't make sense to re-list all objects because most likely we will be able to restart
  			// watch where we ended.
  			// If that's the case begin exponentially backing off and resend watch request.
  			// Do the same for "429" errors.
  			if utilnet.IsConnectionRefused(err) || apierrors.IsTooManyRequests(err) {
  				<-r.initConnBackoffManager.Backoff().C()
  				continue
  			}
  			return err
  		}
  
  		err = watchHandler(start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.expectedTypeName, r.setLastSyncResourceVersion, r.clock, resyncerrc, stopCh)
  		retry.After(err)
  		if err != nil {
  			if err != errorStopRequested {
  				switch {
  				case isExpiredError(err):
  					// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
  					// has a semantic that it returns data at least as fresh as provided RV.
  					// So first try to LIST with setting RV to resource version of last observed object.
  					klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
  				case apierrors.IsTooManyRequests(err):
  					klog.V(2).Infof("%s: watch of %v returned 429 - backing off", r.name, r.expectedTypeName)
  					<-r.initConnBackoffManager.Backoff().C()
  					continue
  				case apierrors.IsInternalError(err) && retry.ShouldRetry():
  					klog.V(2).Infof("%s: retrying watch of %v internal error: %v", r.name, r.expectedTypeName, err)
  					continue
  				default:
  					klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
  				}
  			}
  			return nil
  		}
  	}
  }