Pod加入后端池

1.问题

1.1.现象

• Pod对应的readness探针设置时间比较大，readness prober还未探测Pod就绪状态时，正常启动处于Running状态的Pod无法基于Service负载访问，会出现流量无法转到目标应用现象。

• 

1.2.分析

• 目前分析，Service流量会由kube-proxy或CNI接管，对于clusterIP类型的Service本质是分了一个VIP作为后端池代理，流量到达时会根据负载策略选出一个endpoint转发，Service流量接管及PodIP直接访问正常情况下，问题应该出现在ep/kubeproxy监听及修改时机。

• 

• 这里是理论分析，具体原因请查看下边的代码定位

2.kubelet

2.1.syncLoop

• Pod生命周期由kubelet主循环驱动，根据提交到APIServer的Pod变化决定创建、销毁及重建，感兴趣可以看kubelet章节，这里不再赘述。

• func (kl *Kubelet) syncLoopIteration(...) bool {
  	select {
      // watch不同来源的pod信息变化(file、http、apiserver)
  	case u, open := <-configCh:
  		...
  		switch u.Op {
  		case kubetypes.ADD:
              // 新增
  			handler.HandlePodAdditions(u.Pods)
  		case kubetypes.UPDATE:
              // 更新
  			handler.HandlePodUpdates(u.Pods)
  		case kubetypes.REMOVE:
              // 移除
  			handler.HandlePodRemoves(u.Pods)
  		case kubetypes.RECONCILE:
              // 重新协调
  			handler.HandlePodReconcile(u.Pods)
  		case kubetypes.DELETE:
              // 优雅删除
  			handler.HandlePodUpdates(u.Pods)
  		...
  		}
      // 2.pleg.start()每秒reList容器状态,根据最新的PodStatus生成PodLifeCycleEvent存入PLEChan
  	case e := <-plegCh:
          // 更新容器最后一次启动时间
  		if e.Type == pleg.ContainerStarted {
  			kl.lastContainerStartedTime.Add(e.ID, time.Now())
  		}
          // 容器状态更新
  		if isSyncPodWorthy(e) {
              handler.HandlePodSyncs([]*v1.Pod{pod})
  		}
  		// 容器退出
  		if e.Type == pleg.ContainerDied {
  			if containerID, ok := e.Data.(string); ok {
  				kl.cleanUpContainersInPod(e.ID, containerID)
  			}
  		}
      // 3.每秒周期执行
  	case <-syncCh:
          // 获取所有待同步的pod(运行正常的pod/内部模块请求的pod)
  		podsToSync := kl.getPodsToSync()
  		...
          // 同步最新的Pod状态
  		handler.HandlePodSyncs(podsToSync)
      // 4.liveness事件处理
  	case update := <-kl.livenessManager.Updates():
          // 如果探针检测失败,触发重建
  		if update.Result == proberesults.Failure {
  			handleProbeSync(kl, update, handler, "liveness", "unhealthy")
  		}
      // 5.readiness事件处理
  	case update := <-kl.readinessManager.Updates():
          // 获取readiness探测状态
  		ready := update.Result == proberesults.Success
          // 更新statusManager的Pod containerStatus
  		kl.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
  		if ready {
  			status = "ready"
  		}
          // 触发探测同步
  		handleProbeSync(kl, update, handler, "readiness", status)
      // 6.启动状态变化
  	case update := <-kl.startupManager.Updates():
  		started := update.Result == proberesults.Success
          // 更新容器状态
  		kl.statusManager.SetContainerStartup(update.PodUID, update.ContainerID, started)
  		if started {
  			status = "started"
  		}
          // 触发探测同步
  		handleProbeSync(kl, update, handler, "startup", status)
      // 7.每2s执行一次GC
  	case <-housekeepingCh:
          // 所有配置源就绪才触发
  		if kl.sourcesReady.AllReady() {
              // 家务进程,清理孤儿容器、回收卷、清理终止pod
  			handler.HandlePodCleanups()
  		}
  	}
  	return true
  }
  

• 

• 注意

  这里不同的handler基本都会进入dispatch分发，根据Pod状态或孤儿标记走到SyncPod/syncTerminatingPod/syncTerminatedPod/

2.2.syncPod

• kubelet会在SyncPod阶段会进行资源准入检查、网络插件检查、旧容器清理、volume附着和挂载等待、prober探针注册及Pod正式调度。

• func (kl *Kubelet) SyncPod(...) (isTerminal bool, err error) {
  	...
  
  	// 生成APIServer层面的PodStatus,这里会调用一下probeManager.UpdatePodStatus合并探针结果
  	apiPodStatus := kl.generateAPIPodStatus(pod, podStatus, false)
      // 这个podStatus是内存侧的状态,对应的是pleg模块缓存的PodStatus
  	podStatus.IPs = make([]string, 0, len(apiPodStatus.PodIPs))
      
      // 更新一下内存侧PodStatus地址
  	for _, ipInfo := range apiPodStatus.PodIPs {
  		podStatus.IPs = append(podStatus.IPs, ipInfo.IP)
  	}
  	if len(podStatus.IPs) == 0 && len(apiPodStatus.PodIP) > 0 {
  		podStatus.IPs = []string{apiPodStatus.PodIP}
  	}
  
  	// 终态的Pod更新一下statusMgr就结束了
  	if apiPodStatus.Phase == v1.PodSucceeded || apiPodStatus.Phase == v1.PodFailed {
  		kl.statusManager.SetPodStatus(pod, apiPodStatus)
  		isTerminal = true
  		return isTerminal, nil
  	}
  
  	// 准入检查,覆盖节点资源压力、白名单、CPU/Mem/GPU资源检查...
  	runnable := kl.canRunPod(pod)
      // 资源不足
  	if !runnable.Admit {
  		// APIServer侧的PodStatus设置为Pending
  		if apiPodStatus.Phase != v1.PodFailed && apiPodStatus.Phase != v1.PodSucceeded {
  			apiPodStatus.Phase = v1.PodPending
  		}
  		apiPodStatus.Reason = runnable.Reason
  		apiPodStatus.Message = runnable.Message
  		
          // Waiting containers are not creating.
  		const waitingReason = "Blocked"
  		for _, cs := range apiPodStatus.InitContainerStatuses {
  			if cs.State.Waiting != nil {
  				cs.State.Waiting.Reason = waitingReason
  			}
  		}
  		for _, cs := range apiPodStatus.ContainerStatuses {
  			if cs.State.Waiting != nil {
  				cs.State.Waiting.Reason = waitingReason
  			}
  		}
  	}
  	...
  
      // APIServer侧的PodStatus设置到statusManager,后续的状态更新以这里的为主,会异步Patch到APIServer
  	kl.statusManager.SetPodStatus(pod, apiPodStatus)
  
  	// 准入不通过,仅保留Pod对象,kill掉已经运行的容器
  	if !runnable.Admit {
  		...
  		p := kubecontainer.ConvertPodStatusToRunningPod(kl.getRuntime().Type(), podStatus)
  		kl.killPod(ctx, pod, p, nil)
          ...
          
  		return false, syncErr
  	}
  
  	// 确保节点存在合适的cni plugin
  	kl.runtimeState.networkErrors()
      ...
  
  	// 这里获取的是cgroupMgr,根据container qos等级将Pod挂载不同的cgroup树限制资源
  	pcm := kl.containerManager.NewPodContainerManager()
  	if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
  		...
          
          // Pod不在正确的qos cgroup目录
          // kubepods/
          // |
          // +-- Guaranteed
          // |   |- podXXX
          // +-- Burstable
          // |   |- podXXX
          // +-- BestEffort
          //     |- podXXX
  		if !pcm.Exists(pod) && !firstSync {
              // kill掉容器重建
  			p := kubecontainer.ConvertPodStatusToRunningPod(kl.getRuntime().Type(), podStatus)
  			kl.killPod(ctx, pod, p, nil)
              ...
  		}
          
          // Pod设置了重启策略
  		if !(podKilled && pod.Spec.RestartPolicy == v1.RestartPolicyNever) {
              // pod qos cgroup还不存在
  			if !pcm.Exists(pod) {
                  // 先计算qos目录该占多少CPU/Mem,更新qos cgroup的资源限制文件
  				kl.containerManager.UpdateQOSCgroups()
                  ...
                  // 重建pod qos cgroup目录,容器放在这个目录下可以实现资源限制
  				pcm.EnsureExists(pod)
                  ...
  			}
  		}
  	}
  
  	// Create Mirror Pod for Static Pod if it doesn't already exist
  	if kubetypes.IsStaticPod(pod) {
  		...
  		if mirrorPod != nil {
              // static Pod删除或异常
  			if mirrorPod.DeletionTimestamp != nil || !kl.podManager.IsMirrorPodOf(mirrorPod, pod) {
  				// 清理mirrorPod对象及缓存
  				podFullName := kubecontainer.GetPodFullName(pod)
  				kl.podManager.DeleteMirrorPod(podFullName, &mirrorPod.ObjectMeta.UID)
  				...
  			}
  		}
          
          // mirrorPod重建
  		if mirrorPod == nil || deleted {
  			...
              // 重建static Pod
  			kl.podManager.CreateMirrorPod(pod)
              ...
  		}
  	}
  
  	// 检查及创建/var/lib/kubelet下的pod目录
  	kl.makePodDataDirs(pod)
      ...
  
  	// volume附着及挂载节点,Pod创建前确保node已经mount volume至宿主机目录
  	if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
  		// Wait for volumes to attach/mount
  		kl.volumeManager.WaitForAttachAndMount(pod)
          ...
  	}
  	...
  
  	// 来看这里,注册prober探针
  	kl.probeManager.AddPod(pod)
  
      // 交互运行时管理Pod容器(pause/init/container...)
  	kl.containerRuntime.SyncPod(ctx, pod, podStatus, pullSecrets, kl.backOff)
  	...
  
  	return false, nil
  }
  
  // creates the final API pod status for a pod, given the internal pod status. This method should only be called // from within syncPod methods.
  func (kl *Kubelet) generateAPIPodStatus(...) v1.PodStatus {
  	...
  
  	// ensure the probe managers have up to date status for containers
  	kl.probeManager.UpdatePodStatus(pod.UID, s)
  	...
  
  	return *s
  }
  

• 

• 这里做了很多Pod管理的前置准备工作，重点看kl.probeManager.AddPod(pod)注册的探针怎么工作的

3.probeMgr

3.1.manager

• Pod设置的探针会以prober形式注册到probeMgr.worker，这些prober会将异步探测的结果存到对应cache，向kubelet主循环发送信号。

• func (m *manager) AddPod(pod *v1.Pod) {
  	m.workerLock.Lock()
  	defer m.workerLock.Unlock()
  
  	key := probeKey{podUID: pod.UID}
  	for _, c := range pod.Spec.Containers {
  		key.containerName = c.Name
  
  		if c.StartupProbe != nil {
  			key.probeType = startup
  			if _, ok := m.workers[key]; ok {
  				return
  			}
  			w := newWorker(m, startup, pod, c)
  			m.workers[key] = w
  			go w.run()
  		}
  
  		if c.ReadinessProbe != nil {
  			key.probeType = readiness
  			if _, ok := m.workers[key]; ok {
  				return
  			}
  			w := newWorker(m, readiness, pod, c)
  			m.workers[key] = w
  			go w.run()
  		}
  
  		if c.LivenessProbe != nil {
  			key.probeType = liveness
  			if _, ok := m.workers[key]; ok {
  				return
  			}
  			w := newWorker(m, liveness, pod, c)
  			m.workers[key] = w
  			go w.run()
  		}
  	}
  }
  

• prober worker是探针的抽象，会基于探针声明执行http/tcp/exec/grpc请求或命令，将结果保存到resultManager供kubelet更新状态用。

• // Creates and starts a new probe worker.
  func newWorker(m *manager, probeType probeType, pod *v1.Pod, container v1.Container) *worker {
  	w := &worker{
  		stopCh:          make(chan struct{}, 1), // Buffer so stop() can be non-blocking.
  		manualTriggerCh: make(chan struct{}, 1), // Buffer so can do non-blocking calls to doProbe.
  		pod:             pod,
  		container:       container,
  		probeType:       probeType,
  		probeManager:    m,
  	}
  
  	switch probeType {
  	case readiness:
  		w.spec = container.ReadinessProbe
  		w.resultsManager = m.readinessManager
  		w.initialValue = results.Failure
  	case liveness:
  		w.spec = container.LivenessProbe
  		w.resultsManager = m.livenessManager
  		w.initialValue = results.Success
  	case startup:
  		w.spec = container.StartupProbe
  		w.resultsManager = m.startupManager
  		w.initialValue = results.Unknown
  	}
  
  	return w
  }
  

• worker.run()本质上会不停执行w.doProbe()循环探测，命中某个case触发一次，case信号来源是策略定时器或外部主动请求的探测信号。

• // run periodically probes the container.
  func (w *worker) run() {
  	ctx := context.Background()
      // 这里就是探针设置的时间间隔
  	probeTickerPeriod := time.Duration(w.spec.PeriodSeconds) * time.Second
  
  	// kubelet刚重启,加抖动避免流量洪峰
  	if probeTickerPeriod > time.Since(w.probeManager.start) {
  		time.Sleep(time.Duration(rand.Float64() * float64(probeTickerPeriod)))
  	}
  
      // 探针定时器
  	probeTicker := time.NewTicker(probeTickerPeriod)
  
  	defer func() {
  		// Clean up.
  		probeTicker.Stop()
  		if !w.containerID.IsEmpty() {
  			w.resultsManager.Remove(w.containerID)
  		}
  		w.probeManager.removeWorker(w.pod.UID, w.container.Name, w.probeType)
  	}()
  
  probeLoop:
  	for w.doProbe(ctx) {
  		// Wait for next probe tick.
  		select {
  		case <-w.stopCh:
  			break probeLoop
  		case <-probeTicker.C:
  		case <-w.manualTriggerCh:
  			// continue
  		}
  	}
  }
  

• 补充

  probe worker本质就是探针管理模块，会缓存声明的探针策略及关联容器，然后执行doProbe—>continue—>doProbe直到退出

3.2.doProbe

• w.doProbe()根据container启动状态执行探测，其实就是生成http/tcp/exec/grpc请求访问对应接口或执行命令，探测结果根据策略生成。

• // doProbe probes the container once and records the result. Returns whether the worker should continue.
  func (w *worker) doProbe(ctx context.Context) (keepGoing bool) {
  	...
      // 获取一下statusMgr缓存的PodStatus,否则说明Pod还未开始调度,等待下一轮
  	status, ok := w.probeManager.statusManager.GetPodStatus(w.pod.UID)
  	if !ok {
  		return true
  	}
  
  	// 终态Pod停止探测
  	if status.Phase == v1.PodFailed || status.Phase == v1.PodSucceeded {
  		return false
  	}
  
      // container对应状态,container没有状态或ID,说明还未启动,等待下一轮
  	c, ok := podutil.GetContainerStatus(status.ContainerStatuses, w.container.Name)
  	if !ok || len(c.ContainerID) == 0 {
  		return true // Wait for more information.
  	}
  
      // 同名容器重启过
  	if w.containerID.String() != c.ContainerID {
          // 清理旧的探测结果
  		if !w.containerID.IsEmpty() {
  			w.resultsManager.Remove(w.containerID)
  		}
          // 更新一下数据
  		w.containerID = kubecontainer.ParseContainerID(c.ContainerID)
  		w.resultsManager.Set(w.containerID, w.initialValue, w.pod)
  		// We've got a new container; resume probing.
  		w.onHold = false
  	}
  
      // liveness/startup探测失败会重启容器,这里会等容器重启直到命中上一步
  	if w.onHold {
  		// Worker is on hold until there is a new container.
  		return true
  	}
  
      // container还未启动,等待下一轮
  	if c.State.Running == nil {
  		if !w.containerID.IsEmpty() {
  			w.resultsManager.Set(w.containerID, results.Failure, w.pod)
  		}
  		// Abort if the container will not be restarted.
  		return c.State.Terminated == nil || w.pod.Spec.RestartPolicy != v1.RestartPolicyNever
  	}
  
  	// Pod优雅退出,停止探测
  	if w.pod.ObjectMeta.DeletionTimestamp != nil && (w.probeType == liveness || w.probeType == startup) {
  		// Set a last result to ensure quiet shutdown.
  		w.resultsManager.Set(w.containerID, results.Success, w.pod)
  		// Stop probing at this point.
  		return false
  	}
  
  	// 容器启动还未达到延迟时间,等待下一轮
  	if int32(time.Since(c.State.Running.StartedAt.Time).Seconds()) < w.spec.InitialDelaySeconds {
  		return true
  	}
  
      // started探针探测完成,等待下一轮
  	if c.Started != nil && *c.Started {
  		// Stop probing for startup once container has started.
  		// we keep it running to make sure it will work for restarted container.
  		if w.probeType == startup {
  			return true
  		}
      // started探针还未成功
  	} else {
  		// 其它探针等待下一轮
  		if w.probeType != startup {
  			return true
  		}
  	}
  
  	// 执行探测命令
  	result, err := w.probeManager.prober.probe(ctx, w.probeType, w.pod, status, w.container, w.containerID)
  	...
  
      // 连续失败或成功未达到阈值,等待下一轮
  	if w.lastResult == result {
  		w.resultRun++
  	} else {
  		w.lastResult = result
  		w.resultRun = 1
  	}
  	if (result == results.Failure && w.resultRun < int(w.spec.FailureThreshold)) ||
  		(result == results.Success && w.resultRun < int(w.spec.SuccessThreshold)) {
  		// Success or failure is below threshold - leave the probe state unchanged.
  		return true
  	}
  
      // 设置结果缓存
  	w.resultsManager.Set(w.containerID, result, w.pod)
  
      // startup/liveness探针失败会重启容器,这里设置挂起,直到重启完成解封
  	if (w.probeType == liveness || w.probeType == startup) && result == results.Failure {
  		// The container fails a liveness/startup check, it will need to be restarted.
  		// Stop probing until we see a new container ID. This is to reduce the
  		// chance of hitting #21751, where running `docker exec` when a
  		// container is being stopped may lead to corrupted container state.
  		w.onHold = true
  		w.resultRun = 0
  	}
  
  	return true
  }
  

• 这里有些乱，基于接口实现三类探针逻辑感觉会清晰很多，可能有其它考量

3.3.probe

• 进入正题，prober.probe()才是真正执行接口请求或exec命令入口，自带重试执行probe探测，这里的事件告警先忽略，看探测逻辑。

• // probe probes the container.
  func (pb *prober) probe(ctx context.Context, probeType probeType, pod *v1.Pod, status v1.PodStatus, container v1.Container, containerID kubecontainer.ContainerID) (results.Result, error) {
  	...
  
      // 探针的probeSpec为空
  	if probeSpec == nil {
  		return results.Success, nil
  	}
  
  	result, output, err := pb.runProbeWithRetries(ctx, probeType, probeSpec, pod, status, container, containerID, maxProbeRetries)
      // 探测失败
  	if err != nil || (result != probe.Success && result != probe.Warning) {
  		return results.Failure, err
  	}
  	return results.Success, nil
  }
  
  // tries to probe the container in a finite loop, it returns the last result if it never succeeds.
  func (pb *prober) runProbeWithRetries(...) (probe.Result, string, error) {
  	...
      
      // 重试3次
  	for i := 0; i < retries; i++ {
  		result, output, err = pb.runProbe(ctx, probeType, p, pod, status, container, containerID)
  		if err == nil {
  			return result, output, nil
  		}
  	}
  	return result, output, err
  }
  
  func (pb *prober) runProbe(...) * time.Second
  	// exec探针
  	if p.Exec != nil {
          // 交互container exec接口执行命令
  		command := kubecontainer.ExpandContainerCommandOnlyStatic(p.Exec.Command, container.Env)
  		return pb.exec.Probe(pb.newExecInContainer(ctx, container, containerID, command, timeout))
  	}
  
  	// http探针
  	if p.HTTPGet != nil {
          // 发http请求(GET http://10.244.1.10:8080/healthz)
  		req, err := httpprobe.NewRequestForHTTPGetAction(p.HTTPGet, &container, status.PodIP, "probe")
  		...
  		return pb.http.Probe(req, timeout)
  	}
  
  	// tcp探针
  	if p.TCPSocket != nil {
  		port, err := probe.ResolveContainerPort(p.TCPSocket.Port, &container)
  		...
  		host := p.TCPSocket.Host
  		if host == "" {
  			host = status.PodIP
  		}
          // 发起tcp connect(nc -zv 10.244.1.10 3306)
  		return pb.tcp.Probe(host, port, timeout)
  	}
  
  	// grpc探针
  	if p.GRPC != nil {
  		host := status.PodIP
  		service := ""
  		if p.GRPC.Service != nil {
  			service = *p.GRPC.Service
  		}
          // 基于unix socket发起grpc调用(grpc.health.v1.Health.Check{service: "user.service"})
  		return pb.grpc.Probe(host, service, int(p.GRPC.Port), timeout)
  	}
  
  	return probe.Unknown, "", fmt.Errorf("missing probe handler for %s:%s", format.Pod(pod), container.Name)
  }
  

• exec探针不是直接执行命令，而是基于runtime向containerd发起grpc调用，fork子进程在容器执行，感兴趣可以看containerd分析。

• // executes a command to check the liveness/readiness of container
  // from executing a command. Returns the Result status, command output, and errors if any.
  func (pr execProber) Probe(e exec.Cmd) (probe.Result, string, error) {
  	...
  	writer := ioutils.LimitWriter(&dataBuffer, maxReadLength)
  	e.SetStderr(writer)
  	e.SetStdout(writer)
      // 执行命令
  	e.Start()
  	...
  	data := dataBuffer.Bytes()
  
  	return probe.Success, string(data), nil
  }
  
  func (eic *execInContainer) Start() error {
  	data, err := eic.run()
  	if eic.writer != nil {
  		// only record the write error, do not cover the command run error
  		eic.writer.Write(data)
          ...
  	}
  	return err
  }
  
  // eic是这里初始化的
  func (pb *prober) newExecInContainer(...) exec.Cmd {
  	return &execInContainer{run: func() ([]byte, error) {
          // 这个runner其实就是kubeGenericRuntimeManager
  		return pb.runner.RunInContainer(ctx, containerID, cmd, timeout)
  	}}
  }
  
  // RunInContainer synchronously executes the command in the container, and returns the output.
  func (m *kubeGenericRuntimeManager) RunInContainer(...) ([]byte, error) {
  	stdout, stderr, err := m.runtimeService.ExecSync(ctx, id.ID, cmd, timeout)
  	return append(stdout, stderr...), err
  }
  
  // ExecSync executes a command in the container, and returns the stdout output.
  // If command exits with a non-zero exit code, an error is returned.
  func (r *remoteRuntimeService) ExecSync(...) (stdout []byte, stderr []byte, err error) {
  	...
  	return r.execSyncV1(ctx, containerID, cmd, timeout)
  }
  
  func (r *remoteRuntimeService) execSyncV1(ctx context.Context, containerID string, cmd []string, timeout time.Duration) (stdout []byte, stderr []byte, err error) {
  	timeoutSeconds := int64(timeout.Seconds())
  	req := &runtimeapi.ExecSyncRequest{
  		ContainerId: containerID,
  		Cmd:         cmd,
  		Timeout:     timeoutSeconds,
  	}
  	resp, err := r.runtimeClient.ExecSync(ctx, req)
  	...
  
  	return resp.Stdout, resp.Stderr, err
  }
  
  func (c *runtimeServiceClient) ExecSync(...) (*ExecSyncResponse, error) {
  	out := new(ExecSyncResponse)
  	err := c.cc.Invoke(ctx, "/runtime.v1.RuntimeService/ExecSync", in, out, opts...)
  	...
      
  	return out, nil
  }
  

• 我不得不说这里真是疯狂套娃，kubelet不同的模块基本都是Func参数注值实现共用

3.4.update

• prober探测完成会将结果缓存，缓存设置的那一刻会向kubelet主循环发个信号，通知探测结果更新了，赶紧设置下PodStatus推到APIServer。

• func (m *manager) Set(id kubecontainer.ContainerID, result Result, pod *v1.Pod) {
      // 设置缓存
  	if m.setInternal(id, result) {
          // 成功发下信号
  		m.updates <- Update{id, result, pod.UID}
  	}
  }
  
  // kubelet主循环会调用一下probeManager.UpdatePodStatus更新一下状态
  func (m *manager) UpdatePodStatus(podUID types.UID, podStatus *v1.PodStatus) {
  	for i, c := range podStatus.ContainerStatuses {
  		var started bool
          // 容器未运行(这个状态是pleg轮询获取containerd容器状态翻译生成)
  		if c.State.Running == nil {
  			started = false
          // 容器已启动,获取startness探针结果
  		} else if result, ok := m.startupManager.Get(kubecontainer.ParseContainerID(c.ContainerID)); ok {
  			started = result == results.Success
          // 否则,prober注册了,就是未启动,未注册就是启动
  		} else {
  			_, exists := m.getWorker(podUID, c.Name, startup)
  			started = !exists
  		}
          // 更新container的启动状态
  		podStatus.ContainerStatuses[i].Started = &started
  
          // 计算ready字段
  		if started {
              ...
  			if c.State.Running == nil {
  				ready = false
              // 获取readness探针结果
  			} else if result, ok := m.readinessManager.Get(kubecontainer.ParseContainerID(c.ContainerID)); ok && result == results.Success {
  				ready = true
  			} else {
  				// The check whether there is a probe which hasn't run yet.
  				w, exists := m.getWorker(podUID, c.Name, readiness)
  				ready = !exists // no readinessProbe -> always ready
                  // 催一下readness prober赶紧探
  				if exists {
  					// Trigger an immediate run of the readinessProbe to update ready state
  					select {
  					case w.manualTriggerCh <- struct{}{}:
  					default: // Non-blocking.
  						klog.InfoS("Failed to trigger a manual run", "probe", w.probeType.String())
  					}
  				}
  			}
              // 更新container的就绪状态
  			podStatus.ContainerStatuses[i].Ready = ready
  		}
  	}
      
  	// init container就绪状态更新
  	for i, c := range podStatus.InitContainerStatuses {
  		var ready bool
          // init容器成功退出就视为就绪
  		if c.State.Terminated != nil && c.State.Terminated.ExitCode == 0 {
  			ready = true
  		}
  		podStatus.InitContainerStatuses[i].Ready = ready
  	}
  }
  

• PodStatus最终会存到statusManager，实现异步Patch到APIServer

4.endpoint

4.1.watch

• 现在可以看endpoint后端池怎么设置的，这里以endpoint为例(endpointSlice类似)，更详细的分析可以看endpoint controller章节。

• 

• 可以看到，Pod变化会被endpoint controller捕获，将关联的service发到workqueue供主循环处理

4.2.syncService

• endpoint controller监听到event会执行e.syncService()同步，结合informer捕获的service/endpoint/pod更新后端池。

• func (e *Controller) processNextWorkItem(ctx context.Context) bool {
      // 获取队列service
  	eKey, quit := e.queue.Get()
  	if quit {
  		return false
  	}
      
      // 标记key已处理(移除processing队列的item)
  	defer e.queue.Done(eKey)
  
      // 执行后端池同步
  	err := e.syncService(ctx, eKey.(string))
      
      // 失败尝试重新入队
  	e.handleErr(err, eKey)
  
  	return true
  }
  
  // syncService update endpoints base service with pods.
  func (e *Controller) syncService(ctx context.Context, key string) error {
  	...
  	// 1.获取service
  	service, err := e.serviceLister.Services(namespace).Get(name)
  	...
  
      // 2.获取service关联pod
  	pods := e.podLister.Pods(service.Namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated())
  	...
  
      // 3.初始化端点
  	for _, pod := range pods {
          // 跳过未分配地址Pod/正在删除Pod/走到终态Pod
  		if !endpointutil.ShouldPodBeInEndpoints(pod, service.Spec.PublishNotReadyAddresses) {
  			continue
  		}
  
          // 构造endpointAddress对象
  		ep, err := podToEndpointAddressForService(service, pod)
  		...
  		epa := *ep
  		...
          
  		// 构造endpointPort对象
  		if len(service.Spec.Ports) == 0 && service.Spec.ClusterIP == api.ClusterIPNone {
              // headless service允许无端口
              subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, service.Spec.PublishNotReadyAddresses)
  		} else {
              // service定义端口
  			for i := range service.Spec.Ports {
  				servicePort := &service.Spec.Ports[i]
                  // 配合container定义端口确认使用端口号
  				portNum, err := podutil.FindPort(pod, servicePort)
  				...
                  
                  // 生成endpointPort
  				epp := endpointPortFromServicePort(servicePort, portNum)
  				...
                  
                  // 更新subsets
  				subsets, readyEps, notReadyEps = addEndpointSubset(subsets, pod, epa, epp, service.Spec.PublishNotReadyAddresses)
  			}
  		}
  	}
      // 计算最终的subsets,去重修正
      // subsets:
      // - Addresses: [{ip: 10.244.1.10}]
      // 	Ports: [{port: 80}]
      // - Addresses: [{ip: 10.244.1.10}]
      // 	Ports: [{port: 443}]
  	subsets = endpoints.RepackSubsets(subsets)
  
  	// 4.获取当前endpoints
  	currentEndpoints, err := e.endpointsLister.Endpoints(service.Namespace).Get(service.Name)
  	...
  	
  	// 5.对比endpoints是否需要更新
  	if !createEndpoints &&
      	// subsets相同
  		endpointutil.EndpointSubsetsEqualIgnoreResourceVersion(currentEndpoints.Subsets, subsets) &&
      	// endpoints label与service label一致(排除headless label)
  		apiequality.Semantic.DeepEqual(compareLabels, service.Labels) &&
      	// subsets未超出容量(1000),无endpoints.kubernetes.io/over-capacity注解
  		capacityAnnotationSetCorrectly(currentEndpoints.Annotations, currentEndpoints.Subsets) {
  		return nil
  	}
      
      // 6.构造新的endpoints
  	newEndpoints := currentEndpoints.DeepCopy()
  	newEndpoints.Subsets = subsets
  	newEndpoints.Labels = service.Labels
  	...
  
      // 7.endpoints subsets容量超出(触发截断,endpoints不再保留完整的后端,endpointslice切分保存)
  	if truncateEndpoints(newEndpoints) {
          // 优先保留ready address,其次考虑截断ready address
  		newEndpoints.Annotations[v1.EndpointsOverCapacity] = truncated
  	} else {
          // 容量未超出,不需要标识注解
  		delete(newEndpoints.Annotations, v1.EndpointsOverCapacity)
  	}
  
      // 8.更新endpoints label(headless类型标注label)
  	if !helper.IsServiceIPSet(service) {
  		newEndpoints.Labels = utillabels.CloneAndAddLabel(newEndpoints.Labels, v1.IsHeadlessService, "")
  	} else {
  		newEndpoints.Labels = utillabels.CloneAndRemoveLabel(newEndpoints.Labels, v1.IsHeadlessService)
  	}
  	
      // 9.创建endpoints
  	if createEndpoints {
  		_, err = e.client.CoreV1().Endpoints(service.Namespace).Create(ctx, newEndpoints, metav1.CreateOptions{})
      // 10.更新endpoints
  	} else {
  		_, err = e.client.CoreV1().Endpoints(service.Namespace).Update(ctx, newEndpoints, metav1.UpdateOptions{})
  	}
  	...
  	return nil
  }
  

• 注意

  这里会匹配出service Pod生成subnets，对比更新endpoint后端池，重点看addEndpointSubset怎么筛地址的

4.3.subnets

• addEndpointSubset()会检查pod.status.ReadyCondition，已就绪的话会将地址放到Addresses，否则放到NotReadyAddresses。

• // add the endpoints addresses and ports to the EndpointSubset.
  func addEndpointSubset(...) ([]v1.EndpointSubset, int, int) {
  	...
      
  	ports := []v1.EndpointPort{}
  	if epp != nil {
  		ports = append(ports, *epp)
  	}
      
      // 这里会匹配pod.status.Readycondition地址
      // 当然你也可以配置service.Spec.PublishNotReadyAddresses发布未就绪地址
  	if tolerateUnreadyEndpoints || podutil.IsPodReady(pod) {
  		subsets = append(subsets, v1.EndpointSubset{
  			Addresses: []v1.EndpointAddress{epa},
  			Ports:     ports,
  		})
  		readyEps++
  	} else { // if it is not a ready address it has to be not ready
  		subsets = append(subsets, v1.EndpointSubset{
  			NotReadyAddresses: []v1.EndpointAddress{epa},
  			Ports:             ports,
  		})
  		notReadyEps++
  	}
      
  	return subsets, readyEps, notReadyEps
  }
  
  // IsPodReady returns true if a pod is ready; false otherwise.
  func IsPodReady(pod *v1.Pod) bool {
  	return IsPodReadyConditionTrue(pod.Status)
  }
  
  // IsPodReadyConditionTrue returns true if a pod is ready; false otherwise.
  func IsPodReadyConditionTrue(status v1.PodStatus) bool {
  	condition := GetPodReadyCondition(status)
  	return condition != nil && condition.Status == v1.ConditionTrue
  }
  

• 配置了service.Spec.PublishNotReadyAddresses也允许将未就绪的Pod加入ready Addresses

4.4.kubeproxy

• kubeproxy监听到endpoint变化会同步iptables/ipvs规则，筛选地址阶段会保留ready的地址，具体内容可以访问kubeproxy章节查看。

• 

• 题外话

  1.不管是iptables分支还是ipvs分支，syncProxyRules同步规则函数写了800多行，我已经不想看第二次了，想具体了解可以点上边链接

  2.另外，kubeproxy不会发布正在终止的Pod地址，启用了这里会正常写到内核，让我疑惑的是，终止的Pod不是被ep controller摘掉了吗

5.总结

• --- kubelet
  	1.kubelet会注册Pod对应的探针至probeManager,执行不同的探测命令缓存结果至resultManager,另外会发信号通知kubelet更新Pod状态
  	2.kubelet主循环开始阶段会根据内存的PodStatus生成APIServer侧PodStatus,这里会基于探针结果更新状态及设置到statusManager
  	3.APIServer侧PodStatus状态更新会计算Ready Condition,只要有一个容器探针结果未就绪,Pod Ready Condition就不是就绪的
  
  --- endpoint controller
  	1.endpoint controller会监听service/pod/endpoint变化,基于selectorCache将相关的service推到workqueue
  	2.主循环基于workqueue获取关联的Pod列表,生成最新的subnets,这里会检查Pod Ready Condition,除非你允许发布未就绪的Pod地址
  	3.最新的subnets会更新到service关联的endpoint对象
  
  --- kube-proxy
  	1.kubeproxy会监听servcie/endpoint/node变化,驱动主循环执行syncService
  	2.不管是iptables分支还是ipvs分支,均会将service关联的endpoints地址拆分,保留subnets.readyAddress
  	3.筛选后的endpoint会随service地址写到endpoint内核,后端池正式生效
  

• 注意

  1.loadbalancer/externalName/NodePort类型的service走的分支不同，这里只展示clusterIP类型处理

  2.kubeproxy的service流量处理会存在性能问题，尤其是iptables实现，大规模集群会出现svc规则爆炸问题，目前已经在很多老集群碰到

  3.kubeproxy的ipvs实现属于过度状态，排查问题会增大成本(规则隐藏)，官方最新版已经逐渐废弃该分支，平替为nftables

  4.service流量处理更建议用cilium/calico或其它类似实现，基于eBPF接管service流量，利用eBPF文件存储规则，提前于协议栈处理