probeManager

1.简介

• probeManager定时监控pod容器的健康状态，感知到的状态调用statusManager更新到apiserver。目前支持三种探针：livenessProbe、readinessProbe和startupProbe，提供http/tcp/exec三种形式探针。

• 

• 注意

  startup/readiness/liveness探测结果会调用statusManager更新到apiserver，影响kubelet主循环的调度事件。

2.probeManager

2.1.初始化

• probeManager持有注册的probeWorker，驱动probeWorker执行探针写入结果至resultManager，探针结果会更新到statusManager。

• // Manager manages pod probing. It creates a probe "worker" for every container that specifies a probe (AddPod). 
  type Manager interface {
  	// 注册pod容器探针
  	AddPod(pod *v1.Pod)
  
  	// 停止存活和启动探针
  	StopLivenessAndStartup(pod *v1.Pod)
  
  	// 删除pod容器探针
  	RemovePod(pod *v1.Pod)
  
  	// 清理不再允许pod
  	CleanupPods(desiredPods map[types.UID]sets.Empty)
  
  	// 更新pod容器探活状态
  	UpdatePodStatus(types.UID, *v1.PodStatus)
  }
  
  type manager struct {
  	// 活跃probeWorker
  	workers map[probeKey]*worker
      
  	// Lock for accessing & mutating workers
  	workerLock sync.RWMutex
  
  	// statusManager提供podID和podIP探活,提供接口更新状态
  	statusManager status.Manager
  
  	// readiness探活结果
  	readinessManager results.Manager
  
  	// liveness探活结果
  	livenessManager results.Manager
  
  	// startup慢启动探活结果
  	startupManager results.Manager
  
  	// prober执行者
  	prober *prober
  
  	start time.Time
  }
  
  // NewManager creates a Manager for pod probing.
  func NewManager(...) Manager {
  	
      // 初始化prober(各类型探针)
  	prober := newProber(runner, recorder)
  	return &manager{
  		statusManager:    statusManager,
  		prober:           prober,
          // 探测结果缓存,map结构
  		readinessManager: readinessManager,
  		livenessManager:  livenessManager,
  		startupManager:   startupManager,
          // probeWorker
  		workers:          make(map[probeKey]*worker),
  		start:            clock.RealClock{}.Now(),
  	}
  }
  

• 

2.2.注册probe

• AddPod()用于遍历Pod所有的container，建立container的probeWorker及启动，定时执行容器探针任务。

• // registry probeWorker
  func (m *manager) AddPod(pod *v1.Pod) {
  	m.workerLock.Lock()
  	defer m.workerLock.Unlock()
  
  	key := probeKey{podUID: pod.UID}
      // 遍历pod container
  	for _, c := range pod.Spec.Containers {
  		key.containerName = c.Name
  
          // 启动探针
  		if c.StartupProbe != nil {
  			key.probeType = startup
              // probeWorker已存在
  			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
              // probeWorker已存在
  			if _, ok := m.workers[key]; ok {
  				return
  			}
  			w := newWorker(m, readiness, pod, c)
  			m.workers[key] = w
  			go w.run()
  		}
  
          // 存活探针(驱动syncLoopPod)
  		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()
  		}
  	}
  }
  
  // run periodically probes the container.
  func (w *worker) run() {
      // 执行周期
  	probeTickerPeriod := time.Duration(w.spec.PeriodSeconds) * time.Second
  
  	// 延迟抖动,避免probe协程同时启动造成探测洪峰
  	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)
  		}
  		// 清理probeWorker
  		w.probeManager.removeWorker(w.pod.UID, w.container.Name, w.probeType)
  		...
  	}()
  
  probeLoop:
      // 执行探测
  	for w.doProbe() {
  		// Wait for next probe tick.
  		select {
          // 退出
  		case <-w.stopCh:
  			break probeLoop
          // 定时器触发
  		case <-probeTicker.C:
          // 外部触发
  		case <-w.manualTriggerCh:
  			// continue
  		}
  	}
  }
  

• 

• doProbe

  检查pod是否存在、是否达到终态、是否终止、容器状态是否存在、容器是否重启、容器是否已启动，通过后才会真正执行探针及记录结果

2.3.状态更新

• probeWorker执行探针更新缓存结果时，会同步信号至readinessManager/livenessManager/startupManager的updates管道，驱动主循环调度pod及更新状态。

• // resultManager update cache.
  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监听探测进行任务分发，执行syncPod/syncTerminatingdPod/syncTerminatedPod进行调度，期间会调用UpdatePodStatus更新容器状态。

• func (m *manager) UpdatePodStatus(podUID types.UID, podStatus *v1.PodStatus) {
  	for i, c := range podStatus.ContainerStatuses {
  		var started bool
          // 容器未运行
  		if c.State.Running == nil {
  			started = false
          // 容器已启动,计算启动状态
  		} else if result, ok := m.startupManager.Get(kubecontainer.ParseContainerID(c.ContainerID)); ok {
  			started = result == results.Success
          // probeWorker运行状态
  		} else {
  			_, exists := m.getWorker(podUID, c.Name, startup)
  			started = !exists
  		}
          // 更新container的启动状态
  		podStatus.ContainerStatuses[i].Started = &started
  
          // 计算ready字段
  		if started {
  			var ready bool
  			if c.State.Running == nil {
  				ready = false
  			} 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
                  // 触发一次readinessProbe
  				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
          // 成功执行,ready为true
  		if c.State.Terminated != nil && c.State.Terminated.ExitCode == 0 {
  			ready = true
  		}
  		podStatus.InitContainerStatuses[i].Ready = ready
  	}
  }
  

•