kubelet 源码分析:statusManager 和 probeManager

「爱情、让人受尽委屈。」 2022-06-07 01:39 167阅读 0赞

kubelet 源码分析:statusManager 和 probeManager

在 kubelet 初始化的时候,会创建 statusManagerprobeManager,两者都和 pod 的状态有关系,因此我们放到一起来讲解。

statusManager 负责维护状态信息,并把 pod 状态更新到 apiserver,但是它并不负责监控 pod 状态的变化,而是提供对应的接口供其他组件调用,比如 probeManagerprobeManager 会定时去监控 pod 中容器的健康状况,一旦发现状态发生变化,就调用 statusManager 提供的方法更新 pod 的状态。

  1. klet.statusManager = status.NewManager(kubeClient, klet.podManager)
  2. klet.probeManager = prober.NewManager(
  3. klet.statusManager,
  4. klet.livenessManager,
  5. klet.runner,
  6. containerRefManager,
  7. kubeDeps.Recorder)

StatusManager

statusManager 对应的代码在 pkg/kubelet/status/status_manager.go 文件中,

  1. type PodStatusProvider interface { GetPodStatus(uid types.UID) (api.PodStatus, bool) }
  2. type Manager interface { PodStatusProvider Start() SetPodStatus(pod *api.Pod, status api.PodStatus) SetContainerReadiness(podUID types.UID, containerID kubecontainer.ContainerID, ready bool) TerminatePod(pod *api.Pod) RemoveOrphanedStatuses(podUIDs map[types.UID]bool) }

这个接口的方法可以分成三组:获取某个 pod 的状态、后台运行 goroutine 执行同步工作、修改 pod 的状态。修改状态的方法有多个,每个都有不同的用途:

  1. SetPodStatus:如果 pod 的状态发生了变化,会调用这个方法,把新状态更新到 apiserver,一般在 kubelet 维护 pod 生命周期的时候会调用
  2. SetContainerReadiness:如果健康检查发现 pod 中容器的健康状态发生变化,会调用这个方法,修改 pod 的健康状态
  3. TerminatePodkubelet 在删除 pod 的时候,会调用这个方法,把 pod 中所有的容器设置为 terminated 状态
  4. RemoveOrphanedStatuses:删除孤儿 pod,直接把对应的状态数据从缓存中删除即可

Start() 方法是在 kubelet 运行的时候调用的,它会启动一个 goroutine 执行更新操作:

  1. const syncPeriod = 10 * time.Second
  2. func (m *manager) Start() {
  3. ......
  4. glog.Info("Starting to sync pod status with apiserver")
  5. syncTicker := time.Tick(syncPeriod)
  6. // syncPod and syncBatch share the same go routine to avoid sync races.
  7. go wait.Forever(func() {
  8. select {
  9. case syncRequest := <-m.podStatusChannel:
  10. m.syncPod(syncRequest.podUID, syncRequest.status)
  11. case <-syncTicker:
  12. m.syncBatch()
  13. }
  14. }, 0)
  15. }

这个 goroutine 就能不断地从两个 channel 监听数据进行处理:syncTicker 是个定时器,也就是说它会定时保证 apiserver 和自己缓存的最新 pod 状态保持一致;podStatusChannel 是所有 pod 状态更新发送到的地方,调用方不会直接操作这个 channel,而是通过调用上面提到的修改状态的各种方法,这些方法内部会往这个 channel 写数据。

m.syncPod 根据参数中的 pod 和它的状态信息对 apiserver 中的数据进行更新,如果发现 pod 已经被删除也会把它从内部数据结构中删除。

ProbeManager

probeManager 检测 pod 中容器的健康状态,目前有两种 probe:readiness 和 liveness。readinessProbe 检测容器是否可以接受请求,如果检测结果失败,则将其从 service 的 endpoints 中移除,后续的请求也就不会发送给这个容器;livenessProbe 检测容器是否存活,如果检测结果失败,kubelet 会杀死这个容器,并重启一个新的(除非 RestartPolicy 设置成了 Never)。

并不是所有的 pod 中的容器都有健康检查的探针,如果没有,则不对容器进行检测,默认认为容器是正常的。在每次创建新 pod 的时候,kubelet 都会调用 probeManager.AddPod(pod) 方法,它对应的实现在 pkg/kubelet/prober/prober_manager.go 文件中:

  1. func (m *manager) AddPod(pod *api.Pod) {
  2. m.workerLock.Lock()
  3. defer m.workerLock.Unlock()
  4. key := probeKey{podUID: pod.UID}
  5. for _, c := range pod.Spec.Containers {
  6. key.containerName = c.Name
  7. if c.ReadinessProbe != nil {
  8. key.probeType = readiness
  9. if _, ok := m.workers[key]; ok {
  10. glog.Errorf("Readiness probe already exists! %v - %v",
  11. format.Pod(pod), c.Name)
  12. return
  13. }
  14. w := newWorker(m, readiness, pod, c)
  15. m.workers[key] = w
  16. go w.run()
  17. }
  18. if c.LivenessProbe != nil {
  19. key.probeType = liveness
  20. if _, ok := m.workers[key]; ok {
  21. glog.Errorf("Liveness probe already exists! %v - %v",
  22. format.Pod(pod), c.Name)
  23. return
  24. }
  25. w := newWorker(m, liveness, pod, c)
  26. m.workers[key] = w
  27. go w.run()
  28. }
  29. }
  30. }

遍历 pod 中的容器,如果其定义了 readiness 或者 liveness,就创建一个 worker,并启动一个 goroutine 在后台运行这个 worker。

pkg/kubelet/prober/worker.go:

  1. func (w *worker) run() {
  2. probeTickerPeriod := time.Duration(w.spec.PeriodSeconds) * time.Second
  3. probeTicker := time.NewTicker(probeTickerPeriod)
  4. defer func() {
  5. probeTicker.Stop()
  6. if !w.containerID.IsEmpty() {
  7. w.resultsManager.Remove(w.containerID)
  8. }
  9. w.probeManager.removeWorker(w.pod.UID, w.container.Name, w.probeType)
  10. }()
  11. time.Sleep(time.Duration(rand.Float64() * float64(probeTickerPeriod)))
  12. probeLoop:
  13. for w.doProbe() {
  14. // Wait for next probe tick.
  15. select {
  16. case <-w.stopCh:
  17. break probeLoop
  18. case <-probeTicker.C:
  19. // continue
  20. }
  21. }
  22. }
  23. func (w *worker) doProbe() (keepGoing bool) {
  24. defer func() { recover() }()
  25. defer runtime.HandleCrash(func(_ interface{}) { keepGoing = true })
  26. // pod 没有被创建,或者已经被删除了,直接跳过检测,但是会继续检测
  27. status, ok := w.probeManager.statusManager.GetPodStatus(w.pod.UID)
  28. if !ok {
  29. glog.V(3).Infof("No status for pod: %v", format.Pod(w.pod))
  30. return true
  31. }
  32. // pod 已经退出(不管是成功还是失败),直接返回,并终止 worker
  33. if status.Phase == api.PodFailed || status.Phase == api.PodSucceeded {
  34. glog.V(3).Infof("Pod %v %v, exiting probe worker",
  35. format.Pod(w.pod), status.Phase)
  36. return false
  37. }
  38. // 容器没有创建,或者已经删除了,直接返回,并继续检测,等待更多的信息
  39. c, ok := api.GetContainerStatus(status.ContainerStatuses, w.container.Name)
  40. if !ok || len(c.ContainerID) == 0 {
  41. glog.V(3).Infof("Probe target container not found: %v - %v",
  42. format.Pod(w.pod), w.container.Name)
  43. return true
  44. }
  45. // pod 更新了容器,使用最新的容器信息
  46. if w.containerID.String() != c.ContainerID {
  47. if !w.containerID.IsEmpty() {
  48. w.resultsManager.Remove(w.containerID)
  49. }
  50. w.containerID = kubecontainer.ParseContainerID(c.ContainerID)
  51. w.resultsManager.Set(w.containerID, w.initialValue, w.pod)
  52. w.onHold = false
  53. }
  54. if w.onHold {
  55. return true
  56. }
  57. if c.State.Running == nil {
  58. glog.V(3).Infof("Non-running container probed: %v - %v",
  59. format.Pod(w.pod), w.container.Name)
  60. if !w.containerID.IsEmpty() {
  61. w.resultsManager.Set(w.containerID, results.Failure, w.pod)
  62. }
  63. // 容器失败退出,并且不会再重启,终止 worker
  64. return c.State.Terminated == nil ||
  65. w.pod.Spec.RestartPolicy != api.RestartPolicyNever
  66. }
  67. // 容器启动时间太短,没有超过配置的初始化等待时间 InitialDelaySeconds
  68. if int32(time.Since(c.State.Running.StartedAt.Time).Seconds()) < w.spec.InitialDelaySeconds {
  69. return true
  70. }
  71. // 调用 prober 进行检测容器的状态
  72. result, err := w.probeManager.prober.probe(w.probeType, w.pod, status, w.container, w.containerID)
  73. if err != nil {
  74. return true
  75. }
  76. if w.lastResult == result {
  77. w.resultRun++
  78. } else {
  79. w.lastResult = result
  80. w.resultRun = 1
  81. }
  82. // 如果容器退出,并且没有超过最大的失败次数,则继续检测
  83. if (result == results.Failure && w.resultRun < int(w.spec.FailureThreshold)) ||
  84. (result == results.Success && w.resultRun < int(w.spec.SuccessThreshold)) {
  85. return true
  86. }
  87. // 保存最新的检测结果
  88. w.resultsManager.Set(w.containerID, result, w.pod)
  89. if w.probeType == liveness && result == results.Failure {
  90. // 容器 liveness 检测失败,需要删除容器并重新创建,在新容器成功创建出来之前,暂停检测
  91. w.onHold = true
  92. }
  93. return true
  94. }

每次检测的时候都会用 w.resultsManager.Set(w.containerID, result, w.pod) 来保存检测结果,resultsManager 的代码在 pkg/kubelet/prober/results/results_manager.go:

  1. func (m *manager) Set(id kubecontainer.ContainerID, result Result, pod *api.Pod) {
  2. if m.setInternal(id, result) {
  3. m.updates <- Update{id, result, pod.UID}
  4. }
  5. }
  6. func (m *manager) setInternal(id kubecontainer.ContainerID, result Result) bool {
  7. m.Lock()
  8. defer m.Unlock()
  9. prev, exists := m.cache[id]
  10. if !exists || prev != result {
  11. m.cache[id] = result
  12. return true
  13. }
  14. return false
  15. }
  16. func (m *manager) Updates() <-chan Update {
  17. return m.updates
  18. }
  19. 它把结果保存在缓存中,并发送到 m.updates 管道。对于 liveness 来说,它的管道消费者是 kubelet,还记得 syncLoopIteration 中的这段代码逻辑吗?
  20. case update := <-kl.livenessManager.Updates():
  21. if update.Result == proberesults.Failure {
  22. // The liveness manager detected a failure; sync the pod.
  23. pod, ok := kl.podManager.GetPodByUID(update.PodUID)
  24. if !ok {
  25. // If the pod no longer exists, ignore the update.
  26. glog.V(4).Infof("SyncLoop (container unhealthy): ignore irrelevant update: %#v", update)
  27. break
  28. }
  29. glog.V(1).Infof("SyncLoop (container unhealthy): %q", format.Pod(pod))
  30. handler.HandlePodSyncs([]*api.Pod{pod})
  31. }
  32. 因为 liveness 关系者 pod 的生死,因此需要 kubelet 的处理逻辑。而 readiness 即使失败也不会重新创建 pod,它的处理逻辑是不同的,它的处理代码同样在 pkg/kubelet/prober/prober_manager.go
  33. func (m *manager) Start() {
  34. go wait.Forever(m.updateReadiness, 0)
  35. }
  36. func (m *manager) updateReadiness() {
  37. update := <-m.readinessManager.Updates()
  38. ready := update.Result == results.Success
  39. m.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
  40. }

proberManager 启动的时候,会运行一个 goroutine 定时读取 readinessManager 管道中的数据,并根据数据调用 statusManager 去更新 apiserver 中 pod 的状态信息。负责 Service 逻辑的组件获取到了这个状态,就能根据不同的值来决定是否需要更新 endpoints 的内容,也就是 service 的请求是否发送到这个 pod。

具体执行检测的代码在 pkg/kubelet/prober/prober.go 文件中,它会根据不同的 prober 方法(exec、HTTP、TCP)调用对应的处理逻辑,而这些具体的逻辑代码是在 pkg/probe/ 文件夹中,三种方法的实现都不复杂,就不再详细解释了。

原文地址
http://cizixs.com/2017/06/12/kubelet-source-code-analysis-part4-status-manager

发表评论

表情:
评论列表 (有 0 条评论,167人围观)

还没有评论,来说两句吧...

相关阅读