kubelet 源码分析：statusManager 和 probeManager-蒲公英云

kubelet 源码分析：statusManager 和 probeManager

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

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

klet.statusManager = status.NewManager(kubeClient, klet.podManager)
klet.probeManager = prober.NewManager(
        klet.statusManager,
        klet.livenessManager,
        klet.runner,
        containerRefManager,
        kubeDeps.Recorder)

StatusManager

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

type PodStatusProvider interface { GetPodStatus(uid types.UID) (api.PodStatus, bool) }
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 的状态。修改状态的方法有多个，每个都有不同的用途：

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

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

const syncPeriod = 10 * time.Second
func (m *manager) Start() {
    ......
    glog.Info("Starting to sync pod status with apiserver")
    syncTicker := time.Tick(syncPeriod)
    // syncPod and syncBatch share the same go routine to avoid sync races.
    go wait.Forever(func() {
        select {
        case syncRequest := <-m.podStatusChannel:
            m.syncPod(syncRequest.podUID, syncRequest.status)
        case <-syncTicker:
            m.syncBatch()
        }
    }, 0)
}

这个 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 文件中：

func (m *manager) AddPod(pod *api.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.ReadinessProbe != nil {
            key.probeType = readiness
            if _, ok := m.workers[key]; ok { 
                glog.Errorf("Readiness probe already exists! %v - %v",
                    format.Pod(pod), c.Name)
                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 { 
                glog.Errorf("Liveness probe already exists! %v - %v",
                    format.Pod(pod), c.Name)
                return
            }
            w := newWorker(m, liveness, pod, c)
            m.workers[key] = w
            go w.run()
        }
    }
}

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

pkg/kubelet/prober/worker.go：

func (w *worker) run() {
    probeTickerPeriod := time.Duration(w.spec.PeriodSeconds) * time.Second
    probeTicker := time.NewTicker(probeTickerPeriod)
    defer func() {
        probeTicker.Stop()
        if !w.containerID.IsEmpty() {
            w.resultsManager.Remove(w.containerID)
        }
        w.probeManager.removeWorker(w.pod.UID, w.container.Name, w.probeType)
    }()
    time.Sleep(time.Duration(rand.Float64() * float64(probeTickerPeriod)))
probeLoop:
    for w.doProbe() {
        // Wait for next probe tick.
        select {
        case <-w.stopCh:
            break probeLoop
        case <-probeTicker.C:
            // continue
        }
    }
}
func (w *worker) doProbe() (keepGoing bool) {
    defer func() { recover() }() 
    defer runtime.HandleCrash(func(_ interface{}) { keepGoing = true })
    // pod 没有被创建，或者已经被删除了，直接跳过检测，但是会继续检测
    status, ok := w.probeManager.statusManager.GetPodStatus(w.pod.UID)
    if !ok {
        glog.V(3).Infof("No status for pod: %v", format.Pod(w.pod))
        return true
    }
    // pod 已经退出（不管是成功还是失败），直接返回，并终止 worker
    if status.Phase == api.PodFailed || status.Phase == api.PodSucceeded {
        glog.V(3).Infof("Pod %v %v, exiting probe worker",
            format.Pod(w.pod), status.Phase)
        return false
    }
    // 容器没有创建，或者已经删除了，直接返回，并继续检测，等待更多的信息
    c, ok := api.GetContainerStatus(status.ContainerStatuses, w.container.Name)
    if !ok || len(c.ContainerID) == 0 {
        glog.V(3).Infof("Probe target container not found: %v - %v",
            format.Pod(w.pod), w.container.Name)
        return true 
    }
    // pod 更新了容器，使用最新的容器信息
    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)
        w.onHold = false
    }
    if w.onHold {
        return true
    }
    if c.State.Running == nil {
        glog.V(3).Infof("Non-running container probed: %v - %v",
            format.Pod(w.pod), w.container.Name)
        if !w.containerID.IsEmpty() {
            w.resultsManager.Set(w.containerID, results.Failure, w.pod)
        }
        // 容器失败退出，并且不会再重启，终止 worker
        return c.State.Terminated == nil ||
            w.pod.Spec.RestartPolicy != api.RestartPolicyNever
    }
    // 容器启动时间太短，没有超过配置的初始化等待时间 InitialDelaySeconds
    if int32(time.Since(c.State.Running.StartedAt.Time).Seconds()) < w.spec.InitialDelaySeconds {
        return true
    }
    // 调用 prober 进行检测容器的状态
    result, err := w.probeManager.prober.probe(w.probeType, w.pod, status, w.container, w.containerID)
    if err != nil {
        return true
    }
    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)) {
        return true
    }
    // 保存最新的检测结果
    w.resultsManager.Set(w.containerID, result, w.pod)
    if w.probeType == liveness && result == results.Failure {
        // 容器 liveness 检测失败，需要删除容器并重新创建，在新容器成功创建出来之前，暂停检测
        w.onHold = true
    }
    return true
}

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

func (m *manager) Set(id kubecontainer.ContainerID, result Result, pod *api.Pod) {
    if m.setInternal(id, result) {
        m.updates <- Update{id, result, pod.UID}
    }
}
func (m *manager) setInternal(id kubecontainer.ContainerID, result Result) bool {
    m.Lock()
    defer m.Unlock()
    prev, exists := m.cache[id]
    if !exists || prev != result {
        m.cache[id] = result
        return true
    }
    return false
}
func (m *manager) Updates() <-chan Update {
    return m.updates
}
它把结果保存在缓存中，并发送到 m.updates 管道。对于 liveness 来说，它的管道消费者是 kubelet，还记得 syncLoopIteration 中的这段代码逻辑吗？
case update := <-kl.livenessManager.Updates():
        if update.Result == proberesults.Failure {
            // The liveness manager detected a failure; sync the pod.
            pod, ok := kl.podManager.GetPodByUID(update.PodUID)
            if !ok {
                // If the pod no longer exists, ignore the update.
                glog.V(4).Infof("SyncLoop (container unhealthy): ignore irrelevant update: %#v", update)
                break
            }
            glog.V(1).Infof("SyncLoop (container unhealthy): %q", format.Pod(pod))
            handler.HandlePodSyncs([]*api.Pod{pod})
        }
因为 liveness 关系者 pod 的生死，因此需要 kubelet 的处理逻辑。而 readiness 即使失败也不会重新创建 pod，它的处理逻辑是不同的，它的处理代码同样在 pkg/kubelet/prober/prober_manager.go：
func (m *manager) Start() {
    go wait.Forever(m.updateReadiness, 0)
}
func (m *manager) updateReadiness() {
    update := <-m.readinessManager.Updates()
    ready := update.Result == results.Success
    m.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
}

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