深入理解Kubelet核心执行框架-创新互联

1.获取Pod配置

Kubelet有多种途径获取本节点需要运行的吊舱的配置信息。最重要的自然是API服务器，其次还能通过指定配置文件的目录以及访问特定的HTTP端口获取.Kubelet会定期对它们进行访问，获取波德配置的更新并及时调整位于本节点的Pod的运行状态。

在Kubelet初始化的时候会创建一个PodConfig对象如下所示：

// kubernetes/pkg/kubelet/config/config.go
type PodConfig struct {
    pods *podStorage
    mux  *config.Mux
    // the channel of denormalized changes passed to listeners
    updates chan kubetypes.PodUpdate
    ...
}

PodConfig本质上是Pod配置信息的一个复用器。内含的mux会对各种Pod配置信息的源（包括apiserver，file以及http）进行监听，定期同步各源当前的Pod配置状态。在pods中则缓存了上次同步时各源的Pod配置状态。mux将两者进行对比之后，即可得到配置发生变化的Pod。接着，它会根据变化类型对不同的Pod进行分类，每个类型的Pod注入一个PodUpdate结构中：

// kubernetes/pkg/kubelet/types/pod_update.go
type PodUpdate struct {
    Pods   []*v1.Pod
    Op     PodOperation
    Source string
}

Op字段即定义了上述的Pod变化类型。例如，它的值可以为ADD或REMOVE，表示对Pods中定义的Pod进行相应的增选。最后，各种类型的PodUpdate都会被注入到PodConfig的updates中。因此，我们只要对updates这个频道进行监听，就能得到所有有关本节点Pod的更新信息。

2.对Pod进行同步

当Kubelet初始化完成之后，最终会调用如下所示的syncLoop函数：

// kubernetes/pkg/kubelet/kubelet.go
// syncLoop is the main loop for processing changes. It watches for changes from
// three channels (file, apiserver, and http) and creates a union of them. For
// any new change seen, will run a sync against desired state and running state. If
// no changes are seen to the configuration, will synchronize the last known desired
// state every sync-frequency seconds. Never returns.
func (kl *Kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler){
    ...
    for {
        if !kl.syncLoopIteration(...) {
            break
        }        
    }
    ...
}

正如它的注释所表明的，syncLoop函数是Kubelet的主循环。它会对updates进行监听，获取吊舱的最新配置，并在当前状态（运行状态）和期望状态（所需状态）之间进行同步，使本节点的pod都按预期状态运行。事实上，syncLoop仅仅对对syncLoopIteration的封装，每一次具体的同步操作都将交由syncLoopIteration完成。

// kubernetes/pkg/kubelet/kubelet.go
func (kl *Kubelet) syncLoopIteration(configCh <-chan kubetypes.PodUpdate ......) bool {
    select {
    case u, open := <-configCh:
        switch u.Op {
        case kubetypes.ADD:
            handler.HandlePodAdditions(u.Pods)
        case kubetypes.UPDATE:
            handler.HandlePodUpdates(u.Pods)
        ...
        }
    case e := <-plegCh:
        ...
        handler.HandlePodSyncs([]*v1.Pod{pod})
        ...
    case <-syncCh:
        podsToSync := kl.getPodsToSync()
        if len(podsToSync) == 0 {
            break
        }
        handler.HandlePodSyncs(podsToSync)
    case update := <-kl.livenessManager.Updates():
        if update.Result == proberesults.Failure {
            ...
            handler.HandlePodSyncs([]*v1.Pod{pod})
        }
    case <-housekeepingCh:
         ...
        handler.HandlePodCleanups()
        ...
    }
}

syncLoopIteration函数的处理逻辑很简单，它会对多个渠道进行监听，一旦从某个渠道中获取到了某类事件，就调用相应的处理函数对其进行处理。下面，我们对各类事件做一个简单的叙述：

1. 从configCh中荚电子杂志配置信息的改变，并根据改变的类型，调用对应的处理函数。例如有新的吊舱绑定到本节点上时，调用就会HandlePodAdditions在本。节点上新建这些吊舱。如果某些荚的配置发生了改变，则会调用HandlePodUpdates对这些pod进行更新。

2. 若荚中有容器的状态发生了变化，例如有新的容器创建并运行，则会向plegCh这个通道发送PodlifecycleEvent这样一个事件，其中包含了事件类型ContainerStarted，该容器的ID，以及它所属Pod的ID，接着syncLoopIteration会调用HandlePodSyncs对该pod进行同步。

3. syncCh其实是一个定时器，Kubelet默认每隔一秒它就会触发一次，对当前节点上所有需要同步的pod进行同步。

4. Kubelet在初始化过程中会创建livenessManager，它会对进行了相关配置的吊舱进行健康检查。一旦检测到吊舱的运行状态出错，同样会调用HandlePodSyncs对相应的吊舱进行同步。关于这部分的内容，我们将在下文中详细描述。

5. houseKeepingCh同样是一个定时器，Kubelet默认每隔两秒它就会触发一次并调用处理函数HandlePodCleanups。简单地说，这就是一个定时清理的机制，每隔一段时间对那些已经结束运行的pod的相关资源进行回收。

podWorkers会在Kubelet初始化的过程中被创建，如下所示：

// kubernetes/pkg/kubelet/pod_workers.go
type podWorkers struct {
    ...
    podUpdates map[types.UID]chan UpdatePodOptions

    isWorking map[types.UID]bool

    lastUndeliveredWorkUpdate map[types.UID]UpdatePodOptions

    workQueue queue.WorkQueue

    syncPodFn syncPodFnType

    podCache kubecontainer.Cache
    ...
}

当Kubelet每创建一个新的吊舱，都会为其配置一个专有的荚工人。每个荚工人其实就是一个够程，它会创建一个缓存大小为1，类型为UpdatePodOptions（一个UpdatePodOptions就是一个荚更新事件）的信道，并不断对其监听来获取pod的更新事件并调用podWorkers中syncPodFn字段指定的同步函数进行具体的同步工作。

同时，pod worker会将该通道注册到podWorkers中的podUpdates这个地图中，从而可以让外部将指定的更新事件发送到对应的pod worker，让它进行处理。

如果pod worker正在处理某个更新，这时候又来了另外一个更新事件怎么办？podWorkers会将其中最新的一个缓存到lastUndeliveredWorkUpdate并在pod worker处理完当前更新事件之后马上对其进行处理。

最后，pod worker每处理完一次更新，都会将pod加入podWorkers的workQueue队列，而且会附加一个时延，只有时延消耗完了，才能将pod从队列中再次取出，进行下一次的同步。在上文中我们提到，每过1秒就会触发一次syncCh，收集本节点上需要进行同步的豆荚调用再HandlePodSyncs进行同步。事实上，那些荚从正是workQueue中的电子杂志，在当前时间节点，时延到期的吊舱。由此，整个pod的同步过程，如下所示，形成了一个闭环。

Kubelet在创建podWorkers对象的时候，会用自己的syncPod方法syncPodFn初始化。不过该方法所做的工作也仅仅是真正进行同步前的一些准备工作。例如将pod的最新状态上传给Apiserver，创建pod的专属目录，获取荚的拉动秘密等等。最终，Kubelet调用会所属其的containerRuntime的SyncPod方法进行同步工作。containerRuntime是Kubelet对底层容器运行时的一种抽象，定义了各种容器运行时需要满足的接口。SyncPod方法就是这些接口中的一个。

Kubelet并不会进行任何具体的容器相关的操作，所谓pod的同步，本质上还是对相关容器状态的改变，而要做到这一点，最终必然只能调用例如PouchContainer这样的底层容器运行时来完成。

下面，将我们展示进入containerRuntime的SyncPod方法，展示真正的同步工作：

// kubernetes/pkg/kubelet/kuberuntime/kuberuntime_manager.go
func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, _ v1.PodStatus, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult)

该函数首先会调用computePodActions(pod, podStatus)，对pod的当前状态podStatus和pod的目标状态pod进行比较，从而计算出我们要进行哪些具体的同步工作。计算结束之后，返回一个PodActions对象如下所示：

// kubernetes/pkg/kubelet/kuberuntime/kuberuntime_manager.go
type podActions struct {
    KillPod bool

    CreateSandbox bool

    SandboxID string

    Attempt uint32

    ContainersToKill map[kubecontainer.ContainerID]containerToKillInfo

    NextInitContainerToStart *v1.Container

    ContainersToStart []int
}

事实上，PodActions就是一个操作列表：

1. KillPod和CreateSandbox的值一般情况下是一致的，表示是否杀死当前Pod的Sandbox（若创建一个新Pod，则该操作为空）而创建一个新的

2. SandboxID用于对pod的创建操作进行标识，若它的值为空，表示第一次创建pod，否则表示杀死原有的sandbox而创建一个新的

3. Attempt表示pod重新创建sandbox的次数，第一次创建pod时，该值为0，作用和SandboxID是类似的

4. ContainersToKill指定了我们需要杀死的pod中的一些容器，之所以要删除它们，可能是因为容器的配置已经发生了变化，或者对它的健康检查失败了

5. 如果pod的init容器还没有全部运行完成或者在运行过程中出现了问题，NextInitContainerToStart表示下一个要创建的init container，创建并启动它，此次同步结束

6. 若荚的沙箱已经创建完成，init container也都运行完毕，则根据ContainersToStart启动pod中还未正常运行的普通容器

有了这样一份操作列表之后，SyncPod剩下的操作就异常简单了，无非是根据配置，按部就班地调用底层容器运行时的相应接口，进行具体的容器增删工作，完成同步。

总的来说，对于pod的同步可以简单归结为：当荚的目标状态发生改变，或者每隔一个同步周期，都会触发对相应pod的同步，而同步的具体内容就是将容器的目标状态和当前状态进行比对计算，生成一张容器的启停清单，根据该清单调用底层的容器运行时接口完成相应容器的启停工作。

总结

如果简单地将容器类比为一个进程的话，那么Kubelet本质上就是一个面向容器的进程监视器。它的任务就是不断地促成本节点pod的运行状态向目标状态转换。转换的方式也非常简单粗暴，如果有不符合要求的容器就直接删除，再根据新的配置重建一个，并不存在对一个已有容器反复修改启停的情况。到此为止，Kubelet核心的处理逻辑阐述完毕。

注：

1. 文中源码对应的Kubernetes版本为v1.9.4，commit：bee2d1505c4fe820744d26d41ecd3fdd4a3d6546

2. Kubernetes详细的源码注释参加我的github

参考文献

• Kubernetes源码

  https://github.com/YaoZengzeng/kubernetes

• 什么甚至是一个kubelet？

  http://kamalmarhubi.com/blog/2015/08/27/what-even-is-a-kubelet/

阿里百万级规模开源容器PouchContainer GA版本已发布。此版本延续以往的节奏，继续在Cloud Native（Kubernetes）生态的支持，以及容器安全隔离等方面做了持续性增强，同时开始孵化PouchContainer的插件机制，使得生态用户可以更加友好便捷地通过自研插件实现容器功能的扩展。

PouchContainer发布GA版本之前，已在阿里巴巴数据中心得到大规模的验证; GA版本发布之后，相信其一系列的突出特性同样可以服务于行业，作为一种开箱即用的系统软件技术，帮助行业服务在推进云原生架构转型上占得先机。

为了分享并促进社区的进步，邀请大家参加2018年9月9日（周日）上海PouchContainer Meetup，扫描上图二维码或者点击阅读原文立即报名。