K8S Pod解析

对于K8S Container中，提到Pod的整个生命周期都是围绕“容器”这个核心进行运转，毕竟，Pod 是 Kubernetes 集群中能够被创建和管理的最小部署单元，只有弄清楚其底层原理以及实现细节，我们才能够对 Kubernetes生态体系有所了解。

当我们借助Kubernetes上创建Deployment时，Deployment会在其中创建带有容器的Pod以承载我们的应用程序实例。Pod从某种意义上来说它是Kubernetes的抽象，代表一组一个或多个应用程序容器（例如Docker）以及这些容器的一些共享资源。每个Pod都绑定到计划的节点上，并保持在那里，直到终止（根据重新启动策略）或删除为止。如果节点发生故障，则会在群集中的其他可用节点上调度相同的Pod。我们先来看一下Pod的模样，以便更清晰地了解其内部实现，具体如下图所示：

从上述图中，我们可以得知 ，Pod的形式有多种类型，有单一的纯容器应用，有挂在存储的应用，也涉及多个容器应用存在于同一个Pod中，具体部署形式应以实际的业务需求进行场景规划。

基于其设计理念，K8S Pod始终在Nodes（节点）上运行。节点是Kubernetes中的工作机，根据集群的不同，它可以是虚拟机也可以是物理机。每个节点由主节点管理。一个节点可以有多个Pod，Kubernetes主节点会自动处理跨集群中所有Node的Pod调度。主节点的自动调度考虑了每个节点上的可用资源。每个Kubernetes节点至少运行：

1、Kubelet，一个负责Kubernetes Master与Node之间通信的过程；它管理Pods和在一台机器上运行的容器。

2、容器运行时（类似Docker）负责从注册表中提取容器映像，解压缩容器并运行应用程序。

接下来，我们简单了解下Pod在Node下的运行信息，具体如下图所示：

基于上图，我们可以看到，所有的Pod均运行在Node（节点）上，只有借助节点，才能使得K8S对其进行相关资源调度，从而达到资源配置。

现在让我们来看下， Kubernetes Pod 是如何使用 基于Yaml 文件进行相关配置描述的，具体可参考如下：

piVersion: v1
kind: Pod
metadata:  
     name: busybox
     labels:   
        app: busybox
spec:  
    containers: 
    - image: busybox   
      command:     
           - sleep     
           - "3600"  
       imagePullPolicy: IfNotPresent  
       name: busybox 
    restartPolicy: Always

基于上述Yaml 文件，其主要描述了一个 Pod 启动时所加载运行的容器和执行命令以及它的重启策略，在当前 Pod 出现错误或者执行结束后是否应该被 Kubernetes 的控制器拉起来，除了这些比较常规的配置之外，元数据 metadata 的配置也非常重要，name 是当前对象在 Kubernetes 集群中的唯一标识符，而标签 labels 可以帮助我们快速选择对象。

接下来，我们再看一下K8S Pod的基本实现原理，再解析原理之前，我们首先了解下Pod的生命周期，只有通过理解Pod 创建、重启和删除的原理，我们才能最终就能够系统地掌握Pod的生命周期与核心原理。在这里我们先看一下 K8S Pod 生命周期流程图，具体如下所示：

基于上述流程图，我们可以看出：K8S Pod 被创建之后，首先会进入健康检查状态，当 Kubernetes 确定当前 Pod 已经能够接受外部的请求时，才会将流量打到新的 Pod 上并继续对外提供服务，在这期间如果发生了错误就可能会触发重启机制，在 Pod 被删除之前都会触发一个 PreStop 的钩子，其中的方法完成之后 Pod 才会被删除，接下来我们按照上述的流程图所述依次解析Pod从出生到消亡的具体过程。

我们知道在Kubelet中，其工作核心围绕着整个syncLoop来完成不同的工作模型。syncLoop会根据不同的上报信息管理Pod的生命周期，具体如下：

syncLoop循环监听管道信息

syncLoop的主要逻辑是在syncLoopIteration中实现，具体代码为pkg/kubelet/kubelet.go所示：

func (kl *Kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler) { 
    ...    
       syncTicker := time.NewTicker(time.Second)  
    defer syncTicker.Stop() 
    housekeepingTicker := time.NewTicker(housekeepingPeriod)  
    defer housekeepingTicker.Stop() 
    plegCh := kl.pleg.Watch() 
    for {   
       ...   
       kl.syncLoopMonitor.Store(kl.clock.Now())   
       if !kl.syncLoopIteration(updates, handler, syncTicker.C, housekeepingTicker.C, plegCh) {     
         break   
       }   
       kl.syncLoopMonitor.Store(kl.clock.Now()) 
    }
}

HandlePodAdditions创建Pod

HandlePodAdditions主要任务是：

1、按照创建时间给Pods进行排序。

2、将Pod添加至其管理器中，若Pod不存在在其管理器中，那么表明此Pod表示已经被删除。

3、校验Pod 是否能在该节点运行，如果不可以直接拒绝。

4、调用DispatchWork把 Pod 分配给给 worker 做异步处理并创建Pod。

5、将Pod添加到probeManager中，如果 Pod 中定义了 readiness 和 liveness 健康检查，启动 goroutine 定期进行检测。

其源码具体如下所示：

func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {  
    start := kl.clock.Now()  sort.
    Sort(sliceutils.PodsByCreationTime(pods)) 
    for _, pod := range pods {   
       existingPods := kl.podManager.GetPods()    
       //将pod添加到pod管理器中，如果有pod不存在在pod管理器中，那么这个pod表示已经被删除了   
       kl.podManager.AddPod(pod)       
       
       if kubetypes.IsMirrorPod(pod) {     
         kl.handleMirrorPod(pod, start)    
         continue   
        } 
        //如果该pod没有被Terminate  
        if !kl.podIsTerminated(pod) {      
          // 获取目前还在active状态的pod     
          activePods := kl.filterOutTerminatedPods(existingPods)     
          
          //验证 pod 是否能在该节点运行，如果不可以直接拒绝     
          if ok, reason, message :=
            kl.canAdmitPod(activePods, pod); !ok {        kl.rejectPod(pod, reason, message)       
            continue    
           }   
         }    
         mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)   
         //把 pod 分配给给 worker 做异步处理,创建pod  
         kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)    
         //在 probeManager 中添加 pod，如果 pod 中定义了 readiness 和 liveness 健康检查，启动 goroutine 定期进行检测  
         kl.probeManager.AddPod(pod)  
    }
}

与上述相关联的源码，大家有兴趣的话，可以去阅源码。现在我们详细解析下Pod不同状态的源码实现，主要涉及Create 、HealthCheak以及Delete操作，具体如下所示：

Create-创建

K8S Pod 的创建都是基于 SyncPod 来实现，其创建过程主要涉及以下步骤：首先计算 Pod 规格和沙箱的变更，然后停止可能影响这一次创建或者更新的容器，最后依次创建沙盒、初始化容器和常规容器。其源码pkg/kubelet/kuberuntime/kuberuntime_manager.go如下所示：

func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {  
    // 计算一下有哪些pod中container有没有变化，有哪些container需要创建,有哪些container需要kill掉
    podContainerChanges := m.computePodActions(pod, podStatus) 
    ...  
    
    // kill掉 sandbox 已经改变的 pod 
    if podContainerChanges.KillPod {  
       ...  
       //kill容器操作   
       killResult := m.killPodWithSyncResult(pod, kubecontainer.ConvertPodStatusToRunningPod(m.runtimeName, podStatus), nil)    
       result.AddPodSyncResult(killResult)    
       ... 
    } else {    
       // kill掉ContainersToKill列表中的container  
       for containerID, containerInfo := range podContainerChanges.ContainersToKill {    
          ...
          if err := m.killContainer(pod, containerID, containerInfo.name, containerInfo.message, nil); err != nil {
             killContainerResult.Fail(kubecontainer.ErrKillContainer, err.Error())        
             klog.Errorf("killContainer %q(id=%q) for pod %q failed: %v", containerInfo.name, containerID, format.Pod(pod), err)      
             return     
           }
        }
    }  
    
    //清理同名的 Init Container 
    m.pruneInitContainersBeforeStart(pod, podStatus)  
    
    var podIPs []string  
    if podStatus != nil {   
      podIPs = podStatus.IPs  
    }   
    podSandboxID := podContainerChanges.SandboxID  
    if podContainerChanges.CreateSandbox {    
      var msg string    
      var err error   
      ...   
      //为pod创建sandbox    
      podSandboxID, msg, err = m.createPodSandbox(pod, podContainerChanges.Attempt)    
      if err != nil {    
        ...     
        return   
       }   
       ...   
    }  
    
    podIP := "" 
    if len(podIPs) != 0 {   
       podIP = podIPs[0] 
    }  
    ...  
    //生成Sandbox的config配置，如pod的DNS、hostName、端口映射  
    podSandboxConfig, err := m.generatePodSandboxConfig(pod, podContainerChanges.Attempt) 
    if err != nil {  
       ...   
       return  
    }  
    
    start := func(typeName string, spec *startSpec) error {  
        ...   
        // 启动容器  
        if msg, err := m.startContainer(podSandboxID, podSandboxConfig, spec, pod, podStatus, pullSecrets, podIP, podIPs); err != nil {    
           ...   
        }    
        return nil  
    }  
    
    // 临时容器相关  
    if utilfeature.DefaultFeatureGate.Enabled(features.EphemeralContainers) {   
       for _, idx := range podContainerChanges.EphemeralContainersToStart {      
           start("ephemeral container", ephemeralContainerStartSpec(&pod.Spec.EphemeralContainers[idx]))   
        }  
     } 
     
     // 启动init container 
     if container := podContainerChanges.NextInitContainerToStart; container != nil {   
        if err := start("init container", containerStartSpec(container)); err != nil {      
          return    
        }    
        
        klog.V(4).Infof("Completed init container %q for pod %q", container.Name, format.Pod(pod))  
       }  
       // 启动containers列表 
       for _, idx := range podContainerChanges.ContainersToStart {    
           start("container", containerStartSpec(&pod.Spec.Containers[idx]))  
        }  
        
        return
}

基于上述的SyncPod 方法，我们可以很好地理解整个 Pod 的 创建 工作流程，而初始化容器和常规容器被调用 startContainer 来启动，具体如下源码所示：

func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandboxConfig *runtimeapi.PodSandboxConfig, container *v1.Container, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, podIP string, containerType kubecontainer.ContainerType) (string, error) { 
    imageRef, _, _ := m.imagePuller.EnsureImageExists(pod, container, pullSecrets)  
    
    // ...  
    containerID, _ := m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)  
    
    m.internalLifecycle.PreStartContainer(pod, container, containerID)  
    
    m.runtimeService.StartContainer(containerID)  
    
    if container.Lifecycle != nil && container.Lifecycle.PostStart != nil {  
       kubeContainerID := kubecontainer.ContainerID{     
           Type: m.runtimeName,    
           ID:   containerID,  
        }    
        msg, _ := m.runner.Run(kubeContainerID, pod, container, container.Lifecycle.PostStart)  
    }  
    
    return "", nil
}

在启动每一个容器的过程中也都按照相同的步骤进行操作，具体：

1、通过镜像拉取器获得当前容器中使用镜像的引用。

2、调用远程的 runtimeService 创建容器。

3、调用内部的生命周期方法 PreStartContainer 为当前的容器设置需要分配的 CPU 等系统资源。

4、调用远程的 runtimeService 开始运行镜像。

5、如果当前的容器包含 PostStart 钩子就会执行该回调。

每次 SyncPod 被调用时不仅仅是创建新的 Pod 对象，还会承担更新、删除和同步 Pod 规格的职能，根据输入的新规格执行相应的操作。

HealthCheak-健康检查

在K8S Pod 被创建或者被移除时，会被加入到当前节点上的 ProbeManager 中，ProbeManager 会负责对这些 Pod 进行健康检查，具体源码如下所示：

func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {  
    start := kl.clock.Now()
    for _, pod := range pods { 
       kl.podManager.AddPod(pod)  
       kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)    
       kl.probeManager.AddPod(pod)  
    }
}  

func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) { 
    start := kl.clock.Now() 
    for _, pod := range pods {    
       kl.podManager.DeletePod(pod) 
       kl.deletePod(pod)   
       kl.probeManager.RemovePod(pod)
    }
}

每一个新的 Pod 都会被调用 ProbeManager 的AddPod 函数，这个方法会初始化一个新的 Goroutine 并在其中运行对当前 Pod 进行健康检查，具体如下：

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

在进行健康检查的过程中，Worker 负责根据当前 Pod 的状态定期触发一次 Probe，它会根据 Pod 的配置分别选择调用 Exec、HTTPGet 或 TCPSocket 三种不同的 Probe 方式，具体如下所示：

func (pb *prober) runProbe(probeType probeType, p *v1.Probe, pod *v1.Pod, status v1.PodStatus, container v1.Container, containerID kubecontainer.ContainerID) (probe.Result, string, error) { 
    timeout := time.Duration(p.TimeoutSeconds) * time.Second 
    if p.Exec != nil {    
      command := kubecontainer.ExpandContainerCommandOnlyStatic(p.Exec.Command, container.Env)   
      return pb.exec.Probe(pb.newExecInContainer(container, containerID, command, timeout)) 
    } 
    if p.HTTPGet != nil {   
      scheme := strings.ToLower(string(p.HTTPGet.Scheme))  
      host := p.HTTPGet.Host  
      port, _ := extractPort(p.HTTPGet.Port, container)  
      path := p.HTTPGet.Path   
      url := formatURL(scheme, host, port, path)   
      headers := buildHeader(p.HTTPGet.HTTPHeaders) 
      if probeType == liveness {     
        return pb.livenessHttp.Probe(url, headers, timeout)  
      } else { // readiness     
         return pb.readinessHttp.Probe(url, headers, timeout)  
      } 
    } 
    if p.TCPSocket != nil {    
      port, _ := extractPort(p.TCPSocket.Port, container)  
      host := p.TCPSocket.Host    
      return pb.tcp.Probe(host, port, timeout) 
    } 
    return probe.Unknown, "", fmt.Errorf("Missing probe handler for %s:%s", format.Pod(pod), container.Name)
}

Kubernetes 在 Pod 启动后的 InitialDelaySeconds 时间内会等待 Pod 的启动和初始化，然后开始进行健康检查，默认的健康检查重试次数是3次，如果健康检查正常则返回一个确定的结果，此时Worker 记录这次的结果，在连续失败 FailureThreshold 次或者成功 SuccessThreshold 次，那么就会改变当前 Pod 的状态，这也是为了避免由于服务不稳定带来的抖动。

Delete-移除

当 Kubelet 在 HandlePodRemoves 方法中接收到来自客户端的Delete请求时，就会通过一个名为 deletePod 的私有方法中的 Channel 将这一事件传递给 PodKiller 进行处理，具体如下：

func (kl *Kubelet) deletePod(pod *v1.Pod) error {  
    kl.podWorkers.ForgetWorker(pod.UID)  
    
    runningPods, _ := kl.runtimeCache.GetPods()  
    runningPod := kubecontainer.Pods(runningPods).FindPod("", pod.UID)  
    podPair := kubecontainer.PodPair{APIPod: pod, RunningPod: &runningPod}  
    
    kl.podKillingCh <- &podPair 
    return nil
}

Kubelet 除了将事件通知给 PodKiller 之外，还需要将当前 Pod 对应的 Worker 从持有的 podWorkers 中移除；PodKiller 其实就是 Kubelet 持有的一个 Goroutine，它会在后台持续运行并监听来自 podKillingCh 的事件。

经过一系列的方法调用之后，最终调用容器运行时的 killContainersWithSyncResult 方法，这个方法会同步地杀掉当前 Pod 中全部的容器，具体如下：

func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(pod *v1.Pod, runningPod kubecontainer.Pod, gracePeriodOverride *int64) (syncResults []*kubecontainer.SyncResult) { 
    containerResults := make(chan *kubecontainer.SyncResult, len(runningPod.Containers))  
    
    for _, container := range runningPod.Containers {    
       go func(container *kubecontainer.Container) {      
           killContainerResult := kubecontainer.NewSyncResult(kubecontainer.KillContainer, container.Name)      
           m.killContainer(pod, container.ID, container.Name, "Need to kill Pod", gracePeriodOverride)      
           containerResults <- killContainerResult   
        }(container) 
      } 
      close(containerResults)  
      
      for containerResult := range containerResults {    
         syncResults = append(syncResults, containerResult) 
      }  
      return
}

在K8S设计理念中，对于每一个容器，在被停止之前都会先调用 PreStop 的钩子方法，让容器中的应用程序能够有时间完成一些未处理的操作，随后调用远程的服务停止运行的容器，具体如下：

func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubecontainer.ContainerID, containerName string, reason string, gracePeriodOverride *int64) error {  
containerSpec := kubecontainer.GetContainerSpec(pod, containerName);  

gracePeriod := int64(minimumGracePeriodInSeconds) 
switch {  
case pod.DeletionGracePeriodSeconds != nil:    
    gracePeriod = *pod.DeletionGracePeriodSeconds  
case pod.Spec.TerminationGracePeriodSeconds != nil:   
    gracePeriod = *pod.Spec.TerminationGracePeriodSeconds 
 }  

 m.executePreStopHook(pod, containerID, containerSpec, gracePeriod)  
 m.internalLifecycle.PreStopContainer(containerID.ID)  
 m.runtimeService.StopContainer(containerID.ID, gracePeriod)  
 m.containerRefManager.ClearRef(containerID)  

 return err
}

从上述源码可以获知，Pod移除操作的基本原理：先从 Pod 的规格中计算出当前停止所需要的时间，然后运行钩子方法和内部的生命周期方法，最后将容器停止并清除引用。

至此，关于Kubernetes Pod基本原理解析到此为止，大家有什么问题或者建议，欢迎随时留言沟通。