上篇文章我们在解析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基本原理解析到此为止，大家有什么问题或者建议，欢迎随时留言沟通。
