作者Luga Lee

·2023-05-05 09:52

系统架构师·None

关注46

K8S Pod解析

字数 14271阅读 985评论 0赞 5

内容摘要：

Kubernetes中最小的部署单元是Pod，包括应用程序容器和共享资源，生命周期包括创建、重启和删除，实现原理涉及SyncPod、健康检查和移除等操作。

当前摘要为AI生成

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

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

![](/home/attachment/202305/05/162937_168324966712378.png)

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

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

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

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

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

![](/home/attachment/202305/05/162937_168324968057576.png)

基于上图，我们可以看到，所有的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 生命周期流程图，具体如下所示：

![](/home/attachment/202305/05/162937_168324976933810.png)

基于上述流程图，我们可以看出：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基本原理解析到此为止，大家有什么问题或者建议，欢迎随时留言沟通。

容器 Kubernetes

著作权归作者所有

如果觉得我的文章对您有用，请点赞。您的支持将鼓励我继续创作！

添加新评论0 条评论

Ctrl+Enter 发表

匿名评论

K8S Pod解析

添加新评论0 条评论

文章目录展开目录

作者其他文章

相关文章

相关问题

相关资料