详解 Kubernetes Pod 的实现原理

阅读数:13 2019 年 12 月 4 日 09:51

详解 Kubernetes Pod 的实现原理

Pod,Service,Volume 和 Namespace 是 Kubernetes 进行中四大基本对象,它们能够表示系统中部署的应用,工作负载,网络和磁盘资源,共同定义了可用的状态。Kubernetes 中很多其他的资源实际上只对这些基本的对象进行了组合。

详解 Kubernetes Pod 的实现原理

Pod 是 Kubernetes 能够实现创建和管理最小的部署单元,想要彻底和完整的了解 Kubernetes 的实现原理,我们必须清楚 Pod 的实现原理以及最佳实践。

在这里,我们将分两部分对 Pod 进行解析,第一部分主要会从概念入手介绍 Pod 中必须了解的特性,而第二部分会介绍 Pod 从创建到删除的整个生命周期内的重要事件在源码民主党是如何实现的。

概述

作为 Kubernetes 内置中的基本单元,Pod 就是最小和最简单的 Kubernetes 对象,这个简单的对象其实就能够独立启动一个并进并在内部的调用方提供服务。在上一篇文章从 Kubernetes 中的对象谈起
中,我们曾经介绍过简单的 Kubernetes Pod 是如何使用 YAML 进行描述的:

YAML

复制代码
apiVersion: 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
是当前对象在 Kuberentes 扩展中的唯一标识符,而标签labels
可以帮助我们快速选择对象。

在同一个 Pod 中,有几个概念特别值得关注,首先就是容器,在 Pod 中实际上可以同时运行一个或多个容器,这些容器能够共享网络,存储以及 CPU,内存等资源。在这一小节中我们将关注 Pod 中的容器,卷和网络三大概念。

容器

每一个 Kubernetes 的 Pod 其实都具有两个不同的容器,两个不同容器的职责其实十分清晰,一种是InitContainer
,这种容器会在 Pod 启动时运行,主要用于初始化一些配置,另一种是 Pod 在运行状态时内部存活的Container
,它们的主要作用是对外提供服务或者作为工作中断处理异步任务等等。

详解 Kubernetes Pod 的实现原理

通过对不同容器类型的命名我们也可以抛光,InitContainer
会比Container
优先启动,在kubeGenericRuntimeManager.SyncPod
方法中会先后启动两种容器。

复制代码
func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, _ v1.PodStatus, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {
// Step 1: Compute sandbox and container changes.
// Step 2: Kill the pod if the sandbox has changed.
// Step 3: kill any running containers in this pod which are not to keep.
// Step 4: Create a sandbox for the pod if necessary.
// ...
// Step 5: start the init container.
if container := podContainerChanges.NextInitContainerToStart; container != nil {
msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeInit)
}
// Step 6: start containers in podContainerChanges.ContainersToStart.
for _, idx := range podContainerChanges.ContainersToStart {
container := &pod.Spec.Containers[idx]
msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeRegular)
}
return
}

通过分析专有方法startContainer
的实现我们得到的:容器的类型最终只会影响到调试时创建的标签,因此对于 Kubernetes 来说,这是两个容器的启动和执行也就只有顺序先后的不同。

每一个 Pod 中的容器是可以通过卷(Volume)
的方式共享文件目录的,这些 Volume 能够存储持久化的数据;在当前 Pod 出现故障或滚动更新时,对应 Volume 中的数据并不会被清除,甚至会在 Pod 中重新启动后重新挂载到期望的文件目录中:

详解 Kubernetes Pod 的实现原理

kubelet.go 文件中的专有方法syncPod
会调用WaitForAttachAndMount
方法为等待当前 Pod 启动需要的挂载文件:

复制代码
func (vm *volumeManager) WaitForAttachAndMount(pod *v1.Pod) error {
expectedVolumes := getExpectedVolumes(pod)
uniquePodName := util.GetUniquePodName(pod)
vm.desiredStateOfWorldPopulator.ReprocessPod(uniquePodName)
wait.PollImmediate(
podAttachAndMountRetryInterval,
podAttachAndMountTimeout,
vm.verifyVolumesMountedFunc(uniquePodName, expectedVolumes))
return nil
}

我们会在后面的章节
详细地介绍 Kubernetes 中卷的创建,挂载是如何进行的,在这里我们需要知道的是卷的挂载是 Pod 启动之前必须要完成的工作:

复制代码
func (kl *Kubelet) syncPod(o syncPodOptions) error {
// ...
if !kl.podIsTerminated(pod) {
kl.volumeManager.WaitForAttachAndMount(pod)
}
pullSecrets := kl.getPullSecretsForPod(pod)
result := kl.containerRuntime.SyncPod(pod, apiPodStatus, podStatus, pullSecrets, kl.backOff)
kl.reasonCache.Update(pod.UID, result)
return nil
}

在当前 Pod 的卷创建完成之后,将会调用上一段中提到的SyncPod
公有方法继续进行同步 Pod 信息和创建,启动容器的工作。

网路

同一个 Pod 中的多个容器会被共同分配到同一个主机,并且这些 Pod 能够通过 localhost 互相访问到彼此的端口和服务,如果使用了相同的端口也会发生冲突,同一个 Pod 上的所有容器会连接到同一个网络设备上,这个网络设备就是由 Pod Sandbox 中的沙箱容器在RunPodSandbox
方法中启动时创建的:

复制代码
func (ds *dockerService) RunPodSandbox(ctx context.Context, r *runtimeapi.RunPodSandboxRequest) (*runtimeapi.RunPodSandboxResponse, error) {
config := r.GetConfig()
// Step 1: Pull the image for the sandbox.
image := defaultSandboxImage
// Step 2: Create the sandbox container.
createConfig, _ := ds.makeSandboxDockerConfig(config, image)
createResp, _ := ds.client.CreateContainer(*createConfig)
resp := &runtimeapi.RunPodSandboxResponse{PodSandboxId: createResp.ID}
ds.setNetworkReady(createResp.ID, false)
// Step 3: Create Sandbox Checkpoint.
ds.checkpointManager.CreateCheckpoint(createResp.ID, constructPodSandboxCheckpoint(config))
// Step 4: Start the sandbox container.
ds.client.StartContainer(createResp.ID)
// Step 5: Setup networking for the sandbox.
cID := kubecontainer.BuildContainerID(runtimeName, createResp.ID)
networkOptions := make(map[string]string)
ds.network.SetUpPod(config.GetMetadata().Namespace, config.GetMetadata().Name, cID, config.Annotations, networkOptions)
return resp, nil
}

沙箱容器pause
实际上就是容器,上述方法引用的defaultSandboxImage
其实就是官方提供的k8s.gcr.io/pause:3.1
副本,这里会创建沙箱体积和检查点并启动容器。

详解 Kubernetes Pod 的实现原理

每一个上游上都会由 Kubernetes 的网络插件 Kubenet 创建一个基本的cbr0
网桥并为每一个 Pod 创建veth
虚拟网络设备,同一个 Pod 中的所有容器就会通过这个网络设备共享网络,也就是能够通过 localhost 互相访问彼此暴露的端口和服务。

小结

Kubernetes 中的每一个 Pod 都包含多个容器,这些容器在通过 Kubernetes 创建之后可以共享网络和存储,这其实是 Pod 非常重要的特性,我们能通过这个特性实现比较复杂的服务拓扑和依赖关系。

生命周期

想要深入了解 Pod 的实现原理,最好的时间的方法就是从 Pod 的生命周期入手,通过理解 Pod 创建,重新启动和删除的原理我们最终就能能够系统地掌握 Pod 的生命周期与核心原理。

详解 Kubernetes Pod 的实现原理

当 Pod 被创建之后,就会进入健康检查状态,当 Kubernetes 确定当前 Pod 已经能够接受外部的请求时,才会将流量打到新的 Pod 上并继续对外提供服务,在这期间如果发生了错误就可能会触发重启机制,在 Pod 被删除之前都会触发一个PreStop
的钩子,其中的方法之前完成之后 Pod 将会被删除,然后我们就会按照此处的顺序介绍 Pod『从生到死』的过程。

创建

Pod 的创建都是通过SyncPod
来实现的,创建的过程大体上可以分为六个步骤:

  1. 计算 Pod 中沙盒和容器的变更;
  2. 强制停止 Pod 对应的沙盒;
  3. 强制停止所有不应该运行的容器;
  4. 为 Pod 创建新的沙盒;
  5. 创建 Pod 规格中指定的初始化容器;
  6. 依次创建 Pod 规格中指定的常规容器;

我们可以看到 Pod 的创建过程其实是比较简单的,首先计算 Pod 规格和沙箱的变更,然后停止可能影响这一次创建或更新的容器,最后依次创建沙盒,初始化容器和常规容器。

复制代码
func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, _ v1.PodStatus, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {
podContainerChanges := m.computePodActions(pod, podStatus)
if podContainerChanges.CreateSandbox {
ref, _ := ref.GetReference(legacyscheme.Scheme, pod)
}
if podContainerChanges.KillPod {
if podContainerChanges.CreateSandbox {
m.purgeInitContainers(pod, podStatus)
}
} else {
for containerID, containerInfo := range podContainerChanges.ContainersToKill {
m.killContainer(pod, containerID, containerInfo.name, containerInfo.message, nil) }
}
}
podSandboxID := podContainerChanges.SandboxID
if podContainerChanges.CreateSandbox {
podSandboxID, _, _ = m.createPodSandbox(pod, podContainerChanges.Attempt)
}
podSandboxConfig, _ := m.generatePodSandboxConfig(pod, podContainerChanges.Attempt)
if container := podContainerChanges.NextInitContainerToStart; container != nil {
msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeInit)
}
for _, idx := range podContainerChanges.ContainersToStart {
container := &pod.Spec.Containers[idx]
msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeRegular)
}
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 规格的变量,根据输入的新规格执行相应的操作。

健康检查

如果我们符合 Pod 的最佳实践,实际上应该应该按地为每一个 Pod 添加livenessProbe
readinessProbe
的健康检查,这两者能够为 Kubernetes 提供额外的存活信息,如果我们配置了合适的健康检查方法和规则,那么就不会出现服务未启动就被打入流量或者长期未响应依然没有重启等问题。

在 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)
}
}

简化后的HandlePodAdditions
HandlePodRemoves
方法非常直白,我们可以直接来看ProbeManager
如何处理不同例程的健康检查。

详解 Kubernetes 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()
}
}
}

在执行健康检查的过程中,工人只是负责根据当前 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 的启动和初始化,在这之后会开始健康检查,替换的健康检查重试次数是三次,如果健康检查正常运行返回了一个确定的结果,那么 Worker 就是记录这次的结果,在连续失败FailureThreshold
次或成功SuccessThreshold
次,那么就会改变当前 Pod 的状态,这也是为了避免由于服务中断带来的误差。

删除

当 Kubelet 在HandlePodRemoves
方法中接收到来自客户端的删除请求时,就会通过一个称为deletePod
的私有方法中的频道将这一事件传递给 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
的事件:

详解 Kubernetes Pod 的实现原理

经过一系列的方法调用之后,最终调用容器运行时的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
}

对于每一个容器而言,它们在停止之前都会先调用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
}

从这个简化版本的killContainer
方法中,我们可以大致修剪停止运行容器的大致逻辑,先从 Pod 的规格中计算出当前停止所需要的时间,然后运行钩子方法和内部的生命周期方法,最后将容器停止并清除引用。

总结

在这篇文章中,我们已经介绍了 Pod 中的几个重要概念—容器,卷和网络以及从创建到删除整个过程是如何实现的。

Kubernetes 的运行和管理总是与 kubelet 以及它的组件密不可分,后面的文章中也介绍 kubelet 究竟是什么,它在整个 Kubernetes 扮演的角色。

相关文章

参考文献

本文转载自 Draveness 技术博客。
原文链接: https://draveness.me/kubernetes-pod

评论

发布