Shulou(Shulou.com)06/03 Report--

After dispatching a Pod to a Node in K8s, the subsequent state maintenance information is maintained by the kubelet on the corresponding machine. How to feed back the local running state in real time and notify the apiserver is the difficulty of design. This section mainly analyzes the core data structure of the Pod through two processes: sensing the state change and detecting the state change, so as to understand the internal design.

1. State Management 1.1 Static Pods

Static Pod mainly refers to those pods that are not created by apiserver, because apiserver does not contain them, but at the same time it needs to maintain and obtain the state of such pods. K8s has designed the concept of a mirror Pod, which is actually a mirror Pod for static Pod. The main information of this Pod is consistent with that of static Pod, and it is created in apiserver. The mirror Pod that apiserver can perceive reflects the state of real static Pod.

1.2 status data source

statusManager是进行状态同步的关键组件其需要综合当前Pod运行中的数据和apiserver存储的数据，从而决定最终的状态转换, 这里先关注图上画出来的，更多的状态等后续会一一介绍

2. 版本一致性type versionedPodStatus struct { status v1.PodStatus // 单调递增的版本号（每个pod） version uint64 // Pod name & namespace, for sending updates to API server. podName string podNamespace string}

在Kubelet中为保证与apiserver端信息的同步，在本地保存了一个Pod状态版本信息，其里面除了保存当前Pod的状态数据还有一个版本版本号，通过单调递增的版本号的对比来确定是否进行状态的同步

3. 核心源码实现

statusManager的流程其实还是蛮复杂的，今天我们就只讲一个场景，即kubelet通过apiserver感知到一个Pod更新，然后顺着该功能的数据流来进行梳理statusMangaer里面的数据流转

3.1 核心数据结构

manager中的核心状态相关的数据结构可以主要分为两大类：映射数据维护(podManager、podStatuses、apiStatusVersions)数据通信管道(podStatusChannel)， 剩余的则是对与apiserver通信的kublet和进行pod删除检查的 podDeletionSafety

type manager struct { kubeClient clientset.Interface // 管理缓存Pod，包含镜像pod和静态pod的映射 podManager kubepod.Manager // 从pod UID映射到相应pod的版本状态信息 。 podStatuses map[types.UID]versionedPodStatus podStatusesLock sync.RWMutex podStatusChannel chan podStatusSyncRequest // 存储镜像pod的版本 apiStatusVersions map[kubetypes.MirrorPodUID]uint64 podDeletionSafety PodDeletionSafetyProvider}3.2 设置Pod状态

设置Pod状态主要是位于kubelet中的syncPod中，在接收到pod事件变更之后，会与apiserver进行 Pod最新数据的同步从而获取当前pod在apiserver端的最新状态

func (m *manager) SetPodStatus(pod *v1.Pod, status v1.PodStatus) { m.podStatusesLock.Lock() defer m.podStatusesLock.Unlock() for _, c := range pod.Status.Conditions { if !kubetypes.PodConditionByKubelet(c.Type) { klog.Errorf("Kubelet is trying to update pod condition %q for pod %q. "+ "But it is not owned by kubelet.", string(c.Type), format.Pod(pod)) } } // Make sure we're caching a deep copy. status = *status.DeepCopy() // 如果Pod被删除了则需要强制与apiserver进行信息的同步 m.updateStatusInternal(pod, status, pod.DeletionTimestamp != nil)}3.3 更新内部缓存状态产生同步事件

3.3.1 获取缓存状态 var oldStatus v1.PodStatus // 检测之前的本地缓存数据 cachedStatus, isCached := m.podStatuses[pod.UID] if isCached { oldStatus = cachedStatus.status } else if mirrorPod, ok := m.podManager.GetMirrorPodByPod(pod); ok { oldStatus = mirrorPod.Status } else { oldStatus = pod.Status }3.3.2 检测容器状态

检测容器状态主要是针对容器终止状态转发的合法性进行检测，其实就是根据设定的Pod的RestartPolicy来检测针对一个终止的容器是否可以进行重启

if err := checkContainerStateTransition(oldStatus.ContainerStatuses, status.ContainerStatuses, pod.Spec.RestartPolicy); err != nil { klog.Errorf("Status update on pod %v/%v aborted: %v", pod.Namespace, pod.Name, err) return false } if err := checkContainerStateTransition(oldStatus.InitContainerStatuses, status.InitContainerStatuses, pod.Spec.RestartPolicy); err != nil { klog.Errorf("Status update on pod %v/%v aborted: %v", pod.Namespace, pod.Name, err) return false }3.3.3 更新PodCondition最后转换时间

通过最新的status里面的condition设定对应PodCondition的LastTransitionTime更新时间未当前时间

// Set ContainersReadyCondition.LastTransitionTime. updateLastTransitionTime(&status, &oldStatus, v1.ContainersReady) // Set ReadyCondition.LastTransitionTime. updateLastTransitionTime(&status, &oldStatus, v1.PodReady) // Set InitializedCondition.LastTransitionTime. updateLastTransitionTime(&status, &oldStatus, v1.PodInitialized) // Set PodScheduledCondition.LastTransitionTime. updateLastTransitionTime(&status, &oldStatus, v1.PodScheduled)3.3.4 校对时间截断过长信息

首先会根据当前容器的个数，从而决定每个容器最大的字节数大小，然后对容器里面的终止状态里面的Message信息，进行截断，同时进行时间的校对

normalizeStatus(pod, &status)3.3.5 状态更新条件检测

如果之前已经缓存了对应的数据，并且缓存的数据与当前的状态未发生改变，也不需要强制更新，就直接返回

if isCached && isPodStatusByKubeletEqual(&cachedStatus.status, &status) && !forceUpdate { // 如果不强制更新 ，默认是true此处不会成立 klog.V(3).Infof("Ignoring same status for pod %q, status: %+v", format.Pod(pod), status) return false // No new status. }3.3.6 生成同步事件更新缓存

生成最新的状态缓存数据，并且递增本地的版本信息

// 构建新的状态 newStatus := versionedPodStatus{ status: status, version: cachedStatus.version + 1, // 更新器缓存 podName: pod.Name, podNamespace: pod.Namespace, } // 更新新的缓存状态 m.podStatuses[pod.UID] = newStatus select { case m.podStatusChannel

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.