本篇文章没有接上篇继续更新 kube-controller-manager,kube-controller-manager 的源码阅读笔记也会继续更新,笔者会同时阅读多个组件的源码,阅读笔记也会按组件进行交叉更新,交叉更新的目的一是为了加深印象避免阅读完后又很快忘记,二是某些代码的功能难以理解,避免死磕,但整体目标是将每个组件的核心代码阅读完。
在前面的文章中已经介绍过 kubelet 的架构以及启动流程,本章会继续介绍 kubelet 中的核心功能,kubelet 中包含数十个 manager 以及对 CNI、CRI、CSI 的调用。每个 manager 的功能各不相同,manager 之间也会有依赖关系,本文会介绍比较简单的 statusManager。
kubernetes 版本:v1.16
statusManager 的主要功能是将 pod 状态信息同步到 apiserver,statusManage 并不会主动监控 pod 的状态,而是提供接口供其他 manager 进行调用。
kubelet 在启动流程时会在 NewMainKubelet
方法中初始化其核心组件,包括各种 manager。
k8s.io/kubernetes/pkg/kubelet/kubelet.go:335
func NewMainKubelet() (*Kubelet, error) {
......
// statusManager 的初始化
klet.statusManager = status.NewManager(klet.kubeClient, klet.podManager, klet)
......
}
NewManager
是用来初始化 statusManager 对象的,其中参数的功能如下所示:
k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:118
func NewManager(kubeClient clientset.Interface, podManager kubepod.Manager, podDeletionSafety PodDeletionSafetyProvider) Manager {
return &manager{
kubeClient: kubeClient,
podManager: podManager,
podStatuses: make(map[types.UID]versionedPodStatus),
podStatusChannel: make(chan podStatusSyncRequest, 1000),
apiStatusVersions: make(map[kubetypes.MirrorPodUID]uint64),
podDeletionSafety: podDeletionSafety,
}
}
在初始化完成后,kubelet 会在 Run 方法中会以 goroutine 的方式启动 statusManager。
k8s.io/kubernetes/pkg/kubelet/kubelet.go:1397
func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
......
kl.statusManager.Start()
......
}
statusManager 的代码主要在 k8s.io/kubernetes/pkg/kubelet/status/
目录中,其对外暴露的接口有以下几个:
type Manager interface {
// 一个 interface 用来暴露给其他组件获取 pod status 的
PodStatusProvider
// 启动 statusManager 的方法
Start()
// 设置 pod 的状态并会触发一个状态同步操作
SetPodStatus(pod *v1.Pod, status v1.PodStatus)
// 设置 pod .status.containerStatuses 中 container 是否为 ready 状态并触发状态同步操作
SetContainerReadiness(podUID types.UID, containerID kubecontainer.ContainerID, ready bool)
// 设置 pod .status.containerStatuses 中 container 是否为 started 状态并触发状态同步操作
SetContainerStartup(podUID types.UID, containerID kubecontainer.ContainerID, started bool)
// 将 pod .status.containerStatuses 和 .status.initContainerStatuses 中 container 的 state 置为 Terminated 状态并触发状态同步操作
TerminatePod(pod *v1.Pod)
// 从 statusManager 缓存 podStatuses 中删除对应的 pod
RemoveOrphanedStatuses(podUIDs map[types.UID]bool)
}
pod 对应的 status 字段如下所示:
status:
conditions:
......
containerStatuses:
- containerID: containerd://64e9d88459b38e90c2a4b4d87db5acd180c820c855a55aabe38e4e11b9b83576
image: docker.io/library/nginx:1.9
imageID: sha256:f568d3158b1e871b713cb33aca5a9377bc21a1f644addf41368393d28c35e894
lastState: {}
name: nginx-pod
ready: true
restartCount: 0
started: true
state:
running:
startedAt: "2019-12-15T16:13:29Z"
podIP: 10.15.225.15
......
然后继续看 statusManager 的启动方法 start
, 其主要逻辑为:
syncPeriod
默认为 10s;wait.Forever
goroutine 同步 pod 的状态,有两种同步方式,第一种是当监听到某个 pod 状态改变时会调用 m.syncPod
进行同步,第二种是当触发定时器时调用 m.syncBatch
进行批量同步;k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:147
func (m *manager) Start() {
// 1、检查 kubeClient 是否被初始化
if m.kubeClient == nil {
klog.Infof("Kubernetes client is nil, not starting status manager.")
return
}
// 2、设置定时器
syncTicker := time.Tick(syncPeriod)
go wait.Forever(func() {
select {
// 3、监听 m.podStatusChannel channel,当接收到数据时触发同步操作
case syncRequest := <-m.podStatusChannel:
......
m.syncPod(syncRequest.podUID, syncRequest.status)
// 4、定时同步
case <-syncTicker:
m.syncBatch()
}
}, 0)
}
syncPod
是用来同步 pod 最新状态至 apiserver 的方法,主要逻辑为:
m.needsUpdate
判断是否需要同步状态,若 apiStatusVersions
中的 status 版本号小于当前接收到的 status 版本号或者 apistatusVersions
中不存在该 status 版本号则需要同步,若不需要同步则继续检查 pod 是否处于删除状态,若处于删除状态调用 m.podDeletionSafety.PodResourcesAreReclaimed
将 pod 完全删除;oldStatus
与 currentStatus
的 uid 是否相等,若不相等则说明 pod 被重建过;statusutil.PatchPodStatus
同步 pod 最新的 status 至 apiserver,并将返回的 pod 作为 newPod;k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:514
func (m *manager) syncPod(uid types.UID, status versionedPodStatus) {
// 1、判断是否需要同步状态
if !m.needsUpdate(uid, status) {
klog.V(1).Infof("Status for pod %q is up-to-date; skipping", uid)
return
}
// 2、获取 pod 的 oldStatus
pod, err := m.kubeClient.CoreV1().Pods(status.podNamespace).Get(status.podName, metav1.GetOptions{})
if errors.IsNotFound(err) {
return
}
if err != nil {
return
}
translatedUID := m.podManager.TranslatePodUID(pod.UID)
// 3、检查 pod UID 是否已经改变
if len(translatedUID) > 0 && translatedUID != kubetypes.ResolvedPodUID(uid) {
return
}
// 4、同步 pod 最新的 status 至 apiserver
oldStatus := pod.Status.DeepCopy()
newPod, patchBytes, err := statusutil.PatchPodStatus(m.kubeClient, pod.Namespace, pod.Name, *oldStatus, mergePodStatus(*oldStatus, status.status))
if err != nil {
return
}
pod = newPod
m.apiStatusVersions[kubetypes.MirrorPodUID(pod.UID)] = status.version
// 5、若 newPod 处于 terminated 状态则调用 apiserver 删除该 pod,删除后 pod 会重建
if m.canBeDeleted(pod, status.status) {
deleteOptions := metav1.NewDeleteOptions(0)
deleteOptions.Preconditions = metav1.NewUIDPreconditions(string(pod.UID))
err = m.kubeClient.CoreV1().Pods(pod.Namespace).Delete(pod.Name, deleteOptions)
if err != nil {
return
}
// 6、从 cache 中清除
m.deletePodStatus(uid)
}
}
needsUpdate
方法主要是检查 pod 的状态是否需要更新,以及检查当 pod 处于 terminated 状态时保证 pod 被完全删除。
k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:570
func (m *manager) needsUpdate(uid types.UID, status versionedPodStatus) bool {
latest, ok := m.apiStatusVersions[kubetypes.MirrorPodUID(uid)]
if !ok || latest < status.version {
return true
}
pod, ok := m.podManager.GetPodByUID(uid)
if !ok {
return false
}
return m.canBeDeleted(pod, status.status)
}
func (m *manager) canBeDeleted(pod *v1.Pod, status v1.PodStatus) bool {
if pod.DeletionTimestamp == nil || kubepod.IsMirrorPod(pod) {
return false
}
// 此处说明 pod 已经处于删除状态了
return m.podDeletionSafety.PodResourcesAreReclaimed(pod, status)
}
PodResourcesAreReclaimed
检查 pod 在 node 上占用的所有资源是否已经被回收,其主要逻辑为:
k8s.io/kubernetes/pkg/kubelet/kubelet_pods.go:900
func (kl *Kubelet) PodResourcesAreReclaimed(pod *v1.Pod, status v1.PodStatus) bool {
// 1、检查 pod 中的所有 container 是否都处于非 running 状态
if !notRunning(status.ContainerStatuses) {
return false
}
// 2、从 podCache 中获取 podStatus,通过 podStatus 检查 pod 中的 container 是否已被完全删除
runtimeStatus, err := kl.podCache.Get(pod.UID)
if err != nil {
return false
}
if len(runtimeStatus.ContainerStatuses) > 0 {
var statusStr string
for _, status := range runtimeStatus.ContainerStatuses {
statusStr += fmt.Sprintf("%+v ", *status)
}
return false
}
// 3、检查 pod 的 volume 是否被清理
if kl.podVolumesExist(pod.UID) && !kl.keepTerminatedPodVolumes {
return false
}
// 4、检查 pod 的 cgroup 是否被清理
if kl.kubeletConfiguration.CgroupsPerQOS {
pcm := kl.containerManager.NewPodContainerManager()
if pcm.Exists(pod) {
return false
}
}
return true
}
syncBatch
是定期将 statusManager 缓存 podStatuses 中的数据同步到 apiserver 的方法,主要逻辑为:
m.podManager.GetUIDTranslations
从 podManager 中获取 mirrorPod uid 与 staticPod uid 的对应关系;needsUpdate
检查 pod 的状态是否与 apiStatusVersions 中的一致,然后调用 needsReconcile
检查 pod 的状态是否与 podManager 中的一致,若不一致则将需要同步的 pod 加入到 updatedStatuses 列表中;m.syncPod
方法同步状态;syncBatch 主要是将 statusManage cache 中的数据与 apiStatusVersions 和 podManager 中的数据进行对比是否一致,若不一致则以 statusManage cache 中的数据为准同步至 apiserver。
k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:469
func (m *manager) syncBatch() {
var updatedStatuses []podStatusSyncRequest
// 1、获取 mirrorPod 与 staticPod 的对应关系
podToMirror, mirrorToPod := m.podManager.GetUIDTranslations()
func() {
m.podStatusesLock.RLock()
defer m.podStatusesLock.RUnlock()
// 2、从 apiStatusVersions 中清理已经不存在的 pod
for uid := range m.apiStatusVersions {
_, hasPod := m.podStatuses[types.UID(uid)]
_, hasMirror := mirrorToPod[uid]
if !hasPod && !hasMirror {
delete(m.apiStatusVersions, uid)
}
}
// 3、遍历 podStatuses,将需要同步状态的 pod 加入到 updatedStatuses 列表中
for uid, status := range m.podStatuses {
syncedUID := kubetypes.MirrorPodUID(uid)
if mirrorUID, ok := podToMirror[kubetypes.ResolvedPodUID(uid)]; ok {
if mirrorUID == "" {
continue
}
syncedUID = mirrorUID
}
if m.needsUpdate(types.UID(syncedUID), status) {
updatedStatuses = append(updatedStatuses, podStatusSyncRequest{uid, status})
} else if m.needsReconcile(uid, status.status) {
delete(m.apiStatusVersions, syncedUID)
updatedStatuses = append(updatedStatuses, podStatusSyncRequest{uid, status})
}
}
}()
// 4、调用 m.syncPod 同步 pod 状态
for _, update := range updatedStatuses {
m.syncPod(update.podUID, update.status)
}
}
syncBatch
中主要调用了两个方法 needsUpdate
和 needsReconcile
,needsUpdate
在上文中已经介绍过了,下面介绍 needsReconcile
方法的主要逻辑。
needsReconcile
对比当前 pod 的状态与 podManager 中的状态是否一致,podManager 中保存了 node 上 pod 的 object,podManager 中的数据与 apiserver 是一致的,needsReconcile
主要逻辑为:
k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:598
func (m *manager) needsReconcile(uid types.UID, status v1.PodStatus) bool {
// 1、从 podManager 中获取 pod 对象
pod, ok := m.podManager.GetPodByUID(uid)
if !ok {
return false
}
// 2、检查 pod 是否为 static pod,若为 static pod 则获取其对应的 mirrorPod
if kubetypes.IsStaticPod(pod) {
mirrorPod, ok := m.podManager.GetMirrorPodByPod(pod)
if !ok {
return false
}
pod = mirrorPod
}
podStatus := pod.Status.DeepCopy()
// 3、格式化 pod status subResource
normalizeStatus(pod, podStatus)
// 4、检查 podManager 中的 status 与 statusManager cache 中的 status 是否一致
if isPodStatusByKubeletEqual(podStatus, &status) {
return false
}
return true
}
以上就是 statusManager 同步 pod status 的主要逻辑,下面再了解一下 statusManager 对其他组件暴露的方法。
SetPodStatus
是为 pod 设置 status subResource 并会触发同步操作,主要逻辑为:
pod.Status.Conditions
中的类型是否为 kubelet 创建的,kubelet 会创建 ContainersReady
、Initialized
、Ready
、PodScheduled
和 Unschedulable
五种类型的 conditions;m.updateStatusInternal
触发更新操作;k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:179
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))
}
}
status = *status.DeepCopy()
m.updateStatusInternal(pod, status, pod.DeletionTimestamp != nil)
}
statusManager 对外暴露的方法中触发状态同步的操作都是由 updateStatusInternal
完成的,updateStatusInternal
会更新 statusManager 的 cache 并会将 newStatus 发送到 m.podStatusChannel
中,然后 statusManager 会调用 syncPod
方法同步到 apiserver。
ContainerStatuses
和 InitContainerStatuses
是否合法;ContainersReady
、PodReady
、PodInitialized
、PodScheduled
conditions;StartTime
;m.podStatusChannel
中;k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:362
func (m *manager) updateStatusInternal(pod *v1.Pod, status v1.PodStatus, forceUpdate bool) bool {
var oldStatus v1.PodStatus
// 1、从 cache 中获取 oldStatus
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
}
// 2、检查 ContainerStatuses 和 InitContainerStatuses 是否合法
if err := checkContainerStateTransition(oldStatus.ContainerStatuses, status.ContainerStatuses, pod.Spec.RestartPolicy); err != nil {
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、为 status 设置 ContainersReady、PodReady、PodInitialized、PodScheduled conditions
updateLastTransitionTime(&status, &oldStatus, v1.ContainersReady)
updateLastTransitionTime(&status, &oldStatus, v1.PodReady)
updateLastTransitionTime(&status, &oldStatus, v1.PodInitialized)
updateLastTransitionTime(&status, &oldStatus, v1.PodScheduled)
// 4、设置 status 的 StartTime
if oldStatus.StartTime != nil && !oldStatus.StartTime.IsZero() {
status.StartTime = oldStatus.StartTime
} else if status.StartTime.IsZero() {
now := metav1.Now()
status.StartTime = &now
}
// 5、格式化 status
normalizeStatus(pod, &status)
if isCached && isPodStatusByKubeletEqual(&cachedStatus.status, &status) && !forceUpdate {
return false
}
// 6、将 newStatus 添加到 statusManager 的 cache podStatuses 中
newStatus := versionedPodStatus{
status: status,
version: cachedStatus.version + 1,
podName: pod.Name,
podNamespace: pod.Namespace,
}
m.podStatuses[pod.UID] = newStatus
// 7、将 newStatus 发送到 m.podStatusChannel 中
select {
case m.podStatusChannel <- podStatusSyncRequest{pod.UID, newStatus}:
return true
default:
return false
}
}
SetPodStatus
方法主要会用在 kubelet 的主 syncLoop 中,并在 syncPod
方法中创建 pod 时使用。
SetContainerReadiness
方法会设置 pod status subResource 中 container 是否为 ready 状态,其主要逻辑为:
m.podStatuses
获取 oldStatus;m.updateStatusInternal
触发同步操作;k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:198
func (m *manager) SetContainerReadiness(podUID types.UID, containerID kubecontainer.ContainerID, ready bool) {
m.podStatusesLock.Lock()
defer m.podStatusesLock.Unlock()
// 1、获取 pod 对象
pod, ok := m.podManager.GetPodByUID(podUID)
if !ok {
return
}
// 2、从 m.podStatuses 获取 oldStatus
oldStatus, found := m.podStatuses[pod.UID]
if !found {
return
}
// 3、通过 containerID 从 pod 中获取 containerStatus
containerStatus, _, ok := findContainerStatus(&oldStatus.status, containerID.String())
if !ok {
return
}
// 4、若 container status 为 Ready 直接返回,此时该 container 状态无须更新
if containerStatus.Ready == ready {
return
}
status := *oldStatus.status.DeepCopy()
containerStatus, _, _ = findContainerStatus(&status, containerID.String())
containerStatus.Ready = ready
updateConditionFunc := func(conditionType v1.PodConditionType, condition v1.PodCondition) {
conditionIndex := -1
for i, condition := range status.Conditions {
if condition.Type == conditionType {
conditionIndex = i
break
}
}
if conditionIndex != -1 {
status.Conditions[conditionIndex] = condition
} else {
status.Conditions = append(status.Conditions, condition)
}
}
// 5、添加 PodReadyCondition 和 ContainersReadyCondition
updateConditionFunc(v1.PodReady, GeneratePodReadyCondition(&pod.Spec, status.Conditions, status.ContainerStatuses, status.Phase))
updateConditionFunc(v1.ContainersReady, GenerateContainersReadyCondition(&pod.Spec, status.ContainerStatuses, status.Phase))
// 6、调用 m.updateStatusInternal 触发同步操作
m.updateStatusInternal(pod, status, false)
}
SetContainerReadiness
方法主要被用在 proberManager 中,关于 proberManager 的功能会在后文说明。
SetContainerStartup
方法会设置 pod status subResource 中 container 是否为 started 状态,主要逻辑为:
m.updateStatusInternal
触发同步操作;k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:255
func (m *manager) SetContainerStartup(podUID types.UID, containerID kubecontainer.ContainerID, started bool) {
m.podStatusesLock.Lock()
defer m.podStatusesLock.Unlock()
// 1、通过 podUID 从 podManager 中获取 pod 对象
pod, ok := m.podManager.GetPodByUID(podUID)
if !ok {
return
}
// 2、从 statusManager 中获取 pod 的 oldStatus
oldStatus, found := m.podStatuses[pod.UID]
if !found {
return
}
// 3、检查要更新的 container 是否存在
containerStatus, _, ok := findContainerStatus(&oldStatus.status, containerID.String())
if !ok {
klog.Warningf("Container startup changed for unknown container: %q - %q",
format.Pod(pod), containerID.String())
return
}
// 4、检查目标 container 的 started 状态是否已为期望值
if containerStatus.Started != nil && *containerStatus.Started == started {
return
}
// 5、设置目标 container 的 started 状态
status := *oldStatus.status.DeepCopy()
containerStatus, _, _ = findContainerStatus(&status, containerID.String())
containerStatus.Started = &started
// 6、触发同步操作
m.updateStatusInternal(pod, status, false)
}
SetContainerStartup
方法也是主要被用在 proberManager 中。
TerminatePod
方法的主要逻辑是把 pod .status.containerStatuses
和 .status.initContainerStatuses
中 container 的 state 置为 Terminated
状态并触发状态同步操作。
k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:312
func (m *manager) TerminatePod(pod *v1.Pod) {
m.podStatusesLock.Lock()
defer m.podStatusesLock.Unlock()
oldStatus := &pod.Status
if cachedStatus, ok := m.podStatuses[pod.UID]; ok {
oldStatus = &cachedStatus.status
}
status := *oldStatus.DeepCopy()
for i := range status.ContainerStatuses {
status.ContainerStatuses[i].State = v1.ContainerState{
Terminated: &v1.ContainerStateTerminated{},
}
}
for i := range status.InitContainerStatuses {
status.InitContainerStatuses[i].State = v1.ContainerState{
Terminated: &v1.ContainerStateTerminated{},
}
}
m.updateStatusInternal(pod, status, true)
}
TerminatePod
方法主要会用在 kubelet 的主 syncLoop 中。
RemoveOrphanedStatuses
的主要逻辑是从 statusManager 缓存 podStatuses 中删除对应的 pod。
k8s.io/kubernetes/pkg/kubelet/status/status_manager.go:457
func (m *manager) RemoveOrphanedStatuses(podUIDs map[types.UID]bool) {
m.podStatusesLock.Lock()
defer m.podStatusesLock.Unlock()
for key := range m.podStatuses {
if _, ok := podUIDs[key]; !ok {
klog.V(5).Infof("Removing %q from status map.", key)
delete(m.podStatuses, key)
}
}
}
本文主要介绍了 statusManager 的功能以及使用,其功能其实非常简单,当 pod 状态改变时 statusManager 会将状态同步到 apiserver,statusManager 也提供了多个接口供其他组件调用,当其他组件需要改变 pod 的状态时会将 pod 的 status 信息发送到 statusManager 进行同步。