一篇带给你pod创建源码分析

一篇带给你pod创建源码分析

大家好，我是华仔。

接触kubernetes已经4年多了，不过多是停留在能够使用，对其原理、源码不是很熟悉。对于平常执行的命令，它背后执行的流程、逻辑也不是很清楚。所以，最近打算去看看k8s各模块的源码。一来是加深对k8s各模块的理解和认识;二来是方便以后遇到问题好分析问题的根本原因，有理有据，则可以服人;再者后续跳槽也不怕被面试官的技术问题所难到了。那么今天，就来简单说一说pod创建的源码吧。文章有错误的地方还请指正，轻喷。首先，k8s的源码在github上即可获取。本次我看的是1.21.3。另外，很多翻译都是直译或翻译软件翻译的。请谅解。

正文

1、k8s源码中针对pod的增删改查是在源码包/pkg/kubelet/kubelet.go中的syncLoop()进行。如下所示：

在syncLoop()中则通过kl.syncLoopIteration()针对pod具体执行具体的操作。

kl.syncLoopMonitor.Store(kl.clock.Now()) if !kl.syncLoopIteration(updates, handler, syncTicker.C, housekeepingTicker.C, plegCh) {  break }

2、在syncLoopIteration有几个重要的参数，如下所示：

// Arguments: // 1.  configCh:       a channel to read config events from // 2.  handler:        the SyncHandler to dispatch pods to // 3.  syncCh:         a channel to read periodic sync events from // 4.  housekeepingCh: a channel to read housekeeping events from // 5.  plegCh:         a channel to read PLEG updates from  // * configCh: dispatch the pods for the config change to the appropriate //             handler callback for the event type // * plegCh: update the runtime cache; sync pod // * syncCh: sync all pods waiting for sync // * housekeepingCh: trigger cleanup of pods // * health manager: sync pods that have failed or in which one or more //                     containers have failed health checks  func (kl *Kubelet) syncLoopIteration(configCh <-chan kubetypes.PodUpdate, handler SyncHandler,  syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool {  select {  case u, open := <-configCh:   // Update from a config source; dispatch it to the right handler   // callback.   if !open {    klog.ErrorS(nil, "Update channel is closed, exiting the sync loop")    return false   }

SyncHandler是一个interface。包含对pod常见操作的几个方法。该接口由kubelet来实现。如下所示：

// SyncHandler is an interface implemented by Kubelet, for testability # pod创建、更新、 删除... type SyncHandler interface {  HandlePodAdditions(pods []*v1.Pod)   HandlePodUpdates(pods []*v1.Pod)   HandlePodRemoves(pods []*v1.Pod)  HandlePodReconcile(pods []*v1.Pod)  HandlePodSyncs(pods []*v1.Pod)  HandlePodCleanups() error }

3、针对pod可进行的操作如下，每个操作都有对应的方法。比如ADD，就会去执行HandlePodAdditions方法

// These constants identify the PodOperations that can be made on a pod configuration. const (  // SET is the current pod configuration.  SET PodOperation = iota  // ADD signifies pods that are new to this source.  ADD  // DELETE signifies pods that are gracefully deleted from this source.  DELETE  // REMOVE signifies pods that have been removed from this source.  REMOVE  // UPDATE signifies pods have been updated in this source.  UPDATE  // RECONCILE signifies pods that have unexpected status in this source,  // kubelet should reconcile status with this source.  RECONCILE )   switch u.Op {   case kubetypes.ADD:    klog.V(2).InfoS("SyncLoop ADD", "source", u.Source, "pods", format.Pods(u.Pods))    // After restarting, kubelet will get all existing pods through    // ADD as if they are new pods. These pods will then go through the    // admission process and *may* be rejected. This can be resolved    // once we have checkpointing.    handler.HandlePodAdditions(u.Pods)

4、HandlePodAdditions又是如何去执行创建pod的呢?主要有以下几个操作：

1. 根据pod的创建时间进行排序 sort.Sort(sliceutils.PodsByCreationTime(pods))  2. 将pod添加到podmanager中.因为kubelet它会依赖这个pod manager作为期望状态的一个凭证。 如果一个在pod manager中无法查询，那么就意味着它已经被apiserver删除了，不再需要其他操作 // Always add the pod to the pod manager. Kubelet relies on the pod // manager as the source of truth for the desired state. If a pod does // not exist in the pod manager, it means that it has been deleted in // the apiserver and no action (other than cleanup) is required. kl.podManager.AddPod(pod)  3. 判断pod是不是静态pod mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)  4. 通过dispatchWork分发任务 kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)  5. 将pod加入到probe manager，即健康检查.包括startup probe、liveness probe、readiness probe。  kl.probeManager.AddPod(pod)

dispatchWork又做了哪些事情呢?如下：

// Run the sync in an async worker. 在一个异步worker中执行同步  kl.podWorkers.UpdatePod(&UpdatePodOptions{   Pod:        pod,   MirrorPod:  mirrorPod,   UpdateType: syncType,   OnCompleteFunc: func(err error) {    if err != nil {     metrics.PodWorkerDuration.WithLabelValues(syncType.String()).Observe(metrics.SinceInSeconds(start))    }   },  })

那么UpdatePod()又做哪些事情呢?

managePodLoop()去执行同步。

for update := range podUpdates {   err := func() error {    podUID := update.Pod.UID    // This is a blocking call that would return only if the cache    // has an entry for the pod that is newer than minRuntimeCache    // Time. This ensures the worker doesn't start syncing until    // after the cache is at least newer than the finished time of    // the previous sync.    status, err := p.podCache.GetNewerThan(podUID, lastSyncTime)    if err != nil {     // This is the legacy event thrown by manage pod loop     // all other events are now dispatched from syncPodFn     p.recorder.Eventf(update.Pod, v1.EventTypeWarning, events.FailedSync, "error determining status: %v", err)     return err    }       // 这里去做同步    err = p.syncPodFn(syncPodOptions{     mirrorPod:      update.MirrorPod,     pod:            update.Pod,     podStatus:      status,     killPodOptions: update.KillPodOptions,     updateType:     update.UpdateType,    })    lastSyncTime = time.Now()    return err   }()

5、最终调用到pkg/kubelet/kuberuntime/kuberuntime_manager.go中SyncPod()进行pod的创建

6、另外，pod worker还要做以下事情：

# 创建pod数据目录、volume、获取image pull secrets。。。 newPodWorkers(klet.syncPod --->pkg/kubelet/kubelet.go） //通过syncPod                                kubetypes.SyncPodKill                                 kubetypes.SyncPodCreate                                 podStatus.IPs = append(podStatus.IPs, ipInfo.IP)                                 runnable.Admit                                 kubetypes.IsStaticPod(pod)                                kl.makePodDataDirs(pod)                                 kl.volumeManager.WaitForAttachAndMount(pod)                                 kl.getPullSecretsForPod(pod)                                 kl.containerRuntime.SyncPod(pkg/kubelet/container/runtime.go)