1 创建pod流程

2 syncLoop函数

func (kl *Kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
   glog.Info("Starting kubelet main sync loop.")
   // The resyncTicker wakes up kubelet to checks if there are any pod workers
   // that need to be sync'd. A one-second period is sufficient because the
   // sync interval is defaulted to 10s.
   syncTicker := time.NewTicker(time.Second)
   defer syncTicker.Stop()
   housekeepingTicker := time.NewTicker(housekeepingPeriod)
   defer housekeepingTicker.Stop()
   plegCh := kl.pleg.Watch()
   for {
      if rs := kl.runtimeState.runtimeErrors(); len(rs) != 0 {
         glog.Infof("skipping pod synchronization - %v", rs)
         time.Sleep(5 * time.Second)
         continue
      }

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

3 syncLoopIteration函数

func (kl *Kubelet) syncLoopIteration(configCh <-chan kubetypes.PodUpdate, handler SyncHandler,
   syncCh <-chan time.Time, housekeepingCh <-chan time.Time, plegCh <-chan *pleg.PodLifecycleEvent) bool

3.1 HandlePodAddtions函数

func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
   start := kl.clock.Now()
   sort.Sort(sliceutils.PodsByCreationTime(pods))
   for _, pod := range pods {
      existingPods := kl.podManager.GetPods()

      kl.podManager.AddPod(pod)

      if kubepod.IsMirrorPod(pod) {
         kl.handleMirrorPod(pod, start)
         continue
      }

      if !kl.podIsTerminated(pod) {
         activePods := kl.filterOutTerminatedPods(existingPods)

         if ok, reason, message := kl.canAdmitPod(activePods, pod); !ok {
            kl.rejectPod(pod, reason, message)
            continue
         }
      }
      mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
      kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)
      kl.probeManager.AddPod(pod)
   }
}

3.1.1 dispatchWork函数

func (kl *Kubelet) dispatchWork(pod *v1.Pod, syncType kubetypes.SyncPodType, mirrorPod *v1.Pod, start time.Time) {
   if kl.podIsTerminated(pod) {
      if pod.DeletionTimestamp != nil {
         kl.statusManager.TerminatePod(pod)
      }
      return
   }
   // Run the sync in an async worker.
   kl.podWorkers.UpdatePod(&UpdatePodOptions{
      Pod:        pod,
      MirrorPod:  mirrorPod,
      UpdateType: syncType,
      OnCompleteFunc: func(err error) {
         if err != nil {
            metrics.PodWorkerLatency.WithLabelValues(syncType.String()).Observe(metrics.SinceInMicroseconds(start))
         }
      },
   })
   // Note the number of containers for new pods.
   if syncType == kubetypes.SyncPodCreate {
      metrics.ContainersPerPodCount.Observe(float64(len(pod.Spec.Containers)))
   }
}

3.1.1.1 UpdatePod函数

func (p *podWorkers) UpdatePod(options *UpdatePodOptions) {
   pod := options.Pod
   uid := pod.UID
   var podUpdates chan UpdatePodOptions
   var exists bool

   p.podLock.Lock()
   defer p.podLock.Unlock()
   if podUpdates, exists = p.podUpdates[uid]; !exists {
      podUpdates = make(chan UpdatePodOptions, 1)
      p.podUpdates[uid] = podUpdates

      go func() {
         defer runtime.HandleCrash()
         p.managePodLoop(podUpdates)
      }()
   }
   if !p.isWorking[pod.UID] {
      p.isWorking[pod.UID] = true
      podUpdates <- *options
   } else {
      // if a request to kill a pod is pending, we do not let anything overwrite that request.
      update, found := p.lastUndeliveredWorkUpdate[pod.UID]
      if !found || update.UpdateType != kubetypes.SyncPodKill {
         p.lastUndeliveredWorkUpdate[pod.UID] = *options
      }
   }
}

3.1.1.1 managePodLoop函数

func (p *podWorkers) managePodLoop(podUpdates <-chan UpdatePodOptions) {
   var lastSyncTime time.Time
   for update := range podUpdates {
      err := func() error {
         podUID := update.Pod.UID
         status, err := p.podCache.GetNewerThan(podUID, lastSyncTime)
         if err != nil {
            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
      }()
      // notify the call-back function if the operation succeeded or not
      if update.OnCompleteFunc != nil {
         update.OnCompleteFunc(err)
      }
      if err != nil {
         p.recorder.Eventf(update.Pod, v1.EventTypeWarning, events.FailedSync, "Error syncing pod")
      }
      p.wrapUp(update.Pod.UID, err)
   }
}

3.1.1.1.1 syncPod函数

// syncPod is the transaction script for the sync of a single pod.
//
// Arguments:
//
// o - the SyncPodOptions for this invocation
//
// The workflow is:
// * If the pod is being created, record pod worker start latency
// * Call generateAPIPodStatus to prepare an v1.PodStatus for the pod
// * If the pod is being seen as running for the first time, record pod
//   start latency
// * Update the status of the pod in the status manager
// * Kill the pod if it should not be running
// * Create a mirror pod if the pod is a static pod, and does not
//   already have a mirror pod
// * Create the data directories for the pod if they do not exist
// * Wait for volumes to attach/mount
// * Fetch the pull secrets for the pod
// * Call the container runtime's SyncPod callback
// * Update the traffic shaping for the pod's ingress and egress limits
//
// If any step of this workflow errors, the error is returned, and is repeated
// on the next syncPod call.
func (kl *Kubelet) syncPod(o syncPodOptions) error

a. 如删除 pod，执行并返回 

// if we want to kill a pod, do it now!
if updateType == kubetypes.SyncPodKill {
   killPodOptions := o.killPodOptions
   if killPodOptions == nil || killPodOptions.PodStatusFunc == nil {
      return fmt.Errorf("kill pod options are required if update type is kill")
   }
   apiPodStatus := killPodOptions.PodStatusFunc(pod, podStatus)
   kl.statusManager.SetPodStatus(pod, apiPodStatus)
   // we kill the pod with the specified grace period since this is a termination
   if err := kl.killPod(pod, nil, podStatus, killPodOptions.PodTerminationGracePeriodSecondsOverride); err != nil {
      // there was an error killing the pod, so we return that error directly
      utilruntime.HandleError(err)
      return err
   }
   return nil
}

b. 检查 pod ，主要是权限检查（使用主机网络模式，privileged 权限运行等）。如果没有权限，就删除本地旧的 pod 并返回错误信息 

runnable := kl.canRunPod(pod)
if !runnable.Admit {
   // Pod is not runnable; update the Pod and Container statuses to why.
   apiPodStatus.Reason = runnable.Reason
   apiPodStatus.Message = runnable.Message
   // Waiting containers are not creating.
   const waitingReason = "Blocked"
   for _, cs := range apiPodStatus.InitContainerStatuses {
      if cs.State.Waiting != nil {
         cs.State.Waiting.Reason = waitingReason
      }
   }
   for _, cs := range apiPodStatus.ContainerStatuses {
      if cs.State.Waiting != nil {
         cs.State.Waiting.Reason = waitingReason
      }
   }
}

func canRunPod(pod *v1.Pod) error {
   if !capabilities.Get().AllowPrivileged {
      for _, container := range pod.Spec.Containers {
         if securitycontext.HasPrivilegedRequest(&container) {
            return fmt.Errorf("pod with UID %q specified privileged container, but is disallowed", pod.UID)
         }
      }
      for _, container := range pod.Spec.InitContainers {
         if securitycontext.HasPrivilegedRequest(&container) {
            return fmt.Errorf("pod with UID %q specified privileged init container, but is disallowed", pod.UID)
         }
      }
   }

   if pod.Spec.HostNetwork {
      allowed, err := allowHostNetwork(pod)
      if err != nil {
         return err
      }
      if !allowed {
         return fmt.Errorf("pod with UID %q specified host networking, but is disallowed", pod.UID)
      }
   }

   if pod.Spec.HostPID {
      allowed, err := allowHostPID(pod)
      if err != nil {
         return err
      }
      if !allowed {
         return fmt.Errorf("pod with UID %q specified host PID, but is disallowed", pod.UID)
      }
   }

   if pod.Spec.HostIPC {
      allowed, err := allowHostIPC(pod)
      if err != nil {
         return err
      }
      if !allowed {
         return fmt.Errorf("pod with UID %q specified host ipc, but is disallowed", pod.UID)
      }
   }

   return nil
}

c. 如果是 static Pod，就创建或者更新对应的 mirrorPod 

if kubepod.IsStaticPod(pod) {
   podFullName := kubecontainer.GetPodFullName(pod)
   deleted := false
   if mirrorPod != nil {
      if mirrorPod.DeletionTimestamp != nil || !kl.podManager.IsMirrorPodOf(mirrorPod, pod) {
         // The mirror pod is semantically different from the static pod. Remove
         // it. The mirror pod will get recreated later.
         glog.Warningf("Deleting mirror pod %q because it is outdated", format.Pod(mirrorPod))
         if err := kl.podManager.DeleteMirrorPod(podFullName); err != nil {
            glog.Errorf("Failed deleting mirror pod %q: %v", format.Pod(mirrorPod), err)
         } else {
            deleted = true
         }
      }
   }
   if mirrorPod == nil || deleted {
      glog.V(3).Infof("Creating a mirror pod for static pod %q", format.Pod(pod))
      if err := kl.podManager.CreateMirrorPod(pod); err != nil {
         glog.Errorf("Failed creating a mirror pod for %q: %v", format.Pod(pod), err)
      }
   }
}

d. 创建 pod 的数据目录，存放 volume 和 plugin 信息 

 

if err := kl.makePodDataDirs(pod); err != nil {
   glog.Errorf("Unable to make pod data directories for pod %q: %v", format.Pod(pod), err)
   return err
}

func (kl *Kubelet) makePodDataDirs(pod *v1.Pod) error {
   uid := pod.UID
   if err := os.MkdirAll(kl.getPodDir(uid), 0750); err != nil && !os.IsExist(err) {
      return err
   }
   if err := os.MkdirAll(kl.getPodVolumesDir(uid), 0750); err != nil && !os.IsExist(err) {
      return err
   }
   if err := os.MkdirAll(kl.getPodPluginsDir(uid), 0750); err != nil && !os.IsExist(err) {
      return err
   }
   return nil
}

e. 如果 PV，等待所有的 volume mount 完成（volumeManager后台处理） 

 

// Volume manager will not mount volumes for terminated pods
if !kl.podIsTerminated(pod) {
   // Wait for volumes to attach/mount
   if err := kl.volumeManager.WaitForAttachAndMount(pod); err != nil {
      kl.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedMountVolume, "Unable to mount volumes for pod %q: %v", format.Pod(pod), err)
      glog.Errorf("Unable to mount volumes for pod %q: %v; skipping pod", format.Pod(pod), err)
      return err
   }
}

f. 有 image secrets，去 apiserver 获取对应的 secrets 数据 

 

pullSecrets := kl.getPullSecretsForPod(pod)

g. 调用 container runtime 的 SyncPod 方法，去实现真正的容器创建逻辑 

    kl.containerRuntime.SyncPod() 调 runtime 执行具体容器的创建

// Call the container runtime's SyncPod callback
result := kl.containerRuntime.SyncPod(pod, apiPodStatus, podStatus, pullSecrets, kl.backOff)
kl.reasonCache.Update(pod.UID, result)
if err := result.Error(); err != nil {
   return err
}

 

 

4 SyncPod函数

 

// Syncs the running pod into the desired pod.
SyncPod(pod *v1.Pod, apiPodStatus v1.PodStatus, podStatus *PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) PodSyncResult

 

 

 

2 syncLoop函数

 

 

 

pod phase 可能存在的值

•     Pending: Pod 已经被 kubernetes 接收，但有一个或者多个容器镜像尚未创建，等待时间包括调度，Pod的时间和通过网络下载镜像的时间，可能需要些时间
•    Running: 该 Pod 已经绑定到了一个节点上，Pod 中所有的容器都已被创建，至少有一个容器正在运行，或者正处于启动或重启状态
•    Succeeded: Pod中所有的容器都被成功终止，并且不会在重启
•    Failed： Pod中所有容器都已终止，并且至少有一个容器是因为失败终止（容器以非0状态退出或者被系统终止）
•    Unknown: 因为某些原因无法获取 Pod 状态，通常因为与 Pod 所在主机通信失败