Kubelet 创建 pod 工作流程

func (kl *kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
  klog.Info("Starting kubelet main sync loop.")
  // The syncTicker 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()
  const (
    base   = 100 * time.Millisecond
    max    = 5 * time.Second
    factor = 2
  )
  duration := base
  // Responsible for checking limits in resolv.conf
  // The limits do not have anything to do with individual pods
  // Since this is called in syncLoop, we don't need to call it anywhere else
  if kl.dnsConfigurer != nil && kl.dnsConfigurer.ResolverConfig != "" {
    kl.dnsConfigurer.CheckLimitsForResolvConf()
  }

  for {
    if err := kl.runtimeState.runtimeErrors(); err != nil {
      klog.Errorf("skipping pod synchronization - %v", err)
      // exponential backoff
      time.Sleep(duration)
      duration = time.Duration(math.Min(float64(max), factor*float64(duration)))
      continue
    }
    // reset backoff if we have a success
    duration = base

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

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.Errorf("Update channel is closed. Exiting the sync loop.")
      return false
    }

    switch u.Op {
    case kubetypes.ADD:
      klog.V(2).Infof("SyncLoop (ADD, %q): %q", u.Source, 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)
    case kubetypes.UPDATE:
      klog.V(2).Infof("SyncLoop (UPDATE, %q): %q", u.Source, format.PodsWithDeletionTimestamps(u.Pods))
      handler.HandlePodUpdates(u.Pods)
    case kubetypes.REMOVE:
      klog.V(2).Infof("SyncLoop (REMOVE, %q): %q", u.Source, format.Pods(u.Pods))
      handler.HandlePodRemoves(u.Pods)
    case kubetypes.RECONCILE:
      klog.V(4).Infof("SyncLoop (RECONCILE, %q): %q", u.Source, format.Pods(u.Pods))
      handler.HandlePodReconcile(u.Pods)
    case kubetypes.DELETE:
      klog.V(2).Infof("SyncLoop (DELETE, %q): %q", u.Source, format.Pods(u.Pods))
      // DELETE is treated as a UPDATE because of graceful deletion.
      handler.HandlePodUpdates(u.Pods)
    case kubetypes.SET:
      // TODO: Do we want to support this?
      klog.Errorf("Kubelet does not support snapshot update")
    default:
      klog.Errorf("Invalid event type received: %d.", u.Op)
    }

    kl.sourcesReady.AddSource(u.Source)

  case e := <-plegCh:
    if e.Type == pleg.ContainerStarted {
      // record the most recent time we observed a container start for this pod.
      // this lets us selectively invalidate the runtimeCache when processing a delete for this pod
      // to make sure we don't miss handling graceful termination for containers we reported as having started.
      kl.lastContainerStartedTime.Add(e.ID, time.Now())
    }
    if isSyncPodWorthy(e) {
      // PLEG event for a pod; sync it.
      if pod, ok := kl.podManager.GetPodByUID(e.ID); ok {
        klog.V(2).Infof("SyncLoop (PLEG): %q, event: %#v", format.Pod(pod), e)
        handler.HandlePodSyncs([]*v1.Pod{pod})
      } else {
        // If the pod no longer exists, ignore the event.
        klog.V(4).Infof("SyncLoop (PLEG): ignore irrelevant event: %#v", e)
      }
    }

    if e.Type == pleg.ContainerDied {
      if containerID, ok := e.Data.(string); ok {
        kl.cleanUpContainersInPod(e.ID, containerID)
      }
    }
  case <-syncCh:
    // Sync pods waiting for sync
    podsToSync := kl.getPodsToSync()
    if len(podsToSync) == 0 {
      break
    }
    klog.V(4).Infof("SyncLoop (SYNC): %d pods; %s", len(podsToSync), format.Pods(podsToSync))
    handler.HandlePodSyncs(podsToSync)
  case update := <-kl.livenessManager.Updates():
    if update.Result == proberesults.Failure {
      // The liveness manager detected a failure; sync the pod.

      // We should not use the pod from livenessManager, because it is never updated after
      // initialization.
      pod, ok := kl.podManager.GetPodByUID(update.PodUID)
      if !ok {
        // If the pod no longer exists, ignore the update.
        klog.V(4).Infof("SyncLoop (container unhealthy): ignore irrelevant update: %#v", update)
        break
      }
      klog.V(1).Infof("SyncLoop (container unhealthy): %q", format.Pod(pod))
      handler.HandlePodSyncs([]*v1.Pod{pod})
    }
  case <-housekeepingCh:
    if !kl.sourcesReady.AllReady() {
      // If the sources aren't ready or volume manager has not yet synced the states,
      // skip housekeeping, as we may accidentally delete pods from unready sources.
      klog.V(4).Infof("SyncLoop (housekeeping, skipped): sources aren't ready yet.")
    } else {
      klog.V(4).Infof("SyncLoop (housekeeping)")
      if err := handler.HandlePodCleanups(); err != nil {
        klog.Errorf("Failed cleaning pods: %v", err)
      }
    }
  }
  return true
}

func (kl *kubelet) HandlePodAdditions(pods []*v1.Pod) {
  start := kl.clock.Now()
  sort.Sort(sliceutils.PodsByCreationTime(pods))
  for _, pod := range pods {
    existingPods := kl.podManager.GetPods()
    // 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)

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

    if !kl.podIsTerminated(pod) {
      // Only go through the admission process if the pod is not
      // terminated.

      // We failed pods that we rejected, so activePods include all admitted
      // pods that are alive.
      activePods := kl.filterOutTerminatedPods(existingPods)

      // Check if we can admit the pod; if not, reject it.
      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)
  }
}

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

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 {
        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
    }()
    // notify the call-back function if the operation succeeded or not
    if update.OnCompleteFunc != nil {
      update.OnCompleteFunc(err)
    }
    if err != nil {
      // IMPORTANT: we do not log errors here, the syncPodFn is responsible for logging errors
      klog.Errorf("Error syncing pod %s (%q), skipping: %v", update.Pod.UID, format.Pod(update.Pod), err)
    }
    p.wrapUp(update.Pod.UID, err)
  }
}

func (kl *kubelet) syncPod(o syncPodOptions) error {
  // pull out the required options
  pod := o.pod
  mirrorPod := o.mirrorPod
  podStatus := o.podStatus
  updateType := o.updateType

  // if we want to kill a pod, do it now!
  if updateType == kubetypes.SyncPodKill {
    ...
    if err := kl.killPod(pod, nil, podStatus, killPodOptions.PodTerminationGracePeriodSecondsOverride); err != nil {
      kl.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedToKillPod, "error killing pod: %v", err)
      // there was an error killing the pod, so we return that error directly
      utilruntime.HandleError(err)
      return err
    }
    return nil
  }
  ...
  runnable := kl.canRunPod(pod)
  if !runnable.Admit {
    ...
  }
   ...
  // If the network plugin is not ready, only start the pod if it uses the host network
  if err := kl.runtimeState.networkErrors(); err != nil && !kubecontainer.IsHostNetworkPod(pod) {
    kl.recorder.Eventf(pod, v1.EventTypeWarning, events.NetworkNotReady, "%s: %v", NetworkNotReadyErrorMsg, err)
    return fmt.Errorf("%s: %v", NetworkNotReadyErrorMsg, err)
  }
  ...
  if !kl.podIsTerminated(pod) {
    ...
    if !(podKilled && pod.Spec.RestartPolicy == v1.RestartPolicyNever) {
      if !pcm.Exists(pod) {
        if err := kl.containerManager.UpdateQOSCgroups(); err != nil {
          klog.V(2).Infof("Failed to update QoS cgroups while syncing pod: %v", err)
        }
        ...
      }
    }
  }

  // Create Mirror Pod for Static Pod if it doesn't already exist
  if kubetypes.IsStaticPod(pod) {
    ...
    }
    if mirrorPod == nil || deleted {
      node, err := kl.GetNode()
      if err != nil || node.DeletionTimestamp != nil {
        klog.V(4).Infof("No need to create a mirror pod, since node %q has been removed from the cluster", kl.nodeName)
      } else {
        klog.V(4).Infof("Creating a mirror pod for static pod %q", format.Pod(pod))
        if err := kl.podManager.CreateMirrorPod(pod); err != nil {
          klog.Errorf("Failed creating a mirror pod for %q: %v", format.Pod(pod), err)
        }
      }
    }
  }

  // Make data directories for the pod
  if err := kl.makePodDataDirs(pod); err != nil {
    kl.recorder.Eventf(pod, v1.EventTypeWarning, events.FailedToMakePodDataDirectories, "error making pod data directories: %v", err)
    klog.Errorf("Unable to make pod data directories for pod %q: %v", format.Pod(pod), err)
    return err
  }

  // 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 attach or mount volumes: %v", err)
      klog.Errorf("Unable to attach or mount volumes for pod %q: %v; skipping pod", format.Pod(pod), err)
      return err
    }
  }

  // Fetch the pull secrets for the pod
  pullSecrets := kl.getPullSecretsForPod(pod)

  // Call the container runtime's SyncPod callback
  result := kl.containerRuntime.SyncPod(pod, podStatus, pullSecrets, kl.backOff)
  kl.reasonCache.Update(pod.UID, result)
  if err := result.Error(); err != nil {
    // Do not return error if the only failures were pods in backoff
    for _, r := range result.SyncResults {
      if r.Error != kubecontainer.ErrCrashLoopBackOff && r.Error != images.ErrImagePullBackOff {
        // Do not record an event here, as we keep all event logging for sync pod failures
        // local to container runtime so we get better errors
        return err
      }
    }
    return nil
  }

  return nil
}

func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {
  // Step 1: Compute sandbox and container changes.
  podContainerChanges := m.computePodActions(pod, podStatus)
  klog.V(3).Infof("computePodActions got %+v for pod %q", podContainerChanges, format.Pod(pod))
  if podContainerChanges.CreateSandbox {
    ref, err := ref.GetReference(legacyscheme.Scheme, pod)
    if err != nil {
      klog.Errorf("Couldn't make a ref to pod %q: '%v'", format.Pod(pod), err)
    }
    ...
  }

  // Step 2: Kill the pod if the sandbox has changed.
  if podContainerChanges.KillPod {
    killResult := m.killPodWithSyncResult(pod, kubecontainer.ConvertPodStatusToRunningPod(m.runtimeName, podStatus), nil)
    result.AddPodSyncResult(killResult)
    ...

  } else {
    // Step 3: kill any running containers in this pod which are not to keep.
    for containerID, containerInfo := range podContainerChanges.ContainersToKill {
      ...

      if err := m.killContainer(pod, containerID, containerInfo.name, containerInfo.message, nil); err != nil {
        killContainerResult.Fail(kubecontainer.ErrKillContainer, err.Error())
        klog.Errorf("killContainer %q(id=%q) for pod %q failed: %v", containerInfo.name, containerID, format.Pod(pod), err)
        return
      }
    }
  }
  ...
  // Step 4: Create a sandbox for the pod if necessary.
  podSandboxID := podContainerChanges.SandboxID
  if podContainerChanges.CreateSandbox {
    var msg string
    var err error
    ...
    podSandboxID, msg, err = m.createPodSandbox(pod, podContainerChanges.Attempt)
    if err != nil {
      ...
    }
    klog.V(4).Infof("Created PodSandbox %q for pod %q", podSandboxID, format.Pod(pod))
    ...
  }

  ...

  // Step 5: start ephemeral containers
  if utilfeature.DefaultFeatureGate.Enabled(features.EphemeralContainers) {
    for _, idx := range podContainerChanges.EphemeralContainersToStart {
      start("ephemeral container", ephemeralContainerStartSpec(&pod.Spec.EphemeralContainers[idx]))
    }
  }

  // Step 6: start the init container.
  if container := podContainerChanges.NextInitContainerToStart; container != nil {
    // Start the next init container.
    if err := start("init container", containerStartSpec(container)); err != nil {
      return
    }
        ...
  }

  // Step 7: start containers in podContainerChanges.ContainersToStart.
  for _, idx := range podContainerChanges.ContainersToStart {
    start("container", containerStartSpec(&pod.Spec.Containers[idx]))
  }

  return
}

func (m *kubeGenericRuntimeManager) createPodSandbox(pod *v1.Pod, attempt uint32) (string, string, error) {
  podSandboxConfig, err := m.generatePodSandboxConfig(pod, attempt)
  ...

  // Create pod logs directory
  err = m.osInterface.MkdirAll(podSandboxConfig.LogDirectory, 0755)
  ...
  podSandBoxID, err := m.runtimeService.RunPodSandbox(podSandboxConfig, runtimeHandler)
  ...
  return podSandBoxID, "", nil
}

func (m *kubeGenericRuntimeManager) generatePodSandboxConfig(pod *v1.Pod, attempt uint32) (*runtimeapi.PodSandboxConfig, error) {
  podUID := string(pod.UID)
  podSandboxConfig := &runtimeapi.PodSandboxConfig{
    Metadata: &runtimeapi.PodSandboxMetadata{
      Name:      pod.Name,
      Namespace: pod.Namespace,
      Uid:       podUID,
      Attempt:   attempt,
    },
    Labels:      newPodLabels(pod),
    Annotations: newPodAnnotations(pod),
  }

  dnsConfig, err := m.runtimeHelper.GetPodDNS(pod)
  ...
  podSandboxConfig.DnsConfig = dnsConfig

  if !kubecontainer.IsHostNetworkPod(pod) {
    podHostname, podDomain, err := m.runtimeHelper.GeneratePodHostNameAndDomain(pod)
    podHostname, err = util.GetNodenameForKernel(podHostname, podDomain, pod.Spec.SetHostnameAsFQDN)
    podSandboxConfig.Hostname = podHostname
  }

  logDir := BuildPodLogsDirectory(pod.Namespace, pod.Name, pod.UID)
  podSandboxConfig.LogDirectory = logDir

  portMappings := []*runtimeapi.PortMapping{}
  for _, c := range pod.Spec.Containers {
    containerPortMappings := kubecontainer.MakePortMappings(&c)
    ...
  }
  if len(portMappings) > 0 {
    podSandboxConfig.PortMappings = portMappings
  }

  lc, err := m.generatePodSandboxLinuxConfig(pod)
  ...
  podSandboxConfig.Linux = lc

  return podSandboxConfig, nil
}

// generatePodSandboxLinuxConfig generates LinuxPodSandboxConfig from v1.Pod.
func (m *kubeGenericRuntimeManager) generatePodSandboxLinuxConfig(pod *v1.Pod) (*runtimeapi.LinuxPodSandboxConfig, error) {
  cgroupParent := m.runtimeHelper.GetPodCgroupParent(pod)
  lc := &runtimeapi.LinuxPodSandboxConfig{
    CgroupParent: cgroupParent,
    SecurityContext: &runtimeapi.LinuxSandboxSecurityContext{
      Privileged: kubecontainer.HasPrivilegedContainer(pod),
      SeccompProfilePath: v1.SeccompProfileRuntimeDefault,
    },
  }

  sysctls := make(map[string]string)
  if utilfeature.DefaultFeatureGate.Enabled(features.Sysctls) {
    if pod.Spec.SecurityContext != nil {
      for _, c := range pod.Spec.SecurityContext.Sysctls {
        sysctls[c.Name] = c.Value
      }
    }
  }

  lc.Sysctls = sysctls

  if pod.Spec.SecurityContext != nil {
    sc := pod.Spec.SecurityContext
    if sc.RunAsUser != nil {
      lc.SecurityContext.RunAsUser = &runtimeapi.Int64Value{Value: int64(*sc.RunAsUser)}
    }
    if sc.RunAsGroup != nil {
      lc.SecurityContext.RunAsGroup = &runtimeapi.Int64Value{Value: int64(*sc.RunAsGroup)}
    }
    lc.SecurityContext.NamespaceOptions = namespacesForPod(pod)

    if sc.FSGroup != nil {
      lc.SecurityContext.SupplementalGroups = append(lc.SecurityContext.SupplementalGroups, int64(*sc.FSGroup))
    }
    if groups := m.runtimeHelper.GetExtraSupplementalGroupsForPod(pod); len(groups) > 0 {
      lc.SecurityContext.SupplementalGroups = append(lc.SecurityContext.SupplementalGroups, groups...)
    }
    if sc.SupplementalGroups != nil {
      for _, sg := range sc.SupplementalGroups {
        lc.SecurityContext.SupplementalGroups = append(lc.SecurityContext.SupplementalGroups, int64(sg))
      }
    }
    if sc.SELinuxOptions != nil {
      lc.SecurityContext.SelinuxOptions = &runtimeapi.SELinuxOption{
        User:  sc.SELinuxOptions.User,
        Role:  sc.SELinuxOptions.Role,
        Type:  sc.SELinuxOptions.Type,
        Level: sc.SELinuxOptions.Level,
      }
    }
  }

  return lc, nil
}