详解 Kubernetes Pod 的实现原理

apiVersion: v1kind: Podmetadata:  name: busybox  labels:    app: busyboxspec:  containers:  - image: busybox    command:      - sleep      - "3600"    imagePullPolicy: IfNotPresent    name: busybox  restartPolicy: Always

func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, _ v1.PodStatus, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {  // Step 1: Compute sandbox and container changes.  // Step 2: Kill the pod if the sandbox has changed.  // Step 3: kill any running containers in this pod which are not to keep.  // Step 4: Create a sandbox for the pod if necessary.  // ...
  // Step 5: start the init container.  if container := podContainerChanges.NextInitContainerToStart; container != nil {    msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeInit)  }
  // Step 6: start containers in podContainerChanges.ContainersToStart.  for _, idx := range podContainerChanges.ContainersToStart {    container := &pod.Spec.Containers[idx]
    msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeRegular)  }
  return}

func (vm *volumeManager) WaitForAttachAndMount(pod *v1.Pod) error {  expectedVolumes := getExpectedVolumes(pod)  uniquePodName := util.GetUniquePodName(pod)
  vm.desiredStateOfWorldPopulator.ReprocessPod(uniquePodName)
  wait.PollImmediate(    podAttachAndMountRetryInterval,    podAttachAndMountTimeout,    vm.verifyVolumesMountedFunc(uniquePodName, expectedVolumes))
  return nil}

func (kl *Kubelet) syncPod(o syncPodOptions) error {  // ...
  if !kl.podIsTerminated(pod) {    kl.volumeManager.WaitForAttachAndMount(pod)  }
  pullSecrets := kl.getPullSecretsForPod(pod)
  result := kl.containerRuntime.SyncPod(pod, apiPodStatus, podStatus, pullSecrets, kl.backOff)  kl.reasonCache.Update(pod.UID, result)
  return nil}

func (ds *dockerService) RunPodSandbox(ctx context.Context, r *runtimeapi.RunPodSandboxRequest) (*runtimeapi.RunPodSandboxResponse, error) {  config := r.GetConfig()
  // Step 1: Pull the image for the sandbox.  image := defaultSandboxImage
  // Step 2: Create the sandbox container.  createConfig, _ := ds.makeSandboxDockerConfig(config, image)  createResp, _ := ds.client.CreateContainer(*createConfig)
  resp := &runtimeapi.RunPodSandboxResponse{PodSandboxId: createResp.ID}
  ds.setNetworkReady(createResp.ID, false)
  // Step 3: Create Sandbox Checkpoint.  ds.checkpointManager.CreateCheckpoint(createResp.ID, constructPodSandboxCheckpoint(config))
  // Step 4: Start the sandbox container.  ds.client.StartContainer(createResp.ID)
  // Step 5: Setup networking for the sandbox.  cID := kubecontainer.BuildContainerID(runtimeName, createResp.ID)  networkOptions := make(map[string]string)  ds.network.SetUpPod(config.GetMetadata().Namespace, config.GetMetadata().Name, cID, config.Annotations, networkOptions)
  return resp, nil}

func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, _ v1.PodStatus, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {  podContainerChanges := m.computePodActions(pod, podStatus)  if podContainerChanges.CreateSandbox {    ref, _ := ref.GetReference(legacyscheme.Scheme, pod)  }
  if podContainerChanges.KillPod {    if podContainerChanges.CreateSandbox {      m.purgeInitContainers(pod, podStatus)    }  } else {    for containerID, containerInfo := range podContainerChanges.ContainersToKill {      m.killContainer(pod, containerID, containerInfo.name, containerInfo.message, nil)      }    }  }
  podSandboxID := podContainerChanges.SandboxID  if podContainerChanges.CreateSandbox {    podSandboxID, _, _ = m.createPodSandbox(pod, podContainerChanges.Attempt)  }  podSandboxConfig, _ := m.generatePodSandboxConfig(pod, podContainerChanges.Attempt)
  if container := podContainerChanges.NextInitContainerToStart; container != nil {    msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeInit)  }
  for _, idx := range podContainerChanges.ContainersToStart {    container := &pod.Spec.Containers[idx]    msg, _ := m.startContainer(podSandboxID, podSandboxConfig, container, pod, podStatus, pullSecrets, podIP, kubecontainer.ContainerTypeRegular)  }
  return}

func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandboxConfig *runtimeapi.PodSandboxConfig, container *v1.Container, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, podIP string, containerType kubecontainer.ContainerType) (string, error) {  imageRef, _, _ := m.imagePuller.EnsureImageExists(pod, container, pullSecrets)
  // ...  containerID, _ := m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
  m.internalLifecycle.PreStartContainer(pod, container, containerID)
  m.runtimeService.StartContainer(containerID)
  if container.Lifecycle != nil && container.Lifecycle.PostStart != nil {    kubeContainerID := kubecontainer.ContainerID{      Type: m.runtimeName,      ID:   containerID,    }    msg, _ := m.runner.Run(kubeContainerID, pod, container, container.Lifecycle.PostStart)  }
  return "", nil}

func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {  start := kl.clock.Now()  for _, pod := range pods {    kl.podManager.AddPod(pod)    kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)    kl.probeManager.AddPod(pod)  }}
func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) {  start := kl.clock.Now()  for _, pod := range pods {    kl.podManager.DeletePod(pod)    kl.deletePod(pod)    kl.probeManager.RemovePod(pod)  }}

func (m *manager) AddPod(pod *v1.Pod) {  key := probeKey{podUID: pod.UID}  for _, c := range pod.Spec.Containers {    key.containerName = c.Name
    if c.ReadinessProbe != nil {      key.probeType = readiness      w := newWorker(m, readiness, pod, c)      m.workers[key] = w      go w.run()    }
    if c.LivenessProbe != nil {      key.probeType = liveness      w := newWorker(m, liveness, pod, c)      m.workers[key] = w      go w.run()    }  }}

func (pb *prober) runProbe(probeType probeType, p *v1.Probe, pod *v1.Pod, status v1.PodStatus, container v1.Container, containerID kubecontainer.ContainerID) (probe.Result, string, error) {  timeout := time.Duration(p.TimeoutSeconds) * time.Second  if p.Exec != nil {    command := kubecontainer.ExpandContainerCommandOnlyStatic(p.Exec.Command, container.Env)    return pb.exec.Probe(pb.newExecInContainer(container, containerID, command, timeout))  }  if p.HTTPGet != nil {    scheme := strings.ToLower(string(p.HTTPGet.Scheme))    host := p.HTTPGet.Host    port, _ := extractPort(p.HTTPGet.Port, container)    path := p.HTTPGet.Path    url := formatURL(scheme, host, port, path)    headers := buildHeader(p.HTTPGet.HTTPHeaders)    if probeType == liveness {      return pb.livenessHttp.Probe(url, headers, timeout)    } else { // readiness      return pb.readinessHttp.Probe(url, headers, timeout)    }  }  if p.TCPSocket != nil {    port, _ := extractPort(p.TCPSocket.Port, container)    host := p.TCPSocket.Host    return pb.tcp.Probe(host, port, timeout)  }  return probe.Unknown, "", fmt.Errorf("Missing probe handler for %s:%s", format.Pod(pod), container.Name)}

func (kl *Kubelet) deletePod(pod *v1.Pod) error {  kl.podWorkers.ForgetWorker(pod.UID)
  runningPods, _ := kl.runtimeCache.GetPods()  runningPod := kubecontainer.Pods(runningPods).FindPod("", pod.UID)  podPair := kubecontainer.PodPair{APIPod: pod, RunningPod: &runningPod}
  kl.podKillingCh <- &podPair  return nil}

func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(pod *v1.Pod, runningPod kubecontainer.Pod, gracePeriodOverride *int64) (syncResults []*kubecontainer.SyncResult) {  containerResults := make(chan *kubecontainer.SyncResult, len(runningPod.Containers))
  for _, container := range runningPod.Containers {    go func(container *kubecontainer.Container) {      killContainerResult := kubecontainer.NewSyncResult(kubecontainer.KillContainer, container.Name)      m.killContainer(pod, container.ID, container.Name, "Need to kill Pod", gracePeriodOverride)      containerResults <- killContainerResult    }(container)  }  close(containerResults)
  for containerResult := range containerResults {    syncResults = append(syncResults, containerResult)  }  return}

func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubecontainer.ContainerID, containerName string, reason string, gracePeriodOverride *int64) error {  containerSpec := kubecontainer.GetContainerSpec(pod, containerName);
  gracePeriod := int64(minimumGracePeriodInSeconds)  switch {  case pod.DeletionGracePeriodSeconds != nil:    gracePeriod = *pod.DeletionGracePeriodSeconds  case pod.Spec.TerminationGracePeriodSeconds != nil:    gracePeriod = *pod.Spec.TerminationGracePeriodSeconds  }
  m.executePreStopHook(pod, containerID, containerSpec, gracePeriod  m.internalLifecycle.PreStopContainer(containerID.ID)  m.runtimeService.StopContainer(containerID.ID, gracePeriod)  m.containerRefManager.ClearRef(containerID)
  return err}