Kubelet Startup #

// Run starts the kubelet reacting to config updates
func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
  kl.imageManager.Start()
  kl.serverCertificateManager.Start()
  kl.oomWatcher.Start(kl.nodeRef)
  kl.resourceAnalyzer.Start()
  go kl.volumeManager.Run(kl.sourcesReady, wait.NeverStop)
  go wait.Until(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, wait.NeverStop)
  go kl.fastStatusUpdateOnce()
  go kl.nodeLeaseController.Run(wait.NeverStop)
  go wait.Until(kl.updateRuntimeUp, 5*time.Second, wait.NeverStop)
  kl.initNetworkUtil()
  // Start a goroutine responsible for killing pods (that are not properly
  // handled by pod workers).
  go wait.Until(kl.podKiller.PerformPodKillingWork, 1*time.Second, wait.NeverStop)
  // Start component sync loops.
  kl.statusManager.Start()
  kl.probeManager.Start()
  kl.runtimeClassManager.Start(wait.NeverStop)
  // Start the pod lifecycle event generator.
  kl.pleg.Start()
  
  kl.syncLoop(updates, kl)
}

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

// Run starts the kubelet reacting to config updates
func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
  kl.syncLoop(updates, kl)
}

func (kl *Kubelet) syncLoop(updates <-chan kubetypes.PodUpdate, handler SyncHandler) {
  for {
    if !kl.syncLoopIteration(updates, handler, syncTicker.C, housekeepingTicker.C, plegCh) {
      break
    }
  }
}

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:
    switch u.Op {
    case kubetypes.ADD:
      // 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:
      handler.HandlePodUpdates(u.Pods)
    case kubetypes.REMOVE:
      handler.HandlePodRemoves(u.Pods)
    case kubetypes.RECONCILE:
      handler.HandlePodReconcile(u.Pods)
    case kubetypes.DELETE:
      // DELETE is treated as a UPDATE because of graceful deletion.
      handler.HandlePodUpdates(u.Pods)
    default:
      klog.Errorf("Invalid event type received: %d.", u.Op)
    }

    kl.sourcesReady.AddSource(u.Source)
  }
  return true
}

// HandlePodUpdates is the callback in the SyncHandler interface for pods
// being updated from a config source.
func (kl *Kubelet) HandlePodUpdates(pods []*v1.Pod) {
  start := kl.clock.Now()
  for _, pod := range pods {
    kl.podManager.UpdatePod(pod)
    if kubetypes.IsMirrorPod(pod) {
      kl.handleMirrorPod(pod, start)
      continue
    }
    mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
    kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
  }
}

// dispatchWork starts the asynchronous sync of the pod in a pod worker.
// If the pod has completed termination, dispatchWork will perform no action.
func (kl *Kubelet) dispatchWork(pod *v1.Pod, syncType kubetypes.SyncPodType, mirrorPod *v1.Pod, start time.Time) {
  // check whether we are ready to delete the pod from the API server (all status up to date)
  containersTerminal, podWorkerTerminal := kl.podAndContainersAreTerminal(pod)
  if pod.DeletionTimestamp != nil && containersTerminal {
    kl.statusManager.TerminatePod(pod)
    return
  }

  // Run the sync in an async worker.
  kl.podWorkers.UpdatePod(&UpdatePodOptions{
    Pod:        pod,
    MirrorPod:  mirrorPod,
    UpdateType: syncType,
  })
}

// HandlePodAdditions is the callback in SyncHandler for pods being added from
// a config source.
func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
  for _, pod := range pods {
    existingPods := kl.podManager.GetPods() // ? Why we need repeat GetPods in iteration?
    kl.podManager.AddPod(pod)
  }
}

func (m *manager) TerminatePod(pod *v1.Pod) {
  m.podStatusesLock.Lock()
  defer m.podStatusesLock.Unlock()

  m.updateStatusInternal(pod, status, true)
}
// updateStatusInternal updates the internal status cache, and queues an update to the api server if
// necessary. Returns whether an update was triggered.
// This method IS NOT THREAD SAFE and must be called from a locked function.
func (m *manager) updateStatusInternal(pod *v1.Pod, status v1.PodStatus, forceUpdate bool) bool {
  select {
  case m.podStatusChannel <- podStatusSyncRequest{pod.UID, newStatus}:
    return true
  default:
    return false
  }
}

// Apply the new setting to the specified pod.
// If the options provide an OnCompleteFunc, the function is invoked if the update is accepted.
// Update requests are ignored if a kill pod request is pending.
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) // Handler here
    }()
  }
  if !p.isWorking[pod.UID] {
    p.isWorking[pod.UID] = true
    podUpdates <- *options
  }
}

func (p *podWorkers) managePodLoop(podUpdates <-chan UpdatePodOptions) {
  var lastSyncTime time.Time
  for update := range podUpdates {
    err := func() error {
      podUID := update.Pod.UID
      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)
    }
  }
}

// pkg/kubelet/kubelet.go
func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
  ...) (*Kubelet, error) {
  klet.podWorkers = newPodWorkers(klet.syncPod, kubeDeps.Recorder, klet.workQueue, klet.resyncInterval, backOffPeriod, klet.podCache)
}

// kubelet/app/server.go
func RunKubelet(kubeServer *options.KubeletServer, kubeDeps *kubelet.Dependencies, runOnce bool) error {
  // NewMainKubelet should have set up a pod source config if one didn't exist
  // when the builder was run. This is just a precaution.
  if kubeDeps.PodConfig == nil {
    return fmt.Errorf("failed to create kubelet, pod source config was nil")
  }
  podCfg := kubeDeps.PodConfig
  startKubelet(k, podCfg, &kubeServer.KubeletConfiguration, kubeDeps, kubeServer.EnableCAdvisorJSONEndpoints, kubeServer.EnableServer)
}

func startKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, enableCAdvisorJSONEndpoints, enableServer bool) {
  go k.Run(podCfg.Updates())

/// pkg/kubelet/kubelet.go
func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
  kubeDeps *Dependencies,
  ...) {
  if kubeDeps.PodConfig == nil {
    var err error
    kubeDeps.PodConfig, err = makePodSourceConfig(kubeCfg, kubeDeps, nodeName)
    if err != nil {
      return nil, err
    }
  }
}

// makePodSourceConfig creates a config.PodConfig from the given
// KubeletConfiguration or returns an error.
func makePodSourceConfig(kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *Dependencies, nodeName types.NodeName) (*config.PodConfig, error) {
  // source of all configuration
  cfg := config.NewPodConfig(config.PodConfigNotificationIncremental, kubeDeps.Recorder)

  if kubeDeps.KubeClient != nil {
    config.NewSourceApiserver(kubeDeps.KubeClient, nodeName, cfg.Channel(kubetypes.ApiserverSource))
  }
  return cfg, nil
}

// pkg/kubelet/config/config.go
func (c *PodConfig) Channel(source string) chan<- interface{} {
  c.sourcesLock.Lock()
  defer c.sourcesLock.Unlock()
  c.sources.Insert(source)
  return c.mux.Channel(source)
}

// pkg/util/config/config.go
func (m *Mux) Channel(source string) chan interface{} {
  m.sourceLock.Lock()
  defer m.sourceLock.Unlock()
  channel, exists := m.sources[source]
  if exists {
    return channel
  }
  newChannel := make(chan interface{})
  m.sources[source] = newChannel
  go wait.Until(func() { m.listen(source, newChannel) }, 0, wait.NeverStop)
  return newChannel
}

func (m *Mux) listen(source string, listenChannel <-chan interface{}) {
  for update := range listenChannel {
    m.merger.Merge(source, update)
  }
}

// pkg/kubelet/config/config.go
// NewPodConfig creates an object that can merge many configuration sources into a stream
// of normalized updates to a pod configuration.
func NewPodConfig(mode PodConfigNotificationMode, recorder record.EventRecorder) *PodConfig {
  updates := make(chan kubetypes.PodUpdate, 50)
  storage := newPodStorage(updates, mode, recorder)
  podConfig := &PodConfig{
    pods:    storage,
    mux:     config.NewMux(storage),
    updates: updates,
    sources: sets.String{},
  }
  return podConfig
}

// pkg/kubelet/config/apiserver.go
// NewSourceApiserver creates a config source that watches and pulls from the apiserver.
func NewSourceApiserver(c clientset.Interface, nodeName types.NodeName, updates chan<- interface{}) {
  lw := cache.NewListWatchFromClient(c.CoreV1().RESTClient(), "pods", metav1.NamespaceAll, fields.OneTermEqualSelector(api.PodHostField, string(nodeName)))
  newSourceApiserverFromLW(lw, updates)
}

// newSourceApiserverFromLW holds creates a config source that watches and pulls from the apiserver.
func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}) {
  send := func(objs []interface{}) {
    var pods []*v1.Pod
    for _, o := range objs {
      pods = append(pods, o.(*v1.Pod))
    }
    updates <- kubetypes.PodUpdate{Pods: pods, Op: kubetypes.SET, Source: kubetypes.ApiserverSource}
  }
  r := cache.NewReflector(lw, &v1.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc), 0)
  go r.Run(wait.NeverStop)
}