《一起读 kubernetes 源码》probe 监控 pod 状态

type manager struct {
  // Map of active workers for probes
  workers map[probeKey]*worker
  // Lock for accessing & mutating workers
  workerLock sync.RWMutex

  // The statusManager cache provides pod IP and container IDs for probing.
  statusManager status.Manager

  // readinessManager manages the results of readiness probes
  readinessManager results.Manager

  // livenessManager manages the results of liveness probes
  livenessManager results.Manager

  // startupManager manages the results of startup probes
  startupManager results.Manager

  // prober executes the probe actions.
  prober *prober

  start time.Time
}

// Manager manages pod probing. It creates a probe "worker" for every container that specifies a
// probe (AddPod). The worker periodically probes its assigned container and caches the results. The
// manager use the cached probe results to set the appropriate Ready state in the PodStatus when
// requested (UpdatePodStatus). Updating probe parameters is not currently supported.
type Manager interface {
  // AddPod creates new probe workers for every container probe. This should be called for every
  // pod created.
  AddPod(pod *v1.Pod)

  // StopLivenessAndStartup handles stopping liveness and startup probes during termination.
  StopLivenessAndStartup(pod *v1.Pod)

  // RemovePod handles cleaning up the removed pod state, including terminating probe workers and
  // deleting cached results.
  RemovePod(pod *v1.Pod)

  // CleanupPods handles cleaning up pods which should no longer be running.
  // It takes a map of "desired pods" which should not be cleaned up.
  CleanupPods(desiredPods map[types.UID]sets.Empty)

  // UpdatePodStatus modifies the given PodStatus with the appropriate Ready state for each
  // container based on container running status, cached probe results and worker states.
  UpdatePodStatus(*v1.Pod, *v1.PodStatus)
}

// pkg/kubelet/prober/prober_manager.go:185
for _, c := range append(pod.Spec.Containers, getRestartableInitContainers(pod)...) {
    key.containerName = c.Name

    if c.StartupProbe != nil {
      key.probeType = startup
      if _, ok := m.workers[key]; ok {
        klog.V(8).ErrorS(nil, "Startup probe already exists for container",
          "pod", klog.KObj(pod), "containerName", c.Name)
        return
      }
      w := newWorker(m, startup, pod, c)
      m.workers[key] = w
      go w.run() // 重点就是这里了
    }

    if c.ReadinessProbe != nil {
      // 与 StartupProbe 类似...
    }

    if c.LivenessProbe != nil {
      // 与 StartupProbe 类似...
    }
  }

// pkg/kubelet/prober/worker.go:145
// run periodically probes the container.
func (w *worker) run() {
  ctx := context.Background()
  probeTickerPeriod := time.Duration(w.spec.PeriodSeconds) * time.Second

  //...

  probeTicker := time.NewTicker(probeTickerPeriod)

  defer func() {
    // Clean up.
    probeTicker.Stop()
    if !w.containerID.IsEmpty() {
      w.resultsManager.Remove(w.containerID)
    }

    w.probeManager.removeWorker(w.pod.UID, w.container.Name, w.probeType)
    //.....
  }()

probeLoop:
  for w.doProbe(ctx) {
    // Wait for next probe tick.
    select {
    case <-w.stopCh:
      break probeLoop
    case <-probeTicker.C:
    case <-w.manualTriggerCh:
      // continue
    }
  }
}

func (w *worker) doProbe(ctx context.Context) (keepGoing bool) {
  defer func() { recover() }() // Actually eat panics (HandleCrash takes care of logging)
  //.......

  // Note, exec probe does NOT have access to pod environment variables or downward API
  result, err := w.probeManager.prober.probe(ctx, w.probeType, w.pod, status, w.container, w.containerID)
  if err != nil {
    // Prober error, throw away the result.
    return true
  }

  //.......

  w.resultsManager.Set(w.containerID, result, w.pod)

  if (w.probeType == liveness || w.probeType == startup) && result == results.Failure {
    // The container fails a liveness/startup check, it will need to be restarted.
    // Stop probing until we see a new container ID. This is to reduce the
    // chance of hitting #21751, where running `docker exec` when a
    // container is being stopped may lead to corrupted container state.
    w.onHold = true
    w.resultRun = 0
  }

  return true
}

probeLoop:
  for w.doProbe(ctx) {
    // Wait for next probe tick.
    select {
    case <-w.stopCh:
      break probeLoop
    case <-probeTicker.C:
    case <-w.manualTriggerCh:
      // continue
    }
  }
}

for _, probeType := range [...]probeType{readiness, liveness, startup} {
    key.probeType = probeType
    if worker, ok := m.workers[key]; ok {
        worker.stop()
    }
}