《一起读 kubernetes 源码》pod 的资源限制和驱逐

// pkg/kubelet/eviction/types.go:57
// Manager evaluates when an eviction threshold for node stability has been met on the node.
type Manager interface {
  // Start starts the control loop to monitor eviction thresholds at specified interval.
  Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, monitoringInterval time.Duration)

  // IsUnderMemoryPressure returns true if the node is under memory pressure.
  IsUnderMemoryPressure() bool

  // IsUnderDiskPressure returns true if the node is under disk pressure.
  IsUnderDiskPressure() bool

  // IsUnderPIDPressure returns true if the node is under PID pressure.
  IsUnderPIDPressure() bool
}

// pkg/kubelet/eviction/eviction_manager.go:178
// Start starts the control loop to observe and response to low compute resources.
func (m *managerImpl) Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, monitoringInterval time.Duration) {
  thresholdHandler := func(message string) {
    klog.InfoS(message)
    m.synchronize(diskInfoProvider, podFunc)
  }
  if m.config.KernelMemcgNotification {
    for _, threshold := range m.config.Thresholds {
      if threshold.Signal == evictionapi.SignalMemoryAvailable || threshold.Signal == evictionapi.SignalAllocatableMemoryAvailable {
        notifier, err := NewMemoryThresholdNotifier(threshold, m.config.PodCgroupRoot, &CgroupNotifierFactory{}, thresholdHandler)
        if err != nil {
          klog.InfoS("Eviction manager: failed to create memory threshold notifier", "err", err)
        } else {
          go notifier.Start()
          m.thresholdNotifiers = append(m.thresholdNotifiers, notifier)
        }
      }
    }
  }
  // start the eviction manager monitoring
  go func() {
    for {
      if evictedPods := m.synchronize(diskInfoProvider, podFunc); evictedPods != nil {
        klog.InfoS("Eviction manager: pods evicted, waiting for pod to be cleaned up", "pods", klog.KObjSlice(evictedPods))
        m.waitForPodsCleanup(podCleanedUpFunc, evictedPods)
      } else {
        time.Sleep(monitoringInterval)
      }
    }
  }()
}

// pkg/kubelet/eviction/memory_threshold_notifier.go:73
func (m *memoryThresholdNotifier) Start() {
  klog.InfoS("Eviction manager: created memoryThresholdNotifier", "notifier", m.Description())
  for range m.events {
    m.handler(fmt.Sprintf("eviction manager: %s crossed", m.Description()))
  }
}

// pkg/kubelet/eviction/memory_threshold_notifier.go:80
func (m *memoryThresholdNotifier) UpdateThreshold(summary *statsapi.Summary) error {
  // .....
  newNotifier, err := m.factory.NewCgroupNotifier(m.cgroupPath, memoryUsageAttribute, memcgThreshold.Value())
  if err != nil {
    return err
  }
  m.notifier = newNotifier
  go m.notifier.Start(m.events)
  return nil
}

// pkg/kubelet/eviction/threshold_notifier_linux.go:49
// NewCgroupNotifier returns a linuxCgroupNotifier, which performs cgroup control operations required
// to receive notifications from the cgroup when the threshold is crossed in either direction.
func NewCgroupNotifier(path, attribute string, threshold int64) (CgroupNotifier, error) {
  // ....

  var watchfd, eventfd, epfd, controlfd int
  var err error
  watchfd, err = unix.Open(fmt.Sprintf("%s/%s", path, attribute), unix.O_RDONLY|unix.O_CLOEXEC, 0)
  // ....
  controlfd, err = unix.Open(fmt.Sprintf("%s/cgroup.event_control", path), unix.O_WRONLY|unix.O_CLOEXEC, 0)
  // ....
  eventfd, err = unix.Eventfd(0, unix.EFD_CLOEXEC)
  // ....
  epfd, err = unix.EpollCreate1(unix.EPOLL_CLOEXEC)
  
  // ....
  config := fmt.Sprintf("%d %d %d", eventfd, watchfd, threshold)
  _, err = unix.Write(controlfd, []byte(config))
  if err != nil {
    return nil, err
  }
  return &linuxCgroupNotifier{
    eventfd: eventfd,
    epfd:    epfd,
    stop:    make(chan struct{}),
  }, nil
}

# pkg/kubelet/eviction/threshold_notifier_linux.go:110
func (n *linuxCgroupNotifier) Start(eventCh chan<- struct{}) {
  err := unix.EpollCtl(n.epfd, unix.EPOLL_CTL_ADD, n.eventfd, &unix.EpollEvent{
    Fd:     int32(n.eventfd),
    Events: unix.EPOLLIN,
  })
  // ...
  buf := make([]byte, eventSize)
  for {
    select {
    case <-n.stop:
      return
    default:
    }
    event, err := wait(n.epfd, n.eventfd, notifierRefreshInterval)
    // ...
    _, err = unix.Read(n.eventfd, buf)
    // ...
    eventCh <- struct{}{}
  }
}

// pkg/kubelet/eviction/eviction_manager.go:233
func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc) []*v1.Pod {  
    // ...
    // 获取所有 pod 的使用量 summary
    activePods := podFunc()  
    updateStats := true 
    summary, err := m.summaryProvider.Get(ctx, updateStats)  
    if err != nil {  
       klog.ErrorS(err, "Eviction manager: failed to get summary stats")  
       return nil  
    }  
  
    // ...
    // 这里面最关键的就是 observations 和 thresholds 比较二者大小就知道是否满足阈值
    // make observations and get a function to derive pod usage stats relative to those observations.  
    observations, statsFunc := makeSignalObservations(summary)  
    // determine the set of thresholds met independent of grace period  
    thresholds = thresholdsMet(thresholds, observations, false)  
    // node conditions report true if it has been observed within the transition period window  
    nodeConditions = nodeConditionsObservedSince(nodeConditionsLastObservedAt, m.config.PressureTransitionPeriod, now)  
    if len(nodeConditions) > 0 {  
       klog.V(3).InfoS("Eviction manager: node conditions - transition period not met", "nodeCondition", nodeConditions)  
    }  
    // determine the set of thresholds we need to drive eviction behavior (i.e. all grace periods are met)  
    thresholds = thresholdsMetGracePeriod(thresholdsFirstObservedAt, now)  

    // ...
    // 这里一个细节，会优先检查 local volume 是已经超限了
    // evict pods if there is a resource usage violation from local volume temporary storage  
    // If eviction happens in localStorageEviction function, skip the rest of eviction action    
    if m.localStorageCapacityIsolation {  
       if evictedPods := m.localStorageEviction(activePods, statsFunc); len(evictedPods) > 0 {  
          return evictedPods  
       }  
    }  
  
    // 关键来了，按驱逐优先级进行排序所有阈值来得到有无超限
    // rank the thresholds by eviction priority  
    sort.Sort(byEvictionPriority(thresholds))  
    thresholdToReclaim, resourceToReclaim, foundAny := getReclaimableThreshold(thresholds)  
    if !foundAny {  
       return nil  
    }  
    

    // 细节来了，会先进行一次GC，如果 GC 之后能满足条件则就不需要驱逐了
    // check if there are node-level resources we can reclaim to reduce pressure before evicting end-user pods.  
    if m.reclaimNodeLevelResources(ctx, thresholdToReclaim.Signal, resourceToReclaim) {  
       return nil  
    }  

    // GC 之后再来排序 pod，被优先驱逐的肯定是 “大头”
    // rank the running pods for eviction for the specified resource  
    rank(activePods, statsFunc)  

    // 最后 for 循环出第一个需要驱逐的 pod 包装一下就可以返回了
    // we kill at most a single pod during each eviction interval  
    for i := range activePods {  
       pod := activePods[i]  
       gracePeriodOverride := int64(0)  
       if !isHardEvictionThreshold(thresholdToReclaim) {  
          gracePeriodOverride = m.config.MaxPodGracePeriodSeconds  
       }  
       message, annotations := evictionMessage(resourceToReclaim, pod, statsFunc, thresholds, observations)  
       var condition *v1.PodCondition  
       if utilfeature.DefaultFeatureGate.Enabled(features.PodDisruptionConditions) {  
          condition = &v1.PodCondition{  
             Type:    v1.DisruptionTarget,  
             Status:  v1.ConditionTrue,  
             Reason:  v1.PodReasonTerminationByKubelet,  
             Message: message,  
          }  
       }  
       if m.evictPod(pod, gracePeriodOverride, message, annotations, condition) {  
          metrics.Evictions.WithLabelValues(string(thresholdToReclaim.Signal)).Inc()  
          return []*v1.Pod{pod}  
       }  
    }  
    klog.InfoS("Eviction manager: unable to evict any pods from the node")  
    return nil  
}

// vendor/k8s.io/kubelet/pkg/apis/stats/v1alpha1/types.go:107
type PodStats struct {
  PodRef PodReference `json:"podRef"`
  StartTime metav1.Time `json:"startTime"`
  Containers []ContainerStats `json:"containers" patchStrategy:"merge" patchMergeKey:"name"`
  CPU *CPUStats `json:"cpu,omitempty"`
  Memory *MemoryStats `json:"memory,omitempty"`
  Network *NetworkStats `json:"network,omitempty"`
  VolumeStats []VolumeStats `json:"volume,omitempty" patchStrategy:"merge" patchMergeKey:"name"`
  EphemeralStorage *FsStats `json:"ephemeral-storage,omitempty"`
  ProcessStats *ProcessStats `json:"process_stats,omitempty"`
  Swap *SwapStats `json:"swap,omitempty"`
}

// NotifierFactory creates CgroupNotifer
type NotifierFactory interface {
  // NewCgroupNotifier creates a CgroupNotifier that creates events when the threshold
  // on the attribute in the cgroup specified by the path is crossed.
  NewCgroupNotifier(path, attribute string, threshold int64) (CgroupNotifier, error)
}