浅析 Kubelet 驱逐机制

type KubeletConfiguration struct {
    ...
  // Map of signal names to quantities that defines hard eviction thresholds. For example: {"memory.available": "300Mi"}.
  EvictionHard map[string]string
  // Map of signal names to quantities that defines soft eviction thresholds.  For example: {"memory.available": "300Mi"}.
  EvictionSoft map[string]string
  // Map of signal names to quantities that defines grace periods for each soft eviction signal. For example: {"memory.available": "30s"}.
  EvictionSoftGracePeriod map[string]string
  // Duration for which the kubelet has to wait before transitioning out of an eviction pressure condition.
  EvictionPressureTransitionPeriod metav1.Duration
  // Maximum allowed grace period (in seconds) to use when terminating pods in response to a soft eviction threshold being met.
  EvictionMaxPodGracePeriod int32
  // Map of signal names to quantities that defines minimum reclaims, which describe the minimum
  // amount of a given resource the kubelet will reclaim when performing a pod eviction while
  // that resource is under pressure. For example: {"imagefs.available": "2Gi"}
  EvictionMinimumReclaim map[string]string
  ...
}

func NewCgroupNotifier(path, attribute string, threshold int64) (CgroupNotifier, error) {
  var watchfd, eventfd, epfd, controlfd int

  watchfd, err = unix.Open(fmt.Sprintf("%s/%s", path, attribute), unix.O_RDONLY|unix.O_CLOEXEC, 0)
  defer unix.Close(watchfd)
  
  controlfd, err = unix.Open(fmt.Sprintf("%s/cgroup.event_control", path), unix.O_WRONLY|unix.O_CLOEXEC, 0)
  defer unix.Close(controlfd)
  
  eventfd, err = unix.Eventfd(0, unix.EFD_CLOEXEC)
  defer func() {
    // Close eventfd if we get an error later in initialization
    if err != nil {
      unix.Close(eventfd)
    }
  }()
  
  epfd, err = unix.EpollCreate1(unix.EPOLL_CLOEXEC)
  defer func() {
    // Close epfd if we get an error later in initialization
    if err != nil {
      unix.Close(epfd)
    }
  }()
  
  config := fmt.Sprintf("%d %d %d", eventfd, watchfd, threshold)
  _, err = unix.Write(controlfd, []byte(config))

  return &linuxCgroupNotifier{
    eventfd: eventfd,
    epfd:    epfd,
    stop:    make(chan struct{}),
  }, nil
}

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

  for {
    select {
    case <-n.stop:
      return
    default:
    }
    event, err := wait(n.epfd, n.eventfd, notifierRefreshInterval)
    if err != nil {
      klog.InfoS("Eviction manager: error while waiting for memcg events", "err", err)
      return
    } else if !event {
      // Timeout on wait.  This is expected if the threshold was not crossed
      continue
    }
    // Consume the event from the eventfd
    buf := make([]byte, eventSize)
    _, err = unix.Read(n.eventfd, buf)
    if err != nil {
      klog.InfoS("Eviction manager: error reading memcg events", "err", err)
      return
    }
    eventCh <- struct{}{}
  }
}

// rankMemoryPressure orders the input pods for eviction in response to memory pressure.
// It ranks by whether or not the pod's usage exceeds its requests, then by priority, and
// finally by memory usage above requests.
func rankMemoryPressure(pods []*v1.Pod, stats statsFunc) {
  orderedBy(exceedMemoryRequests(stats), priority, memory(stats)).Sort(pods)
}

func (m *managerImpl) evictPod(pod *v1.Pod, gracePeriodOverride int64, evictMsg string, annotations map[string]string) bool {
  // If the pod is marked as critical and static, and support for critical pod annotations is enabled,
  // do not evict such pods. Static pods are not re-admitted after evictions.
  // https://github.com/kubernetes/kubernetes/issues/40573 has more details.
  if kubelettypes.IsCriticalPod(pod) {
    klog.ErrorS(nil, "Eviction manager: cannot evict a critical pod", "pod", klog.KObj(pod))
    return false
  }
  // record that we are evicting the pod
  m.recorder.AnnotatedEventf(pod, annotations, v1.EventTypeWarning, Reason, evictMsg)
  // this is a blocking call and should only return when the pod and its containers are killed.
  klog.V(3).InfoS("Evicting pod", "pod", klog.KObj(pod), "podUID", pod.UID, "message", evictMsg)
  err := m.killPodFunc(pod, true, &gracePeriodOverride, func(status *v1.PodStatus) {
    status.Phase = v1.PodFailed
    status.Reason = Reason
    status.Message = evictMsg
  })
  if err != nil {
    klog.ErrorS(err, "Eviction manager: pod failed to evict", "pod", klog.KObj(pod))
  } else {
    klog.InfoS("Eviction manager: pod is evicted successfully", "pod", klog.KObj(pod))
  }
  return true
}

func IsCriticalPod(pod *v1.Pod) bool {
  if IsStaticPod(pod) {
    return true
  }
  if IsMirrorPod(pod) {
    return true
  }
  if pod.Spec.Priority != nil && IsCriticalPodBasedOnPriority(*pod.Spec.Priority) {
    return true
  }
  return false
}

// IsMirrorPod returns true if the passed Pod is a Mirror Pod.
func IsMirrorPod(pod *v1.Pod) bool {
  _, ok := pod.Annotations[ConfigMirrorAnnotationKey]
  return ok
}

// IsStaticPod returns true if the pod is a static pod.
func IsStaticPod(pod *v1.Pod) bool {
  source, err := GetPodSource(pod)
  return err == nil && source != ApiserverSource
}

func IsCriticalPodBasedOnPriority(priority int32) bool {
  return priority >= scheduling.SystemCriticalPriority
}