Kubernetes教程(十九)--- Kubelet 垃圾回收原理

// pkg/kubelet/kubelet.go#L1397
func (kl *Kubelet) StartGarbageCollection() {
    loggedContainerGCFailure := false
    go wait.Until(func() {
       ctx := context.Background()
       if err := kl.containerGC.GarbageCollect(ctx); err != nil {
          klog.ErrorS(err, "Container garbage collection failed")
          kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.ContainerGCFailed, err.Error())
          loggedContainerGCFailure = true
       } else {
          var vLevel klog.Level = 4
          if loggedContainerGCFailure {
             vLevel = 1
             loggedContainerGCFailure = false
          }

          klog.V(vLevel).InfoS("Container garbage collection succeeded")
       }
    }, ContainerGCPeriod, wait.NeverStop) // 每 1 分钟执行一次容器垃圾回收

    // 一个特殊逻辑，如果把参数设置为 100 可以跳过垃圾回收
    // when the high threshold is set to 100, stub the image GC manager
    if kl.kubeletConfiguration.ImageGCHighThresholdPercent == 100 {
       klog.V(2).InfoS("ImageGCHighThresholdPercent is set 100, Disable image GC")
       return
    }

    prevImageGCFailed := false
    go wait.Until(func() {
       ctx := context.Background()
       if err := kl.imageManager.GarbageCollect(ctx); err != nil {
          if prevImageGCFailed {
             klog.ErrorS(err, "Image garbage collection failed multiple times in a row")
             // Only create an event for repeated failures
             kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.ImageGCFailed, err.Error())
          } else {
             klog.ErrorS(err, "Image garbage collection failed once. Stats initialization may not have completed yet")
          }
          prevImageGCFailed = true
       } else {
          var vLevel klog.Level = 4
          if prevImageGCFailed {
             vLevel = 1
             prevImageGCFailed = false
          }

          klog.V(vLevel).InfoS("Image garbage collection succeeded")
       }
    }, ImageGCPeriod, wait.NeverStop) // 每 5 分钟执行一次镜像垃圾回收
}

// ContainerGCPeriod is the period for performing container garbage collection.
ContainerGCPeriod = time.Minute
// ImageGCPeriod is the period for performing image garbage collection.
ImageGCPeriod = 5 * time.Minute

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

func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc) []*v1.Pod {
    // 省略其他逻辑...
    
    // 根据 thresholdToReclaim.Signal 信号进行垃圾回收，如果回收后剩余资源低于阈值，就直接返回 nil，
    // 代表本次不需要驱逐任何 Pod 了
    if m.reclaimNodeLevelResources(ctx, thresholdToReclaim.Signal, resourceToReclaim) {
        klog.InfoS("Eviction manager: able to reduce resource pressure without evicting pods.", "resourceName", resourceToReclaim)
        return nil
    }
    // 否则就在所有 Pod 里找一个进行驱逐，为了保证稳定性，每轮也只会驱逐一个 Pod
    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}
        }
    }
}

在 eviction 功能的主循环里就会根据对应的驱逐信号，执行对应方法，具体对应关系如下：
// buildSignalToNodeReclaimFuncs returns reclaim functions associated with resources.
func buildSignalToNodeReclaimFuncs(imageGC ImageGC, containerGC ContainerGC, withImageFs bool) map[evictionapi.Signal]nodeReclaimFuncs {
    signalToReclaimFunc := map[evictionapi.Signal]nodeReclaimFuncs{}
    // usage of an imagefs is optional
    if withImageFs {
       // with an imagefs, nodefs pressure should just delete logs
       signalToReclaimFunc[evictionapi.SignalNodeFsAvailable] = nodeReclaimFuncs{}
       signalToReclaimFunc[evictionapi.SignalNodeFsInodesFree] = nodeReclaimFuncs{}
       // with an imagefs, imagefs pressure should delete unused images
       signalToReclaimFunc[evictionapi.SignalImageFsAvailable] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
       signalToReclaimFunc[evictionapi.SignalImageFsInodesFree] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
    } else {
       // without an imagefs, nodefs pressure should delete logs, and unused images
       // since imagefs and nodefs share a common device, they share common reclaim functions
       signalToReclaimFunc[evictionapi.SignalNodeFsAvailable] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
       signalToReclaimFunc[evictionapi.SignalNodeFsInodesFree] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
       signalToReclaimFunc[evictionapi.SignalImageFsAvailable] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
       signalToReclaimFunc[evictionapi.SignalImageFsInodesFree] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
    }
    return signalToReclaimFunc
}

func (cgc *containerGC) GarbageCollect(ctx context.Context, gcPolicy kubecontainer.GCPolicy, allSourcesReady bool, evictNonDeletedPods bool) error {
    ctx, otelSpan := cgc.tracer.Start(ctx, "Containers/GarbageCollect")
    defer otelSpan.End()
    errors := []error{}
    // Remove evictable containers
    if err := cgc.evictContainers(ctx, gcPolicy, allSourcesReady, evictNonDeletedPods); err != nil {
       errors = append(errors, err)
    }

    // Remove sandboxes with zero containers
    if err := cgc.evictSandboxes(ctx, evictNonDeletedPods); err != nil {
       errors = append(errors, err)
    }

    // Remove pod sandbox log directory
    if err := cgc.evictPodLogsDirectories(ctx, allSourcesReady); err != nil {
       errors = append(errors, err)
    }
    return utilerrors.NewAggregate(errors)
}

func (cgc *containerGC) evictContainers(ctx context.Context, gcPolicy kubecontainer.GCPolicy, allSourcesReady bool, evictNonDeletedPods bool) error {
    // 找到所有可以驱逐的容器，判断条件为：当前状态不是 RUNNING 并且是在本轮 GC 前创建的容器
    evictUnits, err := cgc.evictableContainers(ctx, gcPolicy.MinAge)
    if err != nil {
       return err
    }

    // allSourceReady 为 true, 则进行回收，这个参数是外部传入的，暂时先不管
    if allSourcesReady {
       for key, unit := range evictUnits {
          if cgc.podStateProvider.ShouldPodContentBeRemoved(key.uid) || (evictNonDeletedPods && cgc.podStateProvider.ShouldPodRuntimeBeRemoved(key.uid)) {
             cgc.removeOldestN(ctx, unit, len(unit)) // Remove all.
             delete(evictUnits, key)
          }
       }
    }
    // MaxPerPodContainer 用于限制每个 Pod 里可以挂掉的容器数，
    // 如果配置了则处理一下，移除其他挂掉的容器，该值默认为 1
    if gcPolicy.MaxPerPodContainer >= 0 {
       cgc.enforceMaxContainersPerEvictUnit(ctx, evictUnits, gcPolicy.MaxPerPodContainer)
    }

    // 和前面类似，这次是限制所有 Pod 累计挂掉的容器数，如果超过了则继续清理
    // 对应配置为  --maximum-dead-containers ，默认为 -1 也就是不限制
    if gcPolicy.MaxContainers >= 0 && evictUnits.NumContainers() > gcPolicy.MaxContainers {
       // Leave an equal number of containers per evict unit (min: 1).
       numContainersPerEvictUnit := gcPolicy.MaxContainers / evictUnits.NumEvictUnits()
       if numContainersPerEvictUnit < 1 {
          numContainersPerEvictUnit = 1
       }
       cgc.enforceMaxContainersPerEvictUnit(ctx, evictUnits, numContainersPerEvictUnit)

       // 如果还不满足条件，则继续清理
       // 因为前面 numContainersPerEvictUnit 可能是小数，但是用的 int 来存储，那么就会出现
       // 每个 Pod 都移除了一个了，但是总数还是超过限制，因为计算出来每个 Pod 可能要移除的是 1.2 个这种
       numContainers := evictUnits.NumContainers()
       if numContainers > gcPolicy.MaxContainers {
          flattened := make([]containerGCInfo, 0, numContainers)
          for key := range evictUnits {
             flattened = append(flattened, evictUnits[key]...)
          }
          // 按照创建时间排序
          sort.Sort(byCreated(flattened))
          // 删除比较旧的容器
          cgc.removeOldestN(ctx, flattened, numContainers-gcPolicy.MaxContainers)
       }
    }
    return nil
}

func (cgc *containerGC) evictSandboxes(ctx context.Context, evictNonDeletedPods bool) error {
    // 获取节点上所有容器
    containers, err := cgc.manager.getKubeletContainers(ctx, true)
    if err != nil {
       return err
    }
    // 获取节点上所有沙箱容器
    sandboxes, err := cgc.manager.getKubeletSandboxes(ctx, true)
    if err != nil {
       return err
    }

    // 计算沙箱容器和普通容器的关系
    sandboxIDs := sets.NewString()
    for _, container := range containers {
       sandboxIDs.Insert(container.PodSandboxId)
    }

    sandboxesByPod := make(sandboxesByPodUID, len(sandboxes))
    for _, sandbox := range sandboxes {
       podUID := types.UID(sandbox.Metadata.Uid)
       sandboxInfo := sandboxGCInfo{
          id:         sandbox.Id,
          createTime: time.Unix(0, sandbox.CreatedAt),
       }

       if sandbox.State == runtimeapi.PodSandboxState_SANDBOX_READY || sandboxIDs.Has(sandbox.Id) {
          sandboxInfo.active = true
       }

       sandboxesByPod[podUID] = append(sandboxesByPod[podUID], sandboxInfo)
    }
   
    for podUID, sandboxes := range sandboxesByPod {
       // 如果和沙箱容器想关联的普通容器都处于驱逐、删除、终止状态
       // 或者外部参数指定要驱逐所有未删除的 Pod 同时所有普通容器处理终止状态
       // 就删除所有沙箱容器
       if cgc.podStateProvider.ShouldPodContentBeRemoved(podUID) || (evictNonDeletedPods && cgc.podStateProvider.ShouldPodRuntimeBeRemoved(podUID)) {
          cgc.removeOldestNSandboxes(ctx, sandboxes, len(sandboxes))
       } else {
          // 否则就保留一个沙箱容器
          cgc.removeOldestNSandboxes(ctx, sandboxes, len(sandboxes)-1)
       }
    }
    return nil
}

func (cgc *containerGC) evictPodLogsDirectories(ctx context.Context, allSourcesReady bool) error {
    osInterface := cgc.manager.osInterface
    if allSourcesReady {
       // 获取 kubelet 日志目录（默认为 /var/log/pods）下的子目录
       dirs, err := osInterface.ReadDir(podLogsRootDirectory)
       if err != nil {
          return fmt.Errorf("failed to read podLogsRootDirectory %q: %v", podLogsRootDirectory, err)
       }
       for _, dir := range dirs {
          name := dir.Name()
          // 从目录名中解析出 pod 的 UID，目录格式为 NAMESPACE_NAME_UID
          podUID := parsePodUIDFromLogsDirectory(name)
          // 这里再判断了一次是否可以清理
          if !cgc.podStateProvider.ShouldPodContentBeRemoved(podUID) {
             continue
          }
          // 条件都满足则直接移除对应目录
          err := osInterface.RemoveAll(filepath.Join(podLogsRootDirectory, name))
          if err != nil {
             klog.ErrorS(err, "Failed to remove pod logs directory", "path", name)
          }
       }
    }

    // 删除对应的软链接
    logSymlinks, _ := osInterface.Glob(filepath.Join(legacyContainerLogsDir, fmt.Sprintf("*.%s", legacyLogSuffix)))
    for _, logSymlink := range logSymlinks {
       if _, err := osInterface.Stat(logSymlink); os.IsNotExist(err) {
          if containerID, err := getContainerIDFromLegacyLogSymlink(logSymlink); err == nil {
             resp, err := cgc.manager.runtimeService.ContainerStatus(ctx, containerID, false)
             if err != nil {
                 klog.InfoS("Error getting ContainerStatus for containerID", "containerID", containerID, "err", err)
             } else {
                status := resp.GetStatus()
                // 如果 container 没有退出则不删除
                if status.State != runtimeapi.ContainerState_CONTAINER_EXITED {
                   continue
                }
             }
          }
          err := osInterface.Remove(logSymlink)
          if err != nil {
             klog.ErrorS(err, "Failed to remove container log dead symlink", "path", logSymlink)
          } else {
             klog.V(4).InfoS("Removed symlink", "path", logSymlink)
          }
       }
    }
    return nil
}

// pkg/kubelet/images/image_gc_manager.go#290
func (im *realImageGCManager) GarbageCollect(ctx context.Context) error {
    ctx, otelSpan := im.tracer.Start(ctx, "Images/GarbageCollect")
    defer otelSpan.End()
    // 获取磁盘状态
    fsStats, err := im.statsProvider.ImageFsStats(ctx)
    if err != nil {
       return err
    }
    // 拿到次哦的容量和可用量
    var capacity, available int64
    if fsStats.CapacityBytes != nil {
       capacity = int64(*fsStats.CapacityBytes)
    }
    if fsStats.AvailableBytes != nil {
       available = int64(*fsStats.AvailableBytes)
    }

    if available > capacity {
       klog.InfoS("Availability is larger than capacity", "available", available, "capacity", capacity)
       available = capacity
    }

    // Check valid capacity.
    if capacity == 0 {
       err := goerrors.New("invalid capacity 0 on image filesystem")
       im.recorder.Eventf(im.nodeRef, v1.EventTypeWarning, events.InvalidDiskCapacity, err.Error())
       return err
    }

    // 计算磁盘使用量，如果超过阈值则进行镜像垃圾回收
    usagePercent := 100 - int(available*100/capacity)
    if usagePercent >= im.policy.HighThresholdPercent {
       // 根据参数计算出需要腾出的空间大小
       amountToFree := capacity*int64(100-im.policy.LowThresholdPercent)/100 - available
       klog.InfoS("Disk usage on image filesystem is over the high threshold, trying to free bytes down to the low threshold", "usage", usagePercent, "highThreshold", im.policy.HighThresholdPercent, "amountToFree", amountToFree, "lowThreshold", im.policy.LowThresholdPercent)
       // 镜像垃圾回收的真正实现在这里
       freed, err := im.freeSpace(ctx, amountToFree, time.Now())
       if err != nil {
          return err
       }
       // 如果最终腾出的空间小于目标值则记录 event 并返回一个 err
       if freed < amountToFree {
          err := fmt.Errorf("Failed to garbage collect required amount of images. Attempted to free %d bytes, but only found %d bytes eligible to free.", amountToFree, freed)
          im.recorder.Eventf(im.nodeRef, v1.EventTypeWarning, events.FreeDiskSpaceFailed, err.Error())
          return err
       }
    }

    return nil
}

func (im *realImageGCManager) freeSpace(ctx context.Context, bytesToFree int64, freeTime time.Time) (int64, error) {
    // 检测镜像状态，拿到当前在使用的镜像列表
    imagesInUse, err := im.detectImages(ctx, freeTime)
    if err != nil {
       return 0, err
    }

    im.imageRecordsLock.Lock()
    defer im.imageRecordsLock.Unlock()

    images := make([]evictionInfo, 0, len(im.imageRecords))
    // 找到可以清理的镜像列表，实际就是从所有镜像里过滤出用到的镜像， 那么剩下的就是可以清理的
    for image, record := range im.imageRecords {
       // 过滤掉使用中的
       if isImageUsed(image, imagesInUse) {
          continue
       }
       // 打上了 pinned 标记的镜像，跳过清理
       if record.pinned {
          continue

       }
       images = append(images, evictionInfo{
          id:          image,
          imageRecord: *record,
       })
    }
    // 根据上次使用时间排个序，优先清理最久未使用的
    sort.Sort(byLastUsedAndDetected(images))


    var deletionErrors []error
    spaceFreed := int64(0)
    for _, image := range images {
       // 如果是新镜像(由前面的 detectImages 方法才发现的镜像)就跳过，本轮不清理
       if image.lastUsed.Equal(freeTime) || image.lastUsed.After(freeTime) {
          continue
       }

       // 再次判断如果镜像的存活时间没有达到配置值也不清理(实际上这里有个 bug)
       if freeTime.Sub(image.firstDetected) < im.policy.MinAge {
          continue
       }

       // 走到这里的都是需要清理的镜像了，调用 containerruntime 移除镜像
       err := im.runtime.RemoveImage(ctx, container.ImageSpec{Image: image.id})
       if err != nil {
          deletionErrors = append(deletionErrors, err)
          continue
       }
       // 从镜像记录中删除
       delete(im.imageRecords, image.id)
       // 累计已经腾出的空间
       spaceFreed += image.size
       // 如果本轮已经清理到了目标值就返回了，不在继续清理
       if spaceFreed >= bytesToFree {
          break
       }
    }

    if len(deletionErrors) > 0 {
       return spaceFreed, fmt.Errorf("wanted to free %d bytes, but freed %d bytes space with errors in image deletion: %v", bytesToFree, spaceFreed, errors.NewAggregate(deletionErrors))
    }
    return spaceFreed, nil
}

// pkg/kubelet/images/image_gc_manager.go#L223
func (im *realImageGCManager) detectImages(ctx context.Context, detectTime time.Time) (sets.String, error) {
    imagesInUse := sets.NewString()

    // 首先是对 sandbox 这个镜像做特殊处理，默认sandbox 在被使用，只要这个镜像存在
    imageRef, err := im.runtime.GetImageRef(ctx, container.ImageSpec{Image: im.sandboxImage})
    if err == nil && imageRef != "" {
       imagesInUse.Insert(imageRef)
    }

    // 然后列出节点上的所有镜像
    images, err := im.runtime.ListImages(ctx)
    if err != nil {
       return imagesInUse, err
    }
    // 列出节点上运行的所有 Pod
    pods, err := im.runtime.GetPods(ctx, true)
    if err != nil {
       return imagesInUse, err
    }

    // 然后把 Pod 里用到的镜像都记录到使用中的镜像列表里
    for _, pod := range pods {
       for _, container := range pod.Containers {
          imagesInUse.Insert(container.ImageID)
       }
    }
    // 到这里主线逻辑就结束了，下面都是在做一个状态维护工作，可以跳过

    now := time.Now()
    currentImages := sets.NewString()
    im.imageRecordsLock.Lock()
    defer im.imageRecordsLock.Unlock()
    for _, image := range images {
       
       currentImages.Insert(image.ID)

       // 如果镜像不在之前的记录列表里，就认为是新镜像，写入记录列表并将当前时间作为第一次检测到的时间
       if _, ok := im.imageRecords[image.ID]; !ok {
          im.imageRecords[image.ID] = &imageRecord{
             firstDetected: detectTime, // 注意，这里和前面提到的那个 bug 有关系
          }
       }

       // 如果镜像在使用就把记录里的使用上次使用时间更新
       if isImageUsed(image.ID, imagesInUse) {
          im.imageRecords[image.ID].lastUsed = now
       }

       // 更新一下镜像的其他信息
       im.imageRecords[image.ID].size = image.Size
       im.imageRecords[image.ID].pinned = image.Pinned
    }

    // 最终把记录里面有但是现在检测到已经不存在的镜像从记录列表中移除
    for image := range im.imageRecords {
       if !currentImages.Has(image) {
          delete(im.imageRecords, image)
       }
    }
    // 最后返回当前正在使用的镜像列表
    return imagesInUse, nil
}