kubelet 保护节点不被容器抢占所有资源 垃圾回收机制的设计与实现 image-gc-high-threshold

1
2
3
4
5
6
7
8
9
10
11
12
13
14

func createAndInitKubelet(......) {k, err = kubelet.NewMainKubelet(......)if err != nil {return nil, err}k.BirthCry()k.StartGarbageCollection()return k, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

func (kl *Kubelet) StartGarbageCollection() {loggedContainerGCFailure := false// 1、启动容器垃圾回收服务go wait.Until(func() {if err := kl.containerGC.GarbageCollect(); err != nil {loggedContainerGCFailure = true} else {var vLevel klog.Level = 4if loggedContainerGCFailure {vLevel = 1loggedContainerGCFailure = false}klog.V(vLevel).Infof("Container garbage collection succeeded")}}, ContainerGCPeriod, wait.NeverStop)// 2、检查 ImageGCHighThresholdPercent 参数的值if kl.kubeletConfiguration.ImageGCHighThresholdPercent == 100 {return}// 3、启动镜像垃圾回收服务prevImageGCFailed := falsego wait.Until(func() {if err := kl.imageManager.GarbageCollect(); err != nil {......prevImageGCFailed = true} else {var vLevel klog.Level = 4if prevImageGCFailed {vLevel = 1prevImageGCFailed = false}}}, ImageGCPeriod, wait.NeverStop)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

func NewMainKubelet(){......// MinAge、MaxPerPodContainer、MaxContainers 分别上文章开头提到的与容器垃圾回收有关的// 三个参数containerGCPolicy := kubecontainer.ContainerGCPolicy{MinAge:             minimumGCAge.Duration,MaxPerPodContainer: int(maxPerPodContainerCount),MaxContainers:      int(maxContainerCount),}// 初始化 containerGC 模块containerGC, err := kubecontainer.NewContainerGC(klet.containerRuntime, containerGCPolicy, klet.sourcesReady)if err != nil {return nil, err}......
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

func NewContainerGC(runtime Runtime, policy ContainerGCPolicy, sourcesReadyProvider SourcesReadyProvider) (ContainerGC, error) {if policy.MinAge < 0 {return nil, fmt.Errorf("invalid minimum garbage collection age: %v", policy.MinAge)}return &realContainerGC{runtime:              runtime,policy:               policy,sourcesReadyProvider: sourcesReadyProvider,}, nil
}func (cgc *realContainerGC) GarbageCollect() error {return cgc.runtime.GarbageCollect(cgc.policy, cgc.sourcesReadyProvider.AllReady(), false)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

func (cgc *containerGC) GarbageCollect(gcPolicy kubecontainer.ContainerGCPolicy, allSourcesReady bool, evictTerminatedPods bool) error {errors := []error{}// 1、回收 pod 中的 containerif err := cgc.evictContainers(gcPolicy, allSourcesReady, evictTerminatedPods); err != nil {errors = append(errors, err)}// 2、回收 pod 中的 sandboxesif err := cgc.evictSandboxes(evictTerminatedPods); err != nil {errors = append(errors, err)}// 3、回收 pod 以及 container 的 log dirif err := cgc.evictPodLogsDirectories(allSourcesReady); err != nil {errors = append(errors, err)}return utilerrors.NewAggregate(errors)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

func (cgc *containerGC) evictContainers(gcPolicy kubecontainer.ContainerGCPolicy, allSourcesReady bool, evictTerminatedPods bool) error {// 1、获取可被回收的容器列表evictUnits, err := cgc.evictableContainers(gcPolicy.MinAge)if err != nil {return err}// 2、回收 Deleted 状态以及 Terminated 状态的 pod，此处 allSourcesReady 指 kubelet //    支持的三种 podSource 是否都可用if allSourcesReady {for key, unit := range evictUnits {if cgc.podStateProvider.IsPodDeleted(key.uid) || (cgc.podStateProvider.IsPodTerminated(key.uid) && evictTerminatedPods) {cgc.removeOldestN(unit, len(unit)) delete(evictUnits, key)}}}// 3、为非 Deleted 状态以及 Terminated 状态的 pod 保留 MaxPerPodContainer 个已经退出的容器if gcPolicy.MaxPerPodContainer >= 0 {cgc.enforceMaxContainersPerEvictUnit(evictUnits, gcPolicy.MaxPerPodContainer)}// 4、若 kubelet 启动时指定了 --maximum-dead-containers（默认为 -1 即不限制）参数，//   此时需要为 node 保留退出的容器数不能超过 --maximum-dead-containers 个if gcPolicy.MaxContainers >= 0 && evictUnits.NumContainers() > gcPolicy.MaxContainers {numContainersPerEvictUnit := gcPolicy.MaxContainers / evictUnits.NumEvictUnits()if numContainersPerEvictUnit < 1 {numContainersPerEvictUnit = 1}cgc.enforceMaxContainersPerEvictUnit(evictUnits, numContainersPerEvictUnit)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(flattened, numContainers-gcPolicy.MaxContainers)}}return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

func (cgc *containerGC) evictSandboxes(evictTerminatedPods bool) error {// 1、获取 node 上所有的 containercontainers, err := cgc.manager.getKubeletContainers(true)if err != nil {return err}// 2、获取 node 上所有的 sandboxessandboxes, err := cgc.manager.getKubeletSandboxes(true)if err != nil {return err}// 3、收集所有 container 的 PodSandboxId sandboxIDs := sets.NewString()for _, container := range containers {sandboxIDs.Insert(container.PodSandboxId)}// 4、构建 sandboxes 与 pod 的对应关系并将其保存在 sandboxesByPodUID 中sandboxesByPod := make(sandboxesByPodUID)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 {sandboxInfo.active = true}if sandboxIDs.Has(sandbox.Id) {sandboxInfo.active = true}sandboxesByPod[podUID] = append(sandboxesByPod[podUID], sandboxInfo)}// 5、对 sandboxesByPod 进行排序for uid := range sandboxesByPod {sort.Sort(sandboxByCreated(sandboxesByPod[uid]))}// 6、遍历 sandboxesByPod，若 sandboxes 所在的 pod 处于 deleted 状态，// 则删除该 pod 中所有的 sandboxes 否则只保留退出时间最短的一个 sandboxesfor podUID, sandboxes := range sandboxesByPod {if cgc.podStateProvider.IsPodDeleted(podUID) || (cgc.podStateProvider.IsPodTerminated(podUID) && evictTerminatedPods) {cgc.removeOldestNSandboxes(sandboxes, len(sandboxes))} else {cgc.removeOldestNSandboxes(sandboxes, len(sandboxes)-1)}}return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

func (cgc *containerGC) evictPodLogsDirectories(allSourcesReady bool) error {osInterface := cgc.manager.osInterface// 1、回收 deleted 状态 pod logs dirif allSourcesReady {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()podUID := parsePodUIDFromLogsDirectory(name)if !cgc.podStateProvider.IsPodDeleted(podUID) {continue}err := osInterface.RemoveAll(filepath.Join(podLogsRootDirectory, name))if err != nil {klog.Errorf("Failed to remove pod logs directory %q: %v", name, err)}}}// 2、回收 deleted 状态 container logs 链接目录logSymlinks, _ := osInterface.Glob(filepath.Join(legacyContainerLogsDir, fmt.Sprintf("*.%s", legacyLogSuffix)))for _, logSymlink := range logSymlinks {if _, err := osInterface.Stat(logSymlink); os.IsNotExist(err) {err := osInterface.Remove(logSymlink)if err != nil {klog.Errorf("Failed to remove container log dead symlink %q: %v", logSymlink, err)}}}return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

func NewMainKubelet(){......    // 初始化时需要指定三个参数，三个参数已经在上文中提到过imageGCPolicy := images.ImageGCPolicy{MinAge:               kubeCfg.ImageMinimumGCAge.Duration,HighThresholdPercent: int(kubeCfg.ImageGCHighThresholdPercent),LowThresholdPercent:  int(kubeCfg.ImageGCLowThresholdPercent),}......imageManager, err := images.NewImageGCManager(klet.containerRuntime, klet.StatsProvider, kubeDeps.Recorder, nodeRef, imageGCPolicy, crOptions.PodSandboxImage)if err != nil {return nil, fmt.Errorf("failed to initialize image manager: %v", err)}klet.imageManager = imageManager......
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

func (im *realImageGCManager) GarbageCollect() error {// 1、获取容器镜像存储目录挂载点文件系统的磁盘信息fsStats, err := im.statsProvider.ImageFsStats()if err != nil {return err}var capacity, available int64if fsStats.CapacityBytes != nil {capacity = int64(*fsStats.CapacityBytes)}if fsStats.AvailableBytes != nil {available = int64(*fsStats.AvailableBytes)}if available > capacity {available = 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}// 2、若使用率大于 HighThresholdPercent，此时需要回收镜像usagePercent := 100 - int(available*100/capacity)if usagePercent >= im.policy.HighThresholdPercent {// 3、计算需要释放的磁盘量amountToFree := capacity*int64(100-im.policy.LowThresholdPercent)/100 - available// 4、调用 im.freeSpace 回收未使用的镜像信息freed, err := im.freeSpace(amountToFree, time.Now())if err != nil {return err}if freed < amountToFree {err := fmt.Errorf("failed to garbage collect required amount of images. Wanted to free %d bytes, but freed %d bytes", amountToFree, freed)im.recorder.Eventf(im.nodeRef, v1.EventTypeWarning, events.FreeDiskSpaceFailed, err.Error())return err}}return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

func (im *realImageGCManager) freeSpace(bytesToFree int64, freeTime time.Time) (int64, error) {// 1、获取已经使用的 images 列表imagesInUse, err := im.detectImages(freeTime)if err != nil {return 0, err}im.imageRecordsLock.Lock()defer im.imageRecordsLock.Unlock()// 2、获取所有未使用的 images 信息images := make([]evictionInfo, 0, len(im.imageRecords))for image, record := range im.imageRecords {if isImageUsed(image, imagesInUse) {klog.V(5).Infof("Image ID %s is being used", image)continue}images = append(images, evictionInfo{id:          image,imageRecord: *record,})}// 3、按镜像使用时间进行排序sort.Sort(byLastUsedAndDetected(images))// 4、回收未使用的镜像var deletionErrors []errorspaceFreed := int64(0)for _, image := range images {if image.lastUsed.Equal(freeTime) || image.lastUsed.After(freeTime) {continue}if freeTime.Sub(image.firstDetected) < im.policy.MinAge {continue}// 5、调用 im.runtime.RemoveImage 删除镜像err := im.runtime.RemoveImage(container.ImageSpec{Image: image.id})if err != nil {deletionErrors = append(deletionErrors, err)continue}delete(im.imageRecords, image.id)spaceFreed += image.sizeif 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
}