删除K8s中镜像的脚本

在使用K8s中发现服务器中的镜像会随着时间推移（发布的版本越来越多），下载的镜像文件会越来越多，然后我们磁盘却是有限的。所以写了一个脚本来定时删除服务器中的镜像：

Docker：

#!/usr/bin/python3
import subprocess
import collections
import tempfile

def deleteImage(image: str, tags: list):
    for tag in tags:
        str_shell = f'docker rmi {image}:{tag}'
        print(str_shell)
        p = subprocess.Popen(args=str_shell, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="UTF-8")
        p.wait()
        if p.stdout is not None:
            print(p.stdout.read())
        if p.stderr is not None:
            print(p.stderr.read())

if __name__ == '__main__':
    out_temp = tempfile.SpooledTemporaryFile(max_size=10*1000000, encoding="utf-8", mode="w+t")
    fileno = out_temp.fileno()

    p = subprocess.Popen('docker image ls | grep softwise', shell=True, stdout=fileno, encoding="UTF-8")
    p.wait()
    out_temp.seek(0)
    images = []
    images_map = collections.OrderedDict()
    for line in out_temp.readlines():
        if 'IMAGE' in line:
            continue

        line_list = line.split()
        
        if images_map.get(line_list[0]) is None:
            temp_list = []
            temp_list.append(line_list[1])
            images_map[line_list[0]] = temp_list
        else:
            temp_list: list = images_map.get(line_list[0])
            temp_list.append(line_list[1])
            images_map[line_list[0]] = temp_list
    # print(images_map)
    for image in images_map.keys():
        # print(images_map.get(image))
        temp_list = images_map.get(image)
        has_test_list = []
        not_has_test_lsit = []
        for tag in temp_list:
            if 'test' in tag:
                has_test_list.append(tag)
            else:
                not_has_test_lsit.append(tag)
        has_test_list.sort(reverse = True)
        not_has_test_lsit.sort(reverse=True)
        deleteImage(image, has_test_list[5:])
        deleteImage(image, not_has_test_lsit[5:])

Containerd：

#!/usr/bin/python3
import subprocess
import collections
import tempfile

def deleteImage(image: str, tags: list):
    for tag in tags:
        str_shell = f'crictl rmi {image}:{tag}'
        print(str_shell)
        p = subprocess.Popen(args=str_shell, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="UTF-8")
        p.wait()
        if p.stdout is not None:
            print(p.stdout.read())
        if p.stderr is not None:
            print(p.stderr.read())

if __name__ == '__main__':
    # 防止执行命令之后产生大量输出，导致p.wait()卡死
    # https://www.cnblogs.com/liuhui0308/p/12464003.html
    # https://www.shuzhiduo.com/A/ke5jrnyj5r/
    out_temp = tempfile.SpooledTemporaryFile(max_size=10*1000000, encoding="utf-8", mode="w+t")
    fileno = out_temp.fileno()

    p = subprocess.Popen('crictl image ls | grep softwise', shell=True, stdout=fileno, encoding="UTF-8")
    p.wait()
    out_temp.seek(0)
    images = []
    images_map = collections.OrderedDict()
    for line in out_temp.readlines():
        if 'IMAGE' in line:
            continue

        line_list = line.split()
        
        if images_map.get(line_list[0]) is None:
            temp_list = []
            temp_list.append(line_list[1])
            images_map[line_list[0]] = temp_list
        else:
            temp_list: list = images_map.get(line_list[0])
            temp_list.append(line_list[1])
            images_map[line_list[0]] = temp_list
    # print(images_map)
    for image in images_map.keys():
        # print(images_map.get(image))
        temp_list = images_map.get(image)
        has_test_list = []
        not_has_test_lsit = []
        for tag in temp_list:
            if 'test' in tag:
                has_test_list.append(tag)
            else:
                not_has_test_lsit.append(tag)
        has_test_list.sort(reverse = True)
        not_has_test_lsit.sort(reverse=True)
        # 保留5份
        deleteImage(image, has_test_list[5:])
        deleteImage(image, not_has_test_lsit[5:])