清理 docker registry image

Bertha 。 2022-02-03 03:49 280阅读 0赞

## 一、代码 ##

#!/usr/bin/env python
    import requests
    from requests.auth import HTTPBasicAuth
    from requests.packages.urllib3.exceptions import InsecureRequestWarning
    import json
    import re
    import argparse
    
    ## this is a registry manipulator, can do following:
    ##   - list all images (including layers)
    ##   - delete images
    ##       - all except last N images
    ##       - all images and/or tags
    ##
    ## run
    ##   registry.py -h
    ##   to get more help
    ##
    ## important: after removing the tags, run the garbage collector
    ## on your registry host:
    ## docker-compose -f [path_to_your_docker_compose_file] run \
    ##     registry bin/registry garbage-collect \
    ##     /etc/docker/registry/config.yml
    ##
    ## or if you are not using docker-compose:
    ## docker run registry:2 bin/registry garbage-collect \
    ##     /etc/docker/registry/config.yml
    ##
    ## for more detail on garbage collection read here:
    ## https://docs.docker.com/registry/garbage-collection/
    
    
    # number of image versions to keep
    CONST_KEEP_LAST_VERSIONS = 10
    
    
    # this class is created for testing
    class Requests:
        def request(self, method, url, **kwargs):
            return requests.request(method, url, **kwargs)
    
    
    def natural_keys(text):
        '''
        alist.sort(key=natural_keys) sorts in human order
        http://nedbatchelder.com/blog/200712/human_sorting.html
        (See Toothy's implementation in the comments)
        '''
    
        def __atoi(text):
            return int(text) if text.isdigit() else text
    
        return [__atoi(c) for c in re.split('(\d+)', text)]
    
    
    # class to manipulate registry
    class Registry:
        # this is required for proper digest processing
        HEADERS = {"Accept":
                       "application/vnd.docker.distribution.manifest.v2+json"}
    
        def __init__(self):
            self.username = None
            self.password = None
            self.hostname = None
            self.no_validate_ssl = False
            self.http = None
            self.last_error = None
    
        def parse_login(self, login):
            if login != None:
    
                if not ':' in login:
                    self.last_error = "Please provide -l in the form USER:PASSWORD"
                    return (None, None)
    
                self.last_error = None
                (username, password) = login.split(':', 1)
                username = username.strip('"').strip("'")
                password = password.strip('"').strip("'")
                return (username, password)
    
            return (None, None)
    
        @staticmethod
        def create(host, login, no_validate_ssl):
            r = Registry()
    
            (r.username, r.password) = r.parse_login(login)
            if r.last_error != None:
                print(r.last_error)
                exit(1)
    
            r.hostname = host
            r.no_validate_ssl = no_validate_ssl
            r.http = Requests()
            return r
    
        def send(self, path, method="GET"):
            # try:
            result = self.http.request(
                method, "{0}{1}".format(self.hostname, path),
                headers=self.HEADERS,
                auth=(None if self.username == ""
                      else (self.username, self.password)),
                verify=not self.no_validate_ssl)
    
            # except Exception as error:
            #     print("cannot connect to {0}\nerror {1}".format(
            #         self.hostname,
            #         error))
            #     exit(1)
            if str(result.status_code)[0] == '2':
                self.last_error = None
                return result
    
            self.last_error = result.status_code
            return None
    
        def list_images(self):
            result = self.send('/v2/_catalog?n=1000')
            if result == None:
                return []
    
            return json.loads(result.text)['repositories']
    
        def list_tags(self, image_name):
            result = self.send("/v2/{0}/tags/list".format(image_name))
            if result == None:
                return []
    
            try:
                tags_list = json.loads(result.text)['tags']
            except ValueError:
                self.last_error = "list_tags: invalid json response"
                return []
    
            if tags_list != None:
                tags_list.sort(key=natural_keys)
    
            return tags_list
    
        # def list_tags_like(self, tag_like, args_tags_like):
        #     for tag_like in args_tags_like:
        #         print("tag like: {0}".format(tag_like))
        #         for tag in all_tags_list:
        #             if re.search(tag_like, tag):
        #                 print("Adding {0} to tags list".format(tag))
    
        def get_tag_digest(self, image_name, tag):
            image_headers = self.send("/v2/{0}/manifests/{1}".format(
                image_name, tag), method="HEAD")
    
            if image_headers == None:
                print("  tag digest not found: {0}".format(self.last_error))
                return None
    
            tag_digest = image_headers.headers['Docker-Content-Digest']
    
            return tag_digest
    
        def delete_tag(self, image_name, tag, dry_run, tag_digests_to_ignore):
            if dry_run:
                print('would delete tag {0}'.format(tag))
                return False
    
            tag_digest = self.get_tag_digest(image_name, tag)
    
            if tag_digest in tag_digests_to_ignore:
                print(
                    "Digest {0} for tag {1} is referenced by another tag or has already been deleted and will be ignored".format(
                        tag_digest, tag))
                return True
    
            if tag_digest == None:
                return False
    
            delete_result = self.send("/v2/{0}/manifests/{1}".format(
                image_name, tag_digest), method="DELETE")
    
            if delete_result == None:
                print("failed, error: {0}".format(self.last_error))
                return False
    
            tag_digests_to_ignore.append(tag_digest)
    
            print("done")
            return True
    
        # this function is not used and thus not tested
        # def delete_tag_layer(self, image_name, layer_digest, dry_run):
        #     if dry_run:
        #         print('would delete layer {0}'.format(layer_digest))
        #         return False
        #
        #     print('deleting layer {0}'.format(layer_digest),)
        #
        #     delete_result = self.send('/v2/{0}/blobs/{1}'.format(
        #         image_name, layer_digest), method='DELETE')
        #
        #     if delete_result == None:
        #         print("failed, error: {0}".format(self.last_error))
        #         return False
        #
        #     print("done")
        #     return True
    
        def list_tag_layers(self, image_name, tag):
            layers_result = self.send("/v2/{0}/manifests/{1}".format(
                image_name, tag))
    
            if layers_result == None:
                print("error {0}".format(self.last_error))
                return []
    
            json_result = json.loads(layers_result.text)
            if json_result['schemaVersion'] == 1:
                layers = json_result['fsLayers']
            else:
                layers = json_result['layers']
    
            return layers
    
    
    def parse_args(args=None):
        parser = argparse.ArgumentParser(
            description="List or delete images from Docker registry",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog=("""
    IMPORTANT: after removing the tags, run the garbage collector
               on your registry host:
       docker-compose -f [path_to_your_docker_compose_file] run \\
           registry bin/registry garbage-collect \\
           /etc/docker/registry/config.yml
    or if you are not using docker-compose:
       docker run registry:2 bin/registry garbage-collect \\
           /etc/docker/registry/config.yml
    for more detail on garbage collection read here:
       https://docs.docker.com/registry/garbage-collection/
                    """))
        parser.add_argument(
            '-l', '--login',
            help="Login and password to access to docker registry",
            required=False,
            metavar="USER:PASSWORD")
    
        parser.add_argument(
            '-r', '--host',
            help="Hostname for registry server, e.g. https://example.com:5000",
            required=True,
            metavar="URL")
    
        parser.add_argument(
            '-d', '--delete',
            help=('If specified, delete all but last {0} tags '
                  'of all images').format(CONST_KEEP_LAST_VERSIONS),
            action='store_const',
            default=False,
            const=True)
    
        parser.add_argument(
            '-n', '--num',
            help=('Set the number of tags to keep'
                  '({0} if not set)').format(CONST_KEEP_LAST_VERSIONS),
            default=CONST_KEEP_LAST_VERSIONS,
            nargs='?',
            metavar='N')
    
        parser.add_argument(
            '--dry-run',
            help=('If used in combination with --delete,'
                  'then images will not be deleted'),
            action='store_const',
            default=False,
            const=True)
    
        parser.add_argument(
            '-i', '--image',
            help='Specify images and tags to list/delete',
            nargs='+',
            metavar="IMAGE:[TAG]")
    
        parser.add_argument(
            '--keep-tags',
            nargs='+',
            help="List of tags that will be omitted from deletion if used in combination with --delete or --delete-all",
            required=False,
            default=[])
    
        parser.add_argument(
            '--tags-like',
            nargs='+',
            help="List of tags (regexp check) that will be handled",
            required=False,
            default=[])
    
        parser.add_argument(
            '--keep-tags-like',
            nargs='+',
            help="List of tags (regexp check) that will be omitted from deletion if used in combination with --delete or --delete-all",
            required=False,
            default=[])
    
        parser.add_argument(
            '--no-validate-ssl',
            help="Disable ssl validation",
            action='store_const',
            default=False,
            const=True)
    
        parser.add_argument(
            '--delete-all',
            help="Will delete all tags. Be careful with this!",
            const=True,
            default=False,
            action="store_const")
    
        parser.add_argument(
            '--layers',
            help=('Show layers digests for all images and all tags'),
            action='store_const',
            default=False,
            const=True)
    
        return parser.parse_args(args)
    
    
    def delete_tags(
            registry, image_name, dry_run, tags_to_delete, tags_to_keep):
        keep_tag_digests = []
    
        if tags_to_keep:
            print("Getting digests for tags to keep:")
            for tag in tags_to_keep:
    
                print("Getting digest for tag {0}".format(tag))
                digest = registry.get_tag_digest(image_name, tag)
                if digest is None:
                    print("Tag {0} does not exist for image {1}. Ignore here.".format(tag, image_name))
                    continue
    
                print("Keep digest {0} for tag {1}".format(digest, tag))
    
                keep_tag_digests.append(digest)
    
        for tag in tags_to_delete:
            if tag in tags_to_keep:
                continue
    
            print("  deleting tag {0}".format(tag))
    
            ##        deleting layers is disabled because
            ##        it also deletes shared layers
            ##
            ##        for layer in registry.list_tag_layers(image_name, tag):
            ##            layer_digest = layer['digest']
            ##            registry.delete_tag_layer(image_name, layer_digest, dry_run)
    
            registry.delete_tag(image_name, tag, dry_run, keep_tag_digests)
    
    
    def get_tags_like(args_tags_like, tags_list):
        result = set()
        for tag_like in args_tags_like:
            print("tag like: {0}".format(tag_like))
            for tag in tags_list:
                if re.search(tag_like, tag):
                    print("Adding {0} to tags list".format(tag))
                    result.add(tag)
        return result
    
    
    def get_tags(all_tags_list, image_name, tags_like):
        # check if there are args for special tags
        result = set()
        if tags_like:
            result = get_tags_like(tags_like, all_tags_list)
        else:
            result.update(all_tags_list)
    
        # get tags from image name if any
        if ":" in image_name:
            (image_name, tag_name) = image_name.split(":")
            result = set([tag_name])
    
        return result
    
    
    def main_loop(args):
        keep_last_versions = int(args.num)
    
        if args.no_validate_ssl:
            requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    
        registry = Registry.create(args.host, args.login, args.no_validate_ssl)
        if args.delete:
            print("Will delete all but {0} last tags".format(keep_last_versions))
    
        if args.image != None:
            image_list = args.image
        else:
            image_list = registry.list_images()
    
        # loop through registry's images
        # or through the ones given in command line
        for image_name in image_list:
            print("---------------------------------")
            print("Image: {0}".format(image_name))
    
            all_tags_list = registry.list_tags(image_name)
    
            if not all_tags_list:
                print("  no tags!")
                continue
    
            tags_list = get_tags(all_tags_list, image_name, args.tags_like)
    
            # print(tags and optionally layers        
            for tag in tags_list:
                print("  tag: {0}".format(tag))
                if args.layers:
                    for layer in registry.list_tag_layers(image_name, tag):
                        if 'size' in layer:
                            print("    layer: {0}, size: {1}".format(
                                layer['digest'], layer['size']))
                        else:
                            print("    layer: {0}".format(
                                layer['blobSum']))
    
            # add tags to "tags_to_keep" list, if we have regexp "tags_to_keep" entries:
            keep_tags = []
            if args.keep_tags_like:
                keep_tags.append(get_tags_like(args.keep_tags_like, tags_list))
    
            # delete tags if told so
            if args.delete or args.delete_all:
                if args.delete_all:
                    tags_list_to_delete = list(tags_list)
                else:
                    tags_list_to_delete = sorted(tags_list, key=natural_keys)[:-keep_last_versions]
    
                    # A manifest might be shared between different tags. Explicitly add those
                    # tags that we want to preserve to the keep_tags list, to prevent
                    # any manifest they are using from being deleted.
                    tags_list_to_keep = [tag for tag in tags_list if tag not in tags_list_to_delete]
                    keep_tags.extend(tags_list_to_keep)
    
                delete_tags(
                    registry, image_name, args.dry_run,
                    tags_list_to_delete, keep_tags)
    
    
    if __name__ == "__main__":
        args = parse_args()
        try:
            main_loop(args)
        except KeyboardInterrupt:
            print("Ctrl-C pressed, quitting")
            exit(1)

## 二、–help ##

[jiankunking@test]# python registry.py --help
    usage: registry.py [-h] [-l USER:PASSWORD] -r URL [-d]
                                           [-n [N]] [--dry-run] [-i IMAGE:[TAG]
                                           [IMAGE:[TAG] ...]]
                                           [--keep-tags KEEP_TAGS [KEEP_TAGS ...]]
                                           [--tags-like TAGS_LIKE [TAGS_LIKE ...]]
                                           [--keep-tags-like KEEP_TAGS_LIKE [KEEP_TAGS_LIKE ...]]
                                           [--no-validate-ssl] [--delete-all]
                                           [--layers]
    
    List or delete images from Docker registry
    
    optional arguments:
      -h, --help            show this help message and exit
      -l USER:PASSWORD, --login USER:PASSWORD
                            Login and password to access to docker registry
      -r URL, --host URL    Hostname for registry server, e.g.
                            https://example.com:5000
      -d, --delete          If specified, delete all but last 10 tags of all
                            images
      -n [N], --num [N]     Set the number of tags to keep(10 if not set)
      --dry-run             If used in combination with --delete,then images will
                            not be deleted
      -i IMAGE:[TAG] [IMAGE:[TAG] ...], --image IMAGE:[TAG] [IMAGE:[TAG] ...]
                            Specify images and tags to list/delete
      --keep-tags KEEP_TAGS [KEEP_TAGS ...]
                            List of tags that will be omitted from deletion if
                            used in combination with --delete or --delete-all
      --tags-like TAGS_LIKE [TAGS_LIKE ...]
                            List of tags (regexp check) that will be handled
      --keep-tags-like KEEP_TAGS_LIKE [KEEP_TAGS_LIKE ...]
                            List of tags (regexp check) that will be omitted from
                            deletion if used in combination with --delete or
                            --delete-all
      --no-validate-ssl     Disable ssl validation
      --delete-all          Will delete all tags. Be careful with this!
      --layers              Show layers digests for all images and all tags
    
    IMPORTANT: after removing the tags, run the garbage collector
               on your registry host:
       docker-compose -f [path_to_your_docker_compose_file] run \
           registry bin/registry garbage-collect \
           /etc/docker/registry/config.yml
    or if you are not using docker-compose:
       docker run registry:2 bin/registry garbage-collect \
           /etc/docker/registry/config.yml
    for more detail on garbage collection read here:
       https://docs.docker.com/registry/garbage-collection/

代码详细用法：[https://github.com/jiankunking/delete-or-list-docker-registry-images][https_github.com_jiankunking_delete-or-list-docker-registry-images]  
个人微信公众号：  
![这里写图片描述][SouthEast]

> 作者：jiankunking 出处：[http://blog.csdn.net/jiankunking][http_blog.csdn.net_jiankunking]

[https_github.com_jiankunking_delete-or-list-docker-registry-images]: https://github.com/jiankunking/delete-or-list-docker-registry-images
[SouthEast]: /images/20220203/91109a76a6a24d6ca7ac3b1d54068ed2.png
[http_blog.csdn.net_jiankunking]: http://blog.csdn.net/jiankunking