Clean Docker Images From Private Registry

clean-docker-registry.py

import requests
import argparse
import functools
import operator
import re

MANIFEST_JSON_ACCEPT_HEADER = 'application/vnd.docker.distribution.manifest.v2+json'

class RegistryClean:
  def __init__(self, registry_url, image_name, retain_count, test):
    self.registry_url = re.sub(r'/$', '', registry_url)
    self.image_name = image_name
    self.retain_count = retain_count
    self.test = test

  def execute(self):
    images = []

    if self.image_name == 'all':
      images = self.get_all_images()
      print('Found %d images to process' % (len(images)))
    else:
      print('Processing on one image, [%s]' % (self.image_name))
      images.append(image_name)

    for image in images:
      self.clean_image(image)


  def get_all_images(self):
    print('Getting list of all images in repository')
    response = requests.get('%s/v2/_catalog' % (self.registry_url))
    response.raise_for_status()

    json = response.json()
    if not 'repositories' in json:
      raise 'Unexpected JSON response to catalog request'

    return json['repositories']


  def clean_image(self, image_name):
    print('Getting list of all tags for [%s]...' % (image_name))
    tags = self.get_all_tags(image_name)

    if 'latest' in tags:
      tags.remove('latest') # Special case
    tags.sort(reverse = True)

    # Convert tags to SHAs and generate unique list
    # We do this because multiple tags can point to one SHA
    # and we want to make sure we keep the latest N *shas*
    # Also keep a mapping of shas to all tags so that we can
    # properly alert the user to what's being deleted
    print('Iterating all tags and gathering info...')
    keep_shas = []
    delete_shas = []
    sha_to_tags = {}

    for tag in tags:
      sha = self.get_tag_sha(image_name, tag)

      if sha not in keep_shas and sha not in delete_shas:
        if len(keep_shas) < self.retain_count:
          keep_shas.append(sha)
        else:
          delete_shas.append(sha)

      if sha not in sha_to_tags:
        sha_to_tags[sha] = []

      sha_to_tags[sha].append(tag)

    print('Found %d SHAs to delete (%d tags), retaining %d SHAs (%d tags)' % (len(delete_shas),
                                                                              functools.reduce(operator.add, map(lambda x: len(sha_to_tags[x]), delete_shas), 0),
                                                                              len(keep_shas),
                                                                              functools.reduce(operator.add, map(lambda x: len(sha_to_tags[x]), keep_shas), 0)))


    for sha in delete_shas:
      print('Deleting %s tagged as %s%s' % (sha, sha_to_tags[sha], ' (Test Only)' if self.test == True else ''))
      if self.test is False:
        self.delete_sha(image_name, sha)


  def get_all_tags(self, image_name):
    response = requests.get('%s/v2/%s/tags/list' % (self.registry_url, image_name))
    response.raise_for_status()

    json = response.json()

    if not 'tags' in json:
      raise 'Unexpected JSON response to tags request'

    return json['tags']


  def get_tag_sha(self, image_name, tag):
    response = requests.head('%s/v2/%s/manifests/%s' % (self.registry_url, image_name, tag), headers={'Accept': MANIFEST_JSON_ACCEPT_HEADER})
    response.raise_for_status()

    if not 'Docker-Content-Digest' in response.headers:
      raise 'No content digest returned for tag %s' % tag

    return response.headers['Docker-Content-Digest']

  def delete_sha(self, image_name, sha):
    response = requests.delete('%s/v2/%s/manifests/%s' % (self.registry_url, image_name, sha), headers={'Accept': MANIFEST_JSON_ACCEPT_HEADER})
    response.raise_for_status()



def main():
  parser = argparse.ArgumentParser(description='Remove Docker tags older than <N> count')
  parser.add_argument('registry_url', help='The URL of the docker registry')
  parser.add_argument('image_name', help='The name of the image to clean, or "all" for all')
  parser.add_argument('retain_count', type=int, help='The count of tags to retain')
  parser.add_argument('--test', help='Perform a test run only', action='store_true')
  args = parser.parse_args()

  executor = RegistryClean(args.registry_url, args.image_name, args.retain_count, args.test)
  executor.execute()


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print("Caught error")
        print(e)

# sudo docker exec -ti registry /bin/sh
# registry garbage-collect /etc/docker/registry/config.yml