hrbor中的镜像清理
背景
现在harbor中的镜像越来越多,目前harbor的硬盘空间不足,需要定期为harbor进行清理
手动清理过程:
1,选择镜像,然后点击删除
2,点击垃圾清理,手动清理垃圾
3,等待垃圾清理任务完成
脚本
import sys
import requests
import argparse
import logging
import json
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(name)s : %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
def return_json(method, url, user, password):
try:
if method == "GET":
response = requests.get(url, auth=(user, password))
response.raise_for_status()
return response.json()
elif method == "DELETE":
response = requests.delete(url, auth=(user, password))
response.raise_for_status()
else:
logger.error("Invalid request method!!")
sys.exit(1)
except requests.exceptions.RequestException as e:
logger.error('Error: {}'.format(e))
sys.exit(1)
def get_project_id(project_name, url, user, password):
logger.info("get the project id for %s" % project_name)
projects = return_json("GET", url+'/api/projects?name='+project_name, user, password)
if project_name != projects[0]['name']:
logger.error('No such project %s found' % project_name)
sys.exit(1)
project_id = projects[0]['project_id']
logger.info("The %s's id: %s" % (project_name, project_id))
return str(project_id)
def get_images_in_project(project_id, url, user, password):
logger.info('get images in %s' % project_id)
images = return_json('GET', url+'/api/repositories?project_id='+project_id, user,password)
if not images:
logger.info("No images found. Nothing to do.")
sys.exit(0)
return images
def get_tags_from_image(project_name, image, url, user, password):
logger.info('get tags in %s%s' % (project_name, image))
tags = return_json('GET', url+"/api/repositories/"+image+"/tags", user,password)
tags_end = sorted(tags, key=lambda keys: keys['created'])
return tags_end
def delete_tags(project_name, image, tag, url, user, password):
return_json("DELETE", url + "/api/repositories/" + image + "/tags/" + tag, user, password)
logger.info("Delete the tags %s:%s:%s" % (project_name, image, tag))
def create_gc_schedule(url, user, password):
gc_url = url + "/api/system/gc/schedule"
headers = {"Content-Type": "application/json"}
data = {
"schedule": {
"type": "Manual"
}
}
try:
response = requests.post(gc_url, data=json.dumps(data),auth=(user, password), headers=headers)
response.raise_for_status()
logger.info("create the gc schedule successful")
except requests.exceptions.RequestException as e:
logger.error('Error: {}'.format(e))
sys.exit(1)
def main():
parser = argparse.ArgumentParser(prog='harbor_clean', description="clean up images in the harbor")
rqs_pars = parser.add_argument_group(title="required arguments")
parser.add_argument('-t', '--dry-run', action='store_true', help='only print the images, do not delete the images')
parser.add_argument('-g', '--gc', action='store_true', help="if create the gc scheduler or not")
rqs_pars.add_argument('-i', '--url', required=True, type=str, help='harbor url')
rqs_pars.add_argument('-u', '--user', required=True, type=str, help="the user of harbor")
rqs_pars.add_argument('-p', '--password', required=True, type=str, help="the password of user")
parser.add_argument('-c', '--keep-counts', type=int, help="how much tags need to keep")
parser.add_argument('-m', '--image', type=str, help="the image name")
parser.add_argument('project', type=str, help="the project which need to clean in harbor")
if sys.argv[1:] == 0:
parser.print_help()
parser.exit()
args = parser.parse_args()
harbor_url = args.url
project_name = args.project
harbor_user = args.user
harbor_pass = args.password
tags_counts = args.keep_counts
image_name = args.image
dry_run = args.dry_run
gc_schedule = args.gc
project_id = get_project_id(project_name, harbor_url, harbor_user, harbor_pass)
if image_name == "ALL" or image_name == "all" or image_name == "All":
images = get_images_in_project(project_id, harbor_url, harbor_user, harbor_pass)
else:
images = [{"name": image_name}]
for image in images:
tags = get_tags_from_image(project_name, image['name'], harbor_url, harbor_user, harbor_pass)
if len(tags) > tags_counts:
tags_to_delete = tags[:-tags_counts]
for tag in tags_to_delete:
if dry_run:
logger.info('we will delte %s:%s:%s' % (project_name, image['name'], tag['name']))
else:
delete_tags(project_name, image['name'], tag['name'], harbor_url, harbor_user, harbor_pass)
else:
logger.info("Do not need to clean up")
if gc_schedule:
create_gc_schedule(harbor_url, harbor_user, harbor_pass)
if __name__ == "__main__":
main()
使用
- 根据日期排序,打印出stjr大于15个tag的项目
python harbor-cleanup.py -i http://harbor-ceshi.sreblog.com -u admin -p harborxxxx -c 15 test -m all -t
清理
python harbor-cleanup.py -i http://harbor-ceshi.sreblog.com -u admin -p harborxxxx -c 15 test -m all
- 清理具体镜像下面的tag
python harbor-cleanup.py -i http://harbor-ceshi.sreblog.com -u admin -p harborxxxx -c 15 test -m test/test:latest
- 清理后做垃圾回收
python harbor-cleanup.py -i http://harbor-ceshi.sreblog.com -u admin -p xxxx -c 15 stjr -m test/test-test -g