golang和python清理harbor镜像

偏执的太偏执、 2023-10-05 17:13 2阅读 0赞

## golang实现（已验证） ##

清理指定项目指定仓库下的镜像（pull\_time时间超过72小时之前）：

package main
    import (
        "fmt"
        "net/http"
        "time"
        "encoding/json"
        "flag"
    )
    
    var url = flag.String("url", "http://192.168.101.123:1603", "输入harbor仓库的url")
    var username = flag.String("username", "admin", "输入harbor仓库的username")
    var password = flag.String("password", "dmGeo1603", "输入harbor仓库的password")
    var projname = flag.String("projname", "dmgeo", "输入harbor仓库的项目名称")
    var reponame = flag.String("reponame", "dmgeo-lictool", "输入harbor仓库的仓库名称")
    
    func main(){
        
        flag.Parse()
       // 定义URL、用户名和密码
        url := *url
        username := *username
        password := *password
        projname := *projname
        reponame := *reponame
        fmt.Println("url=",url)
        fmt.Println("username=",username)
        fmt.Println("password=",password)
        fmt.Println("projname=",projname)
        fmt.Println("reponame=",reponame)
        // 获取API版本
        resp, _ := http.Get(url + "/api/version")
        defer resp.Body.Close()
        
        type Version struct {
        
            Version string `json:"version"`
        }
        encoder := json.NewDecoder(resp.Body)
        version := &Version{
        }
        _ = encoder.Decode(version)
        //fmt.Println(version.Version)
        // 获取“项目/仓库”下的所有Artifact
        req, _ := http.NewRequest(http.MethodGet,
            fmt.Sprintf("%s/api/%s/projects/%s/repositories/%s/artifacts",
                url, version.Version, projname, reponame), nil)
        req.SetBasicAuth(username, password)
        q := req.URL.Query()
        q.Add("page_size","1000")
        req.URL.RawQuery = q.Encode()
        fmt.Println(req.URL.String())
        resp, _ = http.DefaultClient.Do(req)
        defer resp.Body.Close()
        
        type Artifact struct {
        
            Digest   string    `json:"digest"`
            PullTime time.Time `json:"pull_time"`
            //Tags     string    `json:"tags"`
        }
        encoder = json.NewDecoder(resp.Body)
        artifacts := []*Artifact{
        }
        _ = encoder.Decode(&artifacts)
        fmt.Println(len(artifacts))
        t := time.Now().Add(-72 * time.Hour)
        
        // 遍历获取到的所有Artifact
        for _, artifact := range artifacts {
        
            // 判断Artifact最近一次被拉取的时间是不是一天前
            if artifact.PullTime.Before(t) {
        
                // 如果是，则删除Artifact
                req, _ := http.NewRequest(http.MethodDelete,
                    fmt.Sprintf("%s/api/%s/projects/%s/repositories/%s/artifacts/%s",url, version.Version, projname, reponame, artifact.Digest), nil)
                req.SetBasicAuth(username, password)
                resp, _ = http.DefaultClient.Do(req)
                defer resp.Body.Close()
            }
        } 
    }

## python实现（未验证） ##

#! /usr/bin/env python
    # -*- coding:utf-8 -*-
    
    
    import requests
    import json
    
    
    class RequestClient(object):
    
        def __init__(self,login_url, username, password):
            self.username = username
            self.password =  password
            self.login_url = login_url
            self.session = requests.Session()
            self.login()
    
        def login(self):
            self.session.post(self.login_url, params={
        "principal": self.username, "password": self.password})
    
    class ClearHarbor(object):
        
        def __init__(self, harbor_domain, password, schema="https",
                     username="admin"):
            self.schema = schema
            self.harbor_domain = harbor_domain
            self.harbor_url = self.schema + "://" + self.harbor_domain
            self.login_url = self.harbor_url + "/login"
            self.api_url = self.harbor_url + "/api"
            self.pro_url = self.api_url + "/projects"
            self.repos_url = self.api_url + "/repositories"
            self.username = username
            self.password = password
            self.client = RequestClient(self.login_url, self.username, self.password)
    
        def __fetch_pros_obj(self):
            # TODO
            self.pros_obj = self.client.session.get(self.pro_url).json()
            return self.pros_obj
    
        def fetch_pros_id(self):
            self.pros_id = []
            # TODO
            pro_res = self.__fetch_pros_obj()
            for i in pro_res:
                self.pros_id.append(i['project_id'])
            return self.pros_id
    
        def fetch_del_repos_name(self, pro_id):
            self.del_repos_name = []
            repos_res = self.client.session.get(self.repos_url, params={
        "project_id": pro_id})
            # TODO
            for repo in repos_res.json():
                if repo["tags_count"] > 30: 
                    self.del_repos_name.append(repo['name'])
            return self.del_repos_name
    
        def fetch_del_repos(self, repo_name):
            self.del_res = []
            tag_url = self.repos_url + "/" + repo_name + "/tags"
            # TODO
            tags = self.client.session.get(tag_url).json()
            tags_sort = sorted(tags, key=lambda a: a["created"])
            #print(tags_sort) 
            del_tags = tags_sort[0:len(tags_sort) -30]
            #print(del_tags)
            for tag in del_tags:
                del_repo_tag_url = tag_url + "/" + tag['name']
                print(del_repo_tag_url)
                del_res = self.client.session.delete(del_repo_tag_url)
                self.del_res.append(del_res)
    
            return self.del_res
    
    
    if __name__ == "__main__":
       
        harbor_domain = "hub.test.com" 
        password = "xxxxxxx"
        res = ClearHarbor(harbor_domain,password)
        # 循环所有的project id
        for i in res.fetch_pros_id():
            # 获取所有tag超过30的repos
            repos = res.fetch_del_repos_name(i)
            if repos:
                print(repos)   
                for repo in repos:
                    del_repos = res.fetch_del_repos(repo)
                    print(del_repos)

参考链接：  
[Harbor镜像清理python实现][Harbor_python]

[Harbor_python]: https://www.cnblogs.com/breezey/p/10615231.html