Harbor v2.0 镜像回收那些事-低调大师

Harbor v2.0 镜像回收那些事

2020-08-16 827

Harbor 镜像回收流程

Harbor 镜像回收分两部分：

1、删除镜像 TAG
2、执行垃圾清理

早期 Harbor 1.0 版本时，删除镜像TAG可以通过UI界面手动一个个选取删除，也可以调用官方提供的SDK删除（很早以前官方不维护了）。后来官方在 1.9 版本提供TAG保留策略，解决以前镜像Tag回收麻烦的问题。

TAG保留策略有哪些策略，如下图：

Harbor Python SDK

作者写了一个 Python SDK 方法，锦上添花吧！

代码项目地址：https://github.com/yangpeng14/harbor_sdk_v2.0

注意：Harbor v2.0+ 和 Harbor v1 版本 API调用方式发生很大的改变。安全方面：修复跨域攻击问题，API登陆Harbor方法也发现变化。

#!/usr/bin/env python3
# -*- coding=utf8 -*-

import logging
import requests

logging.basicConfig(level=logging.INFO)


class HarborClient(object):
    def __init__(self, host, user, password, protocol="http"):
        self.host = host
        self.user = user
        self.password = password
        self.protocol = protocol

        # 第一次get请求，获取 cookie 信息
        self.cookies, self.headers = self.get_cookie()

        # 获取登陆成功 session
        self.session_id = self.login()

        # 把登陆成功的 sid值 替换 get_cookie 方法中 cookie sid值，用于 delete 操作
        self.cookies_new = self.cookies
        self.cookies_new.update({'sid': self.session_id})

    # def __del__(self):
    #     self.logout()

    def get_cookie(self):
        response = requests.get("{0}://{1}/c/login".format(self.protocol, self.host))
        csrf_cookie = response.cookies.get_dict()
        headers = {'X-Harbor-CSRF-Token': csrf_cookie['__csrf']}
        return csrf_cookie, headers

    def login(self):
        login_data = requests.post('%s://%s/c/login' %
                                   (self.protocol, self.host),
                                   data={'principal': self.user,
                                         'password': self.password}, cookies=self.cookies, headers=self.headers)

        if login_data.status_code == 200:
            session_id = login_data.cookies.get('sid')

            logging.debug("Successfully login, session id: {}".format(
                session_id))
            return session_id
        else:
            logging.error("Fail to login, please try again")
            return None

    def logout(self):
        requests.get('%s://%s/c/logout' % (self.protocol, self.host),
                     cookies={'sid': self.session_id})
        logging.debug("Successfully logout")

    # GET /projects
    def get_projects(self, project_name=None, is_public=None):
        # TODO: support parameter
        result = []
        page = 1
        page_size = 15

        while True:
            path = '%s://%s/api/v2.0/projects?page=%s&page_size=%s' % (self.protocol, self.host, page, page_size)
            response = requests.get(path,
                                    cookies={'sid': self.session_id})
            if response.status_code == 200:
                logging.debug("Successfully get projects result: {}".format(
                    result))
                if isinstance(response.json(), list):
                    result.extend(response.json())
                    page += 1
                else:
                    break
            else:
                logging.error("Fail to get projects result")
                result = None
                break
        return result

    # GET /projects/{project_name}/repositories
    def get_repositories(self, project_name, query_string=None):
        # TODO: support parameter
        result = []
        page = 1
        page_size = 15

        while True:
            path = '%s://%s/api/v2.0/projects/%s/repositories?page=%s&page_size=%s' % (
                self.protocol, self.host, project_name, page, page_size)
            response = requests.get(path,
                                    cookies={'sid': self.session_id})
            if response.status_code == 200:
                logging.debug(
                    "Successfully get repositories with name: {}, result: {}".format(
                        project_name, result))
                if len(response.json()):
                    result.extend(response.json())
                    page += 1
                else:
                    break
            else:
                logging.error("Fail to get repositories result with name: {}".format(
                    project_name))
                result = None
                break
        return result

    # GET /projects/{project_name}/repositories/{repository_name}/artifacts
    # GET /projects/{project_name}/repositories/{repository_name}/artifacts?with_tag=true&with_scan_overview=true&with_label=true&page_size=15&page=1
    def get_repository_artifacts(self, project_name, repository_name):
        result = []
        page = 1
        page_size = 15

        while True:
            path = '%s://%s/api/v2.0/projects/%s/repositories/%s/artifacts?with_tag=true&with_scan_overview=true&with_label=true&page_size=%s&page=%s' % (
                self.protocol, self.host, project_name, repository_name, page_size, page)
            response = requests.get(path,
                                    cookies={'sid': self.session_id}, timeout=60)
            if response.status_code == 200:
                logging.debug(
                    "Successfully get repositories artifacts with name: {}, {}, result: {}".format(
                        project_name, repository_name, result))
                if len(response.json()):
                    result.extend(response.json())
                    page += 1
                else:
                    break
            else:
                logging.error("Fail to get repositories artifacts result with name: {}, {}".format(
                    project_name, repository_name))
                result = None
                break
        return result

    # DELETE /projects/{project_name}/repositories/{repository_name}
    def delete_repository(self, project_name, repository_name, tag=None):
        # TODO: support to check tag
        # TODO: return 200 but the repo is not deleted, need more test
        result = False
        path = '%s://%s/api/v2.0/projects/%s/repositories/%s' % (
            self.protocol, self.host, project_name, repository_name)
        response = requests.delete(path,
                                   cookies=self.cookies_new, headers=self.headers)
        if response.status_code == 200:
            result = True
            print("Delete {} successful!".format(repository_name))
            logging.debug("Successfully delete repository: {}".format(
                repository_name))
        else:
            logging.error("Fail to delete repository: {}".format(repository_name))
        return result

    # Get /projects/{project_name}/repositories/{repository_name}/artifacts/{reference}/tags
    def get_repository_tags(self, project_name, repository_name, reference_hash):
        result = None
        path = '%s://%s/api/v2.0/projects/%s/repositories/%s/artifacts/%s/tags' % (
            self.protocol, self.host, project_name, repository_name, reference_hash)
        response = requests.get(path,
                                cookies={'sid': self.session_id}, timeout=60)
        if response.status_code == 200:
            result = response.json()
            logging.debug(
                "Successfully get tag with repository name: {}, result: {}".format(
                    repository_name, result))
        else:
            logging.error("Fail to get tags with repository name: {}".format(
                repository_name))
        return result

    # Del /projects/{project_name}/repositories/{repository_name}/artifacts/{reference}/tags/{tag_name}
    def del_repository_tag(self, project_name, repository_name, reference_hash, tag):
        result = False
        path = '%s://%s/api/v2.0/projects/%s/repositories/%s/artifacts/%s/tags/%s' % (
            self.protocol, self.host, project_name, repository_name, reference_hash, tag)
        response = requests.delete(path, cookies=self.cookies_new, headers=self.headers)
        if response.status_code == 200:
            result = True
            print("Delete {} {} {} {} successful!".format(project_name, repository_name, reference_hash, tag))
            logging.debug(
                "Successfully delete repository project_name: {}, repository_name: {}, reference_hash: {}, tag: {}".format(
                    project_name, repository_name, reference_hash, tag))
        else:
            logging.error("Fail to delete repository project_name: {}, repository_name: {}, reference_hash: {}, tag: {}".format(
                project_name, repository_name, reference_hash, tag))
        return result

    # Del /projects/{project_name}/repositories/{repository_name}/artifacts/{reference}
    def del_artifacts_hash(self, project_name, repository_name, reference_hash):
        result = False
        path = '%s://%s/api/v2.0/projects/%s/repositories/%s/artifacts/%s' % (
            self.protocol, self.host, project_name, repository_name, reference_hash)
        response = requests.delete(path, cookies=self.cookies_new, headers=self.headers)
        if response.status_code == 200:
            result = True
            print("Delete artifacts hash {} {} {} successful!".format(project_name, repository_name, reference_hash))
            logging.debug(
                "Successfully delete repository project_name: {}, repository_name: {}, artifacts hash: {}".format(
                    project_name, repository_name, reference_hash))
        else:
            logging.error("Fail to delete repository project_name: {}, repository_name: {}, artifacts hash: {}".format(
                project_name, repository_name, reference_hash))
        return result

使用举例

列出 Harbor Projects

import harborclient_modify_v2_0

class GetHarborApi(object):
    def __init__(self, host, user, password, protocol="http"):
        self.host = host
        self.user = user
        self.password = password
        self.protocol = protocol
        self.client = harborclient_modify_v2_0.HarborClient(self.host, self.user, self.password, self.protocol)

    def main(self):
        print(self.client.get_projects())

if __name__ == '__main__':
    host = "harbor.example.com"
    user = "admin"
    password = "******"
    protocol = "https"
    cline_get = GetHarborApi(host, user, password, protocol)
    cline_get.main()

最后

欢迎您加我微信【 ypxiaozhan01】，拉您进技术群，一起交流学习...
欢迎您关注【 YP小站】，学习互联网最流行的技术，做个专业的技术人...

【文章让您有收获，👇 赞或者在看支持我吧】

本文分享自微信公众号 - YP小站（ypxiaozhan）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。

微信关注我们

原文链接：https://my.oschina.net/u/4437985/blog/4499123

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

LeetCode27|最长公共前缀

1，问题简述编写一个函数来查找字符串数组中的最长公共前缀。如果不存在公共前缀，返回空字符串""。 2，示例示例1:输入: ["flower","flow","flight"]输出: "fl"示例2:输入: ["dog","racecar","car"]输出: ""解释: 输入不存在公共前缀。 3，题解思路对比 4，题解程序 import java.util.Arrays;public class LongestCommonPrefixTest { public static void main(String[] args) { String[] strs = {"flower", "flow", "flight"}; String longestCommonPrefix = longestCommonPrefix(strs); System.out.println("longestCommonPrefix = " + longestCommonPrefix); } public static String longestCommonPrefix(String[] strs) {...

2020-08-16

601

123457 导读上一篇文章《ShardingJdbc分库分表实战案例解析（上）》中我们初步介绍了使用ShardingJdbc实现订单数据分散存储的分库分表方法，在本篇文章中将重点介绍在不停服的情况下实现数据分片存储的在线扩容。具体将以如下两个常见的场景进行演示：1)、尚未进行分库分表的单库单表系统如何平稳的实施分库分表方案；2)、已经实施过分库分表方案的系统，由于数据量的持续增长导致原有分库分表不够用了，需要二次扩容的情况。实施方案概述在具体演示之前，我们先简单聊一聊关于分库分表数据迁移中，几种常见的思路，具体如下： 1)、停服迁移停服迁移方案，是数据迁移方案中最常见、也是相对安全，但是也是最不能让人接受的方案。之所以说它安全，是因为停服之后就不会有新的数据写入，这样能保证数据迁移工作能够在一个相对稳定的环境中进行，因而能较大程度上避免迁移过程中产生数据不一致的问题。但停服务方案会比较严重伤害用户体验、降低服务的可用性，而如果数据量较大迁移需要大量的时间的话，长时间的停服会严重影响业务，对于强调"9999"服务体验的互联网公司来说，停服方案绝对是让人不能接受的。一般来说停...

2020-08-17

676

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。