One Hat Cyber Team

Edit File: netdata-pkgcloud-cleanup.py

#!/bin/env python3

import requests
from requests.auth import HTTPBasicAuth
from datetime import date, datetime, timedelta
import os
import sys
import argparse
from pprint import pprint
from datetime import datetime
from dateutil import parser

class PackageCloud:
    NUM_PACKAGE_MINOR_TO_KEEP = 5
    NUM_RETENTION_DAYS = 30
    # number of pages to process. Use '0' to process all
    MAX_PAGES = 0

def __init__(self, repo_type, dry_run=True, auth_token=None):
        self.headers = {
            "Accept" : "application/json",
            "Content-Type" : "application/json",
        }
        self.dry_run = dry_run
        self.repo_type = repo_type
        if repo_type == "stable":
            repo = "netdata/netdata"
        elif repo_type == "devel":
            repo = "netdata/netdata-devel"
        elif repo_type == "edge":
            repo = "netdata/netdata-edge"
        else:
            print(f"ERROR: unknown repo type '{repo_type}'!\nAccepted values are: stable,devel,edge")
            sys.exit(1)
        self.base_url = f"https://packagecloud.io/api/v1/repos/{repo}"
        self.auth = HTTPBasicAuth(username=auth_token, password='') if auth_token else None

def get_all_packages(self):
        page = 1
        all_pkg_list = []
        while True:
            url = f"{self.base_url}/packages.json?page={page}"
            if page > self.MAX_PAGES and self.MAX_PAGES != 0:
                break
            else:
                pkg_list = requests.get(url, auth=self.auth, headers=self.headers).json()
                if len(pkg_list) == 0:
                    break
                else:
                    print(f"Processing page: {page}")
                    for element in pkg_list:
                        self.is_pkg_older_than_days(element, 30)
                        if element['name'] != 'netdata-repo' and element['name'] != 'netdata-repo-edge':
                            all_pkg_list.append(element)
            page += 1
        return all_pkg_list

def delete_package(self, destroy_url):
        if self.dry_run:
            print(f"         - DRY_RUN mode. Not deleting package '{destroy_url}'.")
        else:
            print(f"         - Deleting package: {destroy_url}")
            url = f"https://packagecloud.io{destroy_url}"
            response = requests.delete(url, auth=self.auth, headers=self.headers).json()
            response = None
            if not response:
                print(f"           Package deleted successfully.")
            else:
                print(f"           Failed deleting package!")

def get_destroy_url(self, pkg_url):
        url = f"https://packagecloud.io{pkg_url}"
        response = requests.get(url, auth=self.auth, headers=self.headers)
        response.raise_for_status()
        return response.json()['destroy_url']

def get_packages_for_distro(self, distro, all_pkg_list):
        distro_pkg_list = [ pkg for pkg in all_pkg_list if pkg['distro_version'] == distro ]
        return distro_pkg_list

def get_packages_for_arch(self, arch, all_pkg_list):
        arch_pkg_list = [ pkg for pkg in all_pkg_list if pkg['package_url'].split('/')[11] == arch ]
        return arch_pkg_list

def get_arches(self, pkg_list):
        arches = list(set([pkg['package_url'].split('/')[11] for pkg in pkg_list ]))
        return arches

def get_pkg_list(self, pkg_name, pkg_list):
        filtered_list = [ pkg for pkg in pkg_list if pkg['name'] == pkg_name ]
        return filtered_list

def get_minor_versions(self, all_versions):
        minor_versions = ['.'.join(version.split('.')[:-1]) for version in all_versions ]
        minor_versions = list(set(minor_versions))
        minor_versions.sort()
        return minor_versions

def is_pkg_older_than_days(self, pkg, num_days):
        pkg_create_date = datetime.strptime(pkg['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ')
        time_difference = datetime.now() - pkg_create_date
        return time_difference.days > num_days

def cleanup_repo(self):
        if self.repo_type == 'stable':
            self.cleanup_stable_repo()
        else:
            self.cleanup_edge_repo()

def cleanup_edge_repo(self):
        all_pkg_list = self.get_all_packages()
        pkgs_to_delete = []
        pkgs_to_keep = []
        for package in all_pkg_list:
            if self.is_pkg_older_than_days(package, self.NUM_RETENTION_DAYS):
                pkgs_to_delete.append(package)
            else:
                pkgs_to_keep.append(package)
        print(f"Keeping the following packages (newer than {self.NUM_RETENTION_DAYS} days):")
        for pkg in pkgs_to_keep:
            print(f"    > pkg: {pkg['package_html_url']} / created_at: {pkg['created_at']}")
        print(f"Deleting the following packages (older than {self.NUM_RETENTION_DAYS} days):")
        for pkg in pkgs_to_delete:
            print(f"    > pkg: {pkg['package_html_url']} / created_at: {pkg['created_at']}")
            self.delete_package(pkg['destroy_url'])

def cleanup_stable_repo(self):
        all_pkg_list = self.get_all_packages()
        all_distros = list(set([ pkg['distro_version'] for pkg in all_pkg_list ]))
        all_distros = sorted(all_distros)
        print(f"<> Distributions list: {all_distros}")

for distro in all_distros:
            print(f">> Processing distro: {distro}")
            pkg_list_distro = self.get_packages_for_distro(distro, all_pkg_list)
            arches = self.get_arches(pkg_list_distro)
            print(f"   <> Arch list: {arches}")
            for arch in arches:
                print(f"   >> Processing arch: {distro} -> {arch}")
                pkg_list_arch = self.get_packages_for_arch(arch, pkg_list_distro)
                pkg_names = [pkg['name'] for pkg in pkg_list_arch]
                pkg_names = list(set(pkg_names))
                print(f"     <> Package names: {pkg_names}")
                for pkg_name in pkg_names:
                    print(f"     >> Processing package: {distro} -> {arch} -> {pkg_name}")
                    pkg_list = self.get_pkg_list(pkg_name, pkg_list_arch)
                    pkg_versions = [pkg['version'] for pkg in pkg_list]
                    pkg_minor_versions = self.get_minor_versions(pkg_versions)
                    pkg_minor_to_keep = pkg_minor_versions[-self.NUM_PACKAGE_MINOR_TO_KEEP:]
                    print(f"       <> Minor Package Versions to Keep: {pkg_minor_to_keep}")
                    pkg_minor_to_delete = list(set(pkg_minor_versions) - set(pkg_minor_to_keep))
                    print(f"       <> Minor Package Versions to Delete: {pkg_minor_to_delete}")
                    urls_to_keep = [pkg['package_url'] for pkg in pkg_list if '.'.join(pkg['version'].split('.')[:-1]) in pkg_minor_to_keep]
                    urls_to_delete = [pkg['package_url'] for pkg in pkg_list if '.'.join(pkg['version'].split('.')[:-1]) in pkg_minor_to_delete]
                    for pkg_url in urls_to_delete:
                        destroy_url = self.get_destroy_url(pkg_url)
                        self.delete_package(destroy_url)

def configure():
    parser = argparse.ArgumentParser()
    parser.add_argument('--repo-type', '-r', required=True,
                        help='Repository type against to perform cleanup')
    parser.add_argument('--dry-run', '-d', action='store_true',
                        help='Dry-run Mode')
    args = parser.parse_args()
    try:
        token = os.environ['PKGCLOUD_TOKEN']
    except Exception as e:
        print(f"FATAL: 'PKGCLOUD_TOKEN' environment variable is not set!", file=sys.stderr)
        sys.exit(1)
    repo_type = args.repo_type
    dry_run = args.dry_run
    conf = {
        'repo_type': args.repo_type,
        'dry_run': args.dry_run,
        'token': token
    }
    return conf

def main():
    config = configure()
    pkg_cloud = PackageCloud(config['repo_type'], config['dry_run'], config['token'])
    pkg_cloud.cleanup_repo()

if __name__ == "__main__":
    main()