#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2024 Zentrum für Digitale Souveränität der Öffentlichen Verwaltung (ZenDiS) GmbH # SPDX-License-Identifier: Apache-2.0 import os.path import logging import yaml import sys import shutil import subprocess import configargparse from pathlib import Path from git import Repo p = configargparse.ArgParser() p.add('--branch', env_var='CHART_DEV_BRANCH', help='The branch you want to work with. Will be created by the script if it does not exist yet.') p.add('--git_hostname', env_var='GIT_HOSTNAME', default='git@gitlab.opencode.de', help='Set the hostname for the chart git checkouts.') p.add('--revert', default=False, action='store_true', help='Set this parameter if you want to revert the referencing of the local helm chart checkout paths in the helmfiles.') p.add('--match', default='', help="Clone/pull only charts that contain the given string in their name.") p.add('--pull', default=False, action='store_true', help='Will also pull and unpack Helm charts that are not developed by product development.') p.add('--loglevel', env_var='LOGLEVEL', default='DEBUG', help='Set the loglevel: DEBUG, INFO, WARNING, ERROR, CRITICAL-') options = p.parse_args() script_path = os.path.dirname(os.path.realpath(__file__)) # some static definitions log_path = script_path+'/../logs' charts_yaml = script_path+'/../helmfile/environments/default/charts.yaml.gotmpl' base_repo_path = script_path+'/..' base_helmfile = base_repo_path+'/helmfile_generic.yaml.gotmpl' helmfile_backup_extension = '.bak' Path(log_path).mkdir(parents=True, exist_ok=True) logFormatter = logging.Formatter("%(asctime)s %(levelname)-5.5s %(message)s") rootLogger = logging.getLogger() rootLogger.setLevel(options.loglevel) fileHandler = logging.FileHandler("{0}/{1}.log".format(log_path, os.path.basename(__file__))) fileHandler.setFormatter(logFormatter) rootLogger.addHandler(fileHandler) consoleHandler = logging.StreamHandler() consoleHandler.setFormatter(logFormatter) rootLogger.addHandler(consoleHandler) logging.debug(f"Working with relative paths from script location: {script_path}") logging.debug(f"Log directory: {log_path}") logging.debug(f"charts.yaml.gotmpl: {charts_yaml}") def create_or_switch_branch_base_repo(): base_repo = Repo(path=base_repo_path) current_branch = base_repo.active_branch.name if not options.branch: branch = current_branch logging.debug(f"No branch specified on commandline, working with current branch: {current_branch}") else: branch = options.branch if branch in base_repo.branches: if branch != current_branch: logging.debug(f"Selected {branch} already exists, switching.") # ToDo: Graceful handle: "Please commit your changes or stash them before you switch branches." base_repo.git.switch(branch) else: logging.debug(f"Already on selected brach {branch}") else: logging.debug(f"Creating branch {branch} and switching") base_repo.git.branch(branch) base_repo.git.switch(branch) return branch def create_path_if_not_exists(path): if os.path.isdir(path): logging.warning(f"Path {path} already exists.") else: logging.debug(f"creating directory {path}.") Path(path).mkdir(parents=True, exist_ok=True) def clone_charts_locally(branch, charts): charts_clone_path = script_path+'/../../chart-repo/'+branch.replace('/', '_') charts_pull_path = script_path+'/../../chart-pull/'+branch.replace('/', '_') charts_dict = {} doublette_dict = {} create_path_if_not_exists(charts_clone_path) if options.pull: create_path_if_not_exists(charts_pull_path) for chart in charts['charts']: tag = charts['charts'][chart]['version'] repository = charts['charts'][chart]['repository'] registry = charts['charts'][chart]['registry'] name = charts['charts'][chart]['name'] logging.debug(f"Working on {chart} / tag {tag} / repo {repository}") if not options.match in name: logging.info(f"Chart name {name} does not match {options.match} - skipping...") elif registry == '': logging.info("Empty registry definition - skipping...") elif 'opendesk/components/platform-development/charts' in repository: logging.info("Cloning the charts repo") git_url = options.git_hostname+':'+repository chart_repo_path = charts_clone_path+'/'+charts['charts'][chart]['name'] if git_url in doublette_dict: logging.debug(f"{chart} located at {git_url} is already checked out to {doublette_dict[git_url]}") charts_dict[chart] = doublette_dict[git_url] else: if os.path.isdir(chart_repo_path): logging.debug(f"Already exists {chart_repo_path} leaving it unmodified") else: logging.debug(f"Cloning into {chart_repo_path}") Repo.clone_from(git_url, chart_repo_path) chart_repo = Repo(path=chart_repo_path) chart_repo.git.checkout('v'+charts['charts'][chart]['version']) doublette_dict[git_url] = chart_repo_path charts_dict[chart] = chart_repo_path elif options.pull: logging.info("Pulling the chart") helm_command = f"helm pull oci://{registry}/{repository}/{name} --version {tag} --untar --destination {charts_pull_path}" logging.debug(f"CLI command: {helm_command}") try: output = subprocess.check_output(helm_command, shell = True) except subprocess.CalledProcessError: sys.exit(f"! CLI command '{helm_command}' failed") else: logging.debug("Not a product development chart and `--pull` option not enabled - skipping...") return charts_dict def grep_yaml(file): with open(file, 'r') as file: content = '' for line in file.readlines(): if not ': {{' in line and not '- {{' in line: content += line return yaml.safe_load(content) def get_child_helmfiles(): child_helmfiles = [] root_helmfile = grep_yaml(base_helmfile) for entry in root_helmfile['helmfiles']: child_helmfiles.append(base_repo_path+'/'+entry['path']) return child_helmfiles def process_the_helmfiles(charts_dict, charts): chart_def_prefix = ' chart: "' child_helmfiles = get_child_helmfiles() for child_helmfile in child_helmfiles: child_helmfile_updated = False output = [] with open(child_helmfile, 'r') as file: for line in file: if chart_def_prefix in line: for chart_ident in charts_dict: if '.Values.charts.'+chart_ident+'.name' in line: logging.debug(f"found match with {chart_ident} in {line.strip()}") line = chart_def_prefix+charts_dict[chart_ident]+'/charts/'+charts['charts'][chart_ident]['name']+'" # replaced by local-dev script'+"\n" child_helmfile_updated = True break output.append(line) if child_helmfile_updated: child_helmfile_backup = child_helmfile+helmfile_backup_extension if os.path.isfile(child_helmfile_backup): logging.debug("backup {child_helmfile_backup} already exists, will not create a new one.") else: logging.debug(f"creating backup {child_helmfile_backup}.") shutil.copy2(child_helmfile, child_helmfile_backup) logging.debug(f"Updating {child_helmfile}") with open(child_helmfile, 'w') as file: file.writelines(output) def revert_the_helmfiles(): child_helmfiles = get_child_helmfiles() for child_helmfile in child_helmfiles: child_helmfile_backup = child_helmfile+helmfile_backup_extension if os.path.isfile(child_helmfile_backup): logging.debug(f"Reverting {child_helmfile} from backup {child_helmfile_backup}") os.rename(child_helmfile_backup, child_helmfile) else: logging.debug(f"Did not found the backup file {child_helmfile_backup}") ## ## Main program ## if options.revert: revert_the_helmfiles() else: branch = create_or_switch_branch_base_repo() with open(charts_yaml, 'r') as file: charts = yaml.safe_load(file) charts_dict = clone_charts_locally(branch, charts) process_the_helmfiles(charts_dict, charts)