import argparse import dataclasses import json import os import sys from typing import List from get_robot_token import get_best_robot_token from github_helper import GitHub from ci_utils import Shell from env_helper import GITHUB_REPOSITORY from report import SUCCESS from ci_buddy import CIBuddy from ci_config import CI def parse_args(): parser = argparse.ArgumentParser( "Checks if enough days elapsed since the last release on each release " "branches and do a release in case for green builds." ) parser.add_argument("--token", help="GitHub token, if not set, used from smm") parser.add_argument( "--post-status", action="store_true", help="Post release branch statuses", ) parser.add_argument( "--post-auto-release-complete", action="store_true", help="Post autorelease completion status", ) parser.add_argument( "--prepare", action="store_true", help="Prepare autorelease info", ) parser.add_argument( "--wf-status", type=str, default="", help="overall workflow status [success|failure]", ) return parser.parse_args(), parser MAX_NUMBER_OF_COMMITS_TO_CONSIDER_FOR_RELEASE = 5 AUTORELEASE_INFO_FILE = "/tmp/autorelease_info.json" AUTORELEASE_MATRIX_PARAMS = "/tmp/autorelease_params.json" @dataclasses.dataclass class ReleaseParams: ready: bool ci_status: str num_patches: int release_branch: str commit_sha: str commits_to_branch_head: int latest: bool def to_dict(self): return dataclasses.asdict(self) @dataclasses.dataclass class AutoReleaseInfo: releases: List[ReleaseParams] def add_release(self, release_params: ReleaseParams) -> None: self.releases.append(release_params) def dump(self): print(f"Dump release info into [{AUTORELEASE_INFO_FILE}]") with open(AUTORELEASE_INFO_FILE, "w", encoding="utf-8") as f: print(json.dumps(dataclasses.asdict(self), indent=2), file=f) # dump file for GH action matrix that is similar to the file above but with dropped not ready release branches params = dataclasses.asdict(self) params["releases"] = [ release for release in params["releases"] if release["ready"] ] with open(AUTORELEASE_MATRIX_PARAMS, "w", encoding="utf-8") as f: print(json.dumps(params, indent=2), file=f) @staticmethod def from_file() -> "AutoReleaseInfo": with open(AUTORELEASE_INFO_FILE, "r", encoding="utf-8") as json_file: res = json.load(json_file) releases = [ReleaseParams(**release) for release in res["releases"]] return AutoReleaseInfo(releases=releases) def _prepare(token): assert len(token) > 10 os.environ["GH_TOKEN"] = token Shell.check("gh auth status") gh = GitHub(token) prs = gh.get_release_pulls(GITHUB_REPOSITORY) prs.sort(key=lambda x: x.head.ref) branch_names = [pr.head.ref for pr in prs] print(f"Found release branches [{branch_names}]") repo = gh.get_repo(GITHUB_REPOSITORY) autoRelease_info = AutoReleaseInfo(releases=[]) for pr in prs: print(f"\nChecking PR [{pr.head.ref}]") refs = list(repo.get_git_matching_refs(f"tags/v{pr.head.ref}")) assert refs latest_release_tag_ref = refs[-1] latest_release_tag = repo.get_git_tag(latest_release_tag_ref.object.sha) commits = Shell.get_output_or_raise( f"git rev-list --first-parent {latest_release_tag.tag}..origin/{pr.head.ref}", ).split("\n") commit_num = len(commits) if latest_release_tag.tag.endswith("new"): print("It's a new release branch - skip auto release for it") continue print( f"Previous release [{latest_release_tag.tag}] was [{commit_num}] commits ago, date [{latest_release_tag.tagger.date}]" ) commits_to_check = commits[:-1] # Exclude the version bump commit commit_sha = "" commit_ci_status = "" commits_to_branch_head = 0 for idx, commit in enumerate( commits_to_check[:MAX_NUMBER_OF_COMMITS_TO_CONSIDER_FOR_RELEASE] ): print( f"Check commit [{commit}] [{pr.head.ref}~{idx+1}] as release candidate" ) commit_num -= 1 is_completed = CI.GH.check_wf_completed(token=token, commit_sha=commit) if not is_completed: print(f"CI is in progress for [{commit}] - check previous commit") commits_to_branch_head += 1 continue # TODO: switch to check if CI is entirely green statuses = [ CI.GH.get_commit_status_by_name( token=token, commit_sha=commit, # handle old name for old releases status_name=(CI.JobNames.BUILD_CHECK, "ClickHouse build check"), ), CI.GH.get_commit_status_by_name( token=token, commit_sha=commit, # handle old name for old releases status_name=CI.JobNames.STATELESS_TEST_RELEASE, ), CI.GH.get_commit_status_by_name( token=token, commit_sha=commit, # handle old name for old releases status_name=CI.JobNames.STATEFUL_TEST_RELEASE, ), ] commit_sha = commit if any(status == SUCCESS for status in statuses): commit_ci_status = SUCCESS break print(f"CI status [{statuses}] - skip") commits_to_branch_head += 1 ready = False if commit_ci_status == SUCCESS and commit_sha: print( f"Add release ready info for commit [{commit_sha}] and release branch [{pr.head.ref}]" ) ready = True else: print(f"WARNING: No ready commits found for release branch [{pr.head.ref}]") autoRelease_info.add_release( ReleaseParams( release_branch=pr.head.ref, commit_sha=commit_sha, ready=ready, ci_status=commit_ci_status, num_patches=commit_num, commits_to_branch_head=commits_to_branch_head, latest=False, ) ) if autoRelease_info.releases: autoRelease_info.releases[-1].latest = True autoRelease_info.dump() def main(): args, parser = parse_args() if args.post_status: info = AutoReleaseInfo.from_file() for release_info in info.releases: if release_info.ready: CIBuddy(dry_run=False).post_info( title=f"Auto Release Status for {release_info.release_branch}", body=release_info.to_dict(), ) else: CIBuddy(dry_run=False).post_warning( title=f"Auto Release Status for {release_info.release_branch}", body=release_info.to_dict(), ) elif args.post_auto_release_complete: assert args.wf_status, "--wf-status Required with --post-auto-release-complete" if args.wf_status != SUCCESS: CIBuddy(dry_run=False).post_job_error( error_description="Autorelease workflow failed", job_name="Autorelease", with_instance_info=False, with_wf_link=True, critical=True, ) else: CIBuddy(dry_run=False).post_info( title=f"Autorelease completed", body="", with_wf_link=True, ) elif args.prepare: _prepare(token=args.token or get_best_robot_token()) else: parser.print_help() sys.exit(2) if __name__ == "__main__": main()