import subprocess from actions_logging.app_logging import logger from github.env import exit_on_error_and_write_summary, write_github_env from terragrunt.tg_common import safe_path from terragrunt.constants import CANDIDATE_SUFFIX def get_candidate_tag(path, version): tag_suffix = safe_path(path) return f"v{version}-{tag_suffix}_{CANDIDATE_SUFFIX}" def get_semver_tag(path, semver): suffix = safe_path(path) return f"v{semver}-{suffix}" def update_git_tags_on_apply(work_dir, plan_version): """ Update Git tags on applying the infrastructure changes. This function handles the creation and updating of Git tags after infrastructure has been applied. It creates a semantic versioning (semver) tag if it doesn't already exist, updates the 'deployed' tag, and deletes candidate tag. Args: work_dir (str): The working directory where the infrastructure code is located. plan_version (str): The version of the plan being applied. Raises: SystemExit: If an error occurs while executing Git commands. """ logger.info(f"Updating git tags for path {work_dir} and version {plan_version} after deploying") tag_prefix = safe_path(work_dir) semver_tag = get_semver_tag(work_dir, plan_version) deployed_tag = f"{tag_prefix}_deployed" try: logger.info(f"tagging with semver {semver_tag}") semver_tag_exists = subprocess.run(["git", "tag", semver_tag], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if semver_tag_exists.returncode > 0: logger.warning(f"tag {semver_tag} already exists. this should be a redeploy... skipping new semver tag creation") else: subprocess.run(["git", "push", "origin", semver_tag], check=True ) # if not there yet (first deploy, or fixes) logger.info(f"Deleting tag {deployed_tag}") deployed_tag_deleted = subprocess.run(["git", "tag", "-d", deployed_tag], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if deployed_tag_deleted.returncode == 0: logger.info(deployed_tag_deleted.stdout) if (deployed_tag_deleted.returncode > 0 and semver_tag_exists.returncode == 0): msg = f"Something went wrong: {semver_tag} exists but {deployed_tag} is not assigned yet, we will forcefully assign it since the infra was applied" logger.error(msg) write_github_env(msg, 'DEPLOY_ERROR') elif deployed_tag_deleted.returncode > 0 and semver_tag_exists.returncode > 0: logger.warning(f"First infra deploment") logger.info(f"Adding deployed tag back {deployed_tag}") subprocess.run(["git", "tag", deployed_tag], check=True ) logger.info(f"Tagged with {deployed_tag}") # Danger zone: forcefully update tags to point to currently checked out commit logger.info(f"Pushing deployed tag to remote") subprocess.run(["git", "push", '--force', "origin", deployed_tag], check=True,) logger.info(f"Pushed deployed tag {deployed_tag} to origin") # this was the first time deploy of this commit and it was tagged with semver so that candidate tag should be removed if semver_tag_exists.returncode == 0: candidate_tag = get_candidate_tag(work_dir, plan_version) logger.info(f"Deleting candidate tag {candidate_tag} locally and on remote") subprocess.run(["git", "tag", "-d", candidate_tag], check=True) logger.info(f"local tag {candidate_tag} deleted") subprocess.run(["git", "push", "origin", f":{candidate_tag}"], check=True) logger.info(f"remote tag {candidate_tag} deleted") except subprocess.CalledProcessError as e: exit_on_error_and_write_summary(f"An error occurred: {e}") def update_git_tags_on_destroy(work_dir): """ Update Git tags on destroying the infrastructure. This function handles the deletion of the 'deployed' tag and creation of the 'destroyed' tag after the infrastructure has been destroyed. It ensures that the tags are updated in the remote repository as well. Args: work_dir (str): The working directory where the infrastructure code is located. Raises: SystemExit: If an error occurs while executing Git commands. """ logger.info(f"Updating git tags for path {work_dir} and version after destroying") tag_prefix = safe_path(work_dir) deployed_tag = f"{tag_prefix}_deployed" destroyed_tag = f"{tag_prefix}_destroyed" try: deployed_tag_deleted = subprocess.run(["git", "tag", "-d", deployed_tag], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if deployed_tag_deleted.returncode == 0: logger.info(deployed_tag_deleted.stdout) if (deployed_tag_deleted.returncode > 0): exit_on_error_and_write_summary(f"Something went wrong: couldn't delete {deployed_tag}") subprocess.run(["git", "tag", destroyed_tag], check=True ) logger.info(f"Tagged with {destroyed_tag}") subprocess.run(["git", "push", '--force', "origin", destroyed_tag], check=True,) logger.info(f"Pushed destroyed tag {destroyed_tag} to origin") subprocess.run(["git", "push", "origin", f":{deployed_tag}"], check=True) logger.info(f"remote deployed tag {deployed_tag} deleted") except subprocess.CalledProcessError as e: exit_on_error_and_write_summary(f"An error occurred: {e}") def tag_exists(tag): try: logger.info(f"Validating tag {tag} exists on remote") tag_exists = subprocess.run(["git", "show-ref", "--tags", f"refs/tags/{tag}"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if tag_exists.returncode == 0: return True if tag_exists.returncode > 0 and not tag_exists.stderr: return False raise RuntimeError(tag_exists.stderr) except Exception as e: exit_on_error_and_write_summary(f"An error occuring during fetching tag {tag}: {e}") def create_or_update_candidate_tag(work_dir, version): """ Create or update a candidate tag to refer to the currently checked out commit. This function deletes any pre-existing candidate tag and creates a new one pointing to the current commit. It then force-pushes the updated candidate tag to the remote repository. Args: work_dir (str): The working directory where the infrastructure code is located. version (str): The version to be used for creating the candidate tag. Raises: SystemExit: If an error occurs while executing Git commands. """ candidate_tag = get_candidate_tag(work_dir, version) try: logger.info(f"Updating {candidate_tag} to refer to currently checked out commit") candidate_tag_deleted = subprocess.run(["git", "tag", "-d", candidate_tag], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if candidate_tag_deleted.returncode == 0: logger.warning(f"Deleted pre-existing candidate tag {candidate_tag}") if (candidate_tag_deleted.returncode > 0): logger.info(f"Candidate tag {candidate_tag} not found on repo") subprocess.run(["git", "tag", candidate_tag], check=True) logger.info(f"Candidate tag {candidate_tag} created") subprocess.run(["git", "push", '--force', "origin", candidate_tag], check=True) logger.info("Candidate tag force-pushed") except subprocess.CalledProcessError as e: exit_on_error_and_write_summary(f"An error occurred during creating candidate tag: {e}") def set_checkout_tag(path, semver): """ Set the Git tag to checkout based on the provided path and semantic version. This function determines the appropriate Git tag to checkout based on the presence of the semantic version tag or the candidate tag. It sets the GitHub environment variable 'CHECKOUT_TAG' accordingly. Args: path (str): The path to the repository. semver (str): The semantic version to use for determining the checkout tag. Raises: SystemExit: If the parameters are empty or if neither the semantic version tag nor the candidate tag is found on the repository. """ if not path or not semver: exit_on_error_and_write_summary("both parameters should be non-empty strings") semver_tag = get_semver_tag(path, semver) candidate_tag = get_candidate_tag(path, semver) semver_tag_exists = tag_exists(semver_tag) if semver_tag_exists: logger.warning(f"PLAN_VERSION {semver} will be re-applied from git") write_github_env(semver_tag, 'CHECKOUT_TAG') elif tag_exists(candidate_tag): logger.warning(f"PLAN_VERSION {semver} corresponding {semver_tag} wasn't found on the repository, will run on git candidate tag: {candidate_tag}") write_github_env(candidate_tag, 'CHECKOUT_TAG') else: exit_on_error_and_write_summary(f"PLAN_VERSION {semver} corresponding {semver_tag} and candidate tag {candidate_tag} were not found on the repository, halting.")