from datetime import datetime from typing import NoReturn from colorama import Fore from gitlab import const from gitlab.exceptions import GitlabGetError from yaml import Dumper, dump from argocd import ArgoCDCenter, helper from common import ( end_collapsible_section, exit_message, get_commit_tag_as_semver, get_env_variable_required, get_last_version, get_version, is_semver, print_message, repository_is_sdl_flex, start_collapsible_section, ) from common.dockerhub_helper import DockerHubHelper from common.file_helpers import get_svc_name from src.enum.project_type import ProjectTypeEnum class UpdateDeploymentsImage(ArgoCDCenter): """ Atualiza as imagens dos deployments do serviço para a versão que está sendo executada. Se não for executado em uma tag será atualizado para uma versão SNAPSHOT do branch. """ def __init__(self) -> NoReturn: super().__init__() ci_project_name = get_env_variable_required("CI_PROJECT_NAME") self.gitlab_user_name = get_env_variable_required("GITLAB_USER_NAME") self.ci_commit_ref_slug = get_env_variable_required("CI_COMMIT_REF_SLUG") self.dockerhub_helper = DockerHubHelper() self.svc_name = self.get_service_name() self.image_name = self.get_image_name() self.version = self._get_version() self.deploy_branch = f"deploy/{ci_project_name}-{self.ci_commit_ref_slug}" self.start_update_process() def _get_version(self) -> str: """ Retrieves the current version of the project, appending the '-SNAPSHOT' suffix if necessary to indicate a development version. https://maven.apache.org/guides/getting-started/#What_is_a_SNAPSHOT_version.3F """ version = get_version() sci_project_type = get_env_variable_required("SCI_PROJECT_TYPE") project_type = ProjectTypeEnum[sci_project_type] if not is_semver(version) and project_type in [ ProjectTypeEnum.MAVEN_SDL, ProjectTypeEnum.MAVEN_APP, ]: return f"{version}-SNAPSHOT" return version def get_service_name(self) -> str: svc_name = get_svc_name() if not svc_name: exit_message( "Não foi possível encontrar o nome da imagem do serviço. " "Por favor entre com contato com a equipe DevOps." ) return svc_name def get_image_name(self) -> str: image_name = ( f"seniorsa/{self.svc_name}-http" if repository_is_sdl_flex() and "flex" in self.manifest_project.name else f"seniorsa/{self.svc_name}" ) image_in_dockerhub = self.dockerhub_helper.get_repo_by_image_name(image_name) if not image_in_dockerhub: exit_message( f"A imagem {image_name} não existe no Docker Hub. " "Certifique-se que o job release snapshot executou com sucesso ou converse com SRE do seu produto." ) return image_name def start_update_process(self) -> None: section_id = start_collapsible_section( f'Começando processo de atualização da imagem "{self.image_name}" para a versão "{self.version}" ' f"no repositório {self.manifest_project.name}.", Fore.CYAN, ) print_message( f'Procurando manifestos que possuem a imagem "{self.image_name}"...' ) blobs_result_search = self.manifest_project.search( const.SEARCH_SCOPE_BLOBS, f"{self.image_name}:" ) has_mr = self.handle_manifest_search(blobs_result_search) if not has_mr: self.create_or_update_branch() self.update_deployment_files(blobs_result_search) end_collapsible_section(section_id) def handle_manifest_search(self, blobs_result_search) -> bool: implementable_environments = ["homologx", "beta"] if not blobs_result_search: environment = self.manifest_project.name.split("-")[0] if self.leaf_manifest_project and environment in implementable_environments: return self.copy_deployment_from_leaf(self.image_name, self.svc_name) else: exit_message( f'Não foi encontrado nenhum manifesto para atualização da imagem "{self.image_name}"', ) print_message(f"Foram encontrados {len(blobs_result_search)} manifestos") return False def create_or_update_branch(self) -> None: try: svc_branch = self.manifest_project.branches.get(self.deploy_branch) print_message( f'Recriando branch "{self.deploy_branch}" no projeto de manifestos...' ) if svc_branch: svc_branch.delete() self.manifest_project.branches.create( { "branch": self.deploy_branch, "ref": self.ci_default_branch, "commit_message": f"Branch criada por {self.gitlab_user_name}", } ) except GitlabGetError: print_message( f'Criando branch "{self.deploy_branch}" no projeto de manifestos...' ) self.manifest_project.branches.create( { "branch": self.deploy_branch, "ref": self.ci_default_branch, "commit_message": f"Branch criada por {self.gitlab_user_name}", } ) def _find_and_update_container(self, containers, image_name, version): for container in containers: container_image_name = container["image"].split(":")[0] if container_image_name == image_name: container["image"] = f"{image_name}:{version}" return True return False def _update_init_containers(self, spec): init_containers = spec.get("initContainers", []) self._find_and_update_container(init_containers, self.image_name, self.version) def _update_timestamp(self, metadata, timestamp): annotations = metadata.get("annotations") if annotations: annotations["timestamp"] = timestamp else: metadata["annotations"] = {"timestamp": timestamp} def _build_commit_action(self, file_path, deployment_yaml): return { "action": "update", "file_path": file_path, "content": dump(deployment_yaml, Dumper=Dumper), } def update_deployment_files(self, blobs_result_search) -> None: commit_actions = [] now_timestamp = datetime.now().strftime("%s") for blob in blobs_result_search: file_path = blob["path"] filename = file_path.split("/").pop() deployment_yaml = self.get_yaml_by_path(file_path) deployment_template = deployment_yaml["spec"]["template"] containers = deployment_template["spec"]["containers"] container_updated = self._find_and_update_container( containers, self.image_name, self.version ) if not container_updated: print_message( f"Container com a imagem {self.image_name} não foi encontrado no manifesto {filename}", Fore.YELLOW, ) continue self._update_init_containers(deployment_template["spec"]) self._update_timestamp(deployment_template["metadata"], now_timestamp) commit_actions.append(self._build_commit_action(file_path, deployment_yaml)) print_message( f'O arquivo "{filename}" será atualizado com a nova imagem "{self.image_name}:{self.version}"' ) self.commit_changes(commit_actions) def commit_changes(self, commit_actions) -> None: data = { "branch": self.deploy_branch, "commit_message": f"Sincronização solicitada por {self.gitlab_user_name}", "actions": commit_actions, } self.manifest_project.commits.create(data) print_message( "Todos os arquivos foram atualizados com sucesso, o deploy deles será feito no próximo job!", Fore.GREEN, )