from colorama import Fore from gitlab import SEARCH_SCOPE_BLOBS, Gitlab, exceptions from helpers import ( get_project_safe, get_version_from_full_scan, get_version_from_search, ) from yaml import dump from argocd.helper import get_deployment_files from common import ( get_env_variable, get_env_variable_required, is_frontend, print_message, repository_is_sdl_flex, ) class CreateSREMergeRequest(object): def __init__(self, repository, sre_team, jira_key) -> None: self.repository = repository self.sre_team = sre_team self.jira_key = jira_key gitlab_protocol = get_env_variable_required("CI_SERVER_PROTOCOL") gitlab_host = get_env_variable_required("CI_SERVER_HOST") self.gitlab_manager = Gitlab( url=f"{gitlab_protocol}://{gitlab_host}", private_token=get_env_variable_required("SCI_ARGOCD_TOKEN"), per_page=100, ) def create(self, is_flex: bool = False): k8s_repo_name = self._get_k8s_repo_sre_name() if is_flex: k8s_repo_name += "-flex" prod_project_name_with_namespace = f"devops/prod-{k8s_repo_name}-k8s-manifests" prod_project_name = get_env_variable("SCI_PROD_MANIFEST_PROJECT") if prod_project_name: prod_project_name_with_namespace = prod_project_name if is_flex and not prod_project_name.endswith("-flex-k8s-manifests"): prod_project_name_with_namespace = prod_project_name.replace( "-k8s-manifests", "-flex-k8s-manifests" ) leaf_project_name_with_namespace = get_env_variable("SCI_MANIFEST_PROJECT") ( files_path, prod_project, environment_project, ) = self._get_project_and_deployment_files_path( prod_project_name_with_namespace, leaf_project_name_with_namespace, should_create_mr_flex=is_flex, ) old_images = [] if not environment_project and not files_path: return None, old_images deployment_files = get_deployment_files(environment_project, files_path) files_changed = [] for deployment_file in deployment_files: deployment_file.content = self._change_image_version( deployment_file.content, self.repository.version, old_images ) files_changed.append(deployment_file) action = "update" if "prod" in environment_project.name else "create" if action == "create": old_images = [] return ( self._commit_changes( action, prod_project, files_changed, ), old_images, ) def _change_image_version(self, deployment_file, version: str, old_images: []): kind = deployment_file.get("kind", "Deployment") try: if kind == "CronJob": containers_path = deployment_file["spec"]["jobTemplate"]["spec"][ "template" ]["spec"]["containers"] else: containers_path = deployment_file["spec"]["template"]["spec"][ "containers" ] except KeyError as e: print_message( f"Estrutura inesperada no manifesto '{kind}'. Erro: {e}", Fore.YELLOW ) return for container in containers_path: image_full = container["image"] if ":" not in image_full: print_message( f"Imagem '{image_full}' não possui tag definida. Usando 'latest' como fallback.", Fore.YELLOW, ) image_name = image_full else: image_name, _ = image_full.split(":") if image_name in [ f"seniorsa/{self.repository.deployment}", f"seniorsa/{self.repository.deployment}-http", f"seniorsa/{get_env_variable_required('CI_PROJECT_NAME')}", ]: if not container["image"] in old_images: old_images.append(container["image"]) container["image"] = f"{image_name}:{version}" return deployment_file def _get_k8s_repo_sre_name(self): sre_team_lowered = self.sre_team.lower() if sre_team_lowered == "sre": return "platform" if "sre" in sre_team_lowered: return sre_team_lowered.replace("sre", "") return sre_team_lowered def _get_project_and_deployment_files_path( self, project_name_with_namespace, project_name_with_namespace_fallback=None, prod_project=None, should_create_mr_flex: bool = False, ): try: environment_project = self.gitlab_manager.projects.get( project_name_with_namespace ) except exceptions.GitlabGetError: print(" ") print_message( ( f"Não foi encontrado o repositório de manifestos {project_name_with_namespace} para esse serviço. " "Por favor entre em contato com o SRE do seu produto!" ), Fore.YELLOW, ) print(" ") return None, None, None image_name = ( f"seniorsa/{self.repository.deployment}-http" if should_create_mr_flex and repository_is_sdl_flex() else f"seniorsa/{self.repository.deployment}" ) files_path = environment_project.search(SEARCH_SCOPE_BLOBS, f"{image_name}:") if not prod_project: prod_project = environment_project if not files_path and project_name_with_namespace_fallback: ( files_path, prod_project, environment_project, ) = self._get_project_and_deployment_files_path( project_name_with_namespace_fallback, prod_project=prod_project, should_create_mr_flex=should_create_mr_flex, ) elif not files_path: print(" ") print_message( ( "Não foi encontrado nenhum manifesto para atualização. " "Por favor entre em contato com o SRE do seu produto!" ), Fore.YELLOW, ) print(" ") return None, None, None return files_path, prod_project, environment_project def get_current_version(self, manifest_project: str) -> str | None: print_message("[HOMOLOGX] Buscando versão no manifest...", Fore.CYAN) project = get_project_safe(self.gitlab_manager, manifest_project) if not project: return None image_name = ( f"seniorsa/{get_env_variable_required('CI_PROJECT_NAME')}" if is_frontend() else f"seniorsa/{self.repository.deployment}" ) if repository_is_sdl_flex(): image_name = f"{image_name}-http" version = get_version_from_search( project, image_name, self._change_image_version ) if version: return version return get_version_from_full_scan( project, image_name, self._change_image_version ) def _commit_changes(self, action: str, gitlab_project, files_to_commit): branch_name = f"{action}/{self.jira_key.lower()}" print_message("Criando/Atualizando branch no repositório de manifestos...") try: gitlab_project.branches.get(branch_name) except exceptions.GitlabGetError: gitlab_project.branches.create({"branch": branch_name, "ref": "develop"}) commit_message = ( f"Atualizando imagem do serviço {self.repository.deployment} " f"para a versão {self.repository.version}." if action == "update" else f"Implantando serviço {self.repository.deployment}" ) actions = [] for file in files_to_commit: actions.append( { "action": action, "file_path": file.path, "content": dump(file.content), }, ) gitlab_project.commits.create( { "branch": branch_name, "commit_message": commit_message, "actions": actions, } ) print_message( "Criando/Atualizando merge request no repositório de manifestos..." ) try: mr_tile = f"#{self.jira_key} - {commit_message}" return gitlab_project.mergerequests.create( { "source_branch": branch_name, "target_branch": "develop", "title": mr_tile, "remove_source_branch": True, } ) except exceptions.GitlabCreateError: mrs = gitlab_project.mergerequests.list( source_branch=branch_name, target_branch="develop" ) return mrs[0]