import boto3 import re from uuid import uuid4 import pathlib from typing import List, Optional, Tuple from master.calculate_version import get_version from aws.env_info import get_svc_static_bucket, get_env_region, get_env_account from aws.s3_apis.s3 import download_file_from_s3 from aws.constants import STG_PROD_ENVS from aws.ecr import EcrConfig from actions_logging.app_logging import logger from aws.constants import ENV_EXCLUDED_SERVICES from env_files.version_history import get_current_env_vars_version_from_s3 from github.env import exit_on_error_and_write_summary from github.github_apis import get_file_content from github.constants import EKS_REPO_NAME, REPOS_YAML_PATH from build_and_deploy.get_service_type import parse_and_validate_repos_yaml from github.get_high_tag import get_git_high_tags FE_VERSION_FILE = "VERSION.txt" LAMBDA_VERSION_TAG = "LAMBDA_VERSION" PIPELINE_VERSION_TAG = "PIPELINE_VERSION" SEMVER_PATTERN = "\d+.\d+.\d+" HOTFIX_PATTERN = "(staging|production)-\d+-[a-z0-9]{7,9}" def get_tag_pattern(env_name: str) -> str: pattern = f"({env_name})-\d+-[a-z0-9]{{7,9}}" if env_name in STG_PROD_ENVS: pattern = HOTFIX_PATTERN return pattern def create_ms_dict_from_repos(d): """ invert the repos yaml to be ms_name as key and repo_name as value :param d: repos dict from yaml :return: dict of {ms_name: {md_data_dict}} """ try: ms_dict = {} for key, value in d.items(): if isinstance(value, list): for ms in value: ms_dict[ms['ms_name']] = ms ms_dict[ms['ms_name']]['repo'] = key ms_dict[ms['ms_name']]['multi_service_repo'] = True else: ms_dict[value['ms_name']] = value ms_dict[value['ms_name']]['repo'] = key return ms_dict except Exception as e: raise RuntimeError(f"failed to create ms dict - {e}") def collect_deployed_services_data(env_name: str, svc_to_ignore: List[str], ignore_errors: bool = False) -> Tuple[dict, list]: services_with_no_version = [] repos_yaml_content = get_file_content(EKS_REPO_NAME, REPOS_YAML_PATH) repos = parse_and_validate_repos_yaml(repos_yaml_content) filtered_repos = filter_excluded_services(repos, svc_to_ignore) svc_data = create_ms_dict_from_repos(filtered_repos) services_statuses = [] for svc_name, value in svc_data.items(): if svc_name.startswith('cd-'): continue svc_type = value['type'] if svc_type == 'lib': continue logger.info(f"fetching version data for {svc_name} of type {svc_type}") version = get_artifact_version(svc_name, env_name, svc_type, ignore_errors) if not version: services_with_no_version.append(svc_name) continue org_repo = f"perimeter-81/{value['repo']}" suffix = svc_name if value.get('multi_service_repo', False) else None latest_version = get_git_high_tags(org_repo, suffix) latest_env_vars_version = get_git_high_tags(org_repo, 'ENV') latest_env_vars_version = latest_env_vars_version if latest_env_vars_version else None env_vars_version = get_current_env_vars_version_from_s3(svc_name, env_name) if svc_type not in ['pipeline'] else None services_statuses.append({ "name": svc_name, "type": svc_type, "repo": value['repo'], "version": version, "env_vars_version": env_vars_version, "latest_version": latest_version, "latest_env_vars_version": latest_env_vars_version,}) return services_statuses, services_with_no_version def get_image_version(svc_name: str, env_name: str) -> str: try: cfg = EcrConfig(env_name) client = boto3.client('ecr', region_name=cfg.ecr_region) response = client.describe_images(repositoryName=f'{cfg.docker_project}{svc_name}', imageIds=[{'imageTag': env_name}]) tags = response['imageDetails'][0]["imageTags"] tag = next((tag for tag in tags if re.search(f".*v{SEMVER_PATTERN}.*", tag)), None) if not tag: # if we didn't find in the environment semver tag look for one with $env_name-buildid-runid pattern = get_tag_pattern(env_name) tag = next((tag for tag in tags if re.search(f'.*{pattern}.*', tag)), None) if tag: logger.info(f'ecr {env_name} version for {svc_name} is {tag}') return tag except Exception as e: logger.error(f"Couldn't fetch image version from {env_name} - {e}") return '' def get_fe_version(svc_name: str, env_name: str): fe_version_file = f"{FE_VERSION_FILE}-{svc_name}-{uuid4()}" try: version = None bucket = get_svc_static_bucket(env_name, svc_name) download_file_from_s3(bucket, FE_VERSION_FILE, fe_version_file) with open(fe_version_file, 'r') as f: version = f.read().split('\n')[0] logger.info(f'frontend {env_name} version for {svc_name} is {version}') pattern = get_tag_pattern(env_name) if not (re.search(f"{SEMVER_PATTERN}.*", version) or re.search(f'{pattern}.*', version)): logger.error(f"version {version} not in correct format") return str(version) except Exception as e: logger.error(f"error geting FE version in {env_name} for ms {svc_name}: {e}") return '' finally: pathlib.Path(fe_version_file).unlink(missing_ok=True) def get_pipeline_version_from_tags(client: boto3.client, connector_arn: str) -> str: tags = client.list_tags_for_resource(resourceArn=connector_arn).get("tags") if tags: pipeline_version = tags.get(PIPELINE_VERSION_TAG, None) return pipeline_version def get_kafkaconnect_plugin_version(plugins: dict, svc_name: str, env_name: str) -> str: plugin_arn = plugins[0]['customPlugin']['customPluginArn'] plugin_name = plugin_arn.split("/")[-2] logger.debug(f"connector plugin name: {plugin_name}") version_raw = plugin_name.split(f"-{svc_name}-")[-1] semver_search = re.search(SEMVER_PATTERN, version_raw.replace("-", ".")) pattern = get_tag_pattern(env_name) dev_version_search = re.search(pattern, version_raw) version = None if semver_search: version = semver_search.group() elif dev_version_search: version = dev_version_search.group() if not version: raise ValueError(f"Couldn't parse version from plugin name {plugin_name}") return version def get_pipeline_version(svc_name: str, env_name: str) -> str: try: region = get_env_region(env_name) connector_name = f'{env_name}-{svc_name}' msk = boto3.client('kafkaconnect', region_name=region) connectors = msk.list_connectors(connectorNamePrefix=connector_name).get("connectors") logger.info(f"kafka connectors data for {connector_name}: {connectors}") if not connectors or len(connectors) > 1: raise ValueError(f"bad connectors list: {len(connectors)} with the prefix {connector_name}") connector_arn = connectors[0]['connectorArn'] version = get_pipeline_version_from_tags(msk, connector_arn) if version: logger.info(f"{connector_name} prod version: {version}") return str(version) # Backward comaptibility way of fetching version from kafka connector logger.debug(f"pipeline tag not found for {connector_arn} falling back to looking version in plugin arn") version = get_kafkaconnect_plugin_version(connectors[0]['plugins'], svc_name, env_name) logger.info(f"{connector_name} prod version: {version}") return str(version) except Exception as e: logger.error(f"Couldn't fetch {env_name} version for yarkon pipeline {svc_name}: {e}") return '' def get_lambda_version(svc_name: str, env_name: str) -> str: try: region = get_env_region(env_name) aws_account = get_env_account(env_name) lambda_client = boto3.client('lambda', region_name=region) lambda_arn = f"arn:aws:lambda:{region}:{aws_account}:function:{env_name}-{svc_name}" response = lambda_client.list_tags(Resource=lambda_arn) lambda_version = response.get('Tags', {}).get(LAMBDA_VERSION_TAG, None) if not lambda_version: raise ValueError(f"Couldn't get current {env_name} lambda version from tags on {svc_name} lambda") logger.info(f'lambda version for {svc_name} in {env_name} is {lambda_version}') return str(lambda_version) except Exception as e: logger.error(e) return '' def get_artifact_version(svc_name: str, env_name: str, svc_type: str, ignore_errors: bool = False) -> Optional[str]: version = '' if svc_type in ['backend', 'core-ecs']: version = get_image_version(svc_name, env_name) elif svc_type == 'frontend': version = get_fe_version(svc_name, env_name) elif svc_type == 'lambda': version = get_lambda_version(svc_name, env_name) elif svc_type == 'pipeline': version = get_pipeline_version(svc_name, env_name) else: msg = f"Unsupported service type {svc_type} provided for {svc_name} - can't get artifact version" if ignore_errors: logger.warning(msg) return None else: exit_on_error_and_write_summary(msg) if not version: msg = f"Couldn't fetch version for {svc_name} in {env_name}" if ignore_errors: logger.warning(msg) return None else: exit_on_error_and_write_summary(msg) return version def filter_excluded_services(repos: dict, svc_to_ignore: List[str], env_name: str = 'solo') -> dict: excluded_services = ENV_EXCLUDED_SERVICES.get(env_name, []) + svc_to_ignore filtered_repos = {} for r, data in repos.items(): if isinstance(data, dict): if data['ms_name'] not in excluded_services: filtered_repos[r] = data elif isinstance(data, list): # If the data is a list (multiple services in one repo) filtered_list = [ms for ms in data if ms['ms_name'] not in excluded_services] if filtered_list: filtered_repos[r] = filtered_list return filtered_repos def log_short_svc_stats(svc_stats: List[dict], env_name: str) -> str: headers = ["Service Name", "Type", "Version", "Env Vars", "Latest Version", "Latest Env Vars"] version_len = len(env_name) + 11 row_format = "{:<50} {:<15} {:<" + str(version_len) + "} {:<" + str(version_len) + "} {:<15} {:<15}" def safe_format(value): return str(value) if value is not None else '-' table_output = row_format.format(*headers) + "\n" table_output += "-" * 150 + "\n" logger.info("\n" + table_output) for svc_data in svc_stats: table_output += row_format.format( safe_format(svc_data.get('name')), safe_format(svc_data.get('type')), safe_format(svc_data.get('version')), safe_format(svc_data.get('env_vars_version')), safe_format(svc_data.get('latest_version')), safe_format(svc_data.get('latest_env_vars_version')) ) + "\n" logger.info(f"\n{table_output}") return table_output