import os from datetime import datetime from pathlib import Path from typing import Union, List, Optional from aws.constants import PROD_ACCOUNT_ENVS from aws.env_info import get_env_bucket from aws.s3_apis.s3 import upload_file_to_s3, download_file_from_s3 from common.common import run_command from actions_logging.app_logging import logger from git.utils import get_current_branch, init_git_user, git_add_commit_and_push from github.env import exit_on_error_and_write_summary, get_required_env_var from master.calculate_version import Version, get_version from master.git_commit_tag_and_push import get_env_files_list from env_files.constants import ( VERSION_HISTORY_FILE, INITIAL_VERSION, INIT_MESSAGE, DATE_TIME_FORMAT, GLOBAL_FILE, VERSION_HISTORY_FOLDER) from env_files.merged_env_vars import get_merged_env_vars_changes, MergedEnvVarsChangeType from uuid import uuid4 import pathlib class VersionHistoryRecord: env_version: Union[Version, str] # the env version, for example v1.0.0. refer to the s3 dir of the .env file git_version: Union[str, None] # the git tag of env, for example v1.0.0-ENV, in SW will be created_at: str updated_at: str branch: str message: str current: bool def __init__(self, env_version: Union[Version, str], branch, git_version=None, first_time: bool = False, global_message: Union[list[str], str] = "", env_message: Union[list[str], str] = "", created_at=None, updated_at=None, message=None, current=False): self.git_version = git_version self.env_version = env_version self.first_time = first_time self.message = message self.created_at = created_at self.updated_at = updated_at self.branch = branch self.current = current now = str(datetime.now().strftime(DATE_TIME_FORMAT)) self.create_message(global_message, env_message) if not self.created_at: self.created_at = now self.updated_at = now def create_line(self): if not self.git_version: self.git_version = "null" props = [ f"git_version: {self.git_version}", f"env_version: {self.env_version}", f"created_at: {self.created_at}", f"updated_at: {self.updated_at}", f"message: {self.message}", f"branch: {self.branch}"] if self.current: props.append("current") history_line = ', '.join(props) return f"{history_line}\n" def create_message(self, global_message, env_message): if self.message: return if self.first_time: self.message = INIT_MESSAGE return self.message = f"global.json: {global_message} env.json: {env_message}" def update_updated_at(self): self.updated_at = str(datetime.now().strftime(DATE_TIME_FORMAT)) def __str__(self): return self.create_line() def __repr__(self): return self.create_line() def upload_version_history_to_s3(envs_to_upload_to, svc_name, version_history_base_path): """ will upload the version history file to the s3 for each env the s3 role will be the default role for deploy flow for master flow we need to assume role for each env :param envs_to_upload_to: :param svc_name: :param version_history_base_path: :return: """ try: logger.info_green(f"will upload version history to s3 for {svc_name} in {envs_to_upload_to} bucket") for env_name in envs_to_upload_to: version_history_file = os.path.join(version_history_base_path, env_name, VERSION_HISTORY_FILE) if not os.path.exists(version_history_file): logger.warning(f"version history file {version_history_file} not found") return bucket_name = get_env_bucket(env_name) file_path_in_s3 = f"{svc_name}/{VERSION_HISTORY_FILE}" logger.info(f"upload version history file {version_history_file} to {bucket_name}/{file_path_in_s3} in s3") env_name_for_role = env_name if env_name not in PROD_ACCOUNT_ENVS else "" upload_file_to_s3(bucket_name=bucket_name, file_name=version_history_file, file_path=file_path_in_s3, env_name=env_name_for_role) logger.info_green(f"upload version history to s3 done for {svc_name} in {envs_to_upload_to} bucket") except Exception as e: raise RuntimeError(f"error in upload_version_history_to_s3: {e}") def upload_services_version_histories_to_s3(is_multi_service_repo: bool, version_history_folder: str = VERSION_HISTORY_FOLDER): """ this function will iterate over env_vars_version_history folder and get every sub folder name as env. for every env or sub dir there should have version_histort.txt related to this env. upload the version_history.txt to the env s3 :return: """ try: logger.info_green(f"will upload services version histories to s3") if not is_multi_service_repo: svc_name = get_required_env_var('SVC_NAME') envs_to_upload_to = [env_name for env_name in os.listdir(version_history_folder)] upload_version_history_to_s3(envs_to_upload_to, svc_name, version_history_folder) return for svc_name in os.listdir(version_history_folder): version_history_svc_dir = os.path.join(version_history_folder, svc_name) envs_to_upload_to = [env_name for env_name in os.listdir(version_history_svc_dir)] logger.info(f"uploading version history for {svc_name} for all the envs {envs_to_upload_to}") upload_version_history_to_s3(envs_to_upload_to, svc_name, os.path.join(version_history_folder, svc_name)) except Exception as e: raise RuntimeError(f"error in upload_services_version_histories_to_s3: {e}") def version_history_record_from_line(line: str) -> VersionHistoryRecord: """ will create a VersionHistoryRecord from a line in the version history file :param line: :return: VersionHistoryRecord object """ fields = ["git_version", "env_version", "created_at", "updated_at", "message", "branch", "current"] current = 'current' line = line.strip() try: record_data = {} for index, key_name in enumerate(fields): if key_name != current and f"{key_name}: " not in line: raise RuntimeError(f"error in version_history_record_from_line: {key_name} not found in {line}") if key_name == current: if line.endswith(key_name): record_data[key_name] = True else: record_data[key_name] = False continue rest_of_line = line.split(f", {fields[index + 1]}: ")[0] data = rest_of_line.split(f"{fields[index]}: ")[1] record_data[key_name] = data versioned_env_version = get_version(record_data.get('env_version')) logger.debug(f"versioned_env_version: {versioned_env_version}") if versioned_env_version: record_data['env_version'] = versioned_env_version return VersionHistoryRecord(**record_data) except Exception as e: raise RuntimeError(f"error in version_history_record_from_line: {e}") def get_latest_env_vars_env_version(history_file_path: str) -> Version: """ will read the lines of the version history file and return the env version of the last record that is semver :param history_file_path: :return: env version """ try: with open(history_file_path, 'r') as f: lines = f.readlines() for line in lines: logger.debug(f"line in {history_file_path}: {line}") record = version_history_record_from_line(line) env_version = record.env_version if isinstance(env_version, Version): return env_version except Exception as e: raise RuntimeError(f"error in get_latest_env_vars_env_version: {e}") def read_version_history(file_path: str): try: with open(file_path, 'r') as file: logger.info_green(f"Reading version history from {file_path}") lines = file.readlines() return [line.strip() for line in lines if line.strip()] except FileNotFoundError as fe: logger.warning(f"YELLOW File {file_path} not found. {fe}") return except Exception as e: raise RuntimeError(f"An error occurred in read_version_history as {e}") def display_version_history(file_path: str, lines_num: int = 10): with open(file_path, "r") as file: lines = file.readlines(lines_num) logger.debug(f"lines in {file_path}:") for line in lines: logger.debug(line) def add_version_history_record(record: VersionHistoryRecord, ms_and_env_name: str, version_history_folder: str = VERSION_HISTORY_FOLDER): """ This function will add a new record to the version history file. :param record: :param ms_and_env_name: :param version_history_folder: :return: """ try: history_file_path = os.path.join(version_history_folder, ms_and_env_name) os.makedirs(history_file_path, exist_ok=True) version_history_full_path = os.path.join(history_file_path, VERSION_HISTORY_FILE) if not os.path.exists(version_history_full_path): logger.info(f"Creating new version history file {version_history_full_path}") with open(version_history_full_path, "w") as file: file.write(record.create_line()) if os.getenv('LOG_LEVEL', '') == 'DEBUG': display_version_history(version_history_full_path) return # This is needed because of decision to pre-pend the new record to the file for better readability. # On multi-ms repos like sdpv2-domain the env_vars version is managed globally for all lambdas. # This will lead to history files bloat in this cases. Reading the whole file can be memory inefficient. logger.info(f"Adding new record {record} to {version_history_full_path}") escaped_line = record.create_line().replace('\n', '\\n').replace("'", "'\\''") # will add the new record to the top of the file if os.uname().sysname == 'Darwin': # for local usage: run_command(f"sed -i '' '1s|^|{escaped_line}|' {version_history_full_path}", return_output=False) else: # for github runner: run_command(f"sed -i '1s|^|{escaped_line}|' {version_history_full_path}", return_output=False) if logger.log_level == 'DEBUG': display_version_history(version_history_full_path) except Exception as e: raise RuntimeError(f"error in update_version_history: {e}") def create_envs_initial_history(env_files: List[str], is_multi_service_repo: bool, env_vars_git_tag: str, version_history_folder: str) -> dict[str, Version]: """ This function will create the initial version history for each env file. :param env_files: list of env files paths :param is_multi_service_repo: :param env_vars_git_tag: the git tag of the env vars for example v1.0.0-ENV :param version_history_folder: the folder where the version history files are stored :return: a dict of env file path as key and Version as value """ try: logger.info("First time run") all_env_files_versions = {} for env_file in env_files: logger.info(f"Creating new version history file for {env_file}") env_version = get_version(INITIAL_VERSION) init_record = VersionHistoryRecord(git_version=env_vars_git_tag, env_version=env_version, branch=get_current_branch(), first_time=True) ms_name = "" if is_multi_service_repo: ms_name = f"{Path(env_file).parents[1].name}/" env_name = Path(env_file).name.split(".json")[0] ms_and_env_name = f"{ms_name}{env_name}" add_version_history_record(record=init_record, ms_and_env_name=ms_and_env_name, version_history_folder=version_history_folder) logger.debug(f"updating all env files versions with {env_file} and version {env_version}") all_env_files_versions[env_file] = env_version return all_env_files_versions except Exception as e: exit_on_error_and_write_summary(f"error in create_envs_initial_history: {e}") def update_envs_history(env_files: List[str], env_files_base_path: str, env_vars_git_tag: str, changes: dict, is_multi_service_repo: bool, version_history_folder: str = VERSION_HISTORY_FOLDER): """ we assume that the list of envs for a microservices is explicitly defined by all env.json files if I don't have the file - environment considered non-existing. means one has to create at least an empty file to get defaults from global.json if env file is deleted the environment for the MS considered deleted too and is not handled """ try: logger.info("Updating version history") global_json = os.path.join(env_files_base_path, GLOBAL_FILE) merged_env_vars_changes = get_merged_env_vars_changes(env_files_path=env_files_base_path) logger.info(f"got merged env vars changes: {merged_env_vars_changes}") all_env_files_versions = {} for env_file in env_files: if GLOBAL_FILE in env_file: logger.debug(f"Skipping global.json file {env_file}") continue ms_name_path = "" if is_multi_service_repo: ms_name = Path(env_file).parents[1].name ms_name_path = f"{ms_name}/" global_json = os.path.join(Path(env_file).parents[1], GLOBAL_FILE) env_name = Path(env_file).name.split(".json")[0] ms_and_env_name = f"{ms_name_path}{env_name}" history_full_path = os.path.join(version_history_folder, ms_and_env_name, VERSION_HISTORY_FILE) # it's a new environment for existing ms: if not os.path.exists(history_full_path): logger.info(f"Creating new version history file for {env_file}") env_version = get_version(INITIAL_VERSION) init_record = VersionHistoryRecord(git_version=env_vars_git_tag, env_version=env_version, branch=get_current_branch(), first_time=True) add_version_history_record(init_record, ms_and_env_name) logger.debug(f"updating all_env_files_versions with {env_file} and version {env_version}") all_env_files_versions[env_file] = env_version continue logger.info(f"file {history_full_path} exists. will generate new line:") merged_env_change_level = merged_env_vars_changes.get(env_file, None) current_env_version = get_latest_env_vars_env_version(history_full_path) what_changed = "key" if merged_env_change_level == MergedEnvVarsChangeType.MINOR else "value" version_bumped = False if merged_env_change_level: logger.info(f"there were {what_changed} changes in {env_file}, {merged_env_change_level.value} bump") if not current_env_version: logger.warning(f"Couldn't find the env version in {history_full_path}, will create the initial one") current_env_version = get_version(INITIAL_VERSION) else: current_env_version.bump_one(merged_env_change_level.value) version_bumped = True env_file_changes = changes.get(env_file, []) global_changes = changes.get(global_json, []) record = VersionHistoryRecord(git_version=env_vars_git_tag, env_version=current_env_version, global_message=global_changes, env_message=env_file_changes, branch=get_current_branch()) add_version_history_record(record, ms_and_env_name) logger.info(f"record {record} added to {history_full_path}") if version_bumped: logger.debug(f"Updating env version for {env_file} to {current_env_version}") all_env_files_versions[env_file] = current_env_version return all_env_files_versions except Exception as e: raise RuntimeError(f"error in update_envs_history: {e}") def commit_history_changes(version_history_folder: str = VERSION_HISTORY_FOLDER): logger.info_green(f"will commit and push changes to version history") init_git_user() # TODO: this is a temporary solution, we need to get the list of files from the version history folder env_files_to_commit = get_env_files_list(version_history_folder) files_to_commit_str = ' '.join(env_files_to_commit).rstrip(' ') ignored_modified_files = ['.npmrc', 'package.json', 'package-lock.json', "yarkon_gw_config.yaml"] git_add_commit_and_push(files_to_commit_str, f'[CI/CD] updated {files_to_commit_str}', get_current_branch(), True, ignored_modified_files) logger.info_green("commit and push done for version history") def delete_current_from_records(history_file: str): current_version_record = run_command(f"grep 'current' {history_file}", return_output=True, exit_on_error=False, print_command=False) logger.info_green(f"current_version_record: {current_version_record}") temp_file = f"{history_file}.tmp" cmd = "awk '{gsub(\", current\", \"\"); print}' %s > %s" % (history_file, temp_file) run_command(cmd, return_output=False) os.rename(temp_file, history_file) logger.info_green("'current' removed from env vars history records") def add_current_to_record(history_file: str, env_version: str = '', git_version: str = ''): version_pattern = '' logger.info_blue(f"env_version: {env_version}, git_version: {git_version}") if not git_version and not env_version: exit_on_error_and_write_summary(f'No git_version and env_version provided.') elif git_version and not env_version: version_pattern = f"git_version: {git_version}" elif env_version and not git_version: version_pattern = f"env_version: {env_version}" else: version_pattern = f"git_version: {git_version}, env_version: {env_version}" logger.info(f"Adding 'current' to env vars history records for {version_pattern}") temp_file = f"{history_file}.tmp" with open(history_file, 'r') as f: origin_lines = f.readlines() logger.info(f'Origin lines:\n{"".join(origin_lines)}') with open(temp_file, 'w') as tmp_f: pattern_found = False for line in origin_lines: if version_pattern in line and not pattern_found: line = version_history_record_from_line(line) line.current = True line.updated_at = str(datetime.now().strftime(DATE_TIME_FORMAT)) logger.info(f'Line after version_history_record_from_line: {line}') line = line.create_line() logger.info(f'Line after create_line: {line}') pattern_found = True tmp_f.write(line) if not pattern_found: logger.warning(f"Couldn't find the record with {version_pattern} in {history_file}, skipping") os.remove(temp_file) return False logger.info_green(f"'current' added to env vars history records for {version_pattern}") os.rename(temp_file, history_file) with open(history_file, 'r') as nf: logger.info_yellow(f"Updated version history file {history_file} content:") for line in nf.readlines(): logger.info_yellow(line.strip()) def set_current_env_vars_version(env_name: str, env_file_version: str, env_version: str, svc_name: str = "", is_multi_service_repo: bool = False, version_history_folder: str = VERSION_HISTORY_FOLDER): """ records for stg/prd envs and sw envs are stored differently: git_version:v2.0.0-ENV, env_version:v1.1.1, 2024-07-17 08:42:07... - stg/prd record git_version:null, env_version:boba-27-ad435b3, 2024-07-17 08:42:07... - SW record """ # handle stg/prd and SW reset try: logger.info_green(f"will add current env vars version to version history for {env_file_version}") history_sub_path = env_name if is_multi_service_repo: history_sub_path = os.path.join(svc_name, history_sub_path) history_file = os.path.join(version_history_folder, history_sub_path, VERSION_HISTORY_FILE) delete_current_from_records(history_file) add_current_to_record(history_file=history_file, git_version=env_file_version, env_version=env_version) logger.info_green(f"current added to version history for {env_file_version}") except Exception as e: raise RuntimeError(f"error in set_current_env_vars_version: {e}") def get_env_version_from_history(version_history_path, git_version): try: with open(version_history_path, 'r') as f: lines = f.readlines() for line in lines: record = version_history_record_from_line(line) if record.git_version == git_version: logger.info_green(f"env version for {git_version} is {record.env_version} (received from version history)") return record.env_version raise RuntimeError(f"couldn't find supplied git tag {git_version} in {version_history_path}") except Exception as e: raise RuntimeError(f"error in get_env_version_from_git_version: {e}") def get_current_env_vars_version_from_history_in_s3(svc_name: str, history_file: str, env_name: str) -> Optional[Version]: # get the first line that contains the current version, ignore the rest current_version_line = None with open(history_file, 'r') as history_f: for line in history_f: if line.strip().endswith('current'): current_version_line = line break if not current_version_line: logger.warning(f"Current env vars version not found in {env_name}, probably initial release with env vars for {svc_name} or something went wrong") return None # history line is of format: git_version: v1.1.1-ENV, env_version: v1.1.1, ... current_version_field = current_version_line.split(',')[0].strip() version_field = current_version_field.split(':')[1].strip() if version_field == "null": logger.error(f"Current env vars version is null for env vars for {svc_name}") return None # Hotfix version if len(version_field.split('-')) == 3: logger.warning(f"Current env vars version is hotfix version for env vars for {svc_name}") return version_field try: version = get_version(version_field.split('-')[0]) logger.info(f"Current {svc_name} env vars version in production: {version}") return version except Exception as e: logger.error(f"Error parsing version from {version_field} for {svc_name}: {e}") return None def get_current_env_vars_version_from_s3(svc_name: str, env_name: str) -> Optional[Version]: logger.info(f"getting {svc_name} env vars version history from {env_name}") bucket_name = get_env_bucket(env_name) src = os.path.join(svc_name, VERSION_HISTORY_FILE) history_file = os.path.join(os.getcwd(), f"{VERSION_HISTORY_FILE}-{svc_name}-{uuid4()}") try: env_for_assume_role = '' if env_name in PROD_ACCOUNT_ENVS else env_name download_file_from_s3(bucket_name=bucket_name, file_name=src, file_name_to_save=history_file, env_name=env_for_assume_role, skip_errors=True) if not os.path.exists(history_file): raise FileNotFoundError(f"File not found: {history_file}") if os.stat(history_file).st_size == 0: raise ValueError(f"{history_file} downloaded for {svc_name} from {bucket_name} is empty") return get_current_env_vars_version_from_history_in_s3(svc_name, history_file, env_name) except Exception as e: logger.warning(f"Failed to get env vars version history for {env_name}: {e}") return None finally: pathlib.Path(history_file).unlink(missing_ok=True)