"""Update containerized lambda. Update ECR repository image tag for lambda function, publish a new version and move a specified alias to it. This is a possible replacement for `update_ecr_code_source.py`. """ import argparse import os import time import boto3 import botocore.exceptions AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1') MAX_ATTEMPTS = int(os.environ.get('RETRY_MAX_ATTEMPTS', '60')) DELAY = int(os.environ.get('RETRY_DELAY', '5')) ECR_REGION = os.environ.get('ECR_REGISTRY_REGION', 'us-east-1') # 620f6ccb5b4e8bd5437577509d364468a2086dba # 127459fc92e29272673d9db584c6e1b35928f47b #python -u update_containerized_lambda.py -f qa-lambda-documents-payoneer-webhooks -t 620f6ccb5b4e8bd5437577509d364468a2086dba -i lambda-documents-payoneer-webhooks -p -a test_alias def main(): """Entrypoint.""" args = get_cli_arguments() function_name, image_tag, publish_version, alias_name, \ image_name, revert, specific_revert_version = ( args.function_name, args.image_tag, args.publish_version, args.alias_name, args.image_name, args.revert, args.specific_revert_version) if not function_name: print('Lambda function name is not specified in the command-line ' 'arguments, trying to generate it using environment variables.') function_name = generate_function_name() if not image_name: print('Image name is not specified in the command-line ' 'arguments, trying to generate it using environment variables.') image_name = generate_image_name() print(f'Processing {function_name} lambda function.') lambda_client = boto3.client('lambda', region_name=AWS_REGION) print('Getting lambda function info.') response = lambda_client.get_function(FunctionName=function_name) if response['Code']['RepositoryType'] != 'ECR': raise SystemExit('\nLambda function is not configured to use ECR') old_image_uri = response['Code']['ImageUri'] ECR_ACCOUNT_ID = os.environ.get( 'ECR_REGISTRY_ACCOUNT_ID', '437795906767') ecr_registry_url = f'{ECR_ACCOUNT_ID}.dkr.ecr.{ECR_REGION}.amazonaws.com' if revert: print('Initiating revert flow. Finding Image URI to use for revert') if specific_revert_version: response = lambda_client.get_function( FunctionName=function_name, Qualifier=specific_revert_version, ) new_image_uri = response['Code']['ImageUri'] else: all_versions = lambda_client.get_paginator( 'list_versions_by_function').paginate( FunctionName=function_name).build_full_result() version_dict = {} for version in all_versions['Versions']: version_dict[version['LastModified']] = version['Version'] version_dict_keys = list(version_dict.keys()) version_dict_keys.sort() # Get version prior to the latest one previous_version = version_dict[version_dict_keys[-2]] response = lambda_client.get_function( FunctionName=function_name, Qualifier=previous_version, ) new_image_uri = response['Code']['ImageUri'] """ Since we are getting the Image URI rather than specifying it, parse out the values of variables needed to verify the image still exists. """ ECR_ACCOUNT_ID = new_image_uri.split('.')[0] image_name = new_image_uri.split('/')[-1].split(':')[0] image_tag = new_image_uri.split(':')[-1] else: new_image_uri = f'{ecr_registry_url}/{image_name}:{image_tag}' print(f'\nValidating image {new_image_uri} exists') ecr_client = boto3.client('ecr', region_name=AWS_REGION) try: ecr_client.describe_images( registryId=ECR_ACCOUNT_ID, repositoryName=image_name, imageIds=[ {'imageTag': image_tag} ] ) except botocore.exceptions.ClientError: raise SystemExit( f'\nFailed to validate existence of image {new_image_uri}. ' f'ECR Repository does not exist or does not permit access.' ) print( f'Updating lambda image URI from {old_image_uri} to {new_image_uri}.') response = lambda_client.update_function_code( FunctionName=function_name, ImageUri=new_image_uri) print('Waiting for lambda function update.') try: waiter = lambda_client.get_waiter('function_updated') waiter.wait(FunctionName=function_name, WaiterConfig={ 'Delay': DELAY, 'MaxAttempts': MAX_ATTEMPTS }) except botocore.exceptions.WaiterError: raise SystemExit('Timeout exceeded while waiting for ' 'lambda function update.') print('Lambda function update successful.') if publish_version: print('Publishing new version of lambda function.') response = lambda_client.publish_version(FunctionName=function_name) try: version = response['Version'] print(f'Published version {version} for lambda function.') except KeyError: raise SystemExit('No published versions for this lambda function.') if alias_name != '': try: prov_conc_config = lambda_client.get_provisioned_concurrency_config( FunctionName=function_name, Qualifier=alias_name) except lambda_client.exceptions.ProvisionedConcurrencyConfigNotFoundException: prov_conc_config = None if prov_conc_config is not None: # If after pervious alias update provisioned concurrency # config is in failed state, it would only update to a # previous healthy version which can be extracted from # AdditionalVersionWeights property of the alias. if prov_conc_config['Status'] == 'FAILED': print(f'Detected a failure of the previous deployment, ' f'reason: {prov_conc_config["StatusReason"]}.') previous_version = get_previous_alias_version( lambda_client,function_name,alias_name) if previous_version is not None: print(f'Rolling alias back on previous healthy version.' f'{previous_version}.') lambda_client.update_alias( FunctionName=function_name, Name=alias_name, FunctionVersion=previous_version, RoutingConfig={}) wait_provisioned_concurrency_config( lambda_client,function_name,alias_name) else: print(f'No previous healthy versions were found for ' f'alias {alias_name}. Skipping rollback.') print(f'Updating alias to new version {version}.') lambda_client.update_alias( FunctionName=function_name, Name=alias_name, FunctionVersion=version, RoutingConfig={}) if prov_conc_config is not None: wait_provisioned_concurrency_config( lambda_client,function_name,alias_name) print('Lambda alias update successful.') def get_cli_arguments(): """Parse command-line arguments.""" parser = argparse.ArgumentParser( description='Update containerized lambda.') parser.add_argument('-f', '--function-name', dest='function_name', default='', help='Lambda Function name.') parser.add_argument('-t', '--image-tag', required=True, dest='image_tag', help='Docker image tag to use.') parser.add_argument('-p', '--publish-version', dest='publish_version', action='store_true', help='Whether to publish a new Lambda version.') parser.add_argument('-a', '--alias-name', dest='alias_name', default='', help='Lambda alias to point out to the new Lambda ' 'version.') parser.add_argument('-i', '--image-name', dest='image_name', default='', help='The name of the image to deploy. ' 'This is equivalent to the ECR repository name.') parser.add_argument('-r', '--revert', dest='revert', action='store_true', help='Whether to revert the function. Optionally ' 'also specify -s with a function version.') parser.add_argument('-s', '--specific-revert-version', dest='specific_revert_version', default='', help='Specific function version to use for revert.') args = parser.parse_args() return args def generate_function_name(): """Generate function name using environment variables.""" environment = os.environ.get('ENV') prefix = os.environ.get('PREFIX', 'lambda') lambda_dir = os.environ.get('LAMBDA_DIR') assert environment, 'Environment variable ENV must be set.' assert prefix, 'Environment variable PREFIX must be set.' assert lambda_dir, 'Environment variable LAMBDA_DIR must be set.' return '-'.join([ environment, prefix, lambda_dir.replace('_', '-') ]) def generate_image_name(): """Generate image name using environment variables.""" prefix = os.environ.get('PREFIX', 'lambda') lambda_dir = os.environ.get('LAMBDA_DIR') assert prefix, 'Environment variable PREFIX must be set.' assert lambda_dir, 'Environment variable LAMBDA_DIR must be set.' return f"{prefix}-{lambda_dir.replace('_', '-')}" def get_previous_alias_version(lambda_client, function_name, alias_name): """Get previous alias version from AdditionalVersionWeights configuration""" response = lambda_client.get_alias(FunctionName=function_name, Name=alias_name) try: # AdditionalVersionWeights contain only a single key with the version number return list(response['RoutingConfig']['AdditionalVersionWeights'].keys())[0] except KeyError: return None def wait_provisioned_concurrency_config( lambda_client, function_name, alias_name): """ Wait until provisioned concurrency config of alias gets to Ready state. Exit if it fails to update. """ print('Waiting for provisioned concurrency config to update to new version,' ' which may take up to 2-3 minutes for provisioned lambdas.') status = 'IN_PROGRESS' while status == 'IN_PROGRESS': time.sleep(DELAY) response = lambda_client.get_provisioned_concurrency_config( FunctionName=function_name, Qualifier=alias_name) status = response['Status'] status_reason = response.get('StatusReason', '') if status != 'READY': raise SystemExit(f'Provisioning concurrency config has ' f'failed to provision environment for ' f'alias {alias_name} with error {status_reason}.') if __name__ == '__main__': main()