"""Deploy ECS tasks and update services.""" import os import boto3 from botocore.exceptions import ClientError from dotenv import load_dotenv from environs import Env from constants import VerifyMode from utils import poll_task_status from docker_image_info import DockerImageInfo # Use env for parsing environment variables env = Env() # Load env file if it exists load_dotenv(verbose=True) task_family = os.environ.get('TASK_FAMILY') image_tag = os.environ.get('GIT_COMMIT', 'latest') deployment_job = os.environ.get('JOB_URL', 'unknown') cluster_name = os.environ.get('CLUSTER_NAME') service_name = os.environ.get('SERVICE_NAME') fargate_service_name = os.environ.get('FARGATE_SERVICE_NAME') container_name = os.environ.get('CONTAINER_NAME') aws_region = os.environ.get('AWS_REGION', 'us-east-1') update_timeout = int(os.environ.get('UPDATE_TIMEOUT', 300)) cloudwatch_downscaling_alarm_name = os.environ.get( 'CLOUDWATCH_DOWNSCALING_ALARM_NAME', f'{fargate_service_name}-down-scaling-alarm') # Set to REVERT to initiate revert workflow deploy_mode = os.environ.get('DEPLOY_MODE', 'NEW_DEPLOYMENT') revert_task_definition_number = os.environ.get('REVERT_TASK_DEFINITION_NUMBER') ecr_account_id = os.environ.get( 'ECR_REGISTRY_ACCOUNT_ID', '437795906767') ecr_region = os.environ.get('ECR_REGISTRY_REGION', 'us-east-1') ecr_registry_url = f'{ecr_account_id}.dkr.ecr.{ecr_region}.amazonaws.com' image_name = os.environ.get('IMAGE_NAME_OVERRIDE', service_name) image_uri = os.environ.get( 'IMAGE_URI', f'{ecr_registry_url}/{image_name}:{image_tag}') defender_container_name = 'TwistlockDefender' defender_entrypoint = ['/var/lib/twistlock/fargate/defender', 'fargate', 'entrypoint'] # When your Fargate task definition includes custom sidecar containers, # set the environment variable: DO_UPDATE_CUSTOM_SIDECARS=1 do_update_custom_sidecars = env.bool('DO_UPDATE_CUSTOM_SIDECARS', False) # When your Fargate task definition includes custom sidecar containers, # set the environment variable CUSTOM_SIDECARS=sidecarcontainer1,sidecarcontainer2,... # CUSTOM_SIDECARS should be a comma-separated list of the names of the # containers in the Fargate task definition. We make the following assumptions: # * the container name used is also the image name # * the docker image is deployed to the same ECR registry url (same region, same account) # as your main container # * the docker image is tagged with the same tag as your main image custom_sidecars = env.list('CUSTOM_SIDECARS', []) # see constants.py for details verify_mode = env.enum( "VERIFY_MODE", enum=VerifyMode, default=VerifyMode.HEALTH_CHECK._name_) # By default, if the service currently has a desired task count of 0, the deployment will fail. # If FORCE_SCALE_OUT is set to True the service will be scaled up to 1 task for the deployment. # This should be used for deployments to worker services which may be scaled down to zero. force_scale_out = env.bool('FORCE_SCALE_OUT', False) if os.environ.get('TASK_DEFINITION_PARAMS'): task_definition_params = os.environ.get( 'TASK_DEFINITION_PARAMS').split(" ") else: task_definition_params = [ 'taskRoleArn', 'executionRoleArn', 'networkMode', 'containerDefinitions', 'volumes', 'placementConstraints', 'requiresCompatibilities', 'cpu', 'memory', 'ephemeralStorage' ] ephemeral_storage_size_default = 20 ephemeral_storage_size = int(os.environ.get('EPHEMERAL_STORAGE_SIZE', ephemeral_storage_size_default)) def disable_downscaling(alarm): """ Before updating service, disable the downscaling alarm and application autoscaling so newly deployed tasks are not inadvertently scaled down during healthcheck, grace, or provisioning periods """ cloudwatch_client = boto3.client('cloudwatch', region_name=aws_region) autoscaling_client = boto3.client( 'application-autoscaling', region_name=aws_region) response = autoscaling_client.describe_scalable_targets( ServiceNamespace='ecs', ResourceIds=[ f'service/{fargate_service_name}/{fargate_service_name}' ], ScalableDimension='ecs:service:DesiredCount', ) if response['ScalableTargets']: print(f'\nDisabling autoscaling for {fargate_service_name}') response = autoscaling_client.register_scalable_target( ServiceNamespace='ecs', ResourceId=f'service/{fargate_service_name}/{fargate_service_name}', ScalableDimension='ecs:service:DesiredCount', SuspendedState={ 'DynamicScalingInSuspended': True, 'ScheduledScalingSuspended': True, } ) if response['ResponseMetadata']['HTTPStatusCode'] != 200: print(f'\nError disabling autoscaling for {fargate_service_name}. Continuing with deployment') return response else: print(f'\nAutoscaling target not found for {fargate_service_name}. Continuing without modifying scaling settings') alarms = cloudwatch_client.describe_alarms(AlarmNames=[alarm]) if alarms['MetricAlarms']: print(f'\nCurrent downscaling alarm action status is {alarms["MetricAlarms"][0]["ActionsEnabled"]}') print(f'\nDisabling downscaling alarm action for {alarm}') response = cloudwatch_client.disable_alarm_actions(AlarmNames=[alarm]) if response['ResponseMetadata']['HTTPStatusCode'] != 200: print(f'\nError disabling downscaling alarm action for {alarm}. Continuing with deployment') return response else: print('\nAlarm not found. Continuing without modifying alarm.') return None def enable_downscaling(alarm): """ Enables downscaling alarm and autoscaling targets. Since this is only called if the alarm exists, we do not need to check for its existence here as well """ cloudwatch_client = boto3.client('cloudwatch', region_name=aws_region) print(f'\nEnabling downscaling alarm action for {alarm}') alarm_response = cloudwatch_client.enable_alarm_actions(AlarmNames=[alarm]) autoscaling_client = boto3.client( 'application-autoscaling', region_name=aws_region) autoscaling_response = autoscaling_client.register_scalable_target( ServiceNamespace='ecs', ResourceId=f'service/{fargate_service_name}/{fargate_service_name}', ScalableDimension='ecs:service:DesiredCount', SuspendedState={ 'DynamicScalingInSuspended': False, 'ScheduledScalingSuspended': False, } ) for response in alarm_response, autoscaling_response: if response['ResponseMetadata']['HTTPStatusCode'] != 200: print(f'\nError enabling downscaling action for {fargate_service_name}') return alarm_response, autoscaling_response def create_new_task_definition(client, task_family_name): """Get latest task definition.""" try: latest_task = client.describe_task_definition( taskDefinition=task_family_name) # Just get the task definition element running_task_definition = latest_task['taskDefinition'] # Check that the value is defined, and if so, add to task dict new_task_definition = dict() for param in task_definition_params: if param in running_task_definition: new_task_definition[param] = running_task_definition[param] # If volumes were not defined, set them to an empty list if 'volumes' not in new_task_definition: new_task_definition['volumes'] = [] container_definitions = new_task_definition['containerDefinitions'] is_defender_enabled = False for container in container_definitions: if container['name'] == defender_container_name: is_defender_enabled = True break for container in container_definitions: """ There may be multiple containers per task. Find the one in question. """ if container['name'] == container_name: """ We need to update container entrypoint only when PrismaCloud Defender is present as a sidecar container """ if is_defender_enabled: docker_image_info = DockerImageInfo(image_name, image_tag, ecr_account_id, ecr_region) entrypoint, cmd = docker_image_info.get_entrypoint_and_cmd() print(f'\nDocker image entrypoint is {entrypoint}') print(f'\nDocker image cmd is {cmd}') container['entryPoint'] = defender_entrypoint + entrypoint + cmd current_image = container['image'] print(f'\nCurrent image is {current_image}') print(f'\nNew image is {image_uri}') container['image'] = image_uri # Return the old and updated task definition return running_task_definition, new_task_definition else: print(f'\n{container_name} not found in current container definition') if container_definitions.index(container) == len(container_definitions)-1: """ It this is the last container and the name is still not found, then this is probably the wrong task definition. """ raise SystemExit(f'{container_name} not found in task definition') except ClientError as error: print(error.response['Error']['Message']) raise SystemExit('Error getting task definition or task parameters') def _update_sidecar_containers(task_definition: dict): """Configure sidecar containers in the task definition.""" if not custom_sidecars: print('\nNo custom sidecars, noop.') return task_definition sidecars = {} for sidecar_name in custom_sidecars: sidecars[sidecar_name] = f'{ecr_registry_url}/{sidecar_name}:{image_tag}' print(f'\nCustom sidecars {sidecars}') container_definitions = task_definition['containerDefinitions'] for container in container_definitions: name = container['name'] sidecar_image_uri = sidecars.get(name, None) if sidecar_image_uri: current_image = container['image'] print(f'\nFound sidecar container {name}') print(f'\nCurrent image is {current_image}') print(f'\nNew image is {sidecar_image_uri}') container['image'] = sidecar_image_uri return task_definition def register_new_task_definition(client, new_task_definition): """Register a new task definition.""" try: service_definition = get_service_definition(client) task_definition_tags = service_definition['services'][0]['tags'] for t in task_definition_tags: if t['key'] == 'terraformed': t['value'] = 'false' task_definition_tags.extend([ { 'key': 'deployed_by', 'value': 'fargate_tools', }, { 'key': 'deployment_job', 'value': deployment_job }, ]) params = { 'family': task_family, 'taskRoleArn': new_task_definition['taskRoleArn'], 'executionRoleArn': new_task_definition['executionRoleArn'], 'networkMode': new_task_definition['networkMode'], 'containerDefinitions': new_task_definition['containerDefinitions'], 'requiresCompatibilities': new_task_definition['requiresCompatibilities'], 'cpu': new_task_definition['cpu'], 'memory': new_task_definition['memory'], 'volumes': new_task_definition['volumes'], 'tags': task_definition_tags } if 'ephemeralStorage' in new_task_definition: params['ephemeralStorage'] = new_task_definition['ephemeralStorage'] if ephemeral_storage_size > ephemeral_storage_size_default: params['ephemeralStorage'] = {'sizeInGiB': ephemeral_storage_size} response = client.register_task_definition(**params) print(f'\nNew task definition ARN is: {response["taskDefinition"]["taskDefinitionArn"]}') return response except ClientError as error: print(error.response['Error']['Message']) raise SystemExit('Error registering task definition') def update_ecs_service(client, **kwargs): """Update service using newly created task definition.""" try: response = client.update_service(**kwargs) return response except ClientError as error: print(f'Error updating service {service_name}: {error.response["Error"]["Message"]}') raise SystemExit(f'Error updating service {service_name}') def get_service_definition(client): return client.describe_services( cluster=cluster_name, services=[ fargate_service_name, ], include=['TAGS'], ) def main(): """Main entrypoint function.""" disable_alarm_response = '' try: update_service_params = {'cluster': cluster_name, 'service': fargate_service_name, 'forceNewDeployment': True, } client = boto3.client('ecs', region_name=aws_region) # Get platform version to be used during service update service_definition = get_service_definition(client) if service_definition['services']: if 'platformVersion' in service_definition['services'][0]: platform_version = service_definition['services'][0][ 'platformVersion'] update_service_params['platformVersion'] = platform_version # Workers do not have load balancer grace periods. if 'healthCheckGracePeriodSeconds' in service_definition['services'][0]: grace_period = \ service_definition['services'][0]['healthCheckGracePeriodSeconds'] else: # If not defined, set to default of 60 seconds grace_period = 60 if service_definition['services'][0]['desiredCount'] == 0: if force_scale_out: print( f'\nService {fargate_service_name} has a desired task ' f'count of 0. Scaling out to 1 task as FORCE_SCALE_OUT ' f'is set to True') update_service_params['desiredCount'] = 1 else: raise SystemExit( f'Cannot deploy to service {fargate_service_name} as the ' f'desired task count is 0. Either scale out the service ' f'or set FORCE_SCALE_OUT to True to automatically scale ' f'out the service as part of the deployment.' ) else: raise SystemExit(f'Service {fargate_service_name} not found.') if deploy_mode == 'REVERT': # If reverting, get the current task definition directly latest_task = client.describe_task_definition( taskDefinition=task_family) current_task_def = latest_task['taskDefinition'] print(f'\nCurrent task definition ARN is {current_task_def["taskDefinitionArn"]}') task_def_arn_params = current_task_def['taskDefinitionArn'].split(':') task_def_arn_params[-1] = revert_task_definition_number new_task_definition_arn = ':'.join( str(param) for param in task_def_arn_params) print(f'\nReverting to task definition ARN {new_task_definition_arn}') update_service_params['taskDefinition'] = new_task_definition_arn new_service = update_ecs_service(client=client, **update_service_params) elif deploy_mode == 'REDEPLOY': new_task_definition_arn = \ service_definition['services'][0]['taskDefinition'] print(f'\nRedeploying task definition ARN {new_task_definition_arn}') update_service_params['taskDefinition'] = new_task_definition_arn new_service = update_ecs_service(client=client, **update_service_params) else: current_task_def, new_task_def = create_new_task_definition( client, task_family) if do_update_custom_sidecars: new_task_def = _update_sidecar_containers(new_task_def) response = register_new_task_definition(client, new_task_def) new_task_definition_arn = response['taskDefinition']['taskDefinitionArn'] # Disable downscaling alarm actions disable_alarm_response = disable_downscaling( alarm=cloudwatch_downscaling_alarm_name) update_service_params['taskDefinition'] = new_task_definition_arn new_service = update_ecs_service(client=client, **update_service_params) if new_service['service']['status']: print(f'\nService status: {new_service["service"]["status"]}') poll_result = poll_task_status( verify_mode=verify_mode, update_timeout=update_timeout, client=client, cluster_name=cluster_name, service_name=fargate_service_name, task_definition_arn=new_task_definition_arn, container_name=container_name, grace_period=grace_period, ) if not poll_result: if deploy_mode == 'REDEPLOY': raise SystemExit(f'Redeployment failed. Check {service_name}') print('Service deployment failed. Initiating rollback...') rollback_task_definition = current_task_def['taskDefinitionArn'] update_service_params[ 'taskDefinition'] = rollback_task_definition rolled_back_service = update_ecs_service(client=client, **update_service_params) if rolled_back_service['service']['status']: print(f'\nService rollback status: {rolled_back_service["service"]["status"]}') rollback_poll_result = poll_task_status( verify_mode=verify_mode, update_timeout=update_timeout, client=client, cluster_name=cluster_name, service_name=fargate_service_name, task_definition_arn=rollback_task_definition, container_name=container_name, grace_period=grace_period, ) if rollback_poll_result is False: raise SystemExit(f'Rollback failed. Check {service_name}') raise SystemExit( f'Rollback succeeded. Service is running task {rollback_task_definition}') else: print('\nStatus not found - dumping raw service output:') for k, v in new_service.items(): print(f'\n{k}: {v}') except Exception as ex: print(f'An error occurred during deployment: {ex}') raise SystemExit(f'Error updating {service_name}') finally: # Re-enable downscaling alarm irrespective of deployment success or failure if disable_alarm_response: enable_downscaling(alarm=cloudwatch_downscaling_alarm_name) if __name__ == "__main__": main()