Exec into Containers on Amazon ECS with aws-cli

ecs-exec

#!/usr/bin/env python3

import sys
import subprocess
import json

def dict_get(dict, fn, other=None):
    try:
        return fn(dict)
    except:
        return other

def print_exec(cmd):
    print("$ " + cmd)
    subprocess.run(cmd, shell=True)

def create_aws_exec(profile):
    def exec(args):
        return subprocess.check_output(["aws", "--profile", profile, "--output", "json"] + args)

    return exec

def get_selected_task_id_from_cluster(aws_exec, cluster_name, service_name):
    aws_res = aws_exec(["ecs", "list-tasks", "--cluster", cluster_name, "--service-name", service_name])
    return dict_get(json.loads(aws_res), lambda x: x["taskArns"][0])

def get_task_info_from_id(aws_exec, cluster_name, task_id):
    res = aws_exec(["ecs", "describe-tasks", "--cluster", cluster_name, "--tasks", task_id])
    res = json.loads(res)
    return (dict_get(res, lambda x: x["tasks"][0]["containerInstanceArn"]), dict_get(res, lambda x: x["tasks"][0]["taskDefinitionArn"]))

def get_ec2_instance_id_from_container_instance_id(aws_exec, cluster_name, container_instance_id):
    res = aws_exec(["ecs", "describe-container-instances", "--cluster", cluster_name, "--container-instances", container_instance_id])
    return dict_get(json.loads(res), lambda x: x["containerInstances"][0]["ec2InstanceId"])

def get_container_image_name_from_task_def_id(aws_exec, task_def_id):
    res = aws_exec(["ecs", "describe-task-definition", "--task-definition", task_def_id])
    return dict_get(json.loads(res), lambda x: x["taskDefinition"]["containerDefinitions"][0]["image"])

def get_ec2_private_ip_from_id(aws_exec, instance_id):
    res = aws_exec(["ec2", "describe-instances", "--instance-ids", instance_id])
    return dict_get(json.loads(res), lambda x: x["Reservations"][0]["Instances"][0]["PrivateIpAddress"])

def get_container_name_from_image(host, image_name):
    return subprocess.check_output(f'DOCKER_HOST={host} docker ps --filter="ancestor={image_name}" --format="{{{{.Names}}}}"', shell=True).decode("utf-8").strip().split("\n")[0]

def get_container_and_host(profile, cluster_name, service_name):
    aws_exec = create_aws_exec(profile)

    selected_task_id = get_selected_task_id_from_cluster(aws_exec, cluster_name, service_name)
    print("Selected Task: " + selected_task_id)
    (container_instance_id, task_definition_id) = get_task_info_from_id(aws_exec, cluster_name, selected_task_id)
    ec2_instance_id = get_ec2_instance_id_from_container_instance_id(aws_exec, cluster_name, container_instance_id)
    container_image_name = get_container_image_name_from_task_def_id(aws_exec, task_definition_id)
    ec2_private_ip = get_ec2_private_ip_from_id(aws_exec, ec2_instance_id)
    container_name = get_container_name_from_image(ec2_private_ip, container_image_name)
    return (container_name, ec2_private_ip)

def handle_exec(argv):
    if len(argv) < 4:
        print("usage: %s exec <profile> <cluster-name> <service-name> ?<user>" % argv[0])
        sys.exit(1)

    (profile, cluster_name, service_name) = argv[1:4]
    user = "www-data" if len(argv) < 5 else argv[4]

    (container_name, ec2_private_ip) = get_container_and_host(profile, cluster_name, service_name)
    print_exec(f'DOCKER_HOST={ec2_private_ip} docker exec -u {user} -it {container_name} bash')

def handle_cp(argv):
    if len(argv) < 6:
        print("usage: %s exec <profile> <cluster-name> <service-name> <src-path> <dst-path>" % argv[0])
        sys.exit(1)

    (profile, cluster_name, service_name, src_path, dst_path) = argv[1:]

    (container_name, ec2_private_ip) = get_container_and_host(profile, cluster_name, service_name)

    if src_path[0] == ":":
        src_path = container_name + src_path
    elif dst_path[0] == ":":
        dst_path = container_name + dst_path
    else:
        print("One of src or dst path must start with ':' in order to designate docker container");
        sys.exit(1)

    print_exec(f'DOCKER_HOST={ec2_private_ip} docker cp {src_path} {dst_path}')

def handle_help(argv):
    print("""
ecs-exec 0.2.0

commands:
  cp    copy files to/from ecs containers
  exec  exec into an ecs container
  help  display this help text
""".strip())

def resolve_command_from_args(argv):
    cmd = argv[1]
    if cmd == "cp":
        return (handle_cp, argv[2:])
    elif cmd == "exec":
        return (handle_exec, argv[2:])
    elif cmd == "help":
        return (handle_help, argv[2:])
    else:
        return (handle_exec, argv[1:])

def main(argv):
    if len(argv) < 2:
        print("usage: %s <command>? ..." % argv[0])
        sys.exit(1)

    (handle_cmd, args) = resolve_command_from_args(argv)
    handle_cmd([argv[0]] + args)


main(sys.argv)

Setup

1. Make sure you have python >=3.7 and aws-cli installed.

2. Configure your ~/.aws/config file with a new profile:

   [profile {profile_name}]
   region = {region}
   output = json
   

3. Add your credentials for that aws profile in ~/.aws/credentials

   [{profile_name}]
   aws_access_key_id = {access_key}
   aws_secret_access_key = {secret_key}
   

4. Copy the gist into your clipboard, and then run the following command:

   cat > /usr/local/bin/ecs-exec; chmod +x /usr/local/bin/ecs-exec;
   

   It should hang at which point you paste in the contents, press enter and ctrl+d.

Usage

# exec into container
ecs-exec exec {profile_name} {ecs_cluster} {ecs_service} ?{user}

# copy file to/from a container
ecs-exec cp {profile_name} {ecs_cluster} {ecs_service} {from_path} {to_path}

Upgrading

Latest Version: 0.2.0

Copy the contents of the gist and save to your local copy of ecs-exec.

Run ecs-exec help to verify the current version.