Upload file to S3 -> Launch EC2 instance

Ambar — Thu, 12 Dec 2019 15:38:40 +0000

Problem Statement

Scenario

A user wants to run a data-processing (e.g. ML) workload on an EC2 instance. The data to be processed in this workload is a CSV file stored in an S3 bucket.

Currently the user has to manually spin up an EC2 instance (with a user-data script that installs the tools and starts the data processing), after uploading the data (a CSV file) to their S3 bucket.

Wouldn't it be great if this could be automated? So that creation of a file in the S3 bucket (from any source) would automatically spin up an appropriate EC2 instance to process the file? And creation of an output file in the same bucket would likewise automatically terminate all relevant (e.g. tagged) EC2 instances?

Requirements

Use Case 1

When a CSV is uploaded to the inputs "directory" in a given S3 bucket, an EC2 "data-processing" instance (i.e. tagged with {"Purpose": "data-processing"}) should be launched, but only if a "data-processing" tagged instance isn't already running. One instance at a time is sufficient for processing workloads.

Use Case 2

When a new file is created in the outputs "directory" in the same S3 bucket, it means the workload has finished processing, and all "data-processing"-tagged instances should now be identified and terminated.

Solution Using S3 Triggers and Lambda

As explained at Configuring Amazon S3 Event Notifications
, S3 can send notifications upon:

object creation
object removal
object restore
replication events
RRS failures

S3 can publish these events to 3 possible destinations:

SNS
SQS
Lambda

We want S3 to push object creation notifications directly to a Lambda function, which will have the necessary logic to process these events and determine further actions.

Here's how to do this using Python3 and boto3 in a simple Lambda function.

AWS Console: Setup the S3 bucket

Go to the AWS S3 console. Go to the bucket you want to use (or create a new one with default settings) for triggering your Lambda function.

Create three "folders" in this bucket ('folders/directories' in S3 are actually 'key-prefixes' or paths) like so:

What are these dirs for?

config : to hold various settings for the Lambda function
inputs : a CSV file uploaded here will trigger the ML workload on an EC2 instance
outputs : any file here indicates completion of the ML workload and should cause any running data-processing (i.e. specially-tagged) EC2 instances to terminate

The `config` folder

We need to supply the EC2 launch information to the Lambda function. By "launch information" we mean, the instance-type, the AMI-Id, the security groups, tags, ssh keypair, user-data, etc.

One way to do this is to hard-code all of this into the Lambda code itself, but this is never a good idea. It's always better to externalize the config, and one way to do this by storing it in the S3 bucket itself like so:

Save this as ec2-launch-config.json in the config folder:

{
    "ami": "ami-0123b531fc646552f",
    "region": "ap-south-1",
    "instance_type": "t2.nano",
    "ssh_key_name": "ssh-connect",
    "security_group_ids": [
        "sg-08b6b31110601e924"
    ],
    "filter_tag_key": "Purpose",
    "filter_tag_value": "data-processing",
    "set_new_instance_tags": [
        {
            "Key": "Purpose",
            "Value": "data-processing"
        },
        {
            "Key": "Name",
            "Value": "ML-runner"
        }
    ]
}

The params are quite self-explanatory, and you can tweak them as you need to.

"filter_tag_key": "Purpose" : "filter_tag_value": "data-processing" --> this is the tag that will be used to identify (i.e. filter) already-running data-processing EC2 instances.

You'll notice that user-data isn't part of the above JSON config. It's read in from a separate file called user-data, just so that it's easier to write and maintain:

#!/bin/bash
apt-get update -y
apt-get install -y apache2
systemctl start apache2
systemctl enable apache2.service
echo "Congrats, your setup is a success!" > /var/www/html/index.html

The above user-data script will install the Apache2 webserver, and writes a congratulatory message that will be served on the instance's public IP address.

Lambda function logic

The Lambda function needs to:

receive an incoming S3 object creation notification JSON object
parse out the S3 bucket name and S3 object key name from the JSON
pull in a JSON-based EC2 launch configuration previously stored in S3
if the S3 object key (i.e. "directory") matches ^inputs/, check if we need to launch a new EC2 instance and if so, launch one
if the S3 object key (i.e. "directory") matches ^outputs/, terminate any running tagged instances

AWS Console: New Lambda function setup

Go to the AWS Lambda console and click the Create function button.

Select Author from scratch, enter a function name, and select Python 3.8 as the Runtime.

Select Create a new role with basic Lambda permissions in the Choose or create an execution role dropdown.

Click the Create function button.

This will take you to the Configuration tab.

Click the Add Trigger button:

Setup an S3 trigger from the bucket in question, like so:

Select your bucket
Select the events you're interested in (All object create events in this case)
Leave Prefix and Suffix empty, as we will take care of prefixes (inputs and outputs bucket paths) in our function
Select Enable trigger

As it says near the bottom of the screenshot:

Lambda will add the necessary permissions for Amazon S3 to invoke your Lambda function from this trigger.
So we don't need to go into S3 to configure notifications separately.

Click the Add button to save this trigger configuration.

Back on the main Lambda designer tab, you'll notice that S3 is now linked up to the our newly-created Lambda function:

Click the name of the Lambda function to open up the Function code editor underneath the Designer pane.

Lambda function code

Copy the following Lambda function code into the Function code editor, then click the Save button on the top-right.

import boto3
import json
import base64
from urllib.parse import unquote_plus


BUCKET_NAME = "YOUR_S3_BUCKET_NAME"
CONFIG_FILE_KEY = "config/ec2-launch-config.json"
USER_DATA_FILE_KEY = "config/user-data"
BUCKET_INPUT_DIR = "inputs"
BUCKET_OUTPUT_DIR = "outputs"


def launch_instance(EC2, config, user_data):
    tag_specs = [{}]
    tag_specs[0]['ResourceType'] = 'instance'
    tag_specs[0]['Tags'] = config['set_new_instance_tags']

    ec2_response = EC2.run_instances(
        ImageId=config['ami'],  # ami-0123b531fc646552f
        InstanceType=config['instance_type'],   # t2.nano
        KeyName=config['ssh_key_name'],  # ambar-default
        MinCount=1,
        MaxCount=1,
        SecurityGroupIds=config['security_group_ids'],  # sg-08b6b31110601e924
        TagSpecifications=tag_specs,
        # UserData=base64.b64encode(user_data).decode("ascii")
        UserData=user_data
    )

    new_instance_resp = ec2_response['Instances'][0]
    instance_id = new_instance_resp['InstanceId']
    # print(f"[DEBUG] Full ec2 instance response data for '{instance_id}': {new_instance_resp}")

    return (instance_id, new_instance_resp)



def lambda_handler(raw_event, context):
    print(f"Received raw event: {raw_event}")
    # event = raw_event['Records']

    for record in raw_event['Records']:
        bucket = record['s3']['bucket']['name']
        print(f"Triggering S3 Bucket: {bucket}")
        key = unquote_plus(record['s3']['object']['key'])
        print(f"Triggering key in S3: {key}")

        # get config from config file stored in S3
        S3 = boto3.client('s3')
        result = S3.get_object(Bucket=BUCKET_NAME, Key=CONFIG_FILE_KEY)
        ec2_config = json.loads(result["Body"].read().decode())
        print(f"Config from S3: {ec2_config}")

        ec2_filters = [
            {
                'Name': f"tag:{ec2_config['filter_tag_key']}",
                'Values':[ ec2_config['filter_tag_value'] ]
            }
        ]

        EC2 = boto3.client('ec2', region_name=ec2_config['region'])

        # launch new EC2 instance if necessary
        if bucket == BUCKET_NAME and key.startswith(f"{BUCKET_INPUT_DIR}/"):
            print("[INFO] Describing EC2 instances with target tags...")
            resp = EC2.describe_instances(Filters=ec2_filters)
            # print(f"[DEBUG] describe_instances response: {resp}")

            if resp["Reservations"] is not []:    # at least one instance with target tags was found
                for reservation in resp["Reservations"] :
                    for instance in reservation["Instances"]:
                        print(f"[INFO] Found '{instance['State']['Name']}' instance '{ instance['InstanceId'] }'"
                            f" having target tags: {instance['Tags']} ")

                        if instance['State']['Code'] == 16: # instance has target tags AND also is in running state
                            print(f"[INFO] instance '{ instance['InstanceId'] }' is already running: so not launching any more instances")
                            return {
                                "newInstanceLaunched": False,
                                "old-instanceId": instance['InstanceId'],
                                "new-instanceId": ""
                            }

            print("[INFO] Could not find even a single running instance matching the desired tag, launching a new one")

            # retrieve EC2 user-data for launch
            result = S3.get_object(Bucket=BUCKET_NAME, Key=USER_DATA_FILE_KEY)
            user_data = result["Body"].read()
            print(f"UserData from S3: {user_data}")

            result = launch_instance(EC2, ec2_config, user_data)
            print(f"[INFO] LAUNCHED EC2 instance-id '{result[0]}'")
            # print(f"[DEBUG] EC2 launch_resp:\n {result[1]}")
            return {
                "newInstanceLaunched": True,
                "old-instanceId": "",
                "new-instanceId": result[0]
            }

        # terminate all tagged EC2 instances
        if bucket == BUCKET_NAME and key.startswith(f"{BUCKET_OUTPUT_DIR}/"):
            print("[INFO] Describing EC2 instances with target tags...")
            resp = EC2.describe_instances(Filters=ec2_filters)
            # print(f"[DEBUG] describe_instances response: {resp}")
            terminated_instance_ids = []

            if resp["Reservations"] is not []:    # at least one instance with target tags was found
                for reservation in resp["Reservations"] :
                    for instance in reservation["Instances"]:
                        print(f"[INFO] Found '{instance['State']['Name']}' instance '{ instance['InstanceId'] }'"
                            f" having target tags: {instance['Tags']} ")

                        if instance['State']['Code'] == 16: # instance has target tags AND also is in running state
                            print(f"[INFO] instance '{ instance['InstanceId'] }' is running: terminating it")
                            terminated_instance_ids.append(instance['InstanceId'])
                            boto3.resource('ec2').Instance(instance['InstanceId']).terminate()

            return {
                "terminated-instance-ids:": terminated_instance_ids
            }

Lambda Execution IAM Role

Our lambda function won't work just yet. It needs S3 access to read in the config and it needs permissions to describe, launch, and terminate EC2 instances.

Scroll down on the Lambda configuration page to the Execution role and click the View role link:

Click on the AWSLambdaBasic... link to edit the lambda function's policy:

Click {} JSON, Edit Policy and then JSON.

Now add the following JSON to the existing Statement Section

        {
            "Effect": "Allow",
            "Action": [
                "s3:*"
            ],
            "Resource": "*"
        },
        {
            "Action": [
                "ec2:RunInstances",
                "ec2:CreateTags",
                "ec2:ReportInstanceStatus",
                "ec2:DescribeInstanceStatus",
                "ec2:DescribeInstances",
                "ec2:TerminateInstances"
            ],
            "Effect": "Allow",
            "Resource": "*"
        }

Click Review Policy and Save changes.

Go!

Our setup's finally complete. We're ready to test it out!

Test Case 1: new instance launch

Upload a file from the S3 console into the inputs folder. If you used the exact same config as above, this would have triggered the Lambda function, which in turn would have launched a new t2.nano instance with the Purpose: data-processing tag on it.

If you put the instance's public IP address into a browser (after giving it a minute or so to boot and warm up), you should also see the test message served to you: which indicates that the user-data did indeed execute successfully upon boot.

Test Case 2: another instance should not launch

As long as there is at least one Purpose: data-processing tagged instance running, another one should not spawn. Let's upload another file to the inputs folder. And indeed the actual behavior matches the expectation.

If we kill the already-running instance, and then upload another file to the inputs folder, it will launch a new instance.

Test Case 3: instance termination condition

Upload a file into the outputs folder. This will trigger the Lambda function into terminating any already-running instances that are tagged with Purpose: data-processing.

Bonus: S3 event object to test your lambda function

S3-object-creation-notification (to test Lambda function)

{
  "Records": [
    {
      "eventVersion": "2.1",
      "eventSource": "aws:s3",
      "awsRegion": "ap-south-1",
      "eventTime": "2019-09-03T19:37:27.192Z",
      "eventName": "ObjectCreated:Put",
      "userIdentity": {
        "principalId": "AWS:AIDAINPONIXQXHT3IKHL2"
      },
      "requestParameters": {
        "sourceIPAddress": "205.255.255.255"
      },
      "responseElements": {
        "x-amz-request-id": "D82B88E5F771F645",
        "x-amz-id-2": "vlR7PnpV2Ce81l0PRw6jlUpck7Jo5ZsQjryTjKlc5aLWGVHPZLj5NeC6qMa0emYBDXOo6QBU0Wo="
      },
      "s3": {
        "s3SchemaVersion": "1.0",
        "configurationId": "828aa6fc-f7b5-4305-8584-487c791949c1",
        "bucket": {
          "name": "YOUR_S3_BUCKET_NAME",
          "ownerIdentity": {
            "principalId": "A3I5XTEXAMAI3E"
          },
          "arn": "arn:aws:s3:::lambda-artifacts-deafc19498e3f2df"
        },
        "object": {
          "key": "b21b84d653bb07b05b1e6b33684dc11b",
          "size": 1305107,
          "eTag": "b21b84d653bb07b05b1e6b33684dc11b",
          "sequencer": "0C0F6F405D6ED209E1"
        }
      }
    }
  ]
}

That's all, folks!

DEV Community: Ambar