DEV Community: Seth M

Real-time S3 File zipping With Lambda and WebSockets in NodeJS

Seth M — Fri, 05 Mar 2021 21:23:24 +0000

It's always amazing to me that AWS S3 hasn't offered a bulk download service from their console. For the systems I'm working on, downloading a set of files is a common activity.

Our original design for bulk downloading was setting up an EC2 machine with the aws cli installed, rabbitmq installed, and a small nodejs service. The node service would consume the messages off rabbitmq, use the aws cli to retrieve files from s3, zip them, and return them to an s3 location. This solution worked ok, but what if there was a way to do it more reliably and more cost effectively?

AWS Labmda and API Gateway to the rescue

We settled on the following strategy:

AWS Lambda would be responsible for zipping
AWS API Gateway using WebSockets would be used for triggering and managing the zipping process
Upon completing creation of the .zip file, get a pre-signed url of the new .zip file and pass it back to the websocket client.

There are some terrific articles and examples that I'll list below, but we ended up using this example from s3-zip as our baseline.

To wrap the zip lambda function in a websocket we used the aws simple-websockets-chat-app as a blueprint.

Finally, to make deployment easier, we created the SAM yaml files for pushing our code to AWS Cloudformation.

S3 Zip Lambda Function

The payload to the Lamda function will be a set of files to zip and the destination instructions for the resulting .zip file

Source code

const payload = {
  region: 'us-east-1',
  bucket: 'demobucket', <-- all assets in same bucket
  folder: 'audio/', <-- must have trailing slash
  files: [
    'file1.mp3',  <-- files in above folder
    'file2.mp3'
  ],
  zipBucket: 'demobucket', <-- bucket for resulting .zip file
  zipFolder: 'temp/', <-- must have trailing slash
  zipFileName: 'demo1.zip',
  signedUrlExpireSeconds: 60 * 60 * 10 <-- expiration time of signedUrl of s3object
}

API Gateway WebSocket

I've setup a simple API Gateway using CloudFormation

  DemoDevS3ZipWebSocket:
    Type: AWS::ApiGatewayV2::Api
    Properties:
      Name: DemoDevS3ZipWebSocket
      ProtocolType: WEBSOCKET
      RouteSelectionExpression: "$request.body.action"

  Deployment:
    Type: AWS::ApiGatewayV2::Deployment
    DependsOn:
    - ZipRoute
    Properties:
      ApiId: !Ref DemoDevS3ZipWebSocket

  Stage:
    Type: AWS::ApiGatewayV2::Stage
    Properties:
      StageName: demo
      Description: Demo Deployment
      DeploymentId: !Ref Deployment
      ApiId: !Ref DemoDevS3ZipWebSocket

The important thing to note here is the RouteSelectionExpression value. The "action" parameter will trigger our "onzip" route.

To actually facilitate the zipping, we are using s3-zip .

When the 'onzip' route is invoked, the params are parsed from the event.body.params :

const params = JSON.parse(event.body).params

Do a quick validation to make sure that there are actual files:

  if (!(files.length > 0)) {
    return {
      statusCode: 500,
      body: JSON.stringify({
        statusCode: 'error',
        message: 'No files to zip'
      })
    }
  }

In order to communicate back through the WebSocket, we needed to use the ApiGatewayManagementApi. The constructor requires the deployed WebSocket address as the endpoint:

  const stage = event.requestContext.stage
  const domainName = event.requestContext.domainName


  // Allows the Lambda function to communicate with the websocket client
  const api = new AWS.ApiGatewayManagementApi({
    endpoint: 'https://' + domainName + '/' + stage
  })

  // event.requestContext.connectionId contains the Websocket id
  const apiParams = {
    ConnectionId: event.requestContext.connectionId,
    Data: null
  }

Once the ApiGatewayManagementApi object is available, we can communicate back through the WebSocket by calling:

apiParams.Data = 'some msg -or- JSON.stringify(obj)'
await api.postToConnection(apiParams).promise()

This call will trigger a WebSocket 'onmessage' event.

Next, the main try/catch block will invoke s3zip and AWS.S3.upload to stream the content into a zip file using the s3zip archive method. Using the archive method allows the process to bypass the need for downloading or storing the target files in any kind of temporary location.

  try {
    const body = s3Zip.archive({ region: region, bucket: bucket }, folder, files)
    const zipKey = zipFolder + zipFileName
    const zipParams = { params: { Bucket: zipBucket, Key: zipKey } }
    const zipFile = new AWS.S3(zipParams)

    const promise = new Promise((resolve, reject) => {
      zipFile.upload({ Body: body })
        .on('httpUploadProgress',
          async function (evt) {
            evt.statusCode = 'progress'
            evt.pctComplete = (100 * evt.loaded / totalBytes)
            apiParams.Data = JSON.stringify(evt)
            // communicate with the Websockets, returning the pctComplete
            await api.postToConnection(apiParams).promise()
          })
        .send(async function (e, r) {
          if (e) {
            e.statusCode = 'error'
            reject(e)
          } else {
            r.statusCode = 'success'
            r.Files = files
            r.Folder = folder
            r.SignedUrl = s3.getSignedUrl('getObject', {
              Bucket: zipBucket,
              Key: r.Key,
              Expires: signedUrlExpireSeconds
            })
            resolve(r)
          }
        })
    })

    // wait for s3zip process to complete
    const res = await promise

    // message push back through websocket with zip results
    apiParams.Data = JSON.stringify(res)
    await api.postToConnection(apiParams).promise()

    return {
      statusCode: 200,
      body: JSON.stringify(res)
    }
  } catch (e) {
    // send error messages back to websocket client
    apiParams.Data = JSON.stringify(e)
    await api.postToConnection(apiParams).promise()
    return {
      statusCode: 500,
      body: JSON.stringify(e)
    }
  }

As the upload to s3 happens, the 'httpUploadProgress' event from the s3 upload method will emit and return a value for event.loaded. We push that event object back through the WebSocket using await api.postToConnection(apiParams).promise(). What's nice is that this allows our web frontend to give the user a real-time progress display on the zipping process.

So far, we haven't seen any issues. We've tested out this process on 5GB worth of target files and haven't seen any issue.

When the zip file is complete, the send() method is invoked. In the send() method we do a quick call to getSignedUrl():

r.SignedUrl = s3.getSignedUrl('getObject', {
  Bucket: zipBucket,
  Key: r.Key,
  Expires: signedUrlExpireSeconds
})

On a successful zip, the signedUrl is generated, and the lambda function will send a message through the websocket. On the calling side, a simple parse can extract the signedUrl and trigger the browser to begin the physical download, as if the user clicked on the .zip file.

The javascript on the caller .html page looks something like this:

socket.onmessage = function (event) {
  if (event.data) {
    var data = JSON.parse(event.data)

    // zip progress
    if (data.statusCode === 'progress') {
      var pct = Math.round(data.pctComplete)
      progressElement.text('Processing ... ' + pct + '%')
    }

    // zip completed
    if (data.statusCode === 'success') {
      if (Object.prototype.hasOwnProperty.call(data, 'SignedUrl')) {
        window.location.assign(data.SignedUrl)
      } else {
        console.error('api did not returned SignedUrl')
      }
      socket.close()
    }
}

So that's it. This satisfied our requirement for having a Lambda based zip function with a progress meter using NodeJS

Have a look at the README in the source. It will give instructions for loading both the AWS Api Gateway Websocket and Lambda function into AWS from the command line using SAM and CloudFormation.

There is also a test script for connecting to the websocket and confirming that the SignedUrl for the zipped assets is being returned correctly.

Plus there is a simple web page with the javascript to show you how to connect to the websocket and invoke the 'onzip' route.

Thank you for reading!

The source code is available from github.

https://github.com/openstepmedia/aws-s3zip-lambda-demo

Using AWS Step Functions with nodejs

Seth M — Wed, 25 Nov 2020 00:08:50 +0000

A recent project required some scalability and specifically needed the ability to put long-running jobs in a queue. The jobs were doing some heavy lifting around transcoding, but really, we needed to come up with a plan for managing this job queue. We've had some experience with tools like rabbitmq and kue/redis, but ultimately, because we're a small team, we wanted to see if we could avoid having to manage all these moving pieces.

Since we use AWS for some of our infrastructure, we chose to use AWS Step Functions as our queue management tool. The core of our transcoding engine had to run in its own process on a windows machine on EC2, so we decided to wrap up the process inside of an API engine that would then hook into the AWS Step Function.

There are plenty of resources on describing what AWS Step Functions are, so I won't go into depth on that. In short, AWS offers the ability to put messages into a queue where they will sit until they are removed from the queue by a consuming application.

This tutorial will focus on demonstrating how to integrate AWS Step Functions as a way to queue large running processes wrapped up in a web API. For our API, we're very comfortable with using AdonisJS, a fantastic nodejs web framework that has been heavily influenced by the Laravel world.

I'll dive right in and assume that you were either able to download the source code to this tutorial or were able to setup a basic AdonisJS project.

Create an AWS Activity

Our example is just a demonstration that will require a single activity step. The goal here is to demonstrate sending data from a web form and adding it to the AWS State Machine that will then act as a queue. A consuming application will then grab the information sitting in the queue and then output what it finds to our console.

Activities are the small tasks that are the "workers" in the state machine.

Let's call our Activity:

DemoWorkerActivity

Create the State Machine

With the ARN of the Activity, we can create a State Machine document like the one below. There is only one step, so we'll call that step "DemoWorkerRun". Inside of the step will be a single activity "DemoWorkerActivity".



{
    "Comment": "Demo State Machine - to show how my application interacts with the queue",
    "StartAt": "DemoWorkerRun",
    "States": {
        "DemoWorkerRun": {
            "Type": "Task",
            "Resource": "arn:aws:states:us-east-1:XXXXXXXXXX:activity:DemoWorkerActivity",
            "End": true
        }
    }
}

Once the state machine is created, AWS will give you a resource ARN for that:

In this example the state machine identifier is:

"arn:aws:states:us-east-1:XXXXXXXXXX:stateMachine:DemoWorkerStateMachine"

Core nodejs components:

QueueService - this service will be responsible for kicking off a state machine "execution" with an initial payload.
DemoWorker - the worker will act as the consumer of the activity the lives within the state machine. It is the point of contact between the payload and your underlying application.

QueueService

In our adonis project, we'll create a Services folder and create the app/Services/QueueService.js class. This class will start a state machine execution with the an initial custom payload. This class requires the aws-sdk to be installed. You can install it using the adonis tools:

adonis install aws-sdk



'use strict'

const AWS = require('aws-sdk')

const Env = use('Env')

class QueueService {

  /**


Returns result of StepFunctions.startExecution

@see https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/StepFunctions.html#startExecution-property

*/

async enqueue (name, payload) {

const awsConfig = new AWS.Config({

  accessKeyId: Env.get('AWS_ACCESS_KEY'),

  secretAccessKey: Env.get('AWS_SECRET_KEY'),

  region: Env.get('AWS_REGION')

})

// Initialize the AWS SDK 

const StepFunctions = new AWS.StepFunctions(awsConfig)

// Executing the State Machine requires a custom name, some custom input and the ID of the State Machine from AWS

var params = {

  name: name,

  input: JSON.stringify(payload),

  stateMachineArn: Env.get('AWS_STATEMACHINE_ARN')

}

// Execute StepFunction

// @see https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/StepFunctions.html#startExecution-property

let result = null

try {

  result = await StepFunctions.startExecution(params).promise()

} catch (e) {

  console.log('Error:', e)

}

return result

}

}

module.exports = QueueService

DemoWorker.js

Let's create a new folder app\Workers and add DemoWorker.js. This class is responsible for consuming the job that is pending inside the state machine. It requires the npm module 'step-function-worker'

Install it into your adonis project:

adonis install step-function-worker



'use strict'

const StepFunctionWorker = require('step-function-worker')

const AWS = require('aws-sdk')

const Env = use('Env')

const Logger = use('Logger')

/*

|--------------------------------------------------------------------------

| DemoWorker

|--------------------------------------------------------------------------

| @see https://github.com/piercus/step-function-worker

|

| For this class to work, a StepFunction activity must first be configured on AWS

*/

class DemoWorker extends StepFunctionWorker {

  constructor () {

    const options = {}

// Configure the connection to AWS
options.awsConfig = new AWS.Config({
  accessKeyId: Env.get('AWS_ACCESS_KEY'),
  secretAccessKey: Env.get('AWS_SECRET_KEY'),
  region: Env.get('AWS_REGION')
})

// Function that interacts directly with the AWS StepFunction this must be defined
// The call back is responsible for letting the State Machine know it can either
// continue to the next step in execution or fail
options.fn = async (input, cb, heartbeat) =&gt; {
  // Respond to StepFunction state machine
  Logger.info('Custom Worker function:', input)
  cb(null, input)
}

// the ID of the Step Function Activity arn:aws:states:us-east-1:XXXXXXXXXXXXX:activity:DemoWorkerActivity
options.activityArn = Env.get('AWS_ACTIVITY_ARN_DEMOWORKERACTIVITY')

super(options)
this._initCallbacks()



}

_initCallbacks () {

    this.on('task', this.task)

    this.on('ready', this.ready)

    this.on('error', this.error)

    this.on('failure', this.failure)

    this.on('success', this.success)

  }

/**


Called when the worker "wakes up"
The StepFunctionWorker parent class will pass in the payload
@param {*} task
*/
task (task) {
// task.input contains the payload from the web
Logger.info('DemoWorker task:', task.input)
}


ready () {

    Logger.info('DemoWorker is ready')

  }

failure (failure) {

    Logger.info('DemoWorker failure:', failure)

  }

success (output) {

    // output.input will contain the payload from the web

    Logger.info('DemoWorker success:', output.input)

  }

error (err) {

    Logger.info('DemoWorker error:', err)

  }

}

module.exports = DemoWorker

Demo Time

I'll skip ahead to our demo. We'll create a form where we will enter in a name for the State Machine Execution. You can customize the name given to the execution job. In our project we used the ObjectID of a MongoDB record that was responsible for maintaining the state of the queued job. Using a naming convention that helped us trace the execution object back to the object in our system that initialized the payload, made it easier to follow our system flow.

Let's fire up out adonis project:

adonis serve --dev

I can now access http://localhost:3333

There I have the following form:

After submitting the form let's take a look at the State Function execution job that has been queued up.

Now we have to fire up our DemoWorker to watch for our custom Activity to get executed.

adonis stepfunctionworker --name=DemoWorker

As soon as the DemoWorker starts, it pulls the pending execution job off the State Function queue.

Let's take another look at the State Machine and see that the execution completed succesfully.

And done! We've accomplished our goal of creating an AWS State Machine and having our custom application initiate and complete a small, but complete job.

Source code is at: https://github.com/openstepmedia/adonisjs-stepfunction-demo