Upload File to AWS S3 in Go

In this post we will review how to upload a file to AWS S3 in Go.

First we create an AWS session. We can use one of the methods specified in this post, but in this case we need to support different AWS session/credentials for each upload, hence we statically supply the AWS session configuration.

import (
   "context"
   "github.com/aws/aws-sdk-go/aws"
   "github.com/aws/aws-sdk-go/aws/credentials"
   "github.com/aws/aws-sdk-go/aws/session"
   "github.com/aws/aws-sdk-go/service/s3/s3manager"
   "strings"
)

func uploadFile() {
   region := "us-east-1"
   accessKey := "AKIAWXXXXXXXXXXXRQWU"
   secretKey := "XXXXKsqJ2inJdxBdXXXXDE0jX+gxxFXXXXRVXX"

   config := aws.Config{
      Region:      aws.String(region),
      Credentials: credentials.NewStaticCredentials(accessKey, secretKey, ""),
   }
   awsSession, err := session.NewSession(&config)
   if err != nil {
      panic(err)
   }

Next we upload the file using S3 manager. The path in bucket can includes folders, and there is no need to create any sub folders, as the S3 does not actually keeps folders, instead it is a key-value implementation, and the folders are only used in the AWS console GUI presentation of the folders files.

   bucketName := "my-bucket"
   keyInBucket := "folder1/my-file.txt"
   fileContent := "this is my data"

   uploader := s3manager.NewUploader(awsSession)

   input := &s3manager.UploadInput{
      Bucket:      aws.String(bucketName),
      Key:         aws.String(keyInBucket),
      Body:        strings.NewReader(fileContent),
      ContentType: aws.String("text/plain"),
   }
   _, err = uploader.UploadWithContext(context.Background(), input)
   if err != nil {
      panic(err)
   }
}

Using AWS SQS in Go

In this post we will review usage of AWS SQS with a Go Application.

Let's have a look at the general structure of the example:

package main

import (
  "fmt"
  "github.com/aws/aws-sdk-go/aws"
  "github.com/aws/aws-sdk-go/aws/session"
  "github.com/aws/aws-sdk-go/service/sqs"
  "strings"
  "sync"
  "time"
)

const queueUrl = "https://sqs.us-east-1.amazonaws.com/YOUR_ACCOUNT_ID/YOUR_QUEUE_NAME"

func main() {
  purge()
  go sendLoop()
  go receiveLoop(1)
  go receiveLoop(2)
  var waitGroup sync.WaitGroup
  waitGroup.Add(1)
  waitGroup.Wait()
}

We start by purging the queue messages, assuring that we start with a clear queue, and not affected by previous runs. 

Notice we can purge queue messages only once in a minute.

func purge() {
  awSession := session.Must(session.NewSession())
  svc := sqs.New(awSession)
  queueInput := &sqs.PurgeQueueInput{
   QueueUrl: aws.String(queueUrl),
  }
  _, err := svc.PurgeQueue(queueInput)
  if err != nil {
   panic(err)
  }
}

We run a go routine to produce messages every ~2 seconds. 

We use DelaySeconds of 3 seconds. This might be required in case the processing cannot start immediately due to some other processing that should be prepared in the background.

func sendLoop() {
  awSession := session.Must(session.NewSession())
  svc := sqs.New(awSession)

  i := 0
  for {
   i++
   messageInput := sqs.SendMessageInput{
    DelaySeconds: aws.Int64(3),
    MessageBody:  aws.String(fmt.Sprintf("%v", i)),
    QueueUrl:     aws.String(queueUrl),
   }

   _, err := svc.SendMessage(&messageInput)
   if err != nil {
    panic(err)
   }
   fmt.Printf("sent message #%v\n", i)

   time.Sleep(2 * time.Second)
  }
}

The last thing we do is to run two consumers as go routine. 

Notice that we configure we're willing to wait up to 1 second for messages, which is good to reduce amount of AWS API calls and enables getting larger bulks of messages, up to MaxNumberOfMessages. 

We also use VisibilityTimeout of 10 seconds that marks the message as not available to other consumers for that period. This means the consumer has up to 10 seconds to process the message and to delete if from the queue.

func receiveLoop(
  receiverId int,
) {
  awSession := session.Must(session.NewSession())
  svc := sqs.New(awSession)

  for {
   messageInput := sqs.ReceiveMessageInput{
    AttributeNames: []*string{
     aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
    },
    MessageAttributeNames: []*string{
     aws.String(sqs.QueueAttributeNameAll),
    },
    QueueUrl:            aws.String(queueUrl),
    MaxNumberOfMessages: aws.Int64(10),
    VisibilityTimeout:   aws.Int64(10),
    WaitTimeSeconds:     aws.Int64(1),
   }

   msgResult, err := svc.ReceiveMessage(&messageInput)
   if err != nil {
    panic(err)
   }

   if len(msgResult.Messages) == 0 {
    fmt.Printf("receiver %v, no messages\n", receiverId)
   } else {
    items := make([]string, 0)
    for _, message := range msgResult.Messages {
     items = append(items, *message.Body)
    }
    fmt.Printf("receiver %v, messages: %v\n",
     receiverId, strings.Join(items, ","))

    for _, message := range msgResult.Messages {
     deleteMessageInput := sqs.DeleteMessageInput{
      QueueUrl:      aws.String(queueUrl),
      ReceiptHandle: message.ReceiptHandle,
     }
     _, err = svc.DeleteMessage(&deleteMessageInput)
     if err != nil {
      panic(err)
     }
    }
   }
  }
}

An example of output is below. 

Notice that a message is indeed consumed only 3 seconds after it was sent.

Also, once one receiver got a message, it is not available for the other consumers.

sent message #1
receiver 2, no messages
receiver 1, no messages
receiver 2, no messages
sent message #2
receiver 1, no messages
receiver 2, no messages
receiver 1, messages: 1
receiver 2, no messages
sent message #3
receiver 1, no messages
receiver 1, messages: 2
receiver 2, no messages
receiver 2, no messages
sent message #4
receiver 1, no messages
receiver 1, messages: 3
receiver 2, no messages
receiver 1, no messages
receiver 2, no messages
sent message #5
receiver 2, messages: 4
receiver 1, no messages
receiver 2, no messages
receiver 1, no messages
sent message #6
receiver 2, messages: 5
receiver 1, no messages
receiver 2, no messages
sent message #7
receiver 1, no messages
receiver 2, messages: 6
receiver 1, no messages
sent message #8
receiver 2, no messages
receiver 1, no messages
receiver 2, messages: 7
receiver 1, no messages
sent message #9
receiver 2, no messages

Let's make our consumer fail to process the messages in a timely fashion in some cases.

We do this by changing the VisibilityTimeout to 5 seconds, and add a random sleep of up to 10 seconds before delete of the message from the queue.

func receiveLoop(
  receiverId int,
) {
  awSession := session.Must(session.NewSession())
  svc := sqs.New(awSession)

  for {
   messageInput := sqs.ReceiveMessageInput{
    AttributeNames: []*string{
     aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
    },
    MessageAttributeNames: []*string{
     aws.String(sqs.QueueAttributeNameAll),
    },
    QueueUrl:            aws.String(queueUrl),
    MaxNumberOfMessages: aws.Int64(10),
    VisibilityTimeout:   aws.Int64(5),
    WaitTimeSeconds:     aws.Int64(1),
   }

   msgResult, err := svc.ReceiveMessage(&messageInput)
   if err != nil {
    panic(err)
   }
   
   if len(msgResult.Messages) == 0 {
    fmt.Printf("receiver %v, no mesages\n", receiverId)
   } else {
    items := make([]string, 0)
    for _, message := range msgResult.Messages {
     items = append(items, *message.Body)
    }
    fmt.Printf("receiver %v, messages: %v\n",
     receiverId, strings.Join(items, ","))
    
    sleepTime := time.Second * time.Duration(rand.Int63n(10))
    fmt.Printf("receiver %v, sleeping %v\n", receiverId, sleepTime)
    time.Sleep(sleepTime)

    for _, message := range msgResult.Messages {
     deleteMessageInput := sqs.DeleteMessageInput{
      QueueUrl:      aws.String(queueUrl),
      ReceiptHandle: message.ReceiptHandle,
     }
     _, err = svc.DeleteMessage(&deleteMessageInput)
     if err != nil {
      panic(err)
     }
    }
   }
  }
}

The result is below.

We can see that some messages that took too long to process are re-consumed by another consumer (see message #4 for example).

sent message #1
receiver 2, no mesages
receiver 1, no mesages
sent message #2
receiver 2, no mesages
receiver 1, no mesages
receiver 2, messages: 1
receiver 2, sleeping 9s
receiver 1, no mesages
sent message #3
receiver 1, messages: 2
receiver 1, sleeping 2s
sent message #4
receiver 1, messages: 3
receiver 1, sleeping 4s
sent message #5
sent message #6
receiver 1, messages: 4
receiver 1, sleeping 8s
receiver 2, messages: 5
receiver 2, sleeping 7s
sent message #7
sent message #8
sent message #9
sent message #10
receiver 2, messages: 4,6,7
receiver 2, sleeping 5s
receiver 1, messages: 8
receiver 1, sleeping 6s
sent message #11
sent message #12
receiver 2, messages: 10
receiver 2, sleeping 8s
sent message #13
receiver 1, messages: 9,11
receiver 1, sleeping 3s
sent message #14
receiver 1, messages: 12
receiver 1, sleeping 3s
sent message #15
sent message #16
receiver 1, messages: 14
receiver 1, sleeping 7s
receiver 2, messages: 13,15
receiver 2, sleeping 2s
sent message #17
receiver 2, messages: 16
receiver 2, sleeping 8s
sent message #18
sent message #19
receiver 1, messages: 18
receiver 1, sleeping 5s

AWS SSQ API, Implementation, and Stub in Go

 

In this post we will present a to send messages to AWS SQS. We will include an interface, an implementation, and a stub. We've previously included an example for producer and consumer in Go, and here we provide a nice API that enables us to use this code both in production and in tests.

The interface is the minimal API required to send a message. In this interface we hide the AWS session connection, as well as the AWS SQS queue name, and include only the message details.

type SqsApi interface {
  SendMessage(
   attributes map[string]string,
   data string,
  )
}

The implementation uses an AWS session. Notice that there are several methods to get an AWS session, but in this implementation we use the simplest method.

package awssqs

import (
  "github.com/aws/aws-sdk-go/aws"
  "github.com/aws/aws-sdk-go/aws/session"
  "github.com/aws/aws-sdk-go/service/sqs"
)

type SqsImpl struct {
  sqsClient *sqs.SQS
  queueUrl  string
}

func ProduceAwsSqsImpl(
  queueUrl string,
) *SqsImpl {
  awsSession, err := session.NewSession()
  if err!= nil{
   panic(err)
  }

  return &SqsImpl{
   sqsClient: sqs.New(awsSession),
   queueUrl:  queueUrl,
  }
}

func (s *SqsImpl) SendMessage(
  attributes map[string]string,
  data string,
) {
  input := sqs.SendMessageInput{
   MessageAttributes: make(map[string]*sqs.MessageAttributeValue),
   MessageBody:       aws.String(data),
   QueueUrl:          aws.String(s.queueUrl),
  }

  for key, value := range attributes {
   input.MessageAttributes[key] = &sqs.MessageAttributeValue{
    DataType:    aws.String("String"),
    StringValue: aws.String(value),
   }
  }

  _, err := s.sqsClient.SendMessage(&input)
  if err!= nil{
   panic(err)
  }
}

We also include a stub that can be used during tests:

type MessageData struct {
  Attributes map[string]string
  Data       string
}

type SqsStub struct {
  Messages []*MessageData
}

func ProduceAwsSqsStub() *SqsStub {
  return &SqsStub{}
}

func (s *SqsStub) SendMessage(
  attributes map[string]string,
  data string,
) {
  messageData := MessageData{
   Attributes: attributes,
   Data:       data,
  }
  s.Messages = append(s.Messages, &messageData)
}

List and Read Files from AWS S3 using GoLang

 

In this post we will review how to list files and read files from AWS S3 using GO.

We will be using the AWS SDK for GO: aws-sdk-go.

To access the AWS S3, you must use valid credentials. The default chain of credentials providers includes the following (quoted from the AWS SDK):

1. Environment Credentials - Set of environment variables that are useful when sub processes are created for specific roles.

2. Shared Credentials file (~/.aws/credentials) - This file stores your credentials based on a profile name and is useful for local development.

3. EC2 Instance Role Credentials - Use EC2 Instance Role to assign credentials to application running on an EC2 instance. This removes the need to manage credential files in production.

AWS documentation recommends using the 3rd method, as it is the best secured alternative, and also automatically manages the credentials. Note that this method can be used only when running your code on an AWS EC2 instance. Trying to run the code on a non EC2 with a managed role, would cause the following error:

panic: NoCredentialProviders: no valid providers in chain. Deprecated.

Let's look at the main logic: we will connect to AWS S3, list files on a specific folder, and then read the first file from the list, and print it to the STDOUT.

The main code is:

package main

import (
   "fmt"
   "github.com/aws/aws-sdk-go/aws"
   "github.com/aws/aws-sdk-go/aws/session"
   "github.com/aws/aws-sdk-go/service/s3"
   "io/ioutil"
   "sort"
)

func main() {
   region := "us-east-1"

   config := aws.Config{
      Region: aws.String(region),
   }
   awsSession, err := session.NewSession(&config)
   if err != nil {
      panic(err)
   }

   s3Client := s3.New(awsSession)

   folder := "my-folder"
   bucket := "my-bucket"

   files := list(s3Client, bucket, folder)
   bytes := read(s3Client, bucket, files[0])
   fmt.Printf("file data is:\n%v\n", string(bytes))
}

The list function receives a bucket name and a folder, and list all the files within this folder. Notice that the list is recursive, which means that all the files in the sub folders are also returned. Notice that the strings array contains the keys for each file. The key is the full path to the file, starting from the bucket root, regardless of the folder used for the list API.

The list function is:

func list(s3Client *s3.S3, bucket string, folder string) []string {
   params := &s3.ListObjectsInput{
      Bucket: aws.String(bucket),
      Prefix: aws.String(folder),
   }

   resp, err := s3Client.ListObjects(params)
   if err != nil {
      panic(err)
   }

   items := make([]string, 0)
   for _, key := range resp.Contents {
      items = append(items, *key.Key)
   }

   sort.Strings(items)
   return items
}

Finally, let review the file read function. It reads a file from the AWS S3, and returns the file content as a bytes array. For large files, that could pose a problem to keep the entire file in the process memory, avoid using this method, and instead consider using the AWS S3 download API.

func read(s3Client *s3.S3, bucket string, file string) []byte {
   getObject := &s3.GetObjectInput{
      Bucket: aws.String(bucket),
      Key:    aws.String(file),
   }

   result, err := s3Client.GetObject(getObject)
   if err != nil {
      panic(err)
   }
   defer result.Body.Close()
   data, err := ioutil.ReadAll(result.Body)
   if err != nil {
      panic(err)
   }

   return data
}

Final Notes

In this post we've used some basic AWS S3 APIs to list files in a bucket, and to read a file content.

For buckets with more than 1K files, a pagination is used, and hence the list API should be repeatedly called and the ListObjectsInput.Marker should be used for pagination.

Using AWS Kinesis in Go

 

In this post we will review a simple Go implementation of AWS kinesis producer and consumer.

The main function starts the producer and the consumer, and then waits forever.

import (
  "fmt"
  "github.com/aws/aws-sdk-go/aws"
  "github.com/aws/aws-sdk-go/aws/session"
  "github.com/aws/aws-sdk-go/service/kinesis"
  "sync"
  "time"
)

func main() {
  go producer()
  go consumer()
  
  waitGroup := sync.WaitGroup{}
  waitGroup.Add(1)
  waitGroup.Wait()
}

The producer writes a record to AWS kinesis stream every second.

func producer() {
  awSession := session.Must(session.NewSession())
  kinesisClient := kinesis.New(awSession)
  for i := 0; ; i++ {
   input := kinesis.PutRecordInput{
    Data:         []byte(fmt.Sprintf("message data %v", i)),
    PartitionKey: aws.String(fmt.Sprintf("key%v", i)),
    StreamName:   aws.String("s1"),
   }
   output, err := kinesisClient.PutRecord(&input)
   if err != nil {
    panic(err)
   }

   fmt.Printf("produced record to shard %v\n", *output.ShardId)
   time.Sleep(time.Second)
  }
}

The consumer is more complex. First we need to list the stream's shards, and start a consumer for each of the shards.

func consumer() {
  awSession := session.Must(session.NewSession())
  kinesisClient := kinesis.New(awSession)

  input := kinesis.DescribeStreamInput{
   StreamName: aws.String("s1"),
  }
  output, err := kinesisClient.DescribeStream(&input)
  if err != nil {
   panic(err)
  }

  fmt.Printf("%v shards loacted\n", len(output.StreamDescription.Shards))

  for i := 0; i < len(output.StreamDescription.Shards); i++ {
   shardId := output.StreamDescription.Shards[i].ShardId
   go consumerShard(kinesisClient, shardId)
  }
}

Each shard consumer has a small trick. First we use iterator to get all the new records. Once we got a record, we modify the iterator to read all records after the last received record.

func consumerShard(
  kinesisClient *kinesis.Kinesis,
  shardId *string,
) {
  iteratorInput := kinesis.GetShardIteratorInput{
   ShardId:           shardId,
   ShardIteratorType: aws.String("LATEST"),
   StreamName:        aws.String("s1"),
  }
  shardIteratorOutput, err := kinesisClient.GetShardIterator(&iteratorInput)
  if err != nil {
   panic(err)
  }

  var lastRecord *string
  for {
   recordsInput := kinesis.GetRecordsInput{
    ShardIterator: shardIteratorOutput.ShardIterator,
   }
   records, err := kinesisClient.GetRecords(&recordsInput)
   if err != nil {
    panic(err)
   }

   fmt.Printf("shard %v got %v records\n", *shardId, len(records.Records))

   for i := 0; i < len(records.Records); i++ {
    record := records.Records[i]
    fmt.Printf("shard %v data: %v\n", *shardId, string(record.Data))
    lastRecord = record.SequenceNumber
   }

   time.Sleep(5 * time.Second)

   if lastRecord != nil {
    iteratorInput.ShardIteratorType = aws.String("AFTER_SEQUENCE_NUMBER")
    iteratorInput.StartingSequenceNumber = lastRecord
    shardIteratorOutput, err = kinesisClient.GetShardIterator(&iteratorInput)
    if err != nil {
     panic(err)
    }
   }
  }
}

An example output for running this is below.

produced record to shard shardId-000000000000
4 shards loacted
shard shardId-000000000000 got 0 records
shard shardId-000000000003 got 0 records
shard shardId-000000000002 got 0 records
shard shardId-000000000001 got 0 records
produced record to shard shardId-000000000003
produced record to shard shardId-000000000001
produced record to shard shardId-000000000000
produced record to shard shardId-000000000003
shard shardId-000000000000 got 1 records
shard shardId-000000000000 data: message data 3
shard shardId-000000000003 got 2 records
shard shardId-000000000003 data: message data 1
shard shardId-000000000003 data: message data 4
shard shardId-000000000001 got 1 records
shard shardId-000000000001 data: message data 2
shard shardId-000000000002 got 0 records
produced record to shard shardId-000000000000
produced record to shard shardId-000000000001
produced record to shard shardId-000000000003
produced record to shard shardId-000000000001
produced record to shard shardId-000000000000
shard shardId-000000000000 got 2 records
shard shardId-000000000000 data: message data 5
shard shardId-000000000000 data: message data 9
shard shardId-000000000003 got 1 records
shard shardId-000000000003 data: message data 7
shard shardId-000000000002 got 0 records
shard shardId-000000000001 got 2 records
shard shardId-000000000001 data: message data 6
shard shardId-000000000001 data: message data 8
produced record to shard shardId-000000000003
produced record to shard shardId-000000000003
produced record to shard shardId-000000000003
produced record to shard shardId-000000000003
shard shardId-000000000000 got 0 records
shard shardId-000000000003 got 4 records
shard shardId-000000000003 data: message data 10
shard shardId-000000000003 data: message data 11
shard shardId-000000000003 data: message data 12
shard shardId-000000000003 data: message data 13
shard shardId-000000000002 got 0 records
shard shardId-000000000001 got 0 records
produced record to shard shardId-000000000002
produced record to shard shardId-000000000001
produced record to shard shardId-000000000003
produced record to shard shardId-000000000002
produced record to shard shardId-000000000000
shard shardId-000000000000 got 1 records
shard shardId-000000000000 data: message data 18
shard shardId-000000000002 got 2 records
shard shardId-000000000002 data: message data 14
shard shardId-000000000002 data: message data 17
shard shardId-000000000003 got 1 records
shard shardId-000000000003 data: message data 16
shard shardId-000000000001 got 1 records
shard shardId-000000000001 data: message data 15
produced record to shard shardId-000000000001
produced record to shard shardId-000000000000

This is just a starting point for using AWS kinesis, as they are many details that affect the stability of a production grade solution. Still, this is a good starting point.

Create AWS DynamoDB using CloudFormation and a Sample Golang Application

 

In this post we will use CloudFormation to setup a DynamoDB table, and then access it using a sample GO application.

To setup the DynamoDB we will use the following CloudFormation stack.

dynamoDBTable:
  Type: AWS::DynamoDB::Table
  Properties:
    BillingMode: PAY_PER_REQUEST
    TableName: "my-table"
    AttributeDefinitions:
      - AttributeName: "mykey"
        AttributeType: "S"
    KeySchema:
      - AttributeName: "mykey"
        KeyType: "HASH"

Notice that we specify the key attribute twice. The first time we define its type, which in this case is a string ("S").  The second time we specify that this is a key attribute. Other attributes will be automatically added by the DynamoDB once objects with new attributes are created. An exception for this is a "RANGE" attribute that if it is required, should be also specified here.

The billing mode is "PER REQUEST", which is great if you have no idea about the expected read/write load on the table.

Any other service that accesses the DynamoDB table, should be granted with permissions to access it, for example, to grant an ECS task role permission to access the DynamoDB table use the following rather too permissive  policy. In case of need, limit the actions to a smaller set, e.g:

• dynamodb.GetItem
• dynamodb.PutItem
• dynamodb.Query

taskIamRole:
  Type: AWS::IAM::Role
  Properties:
    RoleName: my-role
    AssumeRolePolicyDocument:
      Statement:
        - Effect: Allow
          Principal:
            Service: ecs-tasks.amazonaws.com
          Action: 'sts:AssumeRole'

taskIamPolicy:
  Type: AWS::IAM::Policy
  Properties:
    PolicyName: my-policy
    PolicyDocument:
      Version: 2012-10-17
      Statement:
        - Effect: Allow
          Action:
            - dynamodb:*
          Resource: arn:aws:dynamodb:*:*:table/my-table
    Roles:
      - !Ref taskIamRole

To access the DynamoDB from a GO application, first get a DynamoDB API interface:

import (
 "github.com/aws/aws-sdk-go/aws"
 "github.com/aws/aws-sdk-go/aws/session"
 "github.com/aws/aws-sdk-go/service/dynamodb"
 "github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
)

config := aws.Config{
 Region: aws.String("us-east-1"),
}

awsSession, err := session.NewSession(&config)
if err != nil {
 panic(err)
}

dynamoDbApi := dynamodb.New(awsSession)

Now we can add items to the DynamoDB using the PutItem API. Notice that we define a structure with the names of the attributes that we want to have in the DynamoDB table.

item:= Item{
 Key:  "key1",
 Data: 123,
}
mappedItem, err := dynamodbattribute.MarshalMap(item)
if err != nil {
 panic(err)
}

query := dynamodb.PutItemInput{
 Item:      mappedItem,
 TableName: aws.String("my-table"),
}

_, err = dynamoDbApi.PutItem(&query)
if err != nil {
 panic(err)
}

Final Note

While it is not cheap, the AWS DynamoDB supplies an easy API, and great performance for an application. I recommend using it in case your DB API rate is moderate.

AWS Batch in Go

 

In a previous post we've used AWS batch using boto3.

In this post we will wrap usage of AWS batch using golang.

First we'll create the batch wrapper class.

package awsbatch

import (
 "fmt"
 "github.com/aws/aws-sdk-go/aws"
 "github.com/aws/aws-sdk-go/aws/session"
 "github.com/aws/aws-sdk-go/service/batch"
 "time"
)

type BatchWrapper struct {
 batch *batch.Batch
 jobs  []*string
}

func ProduceBatchWrapper() *BatchWrapper {
 awsSession, err := session.NewSession()
 if err != nil {
  panic(err)
 }

 awsBatch := batch.New(awsSession)
 return &BatchWrapper{
  batch: awsBatch,
  jobs:  []*string{},
 }
}

Next, we submit batches, and let them run in the background.

func (b *BatchWrapper) SubmitBatch(
 jobName string,
 environment map[string]string,
) {

 overrides := batch.ContainerOverrides{
  Command:     []*string{aws.String("/simulatorbackend")},
  Environment: []*batch.KeyValuePair{},
 }

 for key, value := range environment {
  overrides.Environment = append(
   overrides.Environment,
   &batch.KeyValuePair{
    Name:  aws.String(key),
    Value: aws.String(value),
   },
  )
 }

 input := batch.SubmitJobInput{
  JobName:            aws.String(jobName),
  JobQueue:           aws.String("my-batch-queue"),
  JobDefinition:      aws.String("my-batch-jobdef"),
  ContainerOverrides: &overrides,
 }

 output, err := b.batch.SubmitJob(&input)
 if err != nil {
  panic(err)
 }

 b.jobs = append(b.jobs, output.JobId)
}

And finally, we wait for all the batches to complete.

func (b *BatchWrapper) WaitForBatches() {
 input := batch.DescribeJobsInput{
  Jobs: b.jobs,
 }

 b.jobs = []*string{}

 for {
  output, err := b.batch.DescribeJobs(&input)
  if err != nil {
   panic(err)
  }

  allDone := true
  for _, job := range output.Jobs {
   if *job.Status == "FAILED" {
    panic("job failed")
   } else if *job.Status == "RUNNING" || *job.Status == "STARTING" || *job.Status == "SUBMITTED" {
    fmt.Printf("job %v id %v status %v\n",
     *job.JobName,
     *job.JobId,
     *job.Status,
    )

    allDone = false
   }
  }
  if allDone {
   return
  }
  time.Sleep(time.Second * 10)
 }
}

Create AWS WebACL for CloudFront

 

NOTICE:

New AWS SDK was published.

You better use it instead of the one specified in the post.

For more details, see here.

In this post we will use AWS Go API to create a WebACL, and then we will associate it with a CloudFront distribution.

Before starting, make sure to setup credentials and region as specified in this post.

We will start by creating an example of WebACL rule to block any request with query string that have a specific suffix.

awsSession, err := session.NewSession(&aws.Config{
   Region: aws.String("us-east-1"),
})
wafClient := wafv2.New(awsSession)
if err != nil {
   panic(err)
}

statement := wafv2.Statement{
   ByteMatchStatement: &wafv2.ByteMatchStatement{
      FieldToMatch: &wafv2.FieldToMatch{
         QueryString: &wafv2.QueryString{},
      },
      PositionalConstraint: aws.String("ENDS_WITH"),
      SearchString:         []byte("/3"),
      TextTransformations: []*wafv2.TextTransformation{
         {
            Priority: aws.Int64(1),
            Type:     aws.String("NONE"),
         },
      },
   },
}

rule := wafv2.Rule{
   Name:     aws.String("rule-1"),
   Priority: aws.Int64(1),
   Action: &wafv2.RuleAction{
      Block: &wafv2.BlockAction{},
   },
   Statement: &statement,
   VisibilityConfig: &wafv2.VisibilityConfig{
      CloudWatchMetricsEnabled: aws.Bool(false),
      SampledRequestsEnabled:   aws.Bool(false),
      MetricName:               aws.String("my-rule-metric"),
   },
}

aclInput := wafv2.CreateWebACLInput{
   DefaultAction: &wafv2.DefaultAction{
      Block: &wafv2.BlockAction{},
   },
   Name:  aws.String("my-webacl"),
   Rules: []*wafv2.Rule{&rule},
   Scope: aws.String("CLOUDFRONT"),
   VisibilityConfig: &wafv2.VisibilityConfig{
      CloudWatchMetricsEnabled: aws.Bool(false),
      SampledRequestsEnabled:   aws.Bool(false),
      MetricName:               aws.String("my-rule-metric"),
   },
}

acl, err := wafClient.CreateWebACL(&aclInput)
if err != nil {
   panic(err)
}

Next we can update the CloudFront distribution to use this WebACL.

Notice that you cannot use the WebACL API to associate the WebACL to the CloudFront distribution (why? ask AWS team. I guess they did not want to make it easy).

getConfigInput := cloudfront.GetDistributionConfigInput{
   Id: aws.String("E1Y55CUPVONMHF"),
}
cloudFrontClient := cloudfront.New(awsSession)

distributionConfigOutput, err := cloudFrontClient.GetDistributionConfig(&getConfigInput)
if err != nil {
   panic(err)
}

distributionInput := cloudfront.UpdateDistributionInput{
   DistributionConfig: distributionConfigOutput.DistributionConfig,
   Id:                 aws.String("E1Y55CUPVONMHF"),
   IfMatch:            distributionConfigOutput.ETag,
}
distributionInput.DistributionConfig.WebACLId = acl.Summary.Id

_, err = cloudFrontClient.UpdateDistribution(&distributionInput)
if err != nil {
   panic(err)
}

That's it, we have a WebACL assigned to our CloudFront distribution.

Final Notes

I've added this post, since there are just no good examples of how to do this.

I hope you will find this useful.

Deploy Python Application to AWS EMR

 

In this post we will review the steps to automatically deploy a python application to spark running on AWS EMR.

Our main function is the following:

import os
import re
import stat
from zipfile import ZipFile

import boto3
import paramiko


def main():
    aws_set_credentials()
    chmod_ssh_key()
    zip_remote_path = copy_source_zip_to_s3()
    main_remote_path = copy_file_to_s3('/git/my-repo/main.py', 's3-source-bucket', 'main.py')
    run_spark_application(zip_remote_path, main_remote_path)

The deploy of the application starts with handling of AWS credentials and the AWS SSH key permissions. Then we create two files in an AWS S3 bucket, that includes our application sources, and finally we run the application by SSH and run command on the AWS EMR master node.

Let examine each of the steps.

def aws_set_credentials():
    credentials_file = '/config/credentials'
    os.environ['AWS_SHARED_CREDENTIALS_FILE'] = credentials_file

The AWS set credentials updates an environment variable to point to the location of our credentials. These will be used for AWS operations, such as update of the S3 bucket. An example of a credentials file is:

[default]
aws_access_key_id=AKIAWJPWYKUU1234567
aws_secret_access_key=rXKlsqJ2inJdxBdJk123456782345678923

Next we update the SSH private key mode:

def chmod_ssh_key():
    private_key_path = '/config/ssh.pem'
    os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)

The SSH private key is the one used to create the EMR master node. We will later SSH to the EMR, hence we want to make sure that SSH private key has permissions only for the owner.

Once the AWS setup is ready, we can copy the source zip file.

def create_zip_file(zip_file_path, add_folder, match_regex):
    pattern = re.compile(match_regex)
    with ZipFile(zip_file_path, 'w') as zip_object:
        for folder_name, sub_folders, file_names in os.walk(add_folder):
            for file_name in file_names:
                file_path = os.path.join(folder_name, file_name)
                if pattern.match(file_path):
                    relative_path = file_path[len(add_folder) + 1:]
                    zip_object.write(file_path, relative_path)


def copy_file_to_s3(local_file_path, bucket_name, remote_file_path):
    remote_path = 's3://{}/{}'.format(bucket_name, remote_file_path)
    session = boto3.Session()
    s3_connection = session.client('s3')
    s3_connection.upload_file(local_file_path, bucket_name, remote_file_path)
    return remote_path


def copy_source_zip_to_s3():
    source_dir = '/git/my-repo'
    zip_file_name = "emr-application.zip"
    local_zip_file_path = os.path.join('tmp', zip_file_name)
    create_zip_file(local_zip_file_path, source_dir, ".*py")
    remote_path = copy_file_to_s3(local_zip_file_path, 's3-source-bucket', zip_file_name)
    os.remove(local_zip_file_path)
    return remote_path

All the related source and dependencies should be zipped and copied to the S3, so the EMR can access it. Notice that this includes the local dependencies, but the main application python file should be copied separately, and hence the main deploy function copies both the sources zip file and the main python file to the S3 bucket.

The last step is the actual run of the application on the EMR.

def get_emr_master_id():
    client = boto3.client('emr')
    response = client.list_clusters(
        ClusterStates=[
            'RUNNING', 'WAITING',
        ],
    )

    emr_cluster_name = 'my-emr'

    for cluster in response['Clusters']:
        if cluster['Name'] == emr_cluster_name:
            return cluster['Id']

    raise Exception('emr cluster {} not located'.format(emr_cluster_name))


def get_emr_master_ip():
    cluster_id = get_emr_master_id()
    client = boto3.client('emr')
    response = client.list_instances(
        ClusterId=cluster_id,
        InstanceGroupTypes=[
            'MASTER',
        ],
        InstanceStates=[
            'RUNNING',
        ]
    )
    instances = response['Instances']
    if len(instances) != 1:
        raise Exception('emr instances count {} is invalid'.format(len(instances)))

    master_instance = instances[0]
    ip = master_instance['PublicIpAddress']
    return ip


def run_ssh_command(host, user, command):
    private_key_path = '/config/ssh.pem'
    private_key = paramiko.RSAKey.from_private_key_file(private_key_path)

    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname=host, username=user, pkey=private_key)
    ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command(command)
    ssh_stderr.channel.recv_exit_status()
    ssh_stdout.channel.recv_exit_status()
    all_err = ssh_stderr.read().decode("utf-8")
    all_out = ssh_stdout.read().decode("utf-8")
    ssh.close()
    return all_err, all_out


def run_spark_application(s3_zip_file_path, main_file_path):
    host_ip = get_emr_master_ip()
    command_sections = [
        'spark-submit',
        '--deploy-mode cluster',
        '--master yarn',
        '--conf spark.yarn.submit.waitAppCompletion=true',
        '--py-files {}'.format(s3_zip_file_path),
        main_file_path,
    ]
    command = ' '.join(command_sections)
    error, output = run_ssh_command(host_ip, 'hadoop', command)
    print(error + '\n' + output)

We start by located the EMR master node public IP using boto3 API. Notice that the master must be in a AWS VPC/subnet that allows SSH to it. After the SSH connection is established, we use the spark submit command to run our code.

The logs of the application can be located in AWS EMR GUI after about 5 minutes, as the EMR periodically updates the status every 5 minutes.