Amazon S3 to AWS StepFunctions pattern
Tags: aws, cdk, serverless
Published: 2021-11-21 05:06:24

---
title: Amazon S3 to AWS StepFunctions pattern
published: true
description: Examples of integrating Amazon S3 and AWS StepFunctions.
tags: aws, cdk, serverless
cover_image: https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7x1t1cw57ailm6zr5fyl.jpeg
---

_Amazon S3_ is easily my top pick from the vast catalog of different AWS services. Considering its [features](https://aws.amazon.com/s3/features/) and the [consistency guarantees](https://aws.amazon.com/s3/consistency/), it really is an engineering marvel.

Since _Amazon S3_ is often used as a starting point for various workloads, one might need to integrate S3 events with a workflow orchestration service - like _AWS StepFunctions_.

This blog post will discuss ways developers might wire the _Amazon S3_ and _AWS StepFunctions_ together to achieve highly scalable workflows.

Let us dive in.

![Mistery](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/swlrbbizfa6bpf9tvdvr.png)

> All the code used in this blog post could be found in [this GitHub repository](https://github.com/WojciechMatuszewski/s3-to-sfn-pattern).

## Using S3 Notifications and AWS Lambda function

Depending on your use case, this might be the most straightforward option available to you. The integration is [well documented](https://docs.aws.amazon.com/lambda/latest/dg/with-s3-example.html) and taught in most AWS-related courses.

![S3 Notification with AWS Lambda](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/rjzatofr09cmfaee0sq1.png)

_AWS CDK_ makes setting this architecture a breeze.

```ts
const machine = new sfn.StateMachine(this, "machine", {
  definition: new sfn.Pass(this, "pass")
});

const bucket = new s3.Bucket(this, "bucket", {
  removalPolicy: cdk.RemovalPolicy.DESTROY
});

/**
 * Handler is responsible for forwarding the S3 event to the state machine.
 */
const notificationHandler = new nodeJsLambda.NodejsFunction(
  this,
  "notificationHandler",
  {
    handler: "handler",
    entry: join(__dirname, "notification-handler.ts"),
    environment: {
      STATE_MACHINE_ARN: machine.stateMachineArn
    }
  }
);
notificationHandler.addToRolePolicy(
  new iam.PolicyStatement({
    actions: ["states:StartExecution"],
    resources: [machine.stateMachineArn],
    effect: iam.Effect.ALLOW
  })
);

bucket.addEventNotification(
  s3.EventType.OBJECT_CREATED_PUT,
  new s3Notifications.LambdaDestination(notificationHandler),
  { prefix: "uploads/" }
);
```

Other prolific authors from the AWS community have written many articles regarding this pattern, so I will not repeat their words here. However, I will remind you about some limitations and gotchas that you might encounter while working with S3 Notifications.

- Make sure you are not going to create an **infinite event loop**. This often happens whenever the target _AWS Lambda_ function causes an S3 event that triggers it again.

- Keep in mind the **limitations of the S3 Notifications event filtering rules**. Here is a [link to the relevant AWS documentation page](https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-how-to-filtering.html).

- Keep in mind the **delivery guarantees**. This point is worth considering regardless of the service. I've read countless _StackOverflow_ questions asking about duplicate events. Consult [this documentation piece](https://docs.aws.amazon.com/AmazonS3/latest/userguide/NotificationHowTo.html) regarding event delivery guarantees.

### Applying the storage-first pattern

In the AWS Serverless space, more and more engineers are talking about _storage-first_ pattern where the data/event is first persisted and then reacted upon. This "store first and react later" approach makes it easy to handle heavy loads (through event batching) and ensures resiliency.

![S3 Notifications storage first](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vpepyqn3bvcfg22ma0i6.png)

Luckily for us, _Amazon S3_ integrates with _Amazon SQS_, making augmenting our current infrastructure a breeze.

```ts
const s3EventsQueue = ...

const notificationHandler = ...

notificationHandler.addEventSource(
  new lambdaEventSources.SqsEventSource(s3EventsQueue, {
    /**
     * Increase this number as you please.
     */
    batchSize: 1,
    enabled: true
  })
);

bucket.addEventNotification(
  s3.EventType.OBJECT_CREATED_PUT,
  // Previously we were targeting the Lambda directly.
  new s3Notifications.SqsDestination(s3EventsQueue),
  { prefix: "uploads/" }
);
```

It is **essential to handle errors correctly in the poller AWS Lambda function**. Remember that **if your Lambda function throws an error, the whole batch of messages will be re-queued**. I have yet to discover a better way of handling this than manually deleting messages when they are processed.

If you are interested in the implementation, [here is mine](https://github.com/WojciechMatuszewski/s3-to-sfn-pattern/blob/main/lib/storage-first-handler.ts#L8).

## Using AWS CloudTrail and Amazon EventBridge

For the longest time, I was confident that the only way of integrating _AWS S3_ with _Amazon StepFunctions_ was through S3 Notifications utilizing an intermediate _AWS Lambda_ function.

I rejoiced when exploring _AWS CloudTrail_ where I learned about **EventSelectors** and **AdvancedEventSelectors**. These would allow me to write _Amazon S3_ and _Amazon EventBridge_ together to achieve a fully "lambda-less" architecture.

![S3 Notification with AWS CloudTrail](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/heikmjmv0veu8sm8nx8f.png)

The following sections touch on the event selectors rather than the _AWS CloudTrail_ service itself. It would be best if you were familiar with _AWS CloudTrail_ before proceeding.

### What are AWS CloudTrail event selectors

_AWS CloudTrail_ event selectors are nothing more than filtering rules one might apply to configure which events get captured by a given trail. It is vital to note that **the event selectors only apply to the data events. To my best knowledge, you cannot filter the management events (only turn them off and on)**.

![CloudTrail event selector](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jhdgmnj38ab59wuc0uyk.png)

**Update 24.11.2021**

As my colleague @rosswilliams pointed out, before embarking on the _CloudTrail_ implementation journey, **keep in mind that _AWS CloudTrail_ might deliver your events with up to 15 minutes delay**. From my personal experience, the delay is relatively insignificant, and events are delivered instantly. Please keep this fact in mind.

For reference, head to the [_AWS CloudTrail_ FAQ](https://aws.amazon.com/cloudtrail/faqs/), mainly the "Event payload, Timelines, and Delivery Frequency" section.

### Utilizing the EventSelectors

The "regular" _AWS CloudTrail_ event selectors allow you to track read and/or write events originating from a specific bucket. The list of all possible events is [rather extensive](https://docs.aws.amazon.com/AmazonS3/latest/userguide/cloudtrail-logging-s3-info.html).

The following is an example of how one might create an _AWS CloudTrail_ event selector for an _Amazon S3_ bucket using _AWS CDK_.

```ts
const bucket = new s3.Bucket(this, "bucket", {
  removalPolicy: cdk.RemovalPolicy.DESTROY
});

const trail = new cloudTrail.Trail(this, "trail", {
  includeGlobalServiceEvents: false,
  isMultiRegionTrail: false
});

trail.addS3EventSelector([{ bucket, objectPrefix: "uploads" }], {
  includeManagementEvents: false,
  readWriteType: cloudTrail.ReadWriteType.WRITE_ONLY
});
```

Just like in the case of S3 Notifications, we can specify the `prefix` parameter. Unlike the S3 Notifications, we cannot specify the `suffix` parameter.

Combined with the broad scope of `WRITE_ONLY` events, this setup forces us to do much of our filtering on the _AWS EventBridge_ side of things. Luckily for us, the service we are sending events to is designed with rich filtering capabilities in mind.

The following code creates an _AWS EventBridge_ rule matching _Amazon S3_ events.

```ts
const bucket = ...

const machine = ...

const assetUploadedRule = new events.Rule(this, "assetUploadedRule", {
  enabled: true,
  eventPattern: {
    source: ["aws.s3"],
    detail: {
      eventSource: ["s3.amazonaws.com"],
      eventName: ["PutObject"],
      requestParameters: {
        bucketName: [bucket.bucketName]
      }
    }
  },
  targets: [new eventsTargets.SfnStateMachine(machine)]
});
```

And that's it! All the `PutObject` events originating from the bucket you have applied the EventSelector on will be pushed to _Amazon EventBridge_ and cause state machine invocation.

### Utilizing the AdvancedEventSelectors

The "advanced" _AWS CloudTrail_ event selectors offer more sophisticated event filtering capabilities. **This is a good thing because every data event incurs _AWS CloudTrail_ cost**. Utilizing the advanced event selectors is a [excellent way to optimize your CloudTrail costs](https://aws.amazon.com/blogs/mt/optimize-aws-cloudtrail-costs-using-advanced-event-selectors/).

The **bad news** is that the **advanced event selectors are not supported via [_AWS CloudFormation_](https://aws.amazon.com/cloudformation/)**, making the _AWS CDK_ code a bit more involved.

The following code creates `AdvancedEventSelector` using [_custom resources_](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/template-custom-resources.html).

```ts
const trail = new cloudTrail.Trail(this, "trail", {
  includeGlobalServiceEvents: false,
  isMultiRegionTrail: false
});
const cfnTrail = trail.node.defaultChild as cloudTrail.CfnTrail;

const advancedEventSelectorResource = new customResources.AwsCustomResource(
  this,
  "advancedEventSelectorResource",
  {
    onCreate: {
      action: "putEventSelectors",
      service: "CloudTrail",
      parameters: {
        AdvancedEventSelectors: [
          {
            FieldSelectors: [
              { Field: "eventCategory", Equals: ["Data"] },
              { Field: "resources.type", Equals: ["AWS::S3::Object"] },
              { Field: "eventName", Equals: ["PutObject"] },
              {
                Field: "resources.ARN",
                StartsWith: [bucket.arnForObjects("uploads/")]
              }
            ],
            Name: "Listens to the PutObject only"
          }
        ],
        TrailName: cfnTrail.ref
      },
      physicalResourceId:
        customResources.PhysicalResourceId.fromResponse("TrailARN")
    },
    onDelete: {
      action: "putEventSelectors",
      service: "CloudTrail",
      parameters: {
        AdvancedSelectors: [],
        TrailName: cfnTrail.ref
      }
    },
    policy: customResources.AwsCustomResourcePolicy.fromSdkCalls({
      resources: [trail.trailArn]
    })
  }
);
```

The _AWS EventBridge_ rule declaration has not changed.

```ts
const bucket = ...

const machine = ...

const assetUploadedRule = new events.Rule(this, "assetUploadedRule", {
  enabled: true,
  eventPattern: {
    source: ["aws.s3"],
    detail: {
      eventSource: ["s3.amazonaws.com"],
      eventName: ["PutObject"],
      requestParameters: {
        bucketName: [bucket.bucketName]
      }
    }
  },
  targets: [new eventsTargets.SfnStateMachine(machine)]
});
```

If I were to choose between the "regular" and "advanced" event selectors, I would personally be inclined to pick the latter - mainly due to cost savings.

## Closing words

There you have it. These were two ways one might integrate _Amazon S3_ events with _AWS StepFunctions_ state machine. Of course, in the world of AWS, there are probably much more ways to do this. I wanted to point you towards the ones I find very useful.

As always, thank you for your valuable time.

You can find me on Twitter - [@wm_matuszewski](https://twitter.com/wm_matuszewski)