iTranslated by AI

The content below is an AI-generated translation. This is an experimental feature, and may contain errors. View original article
🐷

[CDK] Buffering SNS Messages with Kinesis Data Firehose and Outputting to S3

に公開
2022/11/13
AWS
S3
SNS
AWS CDK
kinesis
tech

As the title suggests, this is a practice note for SNS + Kinesis Firehose + S3.

Environment

  • constructs 10.1.43
  • aws-cdk-lib 2.46.0
  • typescript 4.7.4

Architecture Diagram

True to the title, the configuration involves buffering messages sent from an SNS topic via a Kinesis Firehose delivery stream and outputting them as files to an S3 bucket.

CDK Code

The infrastructure is defined using AWS CDK.

[AWS CDK] Recording logs when S3 output of Amazon Kinesis Data Firehose fails | DevelopersIO

I mainly referred to the article above. The following CDK code is a slightly modified version of it, adding the SNS topic and its subscription settings.

import { Construct } from "constructs"
import * as cdk from "aws-cdk-lib"

export class KinesisFirehoseStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props)

    const region = cdk.Stack.of(this).region
    const accountId = cdk.Stack.of(this).account

    // S3 Bucket
    const streamDestinationBucket = new cdk.aws_s3.Bucket(
      this,
      "streamDestinationBucket",
      {
        bucketName: `stream-destination-bucket-${region}-${accountId}`,
        removalPolicy: cdk.RemovalPolicy.DESTROY,
        blockPublicAccess: cdk.aws_s3.BlockPublicAccess.BLOCK_ALL,
        encryption: cdk.aws_s3.BucketEncryption.S3_MANAGED,
      }
    )

    // Log Group for Kinesis Firehose delivery stream
    const deliveryStreamFailLogGroup = new cdk.aws_logs.LogGroup(
      this,
      "deliveryStreamFailLogGroup",
      {
        logGroupName: `/aws/kinesisfirehose/sample-stream-fail-log`,
      }
    )

    // Log Stream for Kinesis Firehose delivery stream
    const deliveryStreamLogStream = new cdk.aws_logs.LogStream(
      this,
      "deliveryStreamLogStream",
      {
        logGroup: deliveryStreamFailLogGroup,
        logStreamName: "logs",
      }
    )

    // Role for Kinesis Firehose delivery stream
    const deliveryStreamRole = new cdk.aws_iam.Role(
      this,
      "deliveryStreamRole",
      {
        assumedBy: new cdk.aws_iam.ServicePrincipal("firehose.amazonaws.com"),
      }
    )
    deliveryStreamRole.addToPolicy(
      new cdk.aws_iam.PolicyStatement({
        actions: [
          "kinesis:DescribeStream",
          "kinesis:GetShardIterator",
          "kinesis:GetRecords",
        ],
        effect: cdk.aws_iam.Effect.ALLOW,
        resources: [`arn:aws:kinesis:${region}:${accountId}:stream/*`],
      })
    )
    deliveryStreamRole.addToPolicy(
      new cdk.aws_iam.PolicyStatement({
        actions: [
          "s3:AbortMultipartUpload",
          "s3:GetBucketLocation",
          "s3:GetObject",
          "s3:ListBucket",
          "s3:ListBucketMultipartUploads",
          "s3:PutObject",
        ],
        effect: cdk.aws_iam.Effect.ALLOW,
        resources: [
          streamDestinationBucket.bucketArn,
          `${streamDestinationBucket.bucketArn}/*`,
        ],
      })
    )
    deliveryStreamRole.addToPolicy(
      new cdk.aws_iam.PolicyStatement({
        actions: ["logs:PutLogEvents"],
        effect: cdk.aws_iam.Effect.ALLOW,
        resources: [
          `arn:aws:logs:${region}:${accountId}:log-group:/aws/kinesisfirehose/*`,
        ],
      })
    )

    // Kinesis Firehose delivery stream
    const deliveryStream = new cdk.aws_kinesisfirehose.CfnDeliveryStream(
      this,
      "deliveryStream",
      {
        deliveryStreamName: "deliveryStream",
        deliveryStreamType: "DirectPut",
        s3DestinationConfiguration: {
          bucketArn: streamDestinationBucket.bucketArn,
          roleArn: deliveryStreamRole.roleArn,
          // Logging configuration for S3 output failure
          cloudWatchLoggingOptions: {
            enabled: true,
            logGroupName: deliveryStreamFailLogGroup.logGroupName,
            logStreamName: "logs",
          },
          compressionFormat: "GZIP",
          prefix: "/",
          errorOutputPrefix: "errorOutput",
          bufferingHints: {
            intervalInSeconds: 60,
          },
        },
      }
    )
    deliveryStream.addDependsOn(
      streamDestinationBucket.node.defaultChild as cdk.CfnResource
    )
    deliveryStream.addDependsOn(
      deliveryStreamFailLogGroup.node.defaultChild as cdk.CfnResource
    )
    deliveryStream.addDependsOn(
      deliveryStreamLogStream.node.defaultChild as cdk.CfnResource
    )

    // SNS Topic
    const topic = new cdk.aws_sns.CfnTopic(this, "topic", {})

    // Role for SNS subscription
    const subscriptionRole = new cdk.aws_iam.Role(this, "subscriptionRole", {
      assumedBy: new cdk.aws_iam.ServicePrincipal("sns.amazonaws.com"),
    })
    subscriptionRole.addToPolicy(
      new cdk.aws_iam.PolicyStatement({
        actions: ["firehose:PutRecord"],
        effect: cdk.aws_iam.Effect.ALLOW,
        resources: [deliveryStream.attrArn],
      })
    )

    // SNS Subscription
    const subscription = new cdk.aws_sns.CfnSubscription(this, "subscription", {
      topicArn: topic.attrTopicArn,
      protocol: "firehose",
      endpoint: deliveryStream.attrArn,
      subscriptionRoleArn: subscriptionRole.roleArn,
    })
    subscription.addDependsOn(topic)
    subscription.addDependsOn(deliveryStream)
    subscription.addDependsOn(
      subscriptionRole.node.defaultChild as cdk.CfnResource
    )
  }
}

Verification

I published the following two messages to the SNS topic from the AWS Management Console.

  • 1st Message
    • Subject: test1
    • Message: test1
  • 2nd Message
    • Subject: test2
    • Message: test2

About a minute later, a file named {year}/{month}/{day}/{number}/xxxxxxxxxxxxxxxxx.gz was created in the S3 bucket.

When I tried to download and extract it from the S3 Management Console, I encountered an error: Unable to expand xxxxxxx.gz (operating on M1 Mac's Finder). This seems to be due to the following circumstances.

How to fix the problem where gzip files saved in S3 with Firehose cannot be extracted normally | DevelopersIO

So, in my case, I was able to verify the content by opening the .gz file directly in VSCode.

{"Type":"Notification","MessageId":"xxxxxxxxxxxxxxxxxxxxxxx","TopicArn":"arn:aws:sns:ap-northeast-1:xxxxxxxxxxxx:KinesisFirehoseStack-topic-xxTXyoP0G1qA","Subject":"test1","Message":"test1","Timestamp":"2022-11-12T10:23:43.286Z","UnsubscribeURL":"https://sns.ap-northeast-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:ap-northeast-1:xxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxxx"}
{"Type":"Notification","MessageId":"xxxxxxxxxxxxxxxxxxxxxxx","TopicArn":"arn:aws:sns:ap-northeast-1:xxxxxxxxxxxx:KinesisFirehoseStack-topic-xxTXyoP0G1qA","Subject":"test2","Message":"test2","Timestamp":"2022-11-12T10:23:49.472Z","UnsubscribeURL":"https://sns.ap-northeast-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:ap-northeast-1:xxxxxxxxxxxx:xxxxxxxxxxxxxxxxxxxxxxx"}

The SNS message JSON was output line by line like this. It results in a JSON Lines format.

Note that delivery data duplication is possible

Unless SNS is set to FIFO, it uses best-effort deduplication, and this point needs attention.

Also, upon investigation, it seems that Kinesis Firehose may also experience data duplication during delivery.

FAQs - Amazon Kinesis Data Firehose | AWS

Amazon Kinesis Data Firehose uses at-least-once semantics for data delivery. In rare cases, such as when a data delivery request attempt times out and Firehose retries the delivery, a previous request might still be executed, resulting in duplicates.

It would be good to establish a policy in advance, such as allowing duplicates as a system specification, maintaining state in a database for application-side deduplication checks, or making the processing idempotent.

Reference

Discussion