Configuring Record Separators for Kinesis Firehose in AWS IoT Core

Takahiro Iwasa

5/13/2020

3 min read

CloudFormation Firehose IoT Kinesis

This note describes how to configure Kinesis Firehose record separator in IoT Core topic rules.

Building

The Separator: |+ <NEW_LINE> configuration on lines 13-14 are important.

1
AWSTemplateFormatVersion: "2010-09-09"
2

3
Resources:
4
  TopicRule:
5
    Type: AWS::IoT::TopicRule
6
    Properties:
7
      RuleName: topic_rule_firehose_separator_test
8
      TopicRulePayload:
9
        Actions:
10
          - Firehose:
11
              DeliveryStreamName: !Ref Firehose
12
              RoleArn: !GetAtt IamTopicRule.Arn
13
              Separator: |+
14

15
        AwsIotSqlVersion: 2016-03-23
16
        RuleDisabled: false
17
        Sql: !Sub
18
          SELECT * FROM 'topic_rule_firehose_separator_test'
19

20
  Firehose:
21
    Type: AWS::KinesisFirehose::DeliveryStream
22
    Properties:
23
      DeliveryStreamName: topic-rule-firehose-separator-test
24
      DeliveryStreamType: DirectPut
25
      S3DestinationConfiguration:
26
        BucketARN: !GetAtt S3.Arn
27
        BufferingHints:
28
          IntervalInSeconds: 60
29
          SizeInMBs: 5
30
        CompressionFormat: GZIP
31
        ErrorOutputPrefix: "error/!{firehose:error-output-type}/!{timestamp:'year='yyyy'/month='MM'/day='dd'/hour='HH}/"
32
        Prefix: "success/!{timestamp:'year='yyyy'/month='MM'/day='dd'/hour='HH}/"
33
        RoleARN: !GetAtt IamFirehose.Arn
34

35
  S3:
36
    Type: AWS::S3::Bucket
37
    Properties:
38
      BucketEncryption:
39
        ServerSideEncryptionConfiguration:
40
          - ServerSideEncryptionByDefault:
41
              SSEAlgorithm: AES256
42
      BucketName: topic-rule-firehose-separator-test
43
      PublicAccessBlockConfiguration:
44
        BlockPublicAcls: TRUE
45
        BlockPublicPolicy: TRUE
46
        IgnorePublicAcls: TRUE
47
        RestrictPublicBuckets: TRUE
48

49
  IamTopicRule:
50
    Type: AWS::IAM::Role
51
    Properties:
52
      AssumeRolePolicyDocument:
53
        Version: '2012-10-17'
54
        Statement:
55
          - Effect: Allow
56
            Principal:
57
              Service: iot.amazonaws.com
58
            Action: sts:AssumeRole
59
      Policies:
60
        - PolicyDocument:
61
            Version: '2012-10-17'
62
            Statement:
63
              - Effect: Allow
64
                Action: firehose:PutRecord
65
                Resource:
66
                  - !GetAtt Firehose.Arn
67
          PolicyName: policy
68
      RoleName: iam-topic-rule
69

70
  IamFirehose:
71
    Type: AWS::IAM::Role
72
    Properties:
73
      AssumeRolePolicyDocument:
74
        Version: '2012-10-17'
75
        Statement:
76
          - Effect: Allow
77
            Principal:
78
              Service: firehose.amazonaws.com
79
            Action: sts:AssumeRole
80
            Condition:
81
              StringEquals:
82
                sts:ExternalId: !Ref AWS::AccountId
83
      Policies:
84
        - PolicyDocument:
85
            Version: '2012-10-17'
86
            Statement:
87
              - Effect: Allow
88
                Action:
89
                  - glue:GetTable
90
                  - glue:GetTableVersion
91
                  - glue:GetTableVersions
92
                Resource: "*"
93
              - Effect: Allow
94
                Action:
95
                  - s3:AbortMultipartUpload
96
                  - s3:GetBucketLocation
97
                  - s3:GetObject
98
                  - s3:ListBucket
99
                  - s3:ListBucketMultipartUploads
100
                  - s3:PutObject
101
                Resource:
102
                  - !GetAtt S3.Arn
103
                  - Fn::Sub:
104
                      - ${arn}/*
105
                      - {arn: !GetAtt S3.Arn}
106
              - Effect: Allow
107
                Action:
108
                  - lambda:InvokeFunction
109
                  - lambda:GetFunctionConfiguration
110
                Resource: !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:%FIREHOSE_DEFAULT_FUNCTION%:%FIREHOSE_DEFAULT_VERSION%
111
              - Effect: Allow
112
                Action:
113
                  - logs:PutLogEvents
114
                Resource:
115
                  - !Sub arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesisfirehose/topic-rule-firehose-separator-test
116
              - Effect: Allow
117
                Action:
118
                  - kinesis:DescribeStream
119
                  - kinesis:GetShardIterator
120
                  - kinesis:GetRecords
121
                Resource: !Sub arn:aws:kinesis:${AWS::Region}:${AWS::AccountId}:stream/%FIREHOSE_STREAM_NAME%
122
              - Effect: Allow
123
                Action:
124
                  - kms:Decrypt
125
                Resource:
126
                  - !Sub arn:aws:kms:${AWS::Region}:${AWS::AccountId}:key/%SSE_KEY_ID%
127
                Condition:
128
                  StringEquals:
129
                    kms:ViaService: kinesis.%REGION_NAME%.amazonaws.com
130
                  StringLike:
131
                    kms:EncryptionContext:aws:kinesis:arn: !Sub arn:aws:kinesis:%REGION_NAME%:${AWS::AccountId}:stream/%FIREHOSE_STREAM_NAME%
132
          PolicyName: policy
133
      RoleName: iam-firehose

Deploy the CloudFormation stack with the following command:

aws cloudformation deploy \
  --template template.yaml \
  --stack-name topic-rule-firehose-separator-test \
  --capabilities CAPABILITY_NAMED_IAM

Run the following command to verify the topic rule configuration:

aws iot get-topic-rule \
  --rule-name topic_rule_firehose_separator_test

The separator configuration should appear on line 12 in the output.

1
{
2
  "ruleArn": "arn:aws:iot:<YOUR_REGION>:<YOUR_ACCOUNT_ID>:rule/topic_rule_firehose_separator_test",
3
  "rule": {
4
    "ruleName": "topic_rule_firehose_separator_test",
5
    "sql": "SELECT * FROM 'topic_rule_firehose_separator_test'",
6
    "createdAt": "2020-05-13T10:29:18+09:00",
7
    "actions": [
8
      {
9
        "firehose": {
10
          "roleArn": "arn:aws:iam::<YOUR_ACCOUNT_ID>:role/iam-topic-rule",
11
          "deliveryStreamName": "topic-rule-firehose-separator-test",
12
          "separator": "\n"
13
        }
14
      }
15
    ],
16
    "ruleDisabled": false,
17
    "awsIotSqlVersion": "2016-03-23"
18
  }
19
}

Testing

Publish test messages to the topic topic_rule_firehose_separator_test:

aws iot-data publish \
  --topic topic_rule_firehose_separator_test \
  --payload '{"id": 1, "message": "Hello from AWS IoT"}' \
  --cli-binary-format raw-in-base64-out

aws iot-data publish \
  --topic topic_rule_firehose_separator_test \
  --payload '{"id": 2, "message": "Hello from AWS IoT"}' \
  --cli-binary-format raw-in-base64-out

Retrieve and inspect the object from the S3 bucket:

# Check an object.
$ aws s3 ls topic-rule-firehose-separator-test --recursive
2020-05-14 11:30:59         68 success/year=2020/month=05/day=14/hour=02/topic-rule-firehose-separator-test-1-2020-05-14-02-29-57-593d65e5-beb6-47b1-8266-83e869b0cccb.gz

# Download the object.
$ aws s3 cp s3://topic-rule-firehose-separator-test/success/year=2020/month=05/day=14/hour=02/topic-rule-firehose-separator-test-1-2020-05-14-02-29-57-593d65e5-beb6-47b1-8266-83e869b0cccb.gz ./result.gz

The output should contain the two records, separated by the configured separator.

# Check the JSON.
$ gunzip -c result.gz > result.json
$ cat result.json
{"id": 1, "message": "Hello from AWS IoT"}
{"id": 2, "message": "Hello from AWS IoT"}

# Delete the results.
$ rm result.gz result.json
$ aws s3 rm s3://topic-rule-firehose-separator-test/success/year=2020/month=05/day=14/hour=02/topic-rule-firehose-separator-test-1-2020-05-14-02-29-57-593d65e5-beb6-47b1-8266-83e869b0cccb.gz

Cleaning Up

Clean up all the AWS resources provisioned during this example with the following command:

aws cloudformation delete-stack --stack-name topic-rule-firehose-separator-test