FluentBit × Kinesis × S3 でログ保存基盤を作る ~Kinesis&S3実装編~

# はじめに
以前書いた[FluentBit × Kinesis × S3 でログ保存基盤を作る ~概要編~](https://qiita.com/Canon11/items/eb5be0a0ffcc1f5c8552)の続きです。

今回はAWS側の実装である、Kinesis Firehose Delivery StreamからS3にデータを流す基盤を作ります。

# リソース作成
今回はCloudFormationを使ってリソースを作成しました。

## 必要なリソース
* S3バケット
* Kinesis Firehose Delivery Stream用のIAMロール
  * S3に書き込みをできる必要がある
  * 必要なIAMポリシーは [公式ページ](https://docs.aws.amazon.com/ja_jp/firehose/latest/dev/controlling-access.html) を参照
* Kinesis Firehose Delivery Stream

※ Kinesis Firehose Delivery Stream が長いので、以降はFirehoseと呼びます。

## テンプレート
上記要件を満たすためのテンプレートが下記です。
バケット名はきっと被ってしまうので、適宜変えるようにするのが良いと思います。
（サンプル用として書いてるのでリソース名の適当さはお許しください...)

```firehose_to_s3.yaml
AWSTemplateFormatVersion: 2010-09-09
Resources:
  LogBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: log-bucket-stream-sample0021
      AccessControl: Private
  LogToS3Stream:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: log-to-s3-stream
      DeliveryStreamType: DirectPut
      ExtendedS3DestinationConfiguration:
        BucketARN: !GetAtt LogBucket.Arn
        RoleARN: !GetAtt FirehoseToS3IamRole.Arn
        BufferingHints:
          IntervalInSeconds: 60
          SizeInMBs: 50
        CompressionFormat: GZIP
  FirehoseToS3IamRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: firehose-to-s3-iam-role
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - firehose.amazonaws.com
            Action:
              - sts:AssumeRole
      ManagedPolicyArns:
        - !Ref FirehoseToS3DeliveryPolicy
  FirehoseToS3DeliveryPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          -
            Sid: AllowFirehoseToS3Delivery
            Effect: Allow
            Action:
              - s3:AbortMultipartUpload
              - s3:GetBucketLocation
              - s3:GetObject
              - s3:ListBucket
              - s3:ListBucketMultipartUploads
              - s3:PutObject
              - kinesis:DescribeStream
              - kinesis:GetShardIterator
              - kinesis:GetRecords
              - kinesis:ListShards
              - kms:Decrypt
              - kms:GenerateDataKey
              - logs:PutLogEvents
              - lambda:InvokeFunction
              - lambda:GetFunctionConfiguration
            Resource: "*"
```

こちらをCloudFormationで流し込みます。
変数は特に埋め込んでいないので、設定は必要ありません。

## リソース確認
### S3
バケットが作成されています。
ここがFirehoseから送られてくるデータが書き込まれる場所になります。
![スクリーンショット 2020-04-29 23.06.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/e14e7221-a926-1d9f-3915-4576632953c8.png)

### Firehose
こんな感じで、リソースが作成されました。
![スクリーンショット 2020-04-29 22.59.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/0b185675-6ed8-46c9-9a21-8a2e4672e747.png)


キャッチしたデータの書き込み先のS3もちゃんと指定されています。
ちなみにここのBuffer conditionsは、Firehoseがデータをキャッチしてから、S3に送るまでの間隔値などのバッファ設定です。
![スクリーンショット 2020-04-29 22.59.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/fa70d077-fe6e-3b87-58d7-102e1189b7cc.png)

# 動かしてみる
## データを送る
Firehoseのコンソール画面にはテストデータを送るようのボタンが設置されているので、これを使ってFirehoseにデータを送ってみます。
![スクリーンショット 2020-04-29 22.59.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/1180c6bc-78d2-332c-4c7c-127eb2a91ccc.png)

「Sending demo data」をクリックすると、データが送られ始めます。
1回押すと1件データが送られるのではなく、何件もデータが連続で送られるみたいです。
そのため、**「Stop sending demo data」をクリックしないとデータの送信が止まらない**のでご注意を！

ちなみに
`{"ticker_symbol":"QXZ", "sector":"HEALTHCARE", "change":-0.05, "price":84.51}`
が何度も送られる形です。JSON文字列ですね。

## Firehoseのモニタリングを確認してみる
「Monitoring」をクリックすると、Firehoseが受信したデータ量、S3に送信したデータ量などのメトリクスを見ることができます。
しばらく待って画面の更新をしてみると、Firehoseがデータをストリームしていることが分かるメトリクスデータが現れます。
![スクリーンショット 2020-04-29 23.08.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/07befe54-5d41-f011-4ba4-62548fd092b8.png)

今回は60個データをFirehoseに流し、それをキャッチしたFirehoseが60個のデータをS3に書き込みました。

## バケットを確認する
S3をみてみると、日時で区切られたディレクトリが自動生成され、その中にFirehoseがストリーミングで書き込んだデータのファイルが配置されています。
![スクリーンショット 2020-04-29 23.16.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/adfccdac-97ee-d505-1bee-a6422f5ee772.png)


gzファイルをダウンロードし、lessコマンド等で確認。
すると、
![スクリーンショット 2020-04-29 23.18.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/154121/f6625206-a1dd-aafe-c006-9fa1e567e357.png)

Firehoseから送られてきたデータを確認することができました。

# 最後に
今回はFirehoseからS3にデータをストリームする基盤を構築しました。
次回はコンテナ側の実装となる、FluentBitからFirehoseにデータを送る部分についてまとめます。