AWS Kinesis を使って S3 にちょっとずつ書き込む。

AWS Kinesis というのは、ちょっとずつデータを集めるための仕組みらしいです。これで試しに S3 にちょっとずつデータを書き込んでみます。

## コツ

* 複数のサービスを組み合わせるので権限の設定がややこしい
* バッファが貯まるまで S3 からデータを取り出せない

## Kinesis Data Stream でデータを集める

データを集める stream を作るには [create-stream](https://docs.aws.amazon.com/cli/latest/reference/kinesis/create-stream.html) コマンドを使います。パラメータは名前と shard 数です。

shared 数一つで 1MiB/秒 のデータ吸い取りと 2MiB/秒のデータ吐き出しが出来ます。([データストリームの作成および更新](https://docs.aws.amazon.com/ja_jp/streams/latest/dev/amazon-kinesis-streams.html))

コマンド例:

    aws kinesis create-stream --stream-name hoge --shard-count 1

それでは作ったストリームに [put-record](https://docs.aws.amazon.com/cli/latest/reference/kinesis/put-record.html) コマンドで書き込みます。`--partition-key` を使ってどの shard に保存するか決めるとあるので、適当な文字列を指定します。

```shell
$ aws kinesis put-record --stream-name hoge --partition-key hoge-key --data HelloWorld
{
    "ShardId": "shardId-000000000000", 
    "SequenceNumber": "49598759924773225241889582386124629810459480330394927106"
}
$ aws kinesis put-record --stream-name hoge --partition-key hoge-key --data ByeWorld
{
    "ShardId": "shardId-000000000000", 
    "SequenceNumber": "49598759924773225241889582386125838736279126639248408578"
}

```

書き込んだデータを読み込みには get-shard-iterator でイテレータを取得します。TRIM_HORIZON といのは古い順に読むということらしい。

```shell
$ aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name hoge
{
    "ShardIterator": "AAAAAAAAAAFECIoPc6qtHLXUSuEyUtRKh0o9iruhc/8JSu+6acPa9pZxTuPLm1Dcn0j7926YtoCBaE2kGXT8G6GRGPOQPADp5uHX9A9pPORDUe5xMbrw8XZmxyHp/GJqy7ihqykXkh+QJXkAgLqpfXM6BhoJ23aQCogbYs5sBCadzaDdt3onf8xWitFkRGX87PuqS5shMnUJZ1EF7n70/ydF41lan9L1"
}
```

イテレータを指定して get-records でデータを取得します。

```shell
$ aws kinesis get-records --shard-iterator AAAAAAAAAAFECIoPc6qtHLXUSuEyUtRKh0o9iruhc/8JSu+6acPa9pZxTuPLm1Dcn0j7926YtoCBaE2kGXT8G6GRGPOQPADp5uHX9A9pPORDUe5xMbrw8XZmxyHp/GJqy7ihqykXkh+QJXkAgLqpfXM6BhoJ23aQCogbYs5sBCadzaDdt3onf8xWitFkRGX87PuqS5shMnUJZ1EF7n70/ydF41lan9L1
{
    "Records": [
        {
            "Data": "SGVsbG9Xb3JsZA==", 
            "PartitionKey": "hoge-key", 
            "ApproximateArrivalTimestamp": 1566379302.427, 
            "SequenceNumber": "49598759924773225241889582386124629810459480330394927106"
        }, 
        {
            "Data": "QnllV29ybGQ=", 
            "PartitionKey": "hoge-key", 
            "ApproximateArrivalTimestamp": 1566379763.774, 
            "SequenceNumber": "49598759924773225241889582386125838736279126639248408578"
        }
    ], 
    "NextShardIterator": "AAAAAAAAAAGlRLCOBUpJHWWqIYeF5K86L9KZ9js+Jo/l40VQdYlp3JGyHOKRZSXuAgbhgN5pEIEmHf/aN3a5dpH+LiI/NRb16Ta7znugIV15+ZO8eT8wlhiO9cICKFbqkqoq0E2DDK7vUf2FgNUP4XVkaV6WM69HvxWXDKbIfdady/ZMkNmCfqGEBMABw6dWgueFCpPX3OD6aYo/MLLwJs4alv09Hku3", 
    "MillisBehindLatest": 0
}
```

内容は Base64 されています。`put-record` には生データを指定するのに、`get-records` では Base64 された物が出てくるという仕様になっています。

```shell
$ echo SGVsbG9Xb3JsZA== | base64 -D -
HelloWorld
```

後始末

```shell
aws kinesis delete-stream --stream-name hoge
```

## Kinesis Data Firehose で S3 にデータを集める

次に、Amazon Kinesis Data Firehose で S3 に書き込みます。Firehose というのは、Kinesis Data Stream で集めたデータを簡単に S3 などに書き込む仕組みです。複数のサービスを関連付けるのはややこしいので Terraform で作ります。Firehose 自体の設定は簡単ですが、必要な権限が沢山あります。

Firehose に付与する権限の設定は [Amazon Kinesis Data Firehose によるアクセスの制御](https://docs.aws.amazon.com/ja_jp/firehose/latest/dev/controlling-access.html) に書いてありますが基本はこんな感じです。

* Firehose 用の IAM Role を作る。
* AssumeRole で自分のアカウントの firehose だけがこの IAM Role を使えるようにする。
* この IAM Role に、Stream からデータを読み込む権限を与える。
* この IAM Role に、S3 にデータを書き込む権限を与える。

Terraform ファイルの例:

```js
// Kinesis Data Stream を作ります。
resource "aws_kinesis_stream" "stream" {
  name        = "hoge"
  shard_count = 1
}

// S3 を作ります。
resource "aws_s3_bucket" "bucket" {
  bucket = "kinesis-bucket-hoge"
}

// Firehose 用の IAM Role を作ります。自分のアカウントの Firehose だけがこの Role を使えるようにします。
data "aws_caller_identity" "current" {}

resource "aws_iam_role" "role" {
  name = "firehose-role-hoge"

  assume_role_policy = <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
        "Effect": "Allow",
        "Action": "sts:AssumeRole",
        "Principal": {
            "Service": "firehose.amazonaws.com"
        },
        "Condition": {
            "StringEquals": {
                "sts:ExternalId": "${data.aws_caller_identity.current.account_id}"
            }
        }
    }
  ]
}
EOF
}

// Stream からデータを取り出して S3 に書き出すための権限を作ります。

resource "aws_iam_policy" "policy" {
  name = "firehose-policy"

  policy = <<EOF
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords"
            ],
            "Resource": "${aws_kinesis_stream.stream.arn}"
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:PutObject"
            ],
            "Resource": [
                "${aws_s3_bucket.bucket.arn}",
                "${aws_s3_bucket.bucket.arn}/*"
            ]
        }
    ]
}
EOF
}

// 権限を Firehose 用の Role に追加します。
resource "aws_iam_role_policy_attachment" "attachment" {
  role       = "${aws_iam_role.role.name}"
  policy_arn = "${aws_iam_policy.policy.arn}"
}

// Firehose を追加して、Kinesis Stream と S3 を繋げます。

resource "aws_kinesis_firehose_delivery_stream" "hose" {
  name        = "kinesis-firehose-hoge"
  destination = "s3"

  kinesis_source_configuration {
    kinesis_stream_arn = "${aws_kinesis_stream.stream.arn}"
    role_arn           = "${aws_iam_role.role.arn}"
  }

  s3_configuration {
    role_arn        = "${aws_iam_role.role.arn}"
    bucket_arn      = "${aws_s3_bucket.bucket.arn}"
    // テスト用にバッファを 60 秒にします。これで put-record したら 60 秒後に S3 で確認出来ます。
    buffer_interval = 60
  }
}
```

テスト

```shell
terraform init
terraform apply
aws kinesis put-record --stream-name hoge --partition-key hoge-key --data HelloWorld
aws kinesis put-record --stream-name hoge --partition-key hoge-key --data ByeWorld
などなど
```

一分後 AWS Console で S3 に書き込まれたことを確認

あと始末

```shell
terraform destroy
```

## 参考

* [Amazon Kinesis Data Streams とは?](https://docs.aws.amazon.com/ja_jp/streams/latest/dev/introduction.html)
* [[AWS]kinesisまとめ](https://qiita.com/yShig/items/500d4139efed91bc432d)