Serverless Framework で Kinesis のプロデューサーを作る

[Serverless Framework](https://serverless.com/)で Kinesis を使います。
プロデューサー側を作るのがメインで、コンシューマー側は簡単なログ出力のみです。

「[Serverless Framework で DynamoDB を使う](http://qiita.com/katsuhiko/items/a2594e73108728a22410)」を参考にすれば、DynamoDB へ反映するコンシューマーを作れると思います。

イベントソーシングをどう使っていくかは、サービス特性を考える必要がありますが、イベントソーシングを使うっと決まったときにすぐに構築できるように準備をしておきます。

# ソースリポジトリ

ソースは以下のリポジトリを参照してください。

- Kinesis のプロデューサー
    - https://github.com/katsuhiko/sls-kinesis
- Kinesis のログ出力コンシューマー
    - https://github.com/katsuhiko/sls-log-from-kinesis

# バージョン

| version | 
|:---|
| node 6.9.1 |
| npm 3.10.8 |
| serverless framework 1.1.0 |

Lambda が Node.js V4.3 なので、合わせるべきかもしれません。今はローカルでは動かしていないので合わしていません。

# Kinesis プロデューサー

## todos.js

「[Serverless Framework で DynamoDB を使う](http://qiita.com/katsuhiko/items/a2594e73108728a22410)」の内容を流用しています。

create / update / delete すべて、Kinesis Stream へ putRecord しているのみです。

Kinesis の Stream名にステージを prefix として追加しています。やり方は他にもあるかもしれませんが、各ステージごとに Kinesis Stream を作る仕組みを盛り込まないと実際に利用するときに困ると思います。今回は、ステージを prefix として Stream名につけることで対応しました。

```js
'use strict';

const uuid = require('uuid'),
      moment = require('moment'),
      streamName = `${process.env.STAGE}-todos`;

module.exports.create = (stream, item, callback) => {
  item.id = uuid.v1();
  item.updatedUtc = moment().utc().toISOString();

  const params = {
    StreamName: streamName,
    PartitionKey: uuid.v1(),
    Data: JSON.stringify({ method: 'CREATE', item: item })
  };

  return stream.putRecord(params, (err, data) => {
    if (err) {
      callback(err);
    } else {
      callback(err, item);
    }
  });
};

module.exports.update = (stream, id, item, callback) => {
  item.id = id;
  item.updatedAt = moment().utc().toISOString();

  const params = {
    StreamName: streamName,
    PartitionKey: uuid.v1(),
    Data: JSON.stringify({ method: 'UPDATE', item: item })
  };

  return stream.putRecord(params, (err, data) => {
    if (err) {
      callback(err);
    } else {
      callback(err, item);
    }
  });
};

module.exports.delete = (stream, id, callback) => {
  const item = { id: id };

  const params = {
    StreamName: streamName,
    PartitionKey: uuid.v1(),
    Data: JSON.stringify({ method: 'DELETE', item: item })
  };

  return stream.putRecord(params, (err, data) => {
    if (err) {
      callback(err);
    } else {
      callback(err, item);
    }
  });
};
```

## handler.js

Lambda Proxy に対応した返却を行っています。`context` ではなく `callback`　　を使っています。しっかりした情報を得ることはできませんでしたが、`callback` を使うのが良いのではないかと思います。

DynamoDB を利用したときとほぼ一緒です。違いは、AWS.DynamoDB.DocumentClient のかわりに AWS.Kinesis を使ってる点です。

```js
'use strict';

const AWS = require('aws-sdk'),
      kinesis = new AWS.Kinesis(),
      env = require('dotenv').config(),
      todos = require('./todos.js');

const createResponse = (statusCode, body) => (
  {
    statusCode,
    headers: {
      'Access-Control-Allow-Origin': '*', // Required for CORS support to work
    },
    body: JSON.stringify(body),
  }
);

module.exports.todosCreate = (event, context, callback) => {
  const item = JSON.parse(event.body);

  todos.create(kinesis, item, (err, result) => {
    if (err) {
      callback(createResponse(500, { message: err.message }));
    } else {
      callback(null, createResponse(201, result));
    }
  });
};

module.exports.todosUpdate = (event, context, callback) => {
  const id = event.pathParameters.id,
        item = JSON.parse(event.body);

  todos.update(kinesis, id, item, (err, result) => {
    if (err) {
      callback(createResponse(500, { message: err.message }));
    } else {
      callback(null, createResponse(200, result));
    }
  });
};

module.exports.todosDelete = (event, context, callback) => {
  const id = event.pathParameters.id;

  todos.delete(kinesis, id, (err, result) => {
    if (err) {
      callback(createResponse(500, { message: err.message }));
    } else {
      callback(null, createResponse(204));
    }
  });
};
```

## serverless.yml

`iamRoleStatements` と `Resources` を使って Kinesis へアクセスする権限と Stream 作成を行っています。

この設定で作られた Kinesis Stream　は `DeletionPolicy: Retain` を指定しているので、`sls remove` を実行しても削除されません。
不要な場合、AWS Console 等を使って別途削除する必要があります。

```yaml
service: sls-kinesis

provider:
  name: aws
  runtime: nodejs4.3
  stage: ${opt:stage, self:custom.defaultStage}
  region: ${opt:region, self:custom.defaultRegion}
  profile: ${self:custom.profiles.${self:provider.stage}}

  iamRoleStatements:
    - Effect: Allow
      Action:
        - kinesis:DescribeStream
        - kinesis:PutRecord
        - kinesis:ListStreams
      Resource: "arn:aws:kinesis:${self:provider.region}:*:stream/${self:provider.stage}-todos"

custom:
  defaultStage: dev
  defaultRegion: ap-northeast-1
  profiles:
    dev: devSls
    prod: prodSls
  writeEnvVars:
    STAGE: ${self:provider.stage}

package:
  exclude:
    - .git/**
    - README.md
    - node_modules/serverless-plugin-write-env-vars/**

plugins:
  - serverless-plugin-write-env-vars

functions:
  todosCreate:
    handler: handler.todosCreate
    events:
      - http:
          path: todos
          method: post
          cors: true

  todosUpdate:
    handler: handler.todosUpdate
    events:
      - http:
          path: todos/{id}
          method: patch
          cors: true

  todosDelete:
    handler: handler.todosDelete
    events:
      - http:
          path: todos/{id}
          method: delete
          cors: true

resources:
  Resources:
    TodosKinesisStream:
      Type: 'AWS::Kinesis::Stream'
      DeletionPolicy: Retain
      Properties:
        Name: "${self:provider.stage}-todos"
        ShardCount: 1
```

# Kinesis コンシューマー

ログ出力（console.log）するのみの簡単なコンシューマーです。

## handler.js

複数のレコードを　Kinesis から取得できるので forEach しています。何件取得するかは serverless.yml で指定できます。

実際に出力される内容を確認すると分かりますが base64 されているので、デコードしています。この後 DynamoDB 等で処理する場合、`JSON.pase()` を使って JSON 形式にする必要があります。

```js
'use strict';

module.exports.todosConsumer = (event, context, callback) => {
  console.log('Event:', JSON.stringify(event));

  event.Records.forEach(function(record) {
    const payload = new Buffer(record.kinesis.data, 'base64').toString('ascii');
    console.log('Decoded payload:', payload);
  });

  callback(null, "message");
};
```

## serverless.yml

`iamRoleStatements` の設定は不要です。`events.stream` を利用すると　Serverless Framework が必要な IAM　Policy を追加してくれます。と思ったのですが、初回のデプロイ時のみ必要になるようです。いずれいらなくなると思います。

Kinesis Stream の arn ですが`Fn::Join`を使って、アカウントIDを埋め込みたかったのですが、エラーが出て動かなかったので環境変数でアカウントIDを渡すようにしています。

- https://github.com/serverless/serverless/issues/2365

この Issue かなっとは思うのですが、なんとも。Kinesis Stream の arn で`Fn::Join`を使う方法がわかれば修正します。ただ、今回に限っては環境変数を serverless.yml に書く方法がわかったので勉強になりました。

`batchSize`が一度の処理で読み込むレコード数です。

`startingPosition`は `TRIM_HORIZON`、`LATEST`を指定することになりますが、どちらが良いか悩ましいところです。
冪等性を考慮して `TRIM_HORIZON` が良いのではないかと思っていますが、考えが変わるかもしれません。

```yaml
service: sls-log-from-kinesis

provider:
  name: aws
  runtime: nodejs4.3
  stage: ${opt:stage, self:custom.defaultStage}
  region: ${opt:region, self:custom.defaultRegion}
  profile: ${self:custom.profiles.${self:provider.stage}}

  iamRoleStatements:
    - Effect: Allow
      Action:
        - kinesis:DescribeStream
        - kinesis:GetRecords
        - kinesis:GetShardIterator
        - kinesis:ListStreams
      Resource: "arn:aws:kinesis:${self:provider.region}:*:stream/${self:provider.stage}-todos"

custom:
  defaultStage: dev
  defaultRegion: ap-northeast-1
  profiles:
    dev: devSls
    prod: prodSls
  writeEnvVars:
    STAGE: ${self:provider.stage}

package:
  exclude:
    - .git/**
    - README.md

functions:
  todosConsumer:
    handler: handler.todosConsumer
    events:
      - stream:
          arn: "arn:aws:kinesis:${self:provider.region}:${env:ACCOUNT_ID}:stream/${self:provider.stage}-todos"
          #arn:
          #  Fn::Join:
          #    - ""
          #    - - "arn:aws:kinesis:${self:provider.region}:"
          #      - Ref: "AWS::AccountId"
          #      - ":stream/${self:provider.stage}-todos"
          batchSize: 1
          startingPosition: TRIM_HORIZON
```

# サンプルを動かしてみる

Serverless Framework のインストール等については「[Serverless Framework で Hello World を作る](http://qiita.com/katsuhiko/items/de971929ffa750ec9b62)」を参照してください。

## サンプルのインストール／デプロイ

Serverless Framework のインストールと AWS への接続ができていたら、以下のコマンドでサンプルをインストール／デプロイできます。

### Kinesis プロデューサーのインストール／デプロイ

```bash
serverless install --url https://github.com/katsuhiko/sls-kinesis
cd sls-kinesis
npm install
serverless deploy -v
```

### Kinesis コンシューマーのインストール／デプロイ

serverless cli を実行するときに環境変数「ACCOUNT_ID」を指定するようにしてください。

```bash
serverless install --url https://github.com/katsuhiko/sls-log-from-kinesis
cd sls-log-from-kinesis
npm install
ACCOUNT_ID=9999 serverless deploy -v
```

## 実行

### Kinesis プロデューサーの実行

curl を使って Kinesis プロデューサーを実行します。「XXXX」部分は各環境に合わして変更してください。

#### Create

```bash
curl -X POST https://XXXX.execute-api.ap-northeast-1.amazonaws.com/dev/todos --data '{ "content" : "Learn Serverless" }'
```

#### Update

```bash
curl -X PATCH https://XXXX.execute-api.ap-northeast-1.amazonaws.com/dev/todos/<id> --data '{ "content" : "Understand Serverless" }'
```

#### Delete

```bash
curl -X DELETE https://XXXX.execute-api.ap-northeast-1.amazonaws.com/dev/todos/<id>
```

### Kinesis コンシューマーの実行

デプロイした時点で、コンシューマーは動いています。ログを参照して本当に動いているかを確認します。

```bash
cd sls-log-from-kinesis
ACCOUNT_ID=9999 serverless logs -f todosConsumer -t
```

## あと片付け

```bash
cd sls-kinesis
serverless remove -v
```

```bash
cd sls-log-from-kinesis
ACCOUNT_ID=9999 sls-log-from-kinesis remove -v
```

上記でも Kinesis Stream が残っているので、AWS Console 等から Kinesis Stream を削除します。


# 感想

serverless.yml の `Resources` は CloudFormation の設定をそのまま記述できる便利さを理解することができました。
Serverless　Framework は aws cli がなくても動く点も良いです。

AWS Console で画面からポチポチリソース作りをしている場合、Serverless Framework を使って yaml でリソース管理をするっという利用方法もあるのではないかと思いました。

Serverless Framework + API Gateway + Lambda + Kinesis + DynamoDB をこんなに簡単に使うことができると、イベントソーシングして CQRS も楽勝です。

イベントソーシングして CQRS が楽勝だと、システムをピタゴラスイッチ化して、新たな混沌を生み出すプロジェクトが、たくさんできそうな気がします。

そうならないように使いどころを見極めていきたいです。