API Gatewayを通じてKinesisにPOSTでデータを入れる方法

API GatewayにPOSTで送ったリクエストを直接Kinesisに入れる方法について解説します。

「あるURLにPOSTメソッドでデータを送ると、Kinesisにそのデータが入れることが可能な仕組み」

を作ることができます。

Kinesisにデータを入れる以外にも、参照/削除もできます。

以下の作業はすべて、AWSのコンソール画面から行うことを想定しています。

目次

• IAMでロールの作成
• Kinesis Streamsの作成
• API Gatewayの設定
• 確認
• エラー集
• 参考

目次としては、以上となります。AWSの3つのサービスを使用します。

IAMでロールの作成

本システムでは、「API Gateway」で設定するロールを一つ用意する必要があります。

IAM上で「AmazonKinesisFullAccess」ポリシーをもったロールを作成すれば大丈夫です。
適当に、KinesisProxyRoleと呼ぶことにします。

ロールを作成したあと、この KinesisProxyRoleを選択し、

[信頼関係] -> [信頼関係の編集]

より「アクセスコントロールポリシードキュメント」を以下のように書き直します。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "apigateway.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}              

Kinesis Streamsの作成

続いて、Kinesisストリームを用意します。
AWSコンソール上で、「Kinesisストリームの作成」をクリックすると、以下の画面が表示されます。

ストリームの名前をつけましょう。test_streamと呼ぶことにします。
この名前は慎重につけましょう。理由としては、このストリーム名がAPI GatewayのURLパターンで参照されるからです。

API Gatewayの設定

API Gatewayの管理画面を開きます。

APIの作成

[APIの作成]より、「新しいAPI」を選択し、API名を入力します。

作るものを確認

さて、これからどういうAPIを作っていくかですが、次のようなURLに対してPOSTで送れるものを作ります。

http://<api gateway url>/<DeployされるStage名>/streams/<Stream名>/record

/streamsの作成

これは、コンソール画面の[アクション] -> [リソースの作成]より、入力します。

	入力内容
リソース名	Streams
リソースパス	streams

ここでは特にメソッドを作成しなくて大丈夫です。

/streams/{stream-name}の作成

/streams配下に、以下のように入力します。

	入力内容
リソース名	stream-name
リソースパス	{stream-name}

この波括弧ですが、stream_nameというなめのパスパラメータを表しており、この部分に、先ほど作ったKinesisの名前を入れます。

このリソース配下でも、特にメソッドを作成しなくて大丈夫です。

/streams/{stream-name}/recordの作成

/streams/{stream-name}配下に、以下のように入力します。

	入力内容
リソース名	Record
リソースパス	record

このリソースでは[メソッドの作成]より、POSTメソッドを作成します。
次に、作成された[POST]という部分をクリックします。
表示された、図中の「統合リクエスト」をクリックします。

	入力内容
統合タイプ	サービス
AWS リージョン	Kinesisと同じregion
AWS サービス	Kinesis
AWS サブドメイン	(空白)
HTTP メソッド	POST
アクション	PutRecord
実行ロール	arn:aws:iam:123456xxxx:role/KinesisProxyRole

と設定します。

続いて同じ画面において、
[HTTP ヘッダー]で

名前	マッピング元
Content-Type	'x-amz-json-1.1'

[本文マッピングテンプレート]で、「テンプレートが定義されていない場合」を選び、[マッピングテンプレートの追加]を押します。
Content-Typeとして、application/json と入力します。

そして、テンプレートの部分に以下をPasteしてください。

{
    "StreamName": "$input.params('stream-name')",
    "Data": "$util.base64Encode($input.json('$.Data'))",
    "PartitionKey": "$input.path('$.PartitionKey')"
}

以上で、設定は完了です。

ここまでの設定を確認するために、[リソース] -> [/recordのPOST]をクリックします。
表示された図中の[テスト]をクリックすると、

指定された入力で メソッドに対してテストコールを行います

という画面に遷移します。

{stream-name}に、test_streamを入力し、リクエスト本文に、

{
   "Data": "some data",
   "PartitionKey": "some key"
}

を貼ります。 [テスト]をクリックしましょう。

すると、

リクエスト: /streams/test_stream/record
ステータス: 200
レイテンシー: 121 ms

レスポンス本文

{
  "SequenceNumber": "1234.......................",
  "ShardId": "shardId-000000000000"
}

という形で、成功が確認できます。

CORSの設定が必要であれば、[アクション]より「CORSの有効化」を押せば設定されます。

確認するために、このAPIをデプロイしましょう。
[アクション] -> [APIのデプロイ]より、ステージを作成しましょう。今回は prodとしましょう。

確認

今回deployにより取得したURLは以下のようになります。

https://hogehoge.fugafuga.amazonaws.com/prod/streams/test_stream/record

こちらをcurlで叩いて見る場合は、以下のコマンドで叩けます。

$ curl -X POST https://hogehoge.fugafuga.amazonaws.com/prod/streams/test_stream/record -H 'Content-type: application/json'  -d '{"Data": "some data","PartitionKey": "some key"}'

JqueryのAjaxで送信する場合は以下のようにします。

js

var data = {"Data": "some data", "PartitionKey": "some key"};
$.ajax({
    type:"POST",
    url:"https://hogehoge.fugafuga.amazonaws.com/prod/streams/test_stream/record",
    data: JSON.stringify(data), // stringifyはお忘れなく
    scriptCharset: 'utf-8',
    contentType: 'application/json',
    success: function(data) {
    }
});

エラー集

やってるときにエラーに遭遇しそうなエラーに関する記述です。

[API Deploy時に] No integration defined for method

これは、

「HTTPメソッドを定義してるけど、integrationに関する情報がない」

というエラーです。つまり、GETやPOSTなどのメソッドを追加しているものの、「Integration Request」の情報の入力がされてないという状態です。

解決方法としては、「追加したもののIntegrationに関する情報を登録してないメソッドは削除する」です。

参考

[KinesisにRequest送信時に] 1 validation error detected: Value '' at 'partitionKey' failed to satisfy constraint: Member must have length greater than or equal to 1

これは、

「partitionKeyは1文字以上必要だ」

というエラーです。起こっているのは、おそらく、Kinesisに対する送信パラメータに「PartitionKey」が足りてない、もしくは、認識されてないということでしょう。

よくあるのは、JavaScriptでKinesisにパラメータを送る場合、PartitionKeyを入れ忘れることがあるので注意しましょう。

JavaScriptから送信する場合は、jsonオブジェクトを JSON.stringifyで文字列化しない場合、こういうエラーが出ます

他のKinesisのストリームにアクセスできてしまう問題

この状態だと、 {stream-name} を変更すると、他のKinesisのストリームにデータを送信できてしまいます。

この対処方法としては、API Gatewayに付与するロールで制限することができます。

例えば、そのロールにアタッチするポリシーを以下のようにして、API GatewayよりアクセスできるKinesisを制限してやることで、他のストリームへのアクセスを禁止することができます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kinesis:*",
      "Resource": "arn:aws:kinesis:<region>:<accountID>:stream/<stream名>"
    }
  ]
}

参考