AWS SDK for Goとgobotを使ってAmazon Kinesisにデータ投入する

佐々木大輔

2015.02.04

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

先日からAWS SDK for Goが(僕の中で)盛り上がってます。今回はAWS SDK for Goとgobotを使って、Intel Edisonからセンサーデータを取得し、Amazon Cognitoで認証した上でAmazon Kinesisにデータ投入してみます。

やってみた

Amazon Cognitoのセットアップ

Amazon Cognitoはまだ東京リージョンで使えないため、us-east-1でセットアップします。AWS管理コンソールからCognito画面を開きます。[Get Started Now]ボタンをクリックします。

[Create New Identity Pool]画面で、[Identity Pool Name]に今回使うアイデンティティプールの名前を入力します。また未認証のゲストユーザーとしてアクセスするために、[Enable Access to Unauthenticated Identities]をチェックし、[Create Pool]ボタンをクリックします。

一緒に認証ユーザー用と未認証ユーザー用のIAM Roleが作られますので、そのまま[Update Roles]ボタンをクリックします。

これでCognitoのセットアップは完了です。

IAMユーザーのセットアップ

上記のCognitoで作られた未認証ユーザー用のIAM Roleは、以下のポリシーしか持っていません。

{
    "Version": "2012-10-17",
    "Statement": [{
        "Action": [
            "mobileanalytics:PutEvents",
            "cognito-sync:*"
        ],
        "Effect": "Allow",
        "Resource": [
            "*"
        ]
    }]
}

このためKinesisにPutRecordできるようにポリシーを追加します。今回はちょっと手を抜いて、Kinesisに対してフルアクセスでポリシーを追加しました。

Kinesisのセットアップ

Kinesisも特に難しいことは無く、[Stream Name]を指定するだけです。またシェード数は1にしておきます。

開発環境のセットアップ

AWS SDK for GoについてはAWS SDK for Go ファーストインプレッションをご覧下さい。

またgobotについてはgobotを使ってGo言語でIntel Edisonを動かすをご覧下さい。

Intel EdisonはA0にGrove - Temperature Sensorを繋いだ状態になっています。

ソースコード

package main

import (
	"fmt"
	"time"

	"github.com/awslabs/aws-sdk-go/aws"
	"github.com/awslabs/aws-sdk-go/gen/cognito/identity"
	"github.com/awslabs/aws-sdk-go/gen/kinesis"

	"github.com/hybridgroup/gobot"
	"github.com/hybridgroup/gobot/platforms/gpio"
	"github.com/hybridgroup/gobot/platforms/intel-iot/edison"
)

const (
	CognitoRegion  = "us-east-1"
	DefaultRegion  = "ap-northeast-1"
	AccountID      = "XXXXXXXXXX"
	IdentityPoolID = "us-east-1:XXXX-XXXX-XXXX-XXXX-XXXX"
)

var awsCreds = aws.Creds("", "", "")

func main() {

	// IdentityID取得
	cognitoCli := cognitoidentity.New(awsCreds, CognitoRegion, nil)

	cognitoGetIdParams := &cognitoidentity.GetIDInput{
		AccountID:      aws.String(AccountID),
		IdentityPoolID: aws.String(IdentityPoolID),
	}

	cognitoGetIDResp, err := cognitoCli.GetID(cognitoGetIdParams)
	if err != nil {
		panic(err)
	}

	identityId := *cognitoGetIDResp.IdentityID

	// Credentials取得
	cognitoGetCredParams := &cognitoidentity.GetCredentialsForIdentityInput{
		IdentityID: aws.String(identityId),
	}

	cognitoGetCredResp, err := cognitoCli.GetCredentialsForIdentity(cognitoGetCredParams)
	if err != nil {
		panic(err)
	}

	accessKey := *cognitoGetCredResp.Credentials.AccessKeyID
	secretKey := *cognitoGetCredResp.Credentials.SecretKey
	sessionToken := *cognitoGetCredResp.Credentials.SessionToken

	// Kinesis操作の準備
	kinesisCreds := aws.Creds(accessKey, secretKey, sessionToken)
	kinesisCli := kinesis.New(kinesisCreds, DefaultRegion, nil)

	// gobot
	gbot := gobot.NewGobot()

	e := edison.NewEdisonAdaptor("edison")
	sensor := gpio.NewAnalogSensorDriver(e, "sensor", "0")

	work := func() {
		gobot.Every(1*time.Second, func() {
			data, err := sensor.Read()
			if err != nil {
				panic(err)
			}

			record := &kinesis.PutRecordInput{
				Data:         []byte{byte(data)},
				PartitionKey: aws.String("temperature"),
				StreamName:   aws.String("gobot"),
			}

			kinesisResp, err := kinesisCli.PutRecord(record)
			if err != nil {
				panic(err)
			}

			fmt.Println(data)
			fmt.Println(*kinesisResp.SequenceNumber)
			fmt.Println(*kinesisResp.ShardID)
		})
	}

	robot := gobot.NewRobot("sensorBot",
		[]gobot.Connection{e},
		[]gobot.Device{sensor},
		work,
	)

	gbot.AddRobot(robot)

	gbot.Start()
}

実行結果

go buildして出来上がったバイナリファイルをscpでIntel Edisonに転送し、Intel Edison上で実行すると、以下のように結果が出力されます。

2206
49547607889839299170133768794339870045608541285689327618
shardId-000000000000

AWS CLIを使ってkinesisにデータが格納されていることを確認します。

$ aws kinesis get-shard-iterator --stream-name edison --shard-id shardId-000000000000  --shard-iterator-type AT_SEQUENCE_NUMBER --region ap-northeast-1 --starting-sequence-number "49547607889839299170133768794339870045608541285689327618"
{
    "ShardIterator": "AAAAAAAAAAE0AVfUb5up3JZwGtL5iGFzbyq4oYMB3IDP0BfW6S+CO57jiPsOe5g1LhLmgXCn8O1uxoA/FCvlHWVMAhMQshFnE+C5WjgzT5jOy28h6+g7a/glbBhTSRw0ya9NI/z5ONN5PW/VFJfELRZkf4+ig1CD6LMNTZozK7uqKuiY1mDmvnE5UZVasma1kKmN2i71L/s="
}

$ aws kinesis get-records --region ap-northeast-1 --shard-iterator "AAAAAAAAAAE0AVfUb5up3JZwGtL5iGFzbyq4oYMB3IDP0BfW6S+CO57jiPsOe5g1LhLmgXCn8O1uxoA/FCvlHWVMAhMQshFnE+C5WjgzT5jOy28h6+g7a/glbBhTSRw0ya9NI/z5ONN5PW/VFJfELRZkf4+ig1CD6LMNTZozK7uqKuiY1mDmvnE5UZVasma1kKmN2i71L/s="
{
    "Records": [
        {
            "PartitionKey": "temperature",
            "Data": "ng==",
            "SequenceNumber": "49547607889839299170133768794339870045608541285689327618"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAEw+BYIGZ4URFQWndzqaSIUSiebXrt1C673wZ2U4u1LibWIGNbFApDWAwxH/E1L38UtyfkXYhbZzTe6QPGK7Sbhxu8C6ljF9QJr/VaWyVwstiOndpbgGzXGtPhpOQ5SuETONigRV0PTGDPfqO2hp825HWoa6RAIVQCMiGTmrO2tDv9pzmbk9/5OYTEln3GkF4E="
}

ちゃんと入ってますね！