Kinesisにリクエストを流してLambdaからS3上のJSを動的に実行する

概要

S3 に JavaScript のモジュールを置き、そのバケット名とオブジェクトキーを Kinesis ストリームに流すことで Lambda から動的に JS を呼び出し実行してみる。
何に使うんでしょうね。

サンプルコード

ローカルに書き出したファイル名に context.invokeid を使うことで require にキャッシュされることはなくなった。
つまり遅い。
つまりステートレスと言っているけど require の結果はキャッシュされているみたい。

Lambda function

handler

var fs = require('fs');
var aws = require('aws-sdk');
var s3 = new aws.S3({apiVersion: '2006-03-01'});

exports.handler = function(event, context) {
    console.log(JSON.stringify(event, null, 2));
    console.log(JSON.stringify(context, null, 2));
    
    var encodedPayload = event.Records[0].kinesis.data;
    var data = JSON.parse(new Buffer(encodedPayload, 'base64').toString('ascii'));
    console.log(JSON.stringify(data, null, 2));
    
    var bucket = data.bucket;
    var key = data.key;
    
    s3.getObject({ Bucket:bucket, Key:key }, function(err, data) {
        if (err) {
            context.done('error getting object', err);
        } else {
            var modPath = '/tmp/' + context.invokeid;
            var w = fs.createWriteStream(modPath);
            w.write(data.Body);
            w.end(function() {
                var mod = require(modPath);
                mod.do(event, function(err, res) {
                    if (err) {
                        context.done('error', err);
                    } else {
                        context.done(null, res);
                    }
                });
            });
        }
    });
};

event

event.Records[0].kinesis.data を decode すると下記のようなJSONになる。

event.Records[0].kinesis.data

{
  "bucket" : "js-bucket",
  "key" : "/path/to/target.js"
}

ちなみに Management Console で Lambda の編集時に表示される Sample event は Kinesis も DynamoDB Streams も実際のデータと違うので注意。
ただ Lambda は Preview なので最終的にどうなるかは分からない。

kinesis

{
  "Records": [
    {
      "kinesis": {
        "partitionKey": "shardId-000000000000",
        "kinesisSchemaVersion": "1.0",
        "data": "eyJidWNrZXQiOiJoaWRlbm8tdGVzdC1pbnZva2UtanMiLCXXXXXXXXXXXXXXXXXX",
        "sequenceNumber": "49545662236882836628328222551726608793875413962349084674"
      },
      "eventSource": "aws:kinesis",
      "eventID": "shardId-000000000000:49545662236882836628328222000000000000000000000000000000",
      "invokeIdentityArn": "arn:aws:iam::000000000000:role/invocation-role",
      "eventVersion": "1.0",
      "eventName": "aws:kinesis:record",
      "eventSourceARN": "arn:aws:kinesis:us-east-1:000000000000:stream/stream-name",
      "awsRegion": "us-east-1"
    }
  ]
}

S3上のJS

do.js

exports.do = function(event, callback) {
    console.log('doing');
    callback(null, 'done');
};

Kinesis と Lambda の紐付け

CLI より pry + SDK のほうが好きなのでそうした。
batch_size はデフォルトで 1 になる。本当は指定し忘れただけだけどそれでいい。

詳細は下記のドキュメントを参照。一連の流れが書いてある。
http://docs.aws.amazon.com/lambda/latest/dg/walkthrough-kinesis-events-adminuser.html

l = Aws::Lambda::Client.new region: 'us-east-1'

es = 'arn:aws:kinesis:us-east-1:000000000000:stream/stream-name'
role = 'arn:aws:iam::000000000000:role/invocation-role'
l.add_event_source event_source: es, function_name: 'invoke-js', role: role

Kinesis にリクエストを流す

k = Aws::Kinesis::Client.new region: 'us-east-1'

data = {'bucket' => 'js-bucket', 'key' => '/path/to/target.js'}.to_json
k.put_record stream_name: 'stream-name', data: data, partition_key: 'shardId-000000000000'

これはテスト実行の結果だけど、こんな感じで CloudWatch Logs にも出力される。

Logs
----
START RequestId: 6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f
2014-12-03T02:23:21.297Z	6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f	{
  "Records": [
    {
      "awsRegion": "us-east-1",
      "sequenceNumber": "196800000000000000000000",
      "partitionKey": "2efdb0ea22685b46993e42a67302a001",
      "eventSource": "aws:kinesis",
      "kinesis": {
        "data": "XXXXXXXXXXXXX............"
      }
    }
  ]
}
2014-12-03T02:23:21.298Z	6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f	{
  "invokeid": "6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f"
}
2014-12-03T02:23:21.298Z	6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f	{
  "bucket": "js-bucket",
  "key": "/path/to/target.js"
}
2014-12-03T02:23:21.379Z	6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f	doing
END RequestId: 6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f
REPORT RequestId: 6cb8aa1e-7a93-11e4-ae0c-cb41dfce093f	Duration: 95.43 ms	Billed Duration: 100 ms 	Memory Size: 128 MB	Max Memory Used: 17 MB	

Message
-------
done

感想

実際使うかと言われれば使わないかなと思う。S3をgitリポジトリにして単発で動かすと開発しやすい、かなあ・・・
Lambda ファンクションに付ける execution role には気をつけよう。

次は Function オブジェクトをシリアライズして Kinesis ストリームに流して Lambda で実行してみたい。