AWS LambdaでSQSのコンシューマを作る

AWS Lambdaの無限ループパターンで何かできないかなと考えてたら思いついたので試してみた話。

構成

KinesisのストリームからAWS Lambdaを起動して、処理が完了したら再度AWS Kinesisにプッシュすることで無限ループを行います。
この無限ループ中でSQSからメッセージを受け取り、メッセージ処理用のAWS Lambdaを起動して、メッセージを処理できたらSQSからメッセージを削除するという動きになります。
もし、失敗した場合はメッセージがそのままSQSに残るので、Dead Letter Queueに入るまでは何度でもリトライ可能になります。

コード

consumer.js

// require
var Promise = require('bluebird');
var AWS     = require('aws-sdk');
var config  = require('config');
var uuid    = require('node-uuid');
var sqs     = Promise.promisifyAll(new AWS.SQS());
var lambda  = Promise.promisifyAll(new AWS.Lambda());
var kinesis = Promise.promisifyAll(new AWS.Kinesis());
/**
 * AWS Lambda entry point
 *
 * @param {Object} event lambda event object
 * @param {Object} context lambda context
 */
exports.handler = function(event, context) {
  // 
  Promise.resolve().then(function() {
    // receive messages from SQS
    var params = {
      QueueUrl: config.get('sqs.url'),
      MaxNumberOfMessages: config.get('sqs.max_number_of_messages'),
      VisibilityTimeout: config.get('sqs.visiblity_timeout'),
      WaitTimeSeconds: config.get('sqs.wait_time_seconds')
    };
    return sqs.receiveMessageAsync(params).then(function(data) {
      return data.Messages || [];
    });

  }).each(function(data) {
    // invoke lambda
    var args = {Records: [
      {
        sqs: {
          QueueUrl: config.get('sqs.url'),
          MessageId: data.MessageId,
          ReceiptHandle: data.ReceiptHandle,
          MD5OfBody: data.MD5OfBody,
          Body: data.Body
        }
      }
    ]};
    var params = {
      FunctionName: config.get('lambda.function_name'),
      InvokeArgs: JSON.stringify(args)
    };
    return lambda.invokeAsyncAsync(params);

  }).then(function() {
    // put to kinesis stream for retry
    var params = {
      Records: [
        {
          PartitionKey: uuid.v4(),
          Data: ''
        }
      ],
      StreamName: config.get('kinesis.stream_name')
    };
    return kinesis.putRecordsAsync(params);

  }).then(function() {
    context.done(null, 'success consumer.');

  }).catch(function(err) {
    context.done(err);

  });

};

コード自体も簡単で、SQSからメッセージを受けとって、AWS Lambdaを起動して、Kinesisnに再プッシュしているだけになります。

ネットワーク障害とかで処理に失敗した場合はAWS Lambda自体のリトライ機能を使って、再実行を行います。
Kinesisをイベントソースにしているためリトライが10回できるので、それで十分かなという感じです。

あとはCloudWatchとかでSQSのMessagesVisibleを見て閾値を超えてたらKinesisに2回プッシュするとかすると、SQSの詰まり具合に合わせてスケールできて良いかもですね。

停止方法

AWS Lambda functionを削除するかKinesisのストリームを削除したら止まります。

感想

パターンとしてはおもしろいですが、Kinesis-Lambdaで安定的に無限ループさせるのが難しそうな雰囲気です。
監視とかパフォーマンスとかスケールとか考えだすと1年間安定運用させる自信がないです。

ちょっと誰か実運用に載せて感想文書いて欲しいです。