CloudWatch LogsをKinesis Stream経由でLambdaからLogentriesに転送する

背景

サーバーレスアーキテクチャに限らず小さなバッチやサービス等で AWS Lambda を使う場面は増えてきていますが、ログの出力と管理については大変です。ログはマネージメントサービスに任せてしまうと楽ですが、そのサービスにどのように送るべきかはログの量によって考慮が必要です。

基本的に Lambda のログは CloudWatch Logs に出力されます。CloudWatch Logs から直接別の Lambda にストリーミングすることもできます。この後述の Lambda からログマネージメントサービスに転送することができます。しかしこの方法はキック頻度が高いため、Lambda の同時実行数が跳ね上がってしまいます。そこで CloudWatch Logs から Kinesis にストリーミングして、さらに Lambda を叩くことを考えました。Kinesis Stream 経由では Lambda の同時実行数をそのシャード数と同じに抑えることができます。どちらがベターというより、リアルタイムをとるかバッファリングをとるかのトレードオフですね。

目的

• 自分の使っている Logentries というログ管理解析サービスに Lambda のログを転送します。
• アーキクチャは、Lambda --stdout-> CloudWatch --subscribe-> Kinesis --trigger-> Lambda --send-> Logentries となります。
• Lambda に限らずその他サービスから出力された CloudWatch Logs の転送にも適用できる事例です。

CloudWatch Logs から Kinesis Stream に転送する設定をする

cloudwatch-logs-stream という名前の Kinesis Stream を ap-northeast-1 (東京リージョン)に作っておきます。

CloudWatch Logs のロール (cwl-to-kinesis) を作成してそのロールに Kinesis Stream への PUT 権限を定義したポリシー (cwl-kinesis-write) を追加します。

CloudWatch Logs のロール (cwl-to-kinesis) を作成

cwl-role.json を作成します。

{
  "Statement": {
    "Effect": "Allow",
    "Principal": { "Service": "logs.ap-northeast-1.amazonaws.com" },
    "Action": "sts:AssumeRole"
  }
}

cwl-to-kinesis ロールを作成します。

$ aws iam create-role --role-name cwl-to-kinesis --assume-role-policy-document file://./cwl-role.json

ロールにポリシー (cwl-kinesis-write) を追加

cwl-kinesis-write.json を作成します。

{
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kinesis:PutRecord",
      "Resource": "arn:aws:kinesis:ap-northeast-1:000000000000:stream/cloudwatch-logs-stream"
    },
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "arn:aws:iam::000000000000:role/cwl-to-kinesis"
    }
  ]
}

cwl-to-kinesis ロールに cwl-kinesis-write ポリシーを追加します。

$ aws iam put-role-policy --role-name cwl-to-kinesis --policy-name cwl-kinesis-write --policy-document file://./cwl-kinesis-write.json

サブスクリプションフィルタを作成

既存の Lambda (example-lambda) のロググループに接続する。

$ aws logs put-subscription-filter \
    --region ap-northeast-1 \
    --log-group-name "/aws/lambda/example-lambda" \
    --filter-name "cloudwatchLogsFilter" \
    --filter-pattern "[timestamp=*Z, request_id="*-*", event]" \
    --role-arn "arn:aws:iam::000000000000:role/cwl-to-kinesis" \
    --destination-arn "arn:aws:kinesis:ap-northeast-1:000000000000:stream/cloudwatch-logs-stream"

Lambda ログの取れ方

Lambda 内で以下の出力を行うと、

console.log({ nonfield: null });

AWS マネージメントコントールで下記のとおり出力されます。

2016-09-17T10:37:27.464Z    ba5c9d30-7cc2-11e6-a2f6-81148a42deb8    { nonfield: null }

これを前述のサブスクリプション フィルタ[timestamp=*Z, request_id="*-*", event] を通して、Lambda で取得すると event オブジェクトが次のようになります。

{
  "Records": [
    {
      "kinesis": {
        "kinesisSchemaVersion": "1.0",
        "partitionKey": "be19c517094e8ba03d1f7fcde1c10c4e",
        "sequenceNumber": "49565835572342545914261768977064411601767982706848694274",
        "data": "H4sIAAAAAAAAADVQy07DMBD8lcriWBPb8TO3SIReyik50VbIcZxi4SRV7BChqv+OoXDceezszhUMNgR9ts3XxYICPJVN+fZS1XW5q8AWTOto5wQrySlDEjHBRIL9dN7N03JJTKbXkHk9tJ3OzOrhhxttcAHGCSaVHePsbLhb6jhbPSQPQZhnSGVYZIeHfdlUdXNC3GLcsp5Zq2iPaUtV19PWaE4w7UWeVoSlDWZ2l+im8dn5aOcAigMwflq6VUfzvp/O4U6A029i9ZkO+BFdgetScE6kyAVSQimOcsIlpYgyoaigPE0C45xSRhRXggqZdIjkQjGSwqNLRUU9pJ9xkmMkMVFS5tv/Av/+gkhBLBqMCiILqh45z1+PUTLDTW8MFMZgiLHlULW8g6hrpVbaEoTQMV434zT2zvqu2IyL95vbcQS30+0b9eSHfKUBAAA=",
        "approximateArrivalTimestamp": 1474108130.452
      },
      "eventSource": "aws:kinesis",
      "eventVersion": "1.0",
      "eventID": "shardId-000000000000:49565835572342545914261768977064411601767982706848694274",
      "eventName": "aws:kinesis:record",
      "invokeIdentityArn": "arn:aws:iam::000000000000:role/lambda_basic_execution",
      "awsRegion": "ap-northeast-1",
      "eventSourceARN": "arn:aws:kinesis:ap-northeast-1:000000000000:stream/cloudwatch-logs-stream"
    }
  ]
}

event.Records の各 kinesis.data が実際のデータ部分で、これは GZip して Base64 エンコードされています。よって Base64 デコードして Gunzip すると下記のような JSON になります。

{
  "messageType": "DATA_MESSAGE",
  "owner": "986450805757",
  "logGroup": "/aws/lambda/cwl-kinesis-to-logentries",
  "logStream": "2016/09/17/[$LATEST]06e11b5f5ee94f14b49df4bca6214f73",
  "subscriptionFilters": [
    "cloudwatchLogsFilter"
  ],
  "logEvents": [
    {
      "id": "32873709799603268440457947462687113445296974780970237952",
      "timestamp": 1474108129883,
      "message": "2016-09-17T10:28:49.663Z\t85c6cfcc-7cc1-11e6-9b6d-0db8a9ae2000\t{ nonfield: null }\n"
    }
  ]
}

ちなみに logEvents の timestamp フィールドを日時を調べたら 2016-09-17T10:28:49.883Z なりました。message 内の timestamp は Lambda 実行コンテナ内での出力時間で、前者が CloudWatch Logs で記録したときの日時だと思われます。

さて、このまま logEvents の message フィールドをログサービスに転送すれば良さそうですが、どうせなら JSON ログで転送したいと思います。

JSON ログを Logentries に転送する

ログを生成する Lambda

bunyan を使ってダミーのログを出力する Lambda (lambda-logentries_generator) を実装してみます。

lambda-logentries_generator

const bunyan = require('bunyan');

exports.handle = (event, context, callback) => {
  const log = bunyan.createLogger({
      name:  'myapp',
      stream: process.stdout,
      level: 'debug'
    });

  for (var i = 0; i < 3; i++) {
    log.debug({ data: { lv: 'debug', value:37 } }, 'Debug sample');
    log.info( { data: { lv: 'info' , value:38 } }, 'Info sample');
    log.error({ data: { lv: 'error', value:39 } }, 'Error sample');
  }

  context.succeed();
}

こちらテスト実行すると、下記のように JSON で出力されているのが確認できます。

先に作ったサブスクリプションフィルタを一旦削除して（ブラウザから可能）、ログレベルが全部マッチするフィルタパターンでサブスクリプションフィルタを再作成します。

$ aws logs put-subscription-filter \
    --region ap-northeast-1 \
    --log-group-name "/aws/lambda/lambda-logentries_generator" \
    --filter-name "cloudwatchLogsFilter" \
    --filter-pattern "{ $.level > 0 }" \
    --role-arn "arn:aws:iam::000000000000:role/cwl-to-kinesis" \
    --destination-arn "arn:aws:kinesis:ap-northeast-1:000000000000:stream/cloudwatch-logs-stream"

ログを Logentries に転送する Lambda

logEvents の message フィールドがまるまる JSON 文字列になっているのでその文字列のまま書き込みます。 Logentries が提供している le_node のロガーを使います。また手前味噌ですが、Kinesis Stream を Node.js の Stream で処理できる NPM モジュール等を kinesis-stream-lambda, promised-lifestream を使っています。

lambda-logentries_cwl-to-logentries

'use strict';

const zlib = require('zlib');
const through2 = require('through2');
const Logger = require('le_node');
const KSL = require('kinesis-stream-lambda');
const PromisedLife = require('promised-lifestream');

exports.handle = (event, context, callback) => {
  const logger = new Logger({ token: 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx' });
  const procStream = through2.obj(
    function(data, enc, cb) {
      data.logEvents.forEach(logevt => {
        logger.log(logevt.message);
      });
      this.push();
      cb();
    });

  PromisedLife([
    KSL.reader(event),
    through2(function (chunk, enc, cb) {
      zlib.gunzip(chunk, cb);
    }),
    KSL.parseJSON(),
    procStream
  ])
  .then(() => {
    return [null, null];
  })
  .catch(err => {
    logger.error(err);
    return [err, null];
  })
  .then(args => {
    logger.end();
    setTimeout(() => {
      logger.closeConnection();
      callback.apply(this, args);
    }, 800);
  });
}

• Logentries のコンソールから [Add a Log] で Token TCP のトークンを取得して使用しています。
• Logger にバッファリング機構が入っているので、最後に end メソッドを呼んだあと0.8秒待機してから、ロガーの接続を閉じて Lambda を終わらせています。

トリガーをセット

lambda-logentries_cwl-to-logentries, lambda-logentries_generator をそれぞれデプロイした後、lambda-logentries_cwl-to-logentries に Kinesis Stream をイベントソースとしてトリガーを作成し有効にします。

転送結果

lambda-logentries_generator の [Test] ボタンでログを出力すると下記のように転送されました。

結び

目的通りの動作が達成できました。この手法は Logentries の類似サービスの Loggly 等でも適用できると思います。またロググループを転送役の Lambda 内で知ることが可能なので、ロググループとTCP Tokenのペアマッチングを設定に持てば Logger クライアントを切り替えてそれぞれログセットに送ることができるでしょう。そうすると Kinesis Stream をいくつも作らない方法も可能でしょう。