はじめに

本連載は以下の3部で構成されています。

AWS X-Rayによる分散トレーシングその1 (概要編)
AWS X-Rayによる分散トレーシングその2 (Java編)
AWS X-Rayによる分散トレーシングその3 (Node.js編) ←今回

X-Rayの基本から学びたいという方はまずその1, その2を参照下さい。

本記事では、LambdaのランタイムにNode.jsを採用した場合にX-Rayに必要な設定やその際に注意すべきことを記載しています。
それでは、早速LambdaでX-Rayを使ってみましょう！

前提構成

今回は、以下のような構成でX-Rayを使ってみます。

API Gatewayでリクエストを受けてLambdaが起動し、DynamoDBアクセス後にSQSにデータを送信します。
SQSのConsumerとしてVPC内でもう一つのLambdaがデータを受け取り、Auroraにアクセスした後、外部のサイトにHTTPSのリクエストを送信します。

各サービスでX-Rayを有効にする方法やSDKでの呼び出し方を確認するのが目的のため、特に意味のある処理を実行しているわけではありません。

この構成で、以下のAWSのドキュメントの様なX-Rayのサービスマップを表示してみましょう。

　AWS X-Ray デベロッパーガイドから抜粋

AWS SAMでのX-Ray設定

今回、サーバレスのサービスのデプロイにはAWS Serverless Application Model (SAM)を使用します。

SAMのテンプレートは以下のようになっており、APIGateway、Lambda、SQS等をデプロイします(X-Rayの設定に影響のないDynamoDBやAuroraは省略しています)。また、テンプレートの中で、X-Rayを使用するための設定はコメント部分の通りです。

template.yaml

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Parameters:
  SecurityGroupId:
    Type: String
  SubnetId:
    Type: String

Resources:
  CloudWatchRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          Action: "sts:AssumeRole"
          Effect: Allow
          Principal:
            Service: apigateway.amazonaws.com
      Path: /
      ManagedPolicyArns:
        - "arn:aws:iam::aws:policy/service-role/AmazonAPIGatewayPushToCloudWatchLogs"
  Account:
    Type: "AWS::ApiGateway::Account"
    Properties:
      CloudWatchRoleArn: !GetAtt CloudWatchRole.Arn
  AccessLogGroup:
    Type: AWS::Logs::LogGroup

  ApiGateway:
    Type: AWS::Serverless::Api
    Properties:
      Name: "XrayTest"
      StageName: "v1"
      AccessLogSetting:
        DestinationArn: !GetAtt AccessLogGroup.Arn
+       # API GatewayのログにX-RayのトレースIDを出力するため、フォーマットを変更
+       Format: "{ 'requestId':'$context.requestId', 'ip': '$context.identity.sourceIp', 'caller':'$context.identity.caller', 'user':'$context.identity.user','requestTime':'$context.requestTime', 'xrayTraceId':'$context.xrayTraceId', 'wafResponseCode':'$context.wafResponseCode', 'httpMethod':'$context.httpMethod','resourcePath':'$context.resourcePath', 'status':'$context.status','protocol':'$context.protocol', 'responseLength':'$context.responseLength' }"
      MethodSettings:
        - MetricsEnabled: True
          ResourcePath: "/*"
          HttpMethod: "*"
+      # API GatewayでのX-Rayトレースを有効にする
+      TracingEnabled: True

  ApiLambda:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda/api.handler
      Runtime: nodejs14.x
      Timeout: 300
      MemorySize: 256
      Policies:
        - AmazonDynamoDBFullAccess
        - AmazonSQSFullAccess
+        - AWSXRayDaemonWriteAccess
      Events:
        APIEvent:
          Type: Api
          Properties:
            Path: /hoge
            Method: post
            RestApiId:
              Ref: ApiGateway
+      # LambdaでのX-Rayトレースを有効にする
+      Tracing: Active

  SqsQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: "XrayQueue"

  SQSLambda:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda/sqs.handler
      Runtime: nodejs14.x
      Timeout: 300
      VpcConfig:
        SecurityGroupIds:
          - !Ref SecurityGroupId
        SubnetIds:
          - !Ref SubnetId
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt SqsQueue.Arn
            BatchSize: 10
      Policies:
+        - AWSXRayDaemonWriteAccess
+      # LambdaでのX-Rayトレースを有効にする
+      Tracing: Active

X-Ray以外のポリシーの設定等はガバガバなのでそのままマネしないでください。

このSAMテンプレートとLambdaのコードをデプロイすると、APIGateway、LambdaでX-Rayが有効になります。

実行結果その1

この時点で一度API Gatwayにリクエストを投げて、X-Rayで実行結果を確認してみましょう。

APIGatewayとLambdaのサービスマップは表示されるようになり、Lambdaの初期化とhandler内部の処理を可視化できるようになりました。
トレースやアナリティクスも取得できています。

が、DynamoDBやAuroraへのアクセスにはトレースが有効になっていないので、ここからさらに分解能を上げる場合はSDKの出番になります。

X-Ray SDK for Node.jsを使ってみる

では次に、LambdaのNode.jsの中からX-Ray SDKを使っていきます。

SDKを使うことによって、すべてのAWSサービスへのアクセス、MySQLやPostgreSQL等へのアクセス、HTTPリクエスト送信、その他任意の処理(カスタムセグメント)、をトレースすることができるようになります。

カスタムセグメントを作らない場合は特に難しいことはなく、require(又はimport)したAWSサービスやnodeモジュールのインスタンスに対して、X-Ray SDKの関数を呼び出すだけです。あら簡単。

API Gatewayから呼ばれるLambdaの変更

APIGatewayから呼ばれる元々Lambdaは以下の様になっていました。

api.js

const DynamoDB = require('aws-sdk/clients/dynamodb');
const SQS = require("aws-sdk/clients/sqs");
const dynamo = new DynamoDB.DocumentClient({
    service: new DynamoDB()
});
const sqs = new SQS();

exports.handler = async (event, context, callback) => {
    //以下略

このコードを以下の様に変更します。

api.js

const DynamoDB = require('aws-sdk/clients/dynamodb');
const SQS = require("aws-sdk/clients/sqs");
const dynamo = new DynamoDB.DocumentClient({
    service: new DynamoDB()
});
- const sqs = new SQS();
+ const AWSXRay = require('aws-xray-sdk');
+ AWSXRay.captureAWSClient(dynamo.service);
+ const sqs = AWSXRay.captureAWSClient(new SQS());

exports.handler = async (event, context, callback) => {
    //以下略

SQSのConsumerのLambdaの変更

SQSのConsumerのLambdaは以下の様になっていました。

sqs.js

const mysql = require("mysql2/promise");
const https = require('https');

exports.handler = async (event, context) => {
  //以下略

こちらのコードは以下の様に変更します。

sqs.js

- const mysql = require("mysql2/promise");
+ const AWSXRay = require('aws-xray-sdk');
+ const mysql = AWSXRay.captureMySQL(require("mysql2/promise"));
const https = require('https');
+ AWSXRay.captureHTTPsGlobal(https);
+ AWSXRay.capturePromise();

exports.handler = async (event, context) => {
  //以下略

実行結果その2

さあ、これで実行してみましょう！

出ました！いい感じです。構成図通りに各処理がサービスマップに登場しています。

が、あれ？

概ねいい感じに出てるのですが、AWSのドキュメントのようにSQSのProducerとConsumerが繋がってくれていません。
この図 ↓ みたいにしたいのですが。。。

　AWS X-Ray デベロッパーガイドから抜粋

何か設定を間違えているのでしょうか？

上記のAWSのドキュメントのJava版のサンプル等を見ると、SQSのConsumer側でSQSのRecordごとにX-RayのセグメントIDを付け替えたりしないとダメなようです。ただし、AWS公式のドキュメントにはNode.js版のやり方が記載されていません。困りました。

さらに検索してみると、Node.jsのSDKではJavaのサンプルのように簡単にはいかない！というコメントが多数ありました。

数年前からずっとこの状態らしく、なかなか改善されていないようです。

但し、コメント欄をよく見ると、ワークアラウンドでがんばっている人もいるようで、カスタムセグメントを駆使してSQSの関連を繋げて可視化できるようにしているお手本があったりします。

SQSのProducerとConsumerの線を繋げてみる

せっかくなので、上記issueのコメントを参考に、SQSのProducerとConsumerを繋げてみましょう！

Consumer側のLambdaで、SQSのRecordごとにセットされているヘッダからトレースIDを取り出し、新しいカスタムセグメントにそれを引き継がせればいいようです。ふむふむ。

ConsumerのLambdaのhandlerはもともと以下の様になっていました。

sqs.js

exports.handler = async (event, context) => {
    const connection = await mysql.createConnection({
        //省略
    });

    for (let record of event.Records) {
        try {
            const message = JSON.parse(record.body);

            // SQL実行
            await connection.execute('INSERT INTO dummy (id, createdAt, n, value) values(?, ?, ?, ?)', [message.id, message.time, message.n, message.value]);

            // HTTPリクエスト
            const response = await new Promise((resolve, reject) => {
                const req = https.request({
                    host: 'www.hitachi.co.jp',
                    protocol: "https:",
                    path: "/products/it/harmonious/cloud/",
                    method: 'GET',
                    rejectUnauthorized: false,
                    requestCert: true,
                }, (res) => {
                    const data = [];
                    res.on('data', function (chunk) {
                        data.push(chunk);
                    }).on('end', function () {
                        resolve(data.join(""));
                    });
                }).on("error", (e) => {
                    reject(e);
                });
                req.end();
            });
        } catch (e) {
            console.error(e);
        }
    }

    if (connection) {
        connection.end();
    }
    return {}
};

ここから、以下の様な変更を加えます。

// ここは先程修正したインポート部分
const AWSXRay = require('aws-xray-sdk');
const mysql = AWSXRay.captureMySQL(require("mysql2/promise"));
const https = require('https');
AWSXRay.captureHTTPsGlobal(https);
AWSXRay.capturePromise();

+ // SQSのrecordごとにカスタムセグメントを作成する関数
+ const startSegment = (record, startTime, context) => {
+     const header = record.attributes.AWSTraceHeader
+     const traceData = AWSXRay.utils.processTraceData(header)
+     const sqsEndTime = Number(record.attributes.ApproximateFirstReceiveTimestamp) / 1000
+     const segment = new AWSXRay.Segment(
+         context.functionName,
+         traceData.root,
+         traceData.parent
+     )
+     segment.origin = "AWS::Lambda::Function"
+     segment.start_time = startTime - (startTime - sqsEndTime)
+     segment.addPluginData({
+         function_arn: context.functionArn,
+         region: record.awsRegion,
+         request_id: context.awsRequestId
+     })
+     AWSXRay.setSegment(segment);
+     return segment
+ }

exports.handler = async (event, context) => {
+     const startTime = Date.now();

    for (let record of event.Records) {
+         // カスタムセグメント開始
+         const segment = startSegment(record, startTime, context);
+
        try {
            const message = JSON.parse(record.body);

            // SQL実行
            await connection.execute('INSERT INTO dummy (id, createdAt, n, value) values(?, ?, ?, ?)', [message.id, message.time, message.n, message.value]);

            // HTTPリクエスト
            const response = await new Promise((resolve, reject) => {
                const req = https.request({
                    host: 'www.hitachi.co.jp',
                    protocol: "https:",
                    path: "/products/it/harmonious/cloud/",
                    method: 'GET',
                    rejectUnauthorized: false,
                    requestCert: true,
+                    // カスタムセグメントごとにhttpsを実行する場合はこのパラメータの追加が必要
+                    XRaySegment: segment
                }, (res) => {
                    const data = [];
                    res.on('data', function (chunk) {
                        data.push(chunk);
                    }).on('end', function () {
                        resolve(data.join(""));
                    });
                }).on("error", (e) => {
                    reject(e);
                });
                req.end();
            });
        } catch (e) {
            console.error(e);
-         }
+        } finally {
+            // カスタムセグメント終了
+            segment.close();
+        }
    }

    if (connection) {
        connection.end();
    }
    return {};
};

実行結果その3

これで実行してみましょう。

なんとかSQSのProducerとConsumerを繋げて可視化することができました！

下にSQS ConsumerのLambda分が2重に出力されてはいますが、これはSAMのSQS Lambdaのトレースを無効にすれば消すことができます

差分ソースのコメントに記載したように、カスタムセグメントを追加する場合は、httpsモジュールにオプションを追加しないとうまくSQS Recordごとのトレースができず、SDKのReadmeを熟読しないとなかなか細かいところがわかりません。

CloudWatch LogsへのトレースID出力

さて、ここまで長々とNode.jsでのサービスマップの出力について書いてきましたが、Node.jsでX-Rayを使う場合には、CloudWatch Logsに出力するログにトレースIDを出力するかどうかも考慮が必要です。

X-Rayで参照できるトレースの情報は通常運用ではサンプリングされていると思うので、障害発生時にログから詳細を追う場合には、CloudWatch Logsのログを参照する必要があります。X-RayのトレースIDが各ログエントリに出力しておけば、CloudWatch Logs Insightsなどで、トレースIDをフィルタ条件に指定して、API GatewayやLambdaなどのログを一気通貫で参照できるようになり、大変便利です。

しかし、SAMでLambdaのX-Rayトレースを有効にしても、自動的に出力されるCloudWatch Logsのログは、Lambda終了時のレポートのログエントリのみです。

その2のJava編に少し記載したように、X-Ray SDK for JavaであればLoggerの設定で比較的簡単にすべてのログエントリにトレースIDを出力できるのですが、残念ながらNode.jsのSDKにはその様な機能がありません。

LambdaにNode.jsを使う場合、以下の様に自分でLambdaのhandlerの中でトレースIDを取得し、consoleやloggerのログエントリごとにトレースIDを付与する必要があります。

api.js

exports.handler = (event, context, callback) => {
    const traceID = event["headers"]["X-Amzn-Trace-Id"];
    console.log(traceID, "message");
    //以下略
}

一通りコードを書ききってから後でX-Rayを導入しようとすると、意外とこういった部分の対応が面倒になることもあると思うので、ロギングの実装前にトレースIDの出力方針も検討しておいたほうが良いと思います。

まとめ

本記事では、Node.jsをランタイムとするLambdaとその関連サービスでX-Rayを使う方法や、X-Ray SDK for Node.jsで注意すべきこと等を記載しました。

これら以外にもX-Ray SDK for Node.jsならでは注意事項やハマりどころがあると思うので、本格的にX-Ray SDK for Node.jsを使う場合は以下のSDKのリポジトリのReadme等を参照して、詳しい使い方を調べておきましょう！

Amazon Web Services、および、その他のAWS 商標は、米国およびその他の諸国におけるAmazon.com, Inc.またはその関連会社の商標です。
Javaは、Oracle Corporation、および、その子会社、関連会社の米国及びその他の国における登録商標です。
その他、本資料に記述してある会社名、製品名は、各社の登録商品または商標です。

AWS DevOps - AWS X-Rayによる分散トレーシング その3 (Node.js編)

はじめに

前提構成

AWS SAMでのX-Ray設定

実行結果その1

X-Ray SDK for Node.jsを使ってみる

API Gatewayから呼ばれるLambdaの変更

SQSのConsumerのLambdaの変更

実行結果その2

SQSのProducerとConsumerの線を繋げてみる

実行結果その3

CloudWatch LogsへのトレースID出力

まとめ

AWS DevOps - AWS X-Rayによる分散トレーシングその3 (Node.js編)