はじめに

AWSについて色々試しながら勉強しています。AWS LambdaからAWS Batchを起動するという流れの処理をおこなう一連のリクエストを、AWS X-Rayで追跡する方法を色々試していたので、備忘録として残したいと思います。

あくまで試行錯誤した結果なので、実際に試す場合は自己責任でお願いします。

AWS Lambdaなど、X-Ray統合されたシステムの場合はリクエストの追跡をAWS側で勝手にやってくれたりしますが、そうでない場合はSDKを使って色々やらなければいけないように思います。

X-Ray SDKを使って実装したセグメントに関する情報を、X-Rayデーモンプロセスを介してAWS X-Rayにデータを送信することで、AWS X-Ray側でリクエストの情報を確認できます。Batch上で動くアプリケーションについて、X-Rayデーモンプロセスを立ち上げてリクエストを追跡してみます。

AWS Batchの作成

今回は、適当に作成したイメージを動かすBatchを作成して試そうと思います。以下の文献などを参考に、簡単にDocker imageをECRにPushしてFargate環境で動かすようなBatchを作成しました。

作成したDockerfileとPythonスクリプトは以下のとおりです。

Dockerfile

FROM amazonlinux:2

RUN yum install -y python3 python3-pip
RUN python3 -m pip install aws_xray_sdk

COPY test.py test.py
CMD python3 test.py

test.py

from aws_xray_sdk.core import xray_recorder

xray_recorder.begin_segment("Test")
print ("Hello, World!")
xray_recorder.end_segment()

AWS Lambda

import json
import boto3

def lambda_handler(event, context):
    
    client = boto3.client('batch')

    JOB_NAME = 'test'
    JOB_QUEUE = "arn:aws:batch:ap-northeast-1:<account id>:job-queue/<queue name>"
    JOB_DEFINITION = "arn:aws:batch:ap-northeast-1:<acount id>:job-definition/<job-definition-name>"

    response = client.submit_job(
        jobName = JOB_NAME,
        jobQueue = JOB_QUEUE,
        jobDefinition = JOB_DEFINITION,
    )
    
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

特に何もしない場合はリクエストの追跡が途切れる結果になります。実際にはAWS Batch上の処理も追跡したいので、このリクエストを追えるようにしたいです。

方法1 Docker image内でAWS X-Ray デーモンプロセスを多重に起動させる

AWS Lambda関数からBatchにJobを投げるようにしてみます。リクエストをつなげるために、Batch上のアプリケーションにそのリクエストのトレースIDを渡したいです。そのため、SQSにトレースIDを渡して、Batch上で取得するようにしてみました。

Dockerfile

FROM amazonlinux:2

RUN yum install -y unzip
RUN curl -o daemon.zip https://s3.us-east-2.amazonaws.com/aws-xray-assets.us-east-2/xray-daemon/aws-xray-daemon-linux-3.x.zip
RUN unzip daemon.zip && cp xray /usr/bin/xray

RUN yum install -y python3 python3-pip
RUN python3 -m pip install aws_xray_sdk boto3

COPY test.py test.py
COPY run.sh run.sh

ENTRYPOINT ["bash", "run.sh"]
EXPOSE 2000/udp
EXPOSE 2000/tcp

test.py

from aws_xray_sdk.core import xray_recorder
import boto3
import json

# SQS
queue_url = 'https://sqs.ap-northeast-1.amazonaws.com/<account id>/test-batch-queue'
sqs_client = boto3.client('sqs')

response = sqs_client.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1)
message = response['Messages'][0]
body = json.loads(message['Body'])
print (body)
receipt_handle = message['ReceiptHandle']

sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)

xray_recorder.begin_segment(name='Test', traceid=body['trace_id'], parent_id=body['parent_id'], sampling=(body['sampled'] == '1'))
print ("Hello, World!")
xray_recorder.end_segment()

run.sh

#!/bin/bash
/usr/bin/xray -t 127.0.0.1:2000 -b 127.0.0.1:2000 -o -n ap-northeast-1 --log-file "/var/log/xray-daemon.log" &
python3 test.py

AWS Lambda

import json
import boto3
import os
import re

def lambda_handler(event, context):
    
    queue_url = 'https://sqs.ap-northeast-1.amazonaws.com/<account id>/test-batch-queue'
    sqs_client = boto3.client('sqs')
    trace_header = os.getenv('_X_AMZN_TRACE_ID')
    trace_id = re.search(r'Root=([^;]*)', trace_header).group(1)
    parent_id = re.search(r'Parent=([^;]*)', trace_header).group(1)
    sampled = re.search(r'Sampled=([^;]*)', trace_header).group(1)

    message_body = json.dumps({
        "trace_id" : trace_id,
        "parent_id": parent_id,
        "sampled" : sampled
    })
    response = sqs_client.send_message(QueueUrl=queue_url, MessageBody=message_body)
    
    client = boto3.client('batch')

    job_name = 'test'
    job_queue = "arn:aws:batch:ap-northeast-1:<account id>:job-queue/getting-started-wizard-job-queue"
    job_definition = "arn:aws:batch:ap-northeast-1:<account id>:job-definition/getting-started-wizard-job-definition:1"

    response = client.submit_job(
        jobName = job_name,
        jobQueue = job_queue,
        jobDefinition = job_definition
    )
    print(response)
    
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

Batch上の処理を追跡できるようになりました。ただ、Docker上で多重にプロセスを立てるのはおそらくあまりよろしくないような気がします。

方法2 別でデーモンを起動させる

デーモンを別で立てて、そこに投げるようにすれば問題ないはずです。今回は簡単にEC2で試してみます。

デーモンをEC2で立てる方法は公式で紹介されています。中で実行させればいいだけなのでどのようにやってもいいと思いますが、今回はDockerと同じようにuser dataで以下のスクリプトを実行させます。

user data

#!/bin/bash
curl -o daemon.zip https://s3.us-east-2.amazonaws.com/aws-xray-assets.us-east-2/xray-daemon/aws-xray-daemon-linux-3.x.zip
unzip daemon.zip
sudo cp xray /usr/bin/xray
/usr/bin/xray -t 0.0.0.0:2000 -b 0.0.0.0:2000 -n ap-northeast-1 --log-file "/var/log/xray-daemon.log"

立てたEC2のaddressをPythonのスクリプト上でそこに投げるように設定します。

test.py

from aws_xray_sdk.core import xray_recorder
import boto3
import json

# SQS
queue_url = 'https://sqs.ap-northeast-1.amazonaws.com/<account id>/test-batch-queue'
sqs_client = boto3.client('sqs')

response = sqs_client.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1)
message = response['Messages'][0]
body = json.loads(message['Body'])
print (body)
receipt_handle = message['ReceiptHandle']

sqs_client.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)

xray_recorder.configure(daemon_address='xx.xx.xx.xx:2000')
xray_recorder.begin_segment(name='Test', traceid=body['trace_id'], parent_id=body['parent_id'], sampling=(body['sampled'] == '1'))
print ("Hello, World!")
xray_recorder.end_segment()

test.pyで、投げ先のEC2のIPアドレスを設定します。結果は以下の画像通りになりました。AWS X-Rayのデーモンを別で立てて、そこに投げるようにすれば問題なくAWS X-Rayで追跡できました。

おわりに

AWS Batch上で動くアプリケーションをAWS X-Rayで追跡するための方法が分からなかったので、試行錯誤的に色々試してみました。AWS LambdaなどX-Ray統合されたシステムではAWS側で勝手にやってくれるので特に何もしなくても追跡してくれますが、そうじゃない場合はちゃんと追えるようにする必要があるので大変でした。