AWS X-Ray

AWS X-Ray란?

AWS X-Ray는 애플리케이션이 처리하는 요청에 대한 데이터를 수집하고, 이를 분석하여 요청 흐름, 지연 시간, 오류 위치 등을 시각화해주는 분산 추적(Distributed Tracing) 서비스입니다. API Gateway, Lambda, ECS, RDS 등 주요 AWS 서비스와 자동 통합되어, 마이크로서비스 아키텍처(MSA) 환경에서도 복잡한 요청 경로를 하나의 트레이스로 묶어 파악할 수 있게 도와줍니다. 서비스 맵과 개별 트레이스 뷰를 제공해 성능 병목이나 장애 원인을 빠르게 찾아낼 수 있으며, 관측 가능성을 강화하는 핵심 도구로 활용됩니다.

[그림 1] AWS X-Ray

관측 가능성(Observability)

관측 가능성(Observability)은 시스템이 외부에 출력하는 로그, 메트릭, 트레이스와 같은 정보를 기반으로 내부 상태를 추론하고 문제를 진단할 수 있는 능력을 의미합니다. 단순히 지표를 모니터링하는 것을 넘어서, 복잡한 분산 환경에서도 문제의 원인을 빠르게 식별하고 대응할 수 있도록 돕는 접근 방식입니다. 특히 MSA나 서버리스 구조처럼 서비스가 분산된 환경에서는, 관측 가능성이 없으면 전체 요청 흐름을 파악하기가 어렵기 때문에 필수적인 개념입니다.

 

[그림 2] 분산 시스템

 

Hands-on

Lambda와 API Gateway, DynamoDB를 사용해서 X-Ray에서 관찰하는 실습을 진행해보겠습니다.

 

1. IAM 설정

각 Lambda 함수 실행을 위한 역할 생성 시 다음 정책을 포함해야합니다.

• AWSXRayDaemonWriteAccess: X-Ray 트레이스 데이터 기록 권한
• AmazonDynamoDBFullAccess: DynamoDB 접근 권한 (실습용 전체 권한)
• AWSLambdaBasicExecutionRole: CloudWatch 로그 기록 권한 포함

2. Lambda 함수 생성

각 서비스(User, Product, Order)에 대해 Lambda 함수 생성합니다.

모두 Python 3.11 런타임 사용해서 코드를 구현하였습니다.

 

user-service

import json
import boto3
import logging
import time
from aws_xray_sdk.core import patch_all, xray_recorder

# X-Ray 및 SDK 패치
patch_all()
xray_recorder.configure(service='user-handler')

# 로깅 설정
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# DynamoDB 연결
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('user-table')

def lambda_handler(event, context):
    method = event.get('requestContext', {}).get('http', {}).get('method', 'GET')
    trace_id = xray_recorder.current_segment().trace_id
    request_id = context.aws_request_id
    start_time = time.time()

    logger.info(json.dumps({
        'trace_id': trace_id,
        'request_id': request_id,
        'message': 'Lambda invoked',
        'method': method,
        'event': event
    }))

    try:
        if method == 'GET':
            query_params = event.get('queryStringParameters') or {}
            user_id = query_params.get('userId')

            if not user_id:
                logger.warning(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'Missing userId in query parameters'
                }))
                return {
                    'statusCode': 400,
                    'body': json.dumps({'message': 'Missing userId'})
                }

            if user_id == "fail":
                raise ValueError("Intentional failure")  # 테스트용 장애 유발

            with xray_recorder.in_subsegment('GetUserFrom_user-table'):
                res = table.get_item(Key={'userId': user_id})
                logger.info(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'DynamoDB GetItem result',
                    'userId': user_id,
                    'result': res
                }))

            return {
                'statusCode': 200,
                'body': json.dumps(res.get('Item', {'message': 'User not found'}))
            }

        elif method == 'POST':
            raw_body = event.get('body')
            logger.info(json.dumps({
                'trace_id': trace_id,
                'request_id': request_id,
                'message': 'Received body',
                'raw_type': str(type(raw_body)),
                'raw_body': raw_body
            }))

            if isinstance(raw_body, str):
                try:
                    body = json.loads(raw_body)
                except Exception:
                    logger.warning(json.dumps({
                        'trace_id': trace_id,
                        'request_id': request_id,
                        'message': 'Invalid JSON',
                        'raw_body': raw_body
                    }))
                    return {
                        'statusCode': 400,
                        'body': json.dumps({'message': 'Invalid JSON in request body'})
                    }
            elif isinstance(raw_body, dict):
                body = raw_body
            else:
                body = {}

            user_id = body.get('userId')
            name = body.get('name')

            logger.info(json.dumps({
                'trace_id': trace_id,
                'request_id': request_id,
                'message': 'Parsed POST body',
                'userId': user_id,
                'name': name
            }))

            if not user_id or not name:
                logger.warning(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'Missing userId or name',
                    'parsed_body': body
                }))
                return {
                    'statusCode': 400,
                    'body': json.dumps({'message': 'Missing userId or name'})
                }

            with xray_recorder.in_subsegment('PutUserTo_user-table'):
                table.put_item(Item={'userId': user_id, 'name': name})
                logger.info(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'DynamoDB PutItem success',
                    'userId': user_id,
                    'name': name
                }))

            return {
                'statusCode': 201,
                'body': json.dumps({'message': 'User created'})
            }

        else:
            logger.error(json.dumps({
                'trace_id': trace_id,
                'request_id': request_id,
                'message': f'Unsupported HTTP method: {method}'
            }))
            return {
                'statusCode': 405,
                'body': json.dumps({'message': f'Method {method} not allowed'})
            }

    except Exception as e:
        # 예외를 X-Ray에 기록해야 에러로 표시됨
        subsegment = xray_recorder.current_subsegment()
        if subsegment:
            subsegment.add_exception(e, stack=True)

        logger.exception(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Unhandled exception occurred',
            'error': str(e)
        }))
        return {
            'statusCode': 500,
            'body': json.dumps({'message': 'Internal server error', 'error': str(e)})
        }

    finally:
        duration = time.time() - start_time
        logger.info(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Lambda execution completed',
            'duration_ms': int(duration * 1000)
        }))

 

product-service

import json
import boto3
import logging
import time
from decimal import Decimal
from aws_xray_sdk.core import patch_all, xray_recorder

# X-Ray 및 SDK 패치
patch_all()
xray_recorder.configure(service='product-handler')

# 로깅 설정
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# DynamoDB 연결
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('product-table')

# Decimal → float 변환 함수
def convert_decimal(obj):
    if isinstance(obj, list):
        return [convert_decimal(i) for i in obj]
    elif isinstance(obj, dict):
        return {k: convert_decimal(v) for k, v in obj.items()}
    elif isinstance(obj, Decimal):
        return float(obj)
    return obj

def lambda_handler(event, context):
    method = event.get('requestContext', {}).get('http', {}).get('method', 'GET')
    trace_id = xray_recorder.current_segment().trace_id
    request_id = context.aws_request_id
    start_time = time.time()

    logger.info(json.dumps({
        'trace_id': trace_id,
        'request_id': request_id,
        'message': 'Lambda invoked',
        'method': method,
        'event_summary': {
            'rawPath': event.get('rawPath'),
            'routeKey': event.get('routeKey'),
            'headers': event.get('headers', {}),
            'body_preview': (event.get('body') or '')[:100]
        }
    }))

    try:
        if method == 'GET':
            query_params = event.get('queryStringParameters') or {}
            product_id = query_params.get('productId')

            if not product_id:
                logger.warning(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'Missing productId in query parameters'
                }))
                return {
                    'statusCode': 400,
                    'body': json.dumps({'message': 'Missing productId'})
                }

            with xray_recorder.in_subsegment('GetProductFrom_product-table'):
                res = table.get_item(Key={'productId': product_id})
                item = res.get('Item', {'message': 'Product not found'})
                item = convert_decimal(item)

                logger.info(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'DynamoDB GetItem result',
                    'productId': product_id,
                    'item': item
                }))

            return {
                'statusCode': 200,
                'body': json.dumps(item)
            }

        elif method == 'POST':
            raw_body = event.get('body') or '{}'
            try:
                body = json.loads(raw_body)
            except json.JSONDecodeError:
                logger.warning(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'Invalid JSON in request body',
                    'body': raw_body
                }))
                return {
                    'statusCode': 400,
                    'body': json.dumps({'message': 'Invalid JSON in request body'})
                }

            product_id = body.get('productId')
            name = body.get('name')
            price = body.get('price')

            logger.info(json.dumps({
                'trace_id': trace_id,
                'request_id': request_id,
                'message': 'Parsed POST body',
                'parsed_body': body
            }))

            if not product_id or not name or price is None:
                logger.warning(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'Missing productId, name, or price',
                    'body': body
                }))
                return {
                    'statusCode': 400,
                    'body': json.dumps({'message': 'Missing productId, name, or price'})
                }

            with xray_recorder.in_subsegment('PutProductTo_product-table'):
                table.put_item(Item={
                    'productId': product_id,
                    'name': name,
                    'price': Decimal(str(price))  # ensure Decimal if float input
                })
                logger.info(json.dumps({
                    'trace_id': trace_id,
                    'request_id': request_id,
                    'message': 'DynamoDB PutItem success',
                    'productId': product_id,
                    'name': name,
                    'price': price
                }))

            return {
                'statusCode': 201,
                'body': json.dumps({'message': 'Product created'})
            }

        else:
            logger.error(json.dumps({
                'trace_id': trace_id,
                'request_id': request_id,
                'message': 'Unsupported HTTP method',
                'method': method
            }))
            return {
                'statusCode': 405,
                'body': json.dumps({'message': f'Method {method} not allowed'})
            }

    except Exception as e:
        logger.exception(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Unhandled exception occurred',
            'error': str(e)
        }))
        return {
            'statusCode': 500,
            'body': json.dumps({'message': 'Internal server error', 'error': str(e)})
        }

    finally:
        duration = time.time() - start_time
        logger.info(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Lambda execution completed',
            'duration_ms': int(duration * 1000)
        }))

 

order-service

import json
import boto3
import logging
import time
from aws_xray_sdk.core import patch_all, xray_recorder
from decimal import Decimal

# AWS SDK 및 X-Ray 패치
patch_all()
xray_recorder.configure(service='order-service')

# 로깅 설정
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Lambda 클라이언트
lambda_client = boto3.client('lambda')

def lambda_handler(event, context):
    method = event.get('requestContext', {}).get('http', {}).get('method', 'GET')
    trace_id = xray_recorder.current_segment().trace_id
    request_id = context.aws_request_id
    start_time = time.time()

    logger.info(json.dumps({
        'trace_id': trace_id,
        'request_id': request_id,
        'message': 'Lambda invoked',
        'method': method,
        'event_summary': {
            'rawPath': event.get('rawPath'),
            'routeKey': event.get('routeKey'),
            'headers': event.get('headers', {}),
            'body_preview': (event.get('body') or '')[:100]
        }
    }))

    try:
        if method != 'POST':
            return {
                'statusCode': 405,
                'body': json.dumps({'message': f'Method {method} not allowed'})
            }

        raw_body = event.get('body') or '{}'
        try:
            body = json.loads(raw_body)
        except json.JSONDecodeError:
            return {
                'statusCode': 400,
                'body': json.dumps({'message': 'Invalid JSON in request body'})
            }

        user_id = body.get('userId')
        product_id = body.get('productId')

        if not user_id or not product_id:
            return {
                'statusCode': 400,
                'body': json.dumps({'message': 'Missing userId or productId'})
            }

        # 사용자 조회
        with xray_recorder.in_subsegment('Invoke_user-service'):
            user_res = lambda_client.invoke(
                FunctionName='user-service',
                InvocationType='RequestResponse',
                Payload=json.dumps({
                    'version': '2.0',
                    'requestContext': {'http': {'method': 'GET'}},
                    'queryStringParameters': {'userId': user_id}
                }).encode('utf-8')
            )
            user_body = json.loads(user_res['Payload'].read().decode('utf-8'))
            user_data = json.loads(user_body.get('body', '{}'))

        # 상품 조회
        with xray_recorder.in_subsegment('Invoke_product-service'):
            product_res = lambda_client.invoke(
                FunctionName='product-service',
                InvocationType='RequestResponse',
                Payload=json.dumps({
                    'version': '2.0',
                    'requestContext': {'http': {'method': 'GET'}},
                    'queryStringParameters': {'productId': product_id}
                }).encode('utf-8')
            )
            product_body = json.loads(product_res['Payload'].read().decode('utf-8'))
            product_data = json.loads(product_body.get('body', '{}'))

        logger.info(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Aggregated user & product data',
            'user': user_data,
            'product': product_data
        }))

        # 성공 응답
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Order fetched successfully',
                'user': user_data,
                'product': product_data
            }, default=str)
        }

    except Exception as e:
        logger.exception(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Unhandled exception occurred',
            'error': str(e)
        }))
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Internal server error',
                'error': str(e)
            })
        }

    finally:
        duration = time.time() - start_time
        logger.info(json.dumps({
            'trace_id': trace_id,
            'request_id': request_id,
            'message': 'Lambda execution completed',
            'duration_ms': int(duration * 1000)
        }))

 

Lambda Config

이때, X-Ray SDK를 포함해야 합니다. Lambda 콘솔 편집기에서는 외부 라이브러리가 자동으로 포함되지 않습니다.

해결 방법으로는 Lambda에 Layer 추가하여 라이브러리 zip 파일을 업로드 하면 됩니다.

 

두 번째로는, Lambda의 최대 실행 시간(timeout) 시간을 늘려줘야 합니다. 기본 시간 3초는 짧기 때문에 10초 이상의 설정을 권장드립니다.

 

세 번째로는, Lambda의 IAM 역할을 부여해야 합니다. Lambda에서 DynamoDB로 직접 접근하고, X-Ray에 데이터를 기록해야 하기 때문에 위에서 추가한 IAM 역할을 부여합니다.

 

네 번째로 아래 그림 3처럼 구성 → 모니터링 및 운영 도구 → 애플리케이션 신호 활성화을 해야 합니다.

[그림 3] X-Ray를 사용하기 위한 Lambda 설정 변경

 

 

3. DynamoDB 테이블 추가

user, product, order table을 생성해두면 모든 준비는 끝납니다!

[그림 4] DynamoDB 테이블

 

4. X-Ray 확인

마지막으로 X-Ray에서 잘 보이는지 확인해봅시다.

[그림 5] X-Ray에서 제공하는 Trace Map

 

클라이언트가 주문을 요청하면, 먼저 order-service가 호출되고, 이 서비스는 사용자 정보를 확인하기 위해 user-service를, 주문한 상품 정보를 조회하기 위해 product-service를 순차적으로 호출합니다. 각 서비스는 자체적으로 DynamoDB 테이블과 연동되어 데이터를 처리하며, 이러한 전체 흐름이 X-Ray를 통해 시각적으로 표현되고 있습니다.

 

[그림 6] 장애 상태를 보여주는 X-Ray

AWS X-Ray는 장애 상황에서도 요청 흐름을 시각적으로 보여주는 기능을 제공합니다. 클라이언트가 호출한 각 서비스는 하나의 노드로 표시되며, 서비스 간 호출 관계는 화살표로 연결되어 트레이스 맵 상에서 흐름을 직관적으로 파악할 수 있습니다. 특히 오류가 발생한 경우, 해당 구간은 빨간색 테두리로 강조되어 500번대 서버 오류가 발생했음을 명확히 알려줍니다.

또한 각 노드의 크기는 해당 서비스가 받은 요청의 양을 의미합니다. 원이 클수록 더 많은 요청을 받은 것이며, 작은 노드는 상대적으로 요청이 적거나 부하가 적은 컴포넌트를 나타냅니다. 이를 통해 시스템 내 트래픽 집중 구간이나 부하 분산 상태를 파악할 수 있고, 성능 튜닝이 필요한 부분도 빠르게 확인할 수 있습니다.

노란색 화살표는 지연 시간이 발생한 호출 경로를 의미하며, 병목 구간을 시각적으로 식별하는 데 유용합니다.

추가로, AWS X-Ray에서는 로그까지 함께 확인할 수 있으며, 오류 로그만 필터링해서 볼 수 있어 문제 추적과 분석에 매우 효과적입니다.