Split View: Serverless 아키텍처 패턴 완전 가이드 2025: Lambda, Step Functions, 이벤트 소싱, 비용 최적화

2014: AWS Lambda 출시 (최초의 FaaS)
2016: Azure Functions, Google Cloud Functions
2017: AWS Step Functions, SAM 출시
2018: Lambda Layers, ALB 지원
2019: Provisioned Concurrency, RDS Proxy
2020: Lambda Container Image, EventBridge
2021: Lambda URL, Graviton2 지원
2022: Lambda SnapStart (Java), 스트리밍 응답
2023: Lambda Advanced Logging, Step Functions 개선
2024: Lambda 성능 최적화, ARM64 전면 지원
2025: Lambda 최대 메모리 10GB, Step Functions Distributed Map 강화

# order_create.py - 주문 생성만 담당
import json
import boto3
import os
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['ORDERS_TABLE'])
sns = boto3.client('sns')

def handler(event, context):
    body = json.loads(event['body'])

    order = {
        'orderId': context.aws_request_id,
        'userId': body['userId'],
        'items': body['items'],
        'total': calculate_total(body['items']),
        'status': 'CREATED',
        'createdAt': datetime.utcnow().isoformat()
    }

    table.put_item(Item=order)

    # 이벤트 발행
    sns.publish(
        TopicArn=os.environ['ORDER_TOPIC'],
        Message=json.dumps(order),
        MessageAttributes={
            'eventType': {
                'DataType': 'String',
                'StringValue': 'OrderCreated'
            }
        }
    )

    return {
        'statusCode': 201,
        'body': json.dumps(order)
    }

def calculate_total(items):
    return sum(item['price'] * item['quantity'] for item in items)

// app.ts - 모놀리식 Lambda
import express from 'express';
import serverless from 'serverless-http';

const app = express();
app.use(express.json());

// 여러 라우트를 하나의 Lambda에서 처리
app.get('/orders', async (req, res) => {
  const orders = await getOrders(req.query);
  res.json(orders);
});

app.post('/orders', async (req, res) => {
  const order = await createOrder(req.body);
  res.status(201).json(order);
});

app.get('/orders/:id', async (req, res) => {
  const order = await getOrder(req.params.id);
  if (!order) return res.status(404).json({ error: 'Not found' });
  res.json(order);
});

app.put('/orders/:id/status', async (req, res) => {
  const order = await updateOrderStatus(req.params.id, req.body.status);
  res.json(order);
});

app.delete('/orders/:id', async (req, res) => {
  await cancelOrder(req.params.id);
  res.status(204).send();
});

export const handler = serverless(app);

# serverless.yml - Fan-out 아키텍처
service: order-processing

provider:
  name: aws
  runtime: nodejs20.x

functions:
  orderReceiver:
    handler: src/receiver.handler
    events:
      - http:
          path: /orders
          method: post
    environment:
      FAN_OUT_TOPIC: !Ref OrderFanOutTopic

  inventoryCheck:
    handler: src/inventory.handler
    events:
      - sns:
          arn: !Ref OrderFanOutTopic
          filterPolicy:
            eventType:
              - OrderCreated

  paymentProcess:
    handler: src/payment.handler
    events:
      - sns:
          arn: !Ref OrderFanOutTopic
          filterPolicy:
            eventType:
              - OrderCreated

  notificationSend:
    handler: src/notification.handler
    events:
      - sns:
          arn: !Ref OrderFanOutTopic
          filterPolicy:
            eventType:
              - OrderCreated

resources:
  Resources:
    OrderFanOutTopic:
      Type: AWS::SNS::Topic
      Properties:
        TopicName: order-fan-out

{
  "Comment": "주문 처리 워크플로",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:validate-order",
      "Next": "CheckInventory",
      "Retry": [
        {
          "ErrorEquals": ["ServiceException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["ValidationError"],
          "Next": "OrderFailed"
        }
      ]
    },
    "CheckInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:check-inventory",
      "Next": "ProcessPaymentOrWait"
    },
    "ProcessPaymentOrWait": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.inventoryAvailable",
          "BooleanEquals": true,
          "Next": "ProcessPayment"
        }
      ],
      "Default": "WaitForInventory"
    },
    "WaitForInventory": {
      "Type": "Wait",
      "Seconds": 300,
      "Next": "CheckInventory"
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:process-payment",
      "Next": "ParallelFulfillment"
    },
    "ParallelFulfillment": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "UpdateDatabase",
          "States": {
            "UpdateDatabase": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:update-db",
              "End": true
            }
          }
        },
        {
          "StartAt": "SendNotification",
          "States": {
            "SendNotification": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:send-notification",
              "End": true
            }
          }
        },
        {
          "StartAt": "InitiateShipping",
          "States": {
            "InitiateShipping": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:initiate-shipping",
              "End": true
            }
          }
        }
      ],
      "Next": "OrderCompleted"
    },
    "OrderCompleted": {
      "Type": "Succeed"
    },
    "OrderFailed": {
      "Type": "Fail",
      "Error": "OrderProcessingFailed",
      "Cause": "Order validation or processing failed"
    }
  }
}

# callback_handler.py - 사람 승인을 기다리는 Lambda
import json
import boto3

sfn = boto3.client('stepfunctions')
ses = boto3.client('ses')

def request_approval(event, context):
    """Step Functions가 태스크 토큰과 함께 호출"""
    task_token = event['taskToken']
    order = event['order']

    # 승인 링크가 포함된 이메일 발송
    approval_url = f"https://api.example.com/approve?token={task_token}"
    reject_url = f"https://api.example.com/reject?token={task_token}"

    ses.send_email(
        Source='noreply@example.com',
        Destination={'ToAddresses': ['manager@example.com']},
        Message={
            'Subject': {'Data': f"주문 승인 요청: {order['orderId']}"},
            'Body': {
                'Html': {
                    'Data': f"""
                    <h2>주문 승인 요청</h2>
                    <p>주문 ID: {order['orderId']}</p>
                    <p>금액: {order['total']}원</p>
                    <a href="{approval_url}">승인</a> |
                    <a href="{reject_url}">거절</a>
                    """
                }
            }
        }
    )

def handle_approval(event, context):
    """승인/거절 콜백 처리"""
    params = event['queryStringParameters']
    task_token = params['token']

    if 'approve' in event['path']:
        sfn.send_task_success(
            taskToken=task_token,
            output=json.dumps({'approved': True})
        )
    else:
        sfn.send_task_failure(
            taskToken=task_token,
            error='Rejected',
            cause='Manager rejected the order'
        )

    return {
        'statusCode': 200,
        'body': json.dumps({'message': '처리 완료'})
    }

# event_store.py
import json
import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
event_store = dynamodb.Table('EventStore')
sns = boto3.client('sns')

def append_event(aggregate_id, event_type, data, version):
    """이벤트를 저장하고 발행"""
    event = {
        'aggregateId': aggregate_id,
        'version': version,
        'eventType': event_type,
        'data': data,
        'timestamp': datetime.utcnow().isoformat(),
        'metadata': {
            'correlationId': data.get('correlationId', ''),
            'causationId': data.get('causationId', '')
        }
    }

    # 낙관적 잠금: version이 이미 존재하면 실패
    event_store.put_item(
        Item=event,
        ConditionExpression='attribute_not_exists(version)'
    )

    # 이벤트 발행
    sns.publish(
        TopicArn='arn:aws:sns:ap-northeast-2:123456789:domain-events',
        Message=json.dumps(event),
        MessageAttributes={
            'eventType': {
                'DataType': 'String',
                'StringValue': event_type
            }
        }
    )

    return event

def replay_events(aggregate_id):
    """특정 Aggregate의 모든 이벤트를 재생"""
    response = event_store.query(
        KeyConditionExpression='aggregateId = :aid',
        ExpressionAttributeValues={':aid': aggregate_id},
        ScanIndexForward=True  # 시간순 정렬
    )
    return response['Items']

{
  "Comment": "주문 Saga - 보상 트랜잭션 포함",
  "StartAt": "ReserveInventory",
  "States": {
    "ReserveInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:reserve-inventory",
      "Next": "ProcessPayment",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "InventoryReservationFailed"
      }]
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:process-payment",
      "Next": "ConfirmOrder",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "RollbackInventory"
      }]
    },
    "ConfirmOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:confirm-order",
      "Next": "SagaCompleted",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "RollbackPayment"
      }]
    },
    "RollbackPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:rollback-payment",
      "Next": "RollbackInventory"
    },
    "RollbackInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-2:123456789:function:rollback-inventory",
      "Next": "SagaFailed"
    },
    "InventoryReservationFailed": {
      "Type": "Fail",
      "Error": "InventoryReservationFailed",
      "Cause": "Could not reserve inventory"
    },
    "SagaCompleted": {
      "Type": "Succeed"
    },
    "SagaFailed": {
      "Type": "Fail",
      "Error": "SagaFailed",
      "Cause": "Order saga failed, all compensations executed"
    }
  }
}

요청 도착
  |
  v
[실행 환경 있음?] --No--> [Cold Start 경로]
  |                           |
  Yes                    1. 실행 환경 프로비저닝
  |                    2. 코드 다운로드 (S3)
  v                    3. 런타임 초기화
[Warm Start]           4. 핸들러 외부 코드 실행
  |                    5. 핸들러 실행
  v                         |
[핸들러 실행]               v
  |                    [응답 반환]
  v
[응답 반환]

# 최적화된 Lambda 함수 구조
import json
import os

# 핸들러 외부에서 초기화 (재사용됨)
# 1. 연결은 전역으로 초기화
import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])

# 2. 불필요한 import 제거
# BAD: import pandas  (패키지 크기 증가)
# GOOD: 필요한 것만 import

# 3. SDK 설정 최적화
from botocore.config import Config
config = Config(
    connect_timeout=5,
    read_timeout=5,
    retries={'max_attempts': 2}
)
s3 = boto3.client('s3', config=config)

def handler(event, context):
    """핸들러는 가능한 가볍게"""
    order_id = event['pathParameters']['orderId']

    response = table.get_item(Key={'orderId': order_id})
    item = response.get('Item')

    if not item:
        return {'statusCode': 404, 'body': json.dumps({'error': 'Not found'})}

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

# SAM template - Provisioned Concurrency 설정
Resources:
  MyFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: app.handler
      Runtime: python3.12
      MemorySize: 512
      AutoPublishAlias: live
      ProvisionedConcurrencyConfig:
        ProvisionedConcurrentExecutions: 10

  # 시간대별 자동 스케일링
  ScalingTarget:
    Type: AWS::ApplicationAutoScaling::ScalableTarget
    Properties:
      MaxCapacity: 100
      MinCapacity: 5
      ResourceId: !Sub function:${MyFunction}:live
      ScalableDimension: lambda:function:ProvisionedConcurrency
      ServiceNamespace: lambda

  ScalingPolicy:
    Type: AWS::ApplicationAutoScaling::ScalingPolicy
    Properties:
      PolicyName: UtilizationScaling
      PolicyType: TargetTrackingScaling
      ScalableTargetId: !Ref ScalingTarget
      TargetTrackingScalingPolicyConfiguration:
        TargetValue: 0.7
        PredefinedMetricSpecification:
          PredefinedMetricType: LambdaProvisionedConcurrencyUtilization

// SnapStart 최적화된 Java Lambda
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import org.crac.Core;
import org.crac.Resource;

public class OrderHandler implements RequestHandler<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent>,
                                     Resource {

    private final DynamoDbClient dynamoDb;
    private final ObjectMapper objectMapper;

    public OrderHandler() {
        // SnapStart: 이 초기화 코드는 스냅샷에 포함됨
        this.dynamoDb = DynamoDbClient.create();
        this.objectMapper = new ObjectMapper();
        Core.getGlobalContext().register(this);
    }

    @Override
    public void beforeCheckpoint(org.crac.Context<? extends Resource> context) {
        // 스냅샷 전: 연결 정리
    }

    @Override
    public void afterRestore(org.crac.Context<? extends Resource> context) {
        // 복원 후: 연결 재설정
        // 고유성 보장 (난수 시드 재설정 등)
    }

    @Override
    public APIGatewayProxyResponseEvent handleRequest(
            APIGatewayProxyRequestEvent event, Context context) {
        // 비즈니스 로직
        return new APIGatewayProxyResponseEvent()
            .withStatusCode(200)
            .withBody("{\"message\": \"OK\"}");
    }
}

# SAM template - REST API
Resources:
  OrdersApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod
      Auth:
        DefaultAuthorizer: CognitoAuthorizer
        Authorizers:
          CognitoAuthorizer:
            UserPoolArn: !GetAtt UserPool.Arn
      # 스로틀링
      MethodSettings:
        - HttpMethod: '*'
          ResourcePath: '/*'
          ThrottlingBurstLimit: 100
          ThrottlingRateLimit: 50
      # CORS
      Cors:
        AllowMethods: "'GET,POST,PUT,DELETE,OPTIONS'"
        AllowHeaders: "'Content-Type,Authorization'"
        AllowOrigin: "'https://example.com'"

  GetOrderFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: src/orders/get.handler
      Runtime: nodejs20.x
      Events:
        GetOrder:
          Type: Api
          Properties:
            RestApiId: !Ref OrdersApi
            Path: /orders/{orderId}
            Method: get

  CreateOrderFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: src/orders/create.handler
      Runtime: nodejs20.x
      Events:
        CreateOrder:
          Type: Api
          Properties:
            RestApiId: !Ref OrdersApi
            Path: /orders
            Method: post

# AppSync 스키마
type Order {
  orderId: ID!
  userId: String!
  items: [OrderItem!]!
  total: Float!
  status: OrderStatus!
  createdAt: AWSDateTime!
}

type OrderItem {
  productId: String!
  name: String!
  quantity: Int!
  price: Float!
}

enum OrderStatus {
  CREATED
  PAID
  SHIPPED
  DELIVERED
  CANCELLED
}

type Query {
  getOrder(orderId: ID!): Order
  listOrders(userId: String!, limit: Int, nextToken: String): OrderConnection!
}

type Mutation {
  createOrder(input: CreateOrderInput!): Order!
  updateOrderStatus(orderId: ID!, status: OrderStatus!): Order!
}

type Subscription {
  onOrderStatusChanged(orderId: ID!): Order
    @aws_subscribe(mutations: ["updateOrderStatus"])
}

# websocket_handler.py
import json
import boto3
import os

dynamodb = boto3.resource('dynamodb')
connections_table = dynamodb.Table(os.environ['CONNECTIONS_TABLE'])

def connect(event, context):
    """WebSocket 연결"""
    connection_id = event['requestContext']['connectionId']
    user_id = event['requestContext']['authorizer']['userId']

    connections_table.put_item(Item={
        'connectionId': connection_id,
        'userId': user_id
    })

    return {'statusCode': 200}

def disconnect(event, context):
    """WebSocket 연결 해제"""
    connection_id = event['requestContext']['connectionId']
    connections_table.delete_item(Key={'connectionId': connection_id})
    return {'statusCode': 200}

def send_message(event, context):
    """메시지 전송"""
    domain = event['requestContext']['domainName']
    stage = event['requestContext']['stage']
    body = json.loads(event['body'])

    apigw = boto3.client(
        'apigatewaymanagementapi',
        endpoint_url=f'https://{domain}/{stage}'
    )

    # 모든 연결에 브로드캐스트
    connections = connections_table.scan()['Items']
    for conn in connections:
        try:
            apigw.post_to_connection(
                ConnectionId=conn['connectionId'],
                Data=json.dumps(body['message']).encode()
            )
        except apigw.exceptions.GoneException:
            connections_table.delete_item(
                Key={'connectionId': conn['connectionId']}
            )

    return {'statusCode': 200}

# DynamoDB Single Table Design
# PK/SK 패턴으로 여러 엔티티를 하나의 테이블에 저장

ENTITY_PATTERNS = {
    'User': {
        'PK': 'USER#user_id',
        'SK': 'PROFILE'
    },
    'Order': {
        'PK': 'USER#user_id',
        'SK': 'ORDER#order_id'
    },
    'OrderItem': {
        'PK': 'ORDER#order_id',
        'SK': 'ITEM#item_id'
    },
    'Product': {
        'PK': 'PRODUCT#product_id',
        'SK': 'DETAIL'
    }
}

# 접근 패턴별 쿼리
def get_user_with_orders(user_id):
    """사용자와 주문 목록을 한 번에 조회"""
    response = table.query(
        KeyConditionExpression='PK = :pk',
        ExpressionAttributeValues={':pk': f'USER#{user_id}'}
    )
    user = None
    orders = []
    for item in response['Items']:
        if item['SK'] == 'PROFILE':
            user = item
        elif item['SK'].startswith('ORDER#'):
            orders.append(item)
    return {'user': user, 'orders': orders}

def get_order_details(order_id):
    """주문 상세와 아이템을 한 번에 조회"""
    response = table.query(
        KeyConditionExpression='PK = :pk',
        ExpressionAttributeValues={':pk': f'ORDER#{order_id}'}
    )
    return response['Items']

# Aurora Serverless v2 + Lambda
Resources:
  AuroraCluster:
    Type: AWS::RDS::DBCluster
    Properties:
      Engine: aurora-postgresql
      EngineVersion: '15.4'
      ServerlessV2ScalingConfiguration:
        MinCapacity: 0.5
        MaxCapacity: 16
      EnableHttpEndpoint: true  # Data API 사용

  AuroraInstance:
    Type: AWS::RDS::DBInstance
    Properties:
      DBClusterIdentifier: !Ref AuroraCluster
      DBInstanceClass: db.serverless
      Engine: aurora-postgresql

  # RDS Proxy로 연결 관리
  RDSProxy:
    Type: AWS::RDS::DBProxy
    Properties:
      DBProxyName: orders-proxy
      EngineFamily: POSTGRESQL
      Auth:
        - AuthScheme: SECRETS
          SecretArn: !Ref DBSecret
          IAMAuth: REQUIRED
      VpcSubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2

# S3 이벤트 -> Lambda -> DynamoDB 파이프라인
import json
import boto3
import csv
import io

s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('ProcessedData')

def process_csv_upload(event, context):
    """S3에 업로드된 CSV를 처리"""
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    # S3에서 파일 읽기
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read().decode('utf-8')

    # CSV 파싱 및 배치 쓰기
    reader = csv.DictReader(io.StringIO(content))

    with table.batch_writer() as batch:
        for row in reader:
            batch.put_item(Item={
                'id': row['id'],
                'data': row,
                'sourceFile': key,
                'processedAt': context.get_remaining_time_in_millis()
            })

    return {
        'statusCode': 200,
        'processedFile': key
    }

메시징 선택 흐름:
1. 실시간 스트리밍이 필요한가?
   -> Yes: Kinesis Data Streams
   -> No: 다음으로

2. 여러 소비자에게 동시 전달?
   -> Yes: 다음으로
   -> No: SQS (단순 큐)

3. 복잡한 이벤트 라우팅/필터링?
   -> Yes: EventBridge
   -> No: SNS

4. 이벤트 재생이 필요한가?
   -> Yes: EventBridge (아카이브) 또는 Kinesis (보존)
   -> No: SNS/SQS

{
  "source": ["com.myapp.orders"],
  "detail-type": ["OrderCreated"],
  "detail": {
    "total": [{"numeric": [">=", 10000]}],
    "status": ["CREATED"],
    "items": {
      "category": ["electronics", "premium"]
    }
  }
}

# Dockerfile - Lambda 컨테이너 이미지
FROM public.ecr.aws/lambda/python:3.12

# 의존성 설치
COPY requirements.txt .
RUN pip install -r requirements.txt

# 애플리케이션 코드
COPY app/ ./app/

# Lambda 핸들러 지정
CMD ["app.main.handler"]

# app/main.py
import json
import numpy as np  # 큰 의존성도 OK (컨테이너 이미지)
from sklearn.ensemble import RandomForestClassifier

# 모델 로드 (콜드 스타트 시 1회)
model = RandomForestClassifier()

def handler(event, context):
    """ML 추론 Lambda"""
    features = np.array(event['features']).reshape(1, -1)
    prediction = model.predict(features)

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

Lambda 비용 = 요청 수 비용 + 실행 시간 비용

요청 수 비용:
  - 월 100만 건 무료
  - 이후 100만 건당 약 0.20 USD

실행 시간 비용 (x86):
  - 128MB: 0.0000000021 USD / ms
  - 512MB: 0.0000000083 USD / ms
  - 1024MB: 0.0000000167 USD / ms
  - 1769MB (1 vCPU): 0.0000000289 USD / ms
  - 10240MB: 0.0000001667 USD / ms

ARM64 (Graviton2) 비용:
  - x86 대비 약 20% 저렴
  - 성능은 동등하거나 우수

Provisioned Concurrency 추가 비용:
  - 프로비저닝: 0.0000041667 USD / GB-초
  - 실행: 0.0000000150 USD / GB-ms (일반보다 저렴)

# AWS Lambda Power Tuning 사용
# Step Functions 기반으로 최적 메모리를 찾아줌
aws stepfunctions start-execution \
  --state-machine-arn arn:aws:states:ap-northeast-2:123456789:stateMachine:powerTuning \
  --input '{
    "lambdaARN": "arn:aws:lambda:ap-northeast-2:123456789:function:my-function",
    "powerValues": [128, 256, 512, 1024, 1769, 3008],
    "num": 50,
    "payload": "{\"test\": true}",
    "parallelInvocation": true,
    "strategy": "cost"
  }'

워크로드 유형 판별:
1. 실행 시간 15분 이상? -> Container
2. 상시 트래픽 (초당 수백 건 이상)? -> Container (비용 효율)
3. 간헐적 트래픽? -> Serverless
4. GPU 필요? -> Container
5. 특수 런타임 필요? -> Container
6. 빠른 프로토타이핑? -> Serverless
7. WebSocket 장기 연결? -> Container
8. 배치 처리 (큰 데이터)? -> Step Functions + Lambda or Container

# Lambda Powertools - 구조화된 로깅, 추적, 메트릭
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

logger = Logger()
tracer = Tracer()
metrics = Metrics()
app = APIGatewayRestResolver()

@app.get("/orders/<order_id>")
@tracer.capture_method
def get_order(order_id: str):
    logger.info("Fetching order", extra={"order_id": order_id})

    order = fetch_order(order_id)

    metrics.add_metric(name="OrderFetched", unit=MetricUnit.Count, value=1)
    metrics.add_dimension(name="Environment", value="production")

    return {"order": order}

@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def handler(event, context):
    return app.resolve(event, context)

# X-Ray SDK로 외부 호출 추적
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

# 모든 AWS SDK 호출 자동 추적
patch_all()

@xray_recorder.capture('process_order')
def process_order(order):
    # 하위 세그먼트 생성
    subsegment = xray_recorder.begin_subsegment('validate')
    try:
        validate_order(order)
        subsegment.put_annotation('valid', True)
    except Exception as e:
        subsegment.put_annotation('valid', False)
        subsegment.add_exception(e)
        raise
    finally:
        xray_recorder.end_subsegment()

    # DynamoDB 호출 (자동 추적)
    save_order(order)

    # SNS 발행 (자동 추적)
    publish_event(order)

Resources:
  # Lambda 에러율 알람
  LambdaErrorAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: lambda-high-error-rate
      MetricName: Errors
      Namespace: AWS/Lambda
      Dimensions:
        - Name: FunctionName
          Value: !Ref MyFunction
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 2
      Threshold: 5
      ComparisonOperator: GreaterThanThreshold
      AlarmActions:
        - !Ref AlertTopic

  # Lambda 스로틀 알람
  LambdaThrottleAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: lambda-throttled
      MetricName: Throttles
      Namespace: AWS/Lambda
      Dimensions:
        - Name: FunctionName
          Value: !Ref MyFunction
      Statistic: Sum
      Period: 60
      EvaluationPeriods: 1
      Threshold: 0
      ComparisonOperator: GreaterThanThreshold
      AlarmActions:
        - !Ref AlertTopic

  # 동시성 사용률 알람
  ConcurrencyAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: lambda-high-concurrency
      MetricName: ConcurrentExecutions
      Namespace: AWS/Lambda
      Dimensions:
        - Name: FunctionName
          Value: !Ref MyFunction
      Statistic: Maximum
      Period: 60
      EvaluationPeriods: 3
      Threshold: 800
      ComparisonOperator: GreaterThanThreshold

# SAM CLI로 로컬 Lambda 실행
sam local invoke MyFunction \
  --event events/api-gateway.json \
  --env-vars env.json

# 로컬 API 서버 실행
sam local start-api --port 3000

# DynamoDB Local과 함께 사용
docker run -p 8000:8000 amazon/dynamodb-local
sam local invoke --docker-network host

# test_integration.py
import boto3
import pytest
import json
import time

STACK_NAME = 'my-serverless-app'
API_URL = None

@pytest.fixture(scope='session', autouse=True)
def setup():
    """CloudFormation 스택에서 API URL 가져오기"""
    global API_URL
    cfn = boto3.client('cloudformation')
    response = cfn.describe_stacks(StackName=STACK_NAME)
    outputs = response['Stacks'][0]['Outputs']
    API_URL = next(o['OutputValue'] for o in outputs if o['OutputKey'] == 'ApiUrl')

def test_create_order():
    """주문 생성 통합 테스트"""
    import requests

    response = requests.post(
        f'{API_URL}/orders',
        json={
            'userId': 'test-user',
            'items': [
                {'productId': 'p1', 'name': 'Widget', 'quantity': 2, 'price': 1000}
            ]
        },
        headers={'Authorization': f'Bearer {get_test_token()}'}
    )

    assert response.status_code == 201
    data = response.json()
    assert 'orderId' in data
    assert data['status'] == 'CREATED'
    assert data['total'] == 2000

def test_get_order():
    """주문 조회 통합 테스트"""
    import requests

    # 먼저 주문 생성
    create_response = requests.post(
        f'{API_URL}/orders',
        json={
            'userId': 'test-user',
            'items': [{'productId': 'p1', 'name': 'Widget', 'quantity': 1, 'price': 500}]
        },
        headers={'Authorization': f'Bearer {get_test_token()}'}
    )
    order_id = create_response.json()['orderId']

    # 조회
    response = requests.get(
        f'{API_URL}/orders/{order_id}',
        headers={'Authorization': f'Bearer {get_test_token()}'}
    )

    assert response.status_code == 200
    assert response.json()['orderId'] == order_id

# test_unit.py
import json
import pytest
from unittest.mock import patch, MagicMock
from moto import mock_dynamodb, mock_sns

@mock_dynamodb
@mock_sns
def test_create_order_handler():
    """Lambda 핸들러 단위 테스트"""
    import boto3

    # DynamoDB 테이블 생성
    dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-2')
    table = dynamodb.create_table(
        TableName='orders',
        KeySchema=[{'AttributeName': 'orderId', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'orderId', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )

    # SNS 토픽 생성
    sns = boto3.client('sns', region_name='ap-northeast-2')
    topic = sns.create_topic(Name='order-events')

    import os
    os.environ['ORDERS_TABLE'] = 'orders'
    os.environ['ORDER_TOPIC'] = topic['TopicArn']

    from src.orders.create import handler

    event = {
        'body': json.dumps({
            'userId': 'user123',
            'items': [{'productId': 'p1', 'name': 'Test', 'quantity': 1, 'price': 1000}]
        })
    }

    context = MagicMock()
    context.aws_request_id = 'test-request-id'

    response = handler(event, context)

    assert response['statusCode'] == 201
    body = json.loads(response['body'])
    assert body['userId'] == 'user123'
    assert body['total'] == 1000

클라이언트
  |
  v
[API Gateway] --> [Lambda: 주문 생성]
                     |
                     v
                  [DynamoDB: 주문 저장]
                     |
                     v
                  [EventBridge: OrderCreated 발행]
                     |
          +----------+----------+
          |          |          |
          v          v          v
     [Lambda:    [Lambda:    [Lambda:
      재고 확인]   결제 처리]   알림 발송]
          |          |
          v          v
     [DynamoDB]  [Stripe API]
          |
          v
     [Step Functions: 배송 워크플로]
          |
          v
     [Lambda: 배송 추적 업데이트]
          |
          v
     [WebSocket -> 클라이언트 실시간 알림]

[S3: 원본 업로드]
  |
  v
[EventBridge: S3 이벤트]
  |
  v
[Step Functions: 미디어 파이프라인]
  |
  +-> [Lambda: 메타데이터 추출]
  |
  +-> [Lambda: 썸네일 생성]
  |
  +-> [Lambda: 비디오 트랜스코딩 시작]
  |      |
  |      v
  |   [MediaConvert]
  |      |
  |      v
  |   [Lambda: 트랜스코딩 완료 처리]
  |
  +-> [Lambda: AI 태깅 (Rekognition)]
  |
  v
[DynamoDB: 메타데이터 저장]
  |
  v
[CloudFront: CDN 배포]