[AWS] 솔루션 설계자 관점의 서버리스 개요

AWS Lambda

기존 방식:
[EC2 서버 24/7 실행] → 비용 높음, 관리 필요

Lambda:
[이벤트 발생] → [Lambda 실행] → [종료]
└─ 사용한 만큼만 과금

• 서버 관리 불필요

• 자동 스케일링

• 밀리초 단위 과금

• 다양한 언어 지원

지원 언어

• Python

• Node.js

• Java

• Go

• C# (.NET)

• Ruby

• Custom Runtime (Rust, PHP 등)

제한사항

메모리: 128MB ~ 10GB
타임아웃: 최대 15분
디스크 (/tmp): 최대 10GB
환경 변수: 4KB
배포 패키지: 50MB (압축) / 250MB (압축 해제)

Lambda 함수 생성

Python 예시

import json

def lambda_handler(event, context):
    # 이벤트 파싱
    name = event.get('name', 'World')

    # 로직 실행
    message = f'Hello, {name}!'

    # 응답 반환
    return {
        'statusCode': 200,
        'body': json.dumps({'message': message})
    }

CLI로 생성

# 1. 함수 코드 작성 (lambda_function.py)

# 2. ZIP 패키징
zip function.zip lambda_function.py

# 3. Lambda 함수 생성
aws lambda create-function \
  --function-name my-function \
  --runtime python3.9 \
  --role arn:aws:iam::account-id:role/lambda-role \
  --handler lambda_function.lambda_handler \
  --zip-file fileb://function.zip \
  --memory-size 256 \
  --timeout 30

# 4. 함수 호출
aws lambda invoke \
  --function-name my-function \
  --payload '{"name": "AWS"}' \
  response.json

환경 변수 및 시크릿

import os
import boto3
import json

def lambda_handler(event, context):
    # 환경 변수
    db_host = os.environ['DB_HOST']

    # Secrets Manager에서 시크릿 가져오기
    secrets = boto3.client('secretsmanager')
    secret = secrets.get_secret_value(SecretId='my-db-password')
    db_password = json.loads(secret['SecretString'])['password']

    # 사용
    return {'statusCode': 200}

# 환경 변수 설정
aws lambda update-function-configuration \
  --function-name my-function \
  --environment "Variables={DB_HOST=db.example.com,ENV=production}"

Lambda 트리거

[S3] → 파일 업로드 시 → Lambda
[DynamoDB Streams] → 데이터 변경 시 → Lambda
[API Gateway] → HTTP 요청 시 → Lambda
[EventBridge] → 스케줄 또는 이벤트 → Lambda
[SNS/SQS] → 메시지 수신 시 → Lambda
[CloudWatch Logs] → 로그 패턴 감지 → Lambda

S3 트리거 예시

def lambda_handler(event, context):
    # S3 이벤트 파싱
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        print(f"Processing: {bucket}/{key}")

        # S3에서 파일 가져오기
        s3 = boto3.client('s3')
        obj = s3.get_object(Bucket=bucket, Key=key)
        data = obj['Body'].read()

        # 처리 로직
        # ...

# S3 버킷에 Lambda 트리거 추가
aws s3api put-bucket-notification-configuration \
  --bucket my-bucket \
  --notification-configuration '{
    "LambdaFunctionConfigurations": [{
      "LambdaFunctionArn": "arn:aws:lambda:...",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [{"Name": "suffix", "Value": ".jpg"}]
        }
      }
    }]
  }'

Lambda Layers

[Lambda Function 1]
[Lambda Function 2] → [Layer: requests 라이브러리]
[Lambda Function 3]

# Layer 생성
mkdir -p layer/python
pip install requests -t layer/python/
cd layer && zip -r layer.zip .

aws lambda publish-layer-version \
  --layer-name my-layer \
  --zip-file fileb://layer.zip \
  --compatible-runtimes python3.9

# Layer 추가
aws lambda update-function-configuration \
  --function-name my-function \
  --layers arn:aws:lambda:region:account-id:layer:my-layer:1

Lambda SnapStart (Java 전용)

aws lambda update-function-configuration \
  --function-name my-java-function \
  --snap-start ApplyOn=PublishedVersions

Lambda@Edge & CloudFront Functions

CloudFront Functions

function handler(event) {
    var request = event.request;
    var uri = request.uri;

    // index.html 추가
    if (uri.endsWith('/')) {
        request.uri += 'index.html';
    }

    return request;
}

• URL 리다이렉트

• 헤더 수정

• A/B 테스트

Lambda@Edge

def lambda_handler(event, context):
    request = event['Records'][0]['cf']['request']

    # 인증 확인
    headers = request['headers']
    if 'authorization' not in headers:
        return {
            'status': '401',
            'statusDescription': 'Unauthorized'
        }

    # 동적 콘텐츠 생성
    # 외부 API 호출
    # ...

    return request

Lambda in VPC

[Lambda in VPC]
    ↓
[Private Subnet]
    ↓
[RDS Database]

aws lambda update-function-configuration \
  --function-name my-function \
  --vpc-config "SubnetIds=subnet-xxx,subnet-yyy,SecurityGroupIds=sg-xxx"

• ENI 생성으로 초기 지연 발생

• NAT Gateway 필요 (인터넷 접근 시)

Amazon DynamoDB

관계형 DB (RDS):
테이블 → 행 → 열

NoSQL (DynamoDB):
테이블 → 항목 (Item) → 속성 (Attribute)

• 완전 관리형

• 밀리초 단위 지연시간

• 자동 스케일링

• 백업 및 복원

• 글로벌 테이블 (다중 리전)

기본 구조

테이블: Users
├─ Partition Key: UserId (필수)
├─ Sort Key: Timestamp (선택)
└─ Attributes: Name, Email, Age, ...

CRUD 작업

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Users')

# Create (항목 추가)
table.put_item(
    Item={
        'UserId': 'user123',
        'Name': 'John Doe',
        'Email': 'john@example.com',
        'Age': 30
    }
)

# Read (항목 조회)
response = table.get_item(
    Key={'UserId': 'user123'}
)
item = response.get('Item')

# Update (항목 수정)
table.update_item(
    Key={'UserId': 'user123'},
    UpdateExpression='SET Age = :age',
    ExpressionAttributeValues={':age': 31}
)

# Delete (항목 삭제)
table.delete_item(
    Key={'UserId': 'user123'}
)

# Query (Partition Key로 조회)
response = table.query(
    KeyConditionExpression='UserId = :uid',
    ExpressionAttributeValues={':uid': 'user123'}
)

# Scan (전체 테이블 스캔 - 비효율적)
response = table.scan(
    FilterExpression='Age > :age',
    ExpressionAttributeValues={':age': 25}
)

용량 모드

On-Demand (온디맨드)

- 자동 스케일링
- 요청당 과금
- 예측 불가능한 워크로드

Provisioned (프로비저닝)

- RCU/WCU 지정
- 저렴 (예측 가능한 워크로드)
- Auto Scaling 가능

RCU (Read Capacity Unit):
- 1 RCU = 4KB 강력한 일관성 읽기/초
- 또는 8KB 최종 일관성 읽기/초

WCU (Write Capacity Unit):
- 1 WCU = 1KB 쓰기/초

Global Secondary Index (GSI)

테이블: Users
- Partition Key: UserId

GSI: EmailIndex
- Partition Key: Email
→ Email로 조회 가능!

# GSI로 쿼리
response = table.query(
    IndexName='EmailIndex',
    KeyConditionExpression='Email = :email',
    ExpressionAttributeValues={':email': 'john@example.com'}
)

DynamoDB Streams

[DynamoDB] → [Streams] → [Lambda]
   데이터 변경      이벤트      처리

# Lambda로 Stream 처리
def lambda_handler(event, context):
    for record in event['Records']:
        if record['eventName'] == 'INSERT':
            new_item = record['dynamodb']['NewImage']
            print(f"New user: {new_item}")
        elif record['eventName'] == 'MODIFY':
            old_item = record['dynamodb']['OldImage']
            new_item = record['dynamodb']['NewImage']
            print(f"Updated: {old_item} → {new_item}")

API Gateway

[Client] → [API Gateway] → [Lambda]
                          → [EC2]
                          → [HTTP Endpoint]

• API 버전 관리

• 인증 및 권한 부여

• 요청 제한 (Throttling)

• 캐싱

• 모니터링

API 생성

# REST API 생성
aws apigateway create-rest-api \
  --name my-api \
  --description "My API"

# 리소스 생성 (GET /users)
aws apigateway create-resource \
  --rest-api-id api-id \
  --parent-id root-id \
  --path-part users

# 메서드 생성
aws apigateway put-method \
  --rest-api-id api-id \
  --resource-id resource-id \
  --http-method GET \
  --authorization-type NONE

# Lambda 통합
aws apigateway put-integration \
  --rest-api-id api-id \
  --resource-id resource-id \
  --http-method GET \
  --type AWS_PROXY \
  --integration-http-method POST \
  --uri arn:aws:apigateway:region:lambda:path/2015-03-31/functions/arn:aws:lambda:.../invocations

# 배포
aws apigateway create-deployment \
  --rest-api-id api-id \
  --stage-name prod

Lambda 통합

# Lambda 프록시 통합
def lambda_handler(event, context):
    # API Gateway 이벤트 파싱
    method = event['httpMethod']
    path = event['path']
    query = event.get('queryStringParameters', {})
    body = event.get('body', '')

    # 로직 처리
    if method == 'GET':
        result = {'message': 'Hello from Lambda!'}

    # API Gateway 응답 형식
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps(result)
    }

인증 및 권한

IAM 인증

import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

# API 호출 (IAM 인증)
session = boto3.Session()
credentials = session.get_credentials()

request = AWSRequest(
    method='GET',
    url='https://api-id.execute-api.region.amazonaws.com/prod/users'
)

SigV4Auth(credentials, 'execute-api', 'region').add_auth(request)
response = requests.get(request.url, headers=dict(request.headers))

Cognito 인증

// 프론트엔드에서 토큰 전송
fetch('https://api-id.execute-api.region.amazonaws.com/prod/users', {
  headers: {
    'Authorization': `Bearer ${idToken}`
  }
})

Lambda Authorizer

def lambda_handler(event, context):
    token = event['authorizationToken']

    # 토큰 검증
    if token == 'valid-token':
        return generate_policy('user', 'Allow', event['methodArn'])
    else:
        return generate_policy('user', 'Deny', event['methodArn'])

def generate_policy(principal_id, effect, resource):
    return {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': effect,
                'Resource': resource
            }]
        }
    }

요청 제한 (Throttling)

# Stage 설정
aws apigateway update-stage \
  --rest-api-id api-id \
  --stage-name prod \
  --patch-operations \
    op=replace,path=/throttle/rateLimit,value=1000 \
    op=replace,path=/throttle/burstLimit,value=2000

캐싱

# 캐시 활성화 (300초)
aws apigateway update-stage \
  --rest-api-id api-id \
  --stage-name prod \
  --patch-operations \
    op=replace,path=/cacheClusterEnabled,value=true \
    op=replace,path=/cacheClusterSize,value=0.5

Step Functions

[Step Functions]
├─ Task 1: Lambda (데이터 검증)
├─ Task 2: Lambda (데이터 처리)
├─ Choice: 조건 분기
│   ├─ Success → Task 3
│   └─ Fail → Task 4
└─ Task 5: Lambda (완료 알림)

{
  "Comment": "Order Processing",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:validate",
      "Next": "ProcessPayment"
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:payment",
      "Next": "IsPaymentSuccess"
    },
    "IsPaymentSuccess": {
      "Type": "Choice",
      "Choices": [{
        "Variable": "$.paymentStatus",
        "StringEquals": "SUCCESS",
        "Next": "ShipOrder"
      }],
      "Default": "PaymentFailed"
    },
    "ShipOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:ship",
      "End": true
    },
    "PaymentFailed": {
      "Type": "Fail",
      "Error": "PaymentFailed"
    }
  }
}

Amazon Cognito

[Frontend] → [Cognito] → [인증 토큰] → [API Gateway]

• 사용자 가입/로그인

• 소셜 로그인 (Google, Facebook)

• MFA

• 사용자 풀 관리

// 프론트엔드 인증
import { CognitoUserPool, CognitoUser, AuthenticationDetails } from 'amazon-cognito-identity-js';

const poolData = {
  UserPoolId: 'us-east-1_xxxxxxxxx',
  ClientId: 'xxxxxxxxxxxxxxxxxxxxxxxxxx'
};

const userPool = new CognitoUserPool(poolData);

// 로그인
const authenticationData = {
  Username: 'user@example.com',
  Password: 'password123',
};

const authenticationDetails = new AuthenticationDetails(authenticationData);
const cognitoUser = new CognitoUser({
  Username: 'user@example.com',
  Pool: userPool
});

cognitoUser.authenticateUser(authenticationDetails, {
  onSuccess: (result) => {
    const idToken = result.getIdToken().getJwtToken();
    // API 호출 시 사용
  },
  onFailure: (err) => {
    console.error(err);
  }
});

실전 서버리스 아키텍처

시나리오 1: 이미지 리사이징

[S3: uploads/] (원본 이미지 업로드)
    ↓ 트리거
[Lambda: Resize]
    ├─ Sharp 라이브러리로 리사이징
    └─ 썸네일 생성
    ↓
[S3: thumbnails/] (썸네일 저장)

시나리오 2: REST API (CRUD)

[API Gateway]
├─ GET /users → Lambda → DynamoDB Scan
├─ GET /users/{id} → Lambda → DynamoDB GetItem
├─ POST /users → Lambda → DynamoDB PutItem
├─ PUT /users/{id} → Lambda → DynamoDB UpdateItem
└─ DELETE /users/{id} → Lambda → DynamoDB DeleteItem

시나리오 3: 실시간 알림

[DynamoDB] (데이터 변경)
    ↓ Streams
[Lambda: Process Change]
    ↓
[SNS Topic]
    ├─ Email
    ├─ SMS
    └─ WebSocket (API Gateway)

모범 사례

1. Lambda 최적화

# ❌ 나쁜 예: 핸들러 내부에서 초기화
def lambda_handler(event, context):
    import boto3
    s3 = boto3.client('s3')  # 매번 생성!
    # ...

# ✅ 좋은 예: 전역으로 초기화
import boto3
s3 = boto3.client('s3')  # 한 번만 생성

def lambda_handler(event, context):
    # s3 사용
    # ...

2. DynamoDB 설계

❌ Scan 피하기 (전체 테이블 스캔)
✅ Query 사용 (Partition Key)
✅ GSI 활용
✅ 배치 작업 (BatchGetItem, BatchWriteItem)

3. API Gateway 캐싱

자주 변경되지 않는 데이터:
- TTL: 300초 (5분)
- 캐시 크기: 0.5GB

동적 데이터:
- 캐싱 비활성화

비용 최적화

Lambda

비용 = 요청 수 + 실행 시간

무료 티어:
- 100만 요청/월
- 40만 GB-초/월

최적화:
1. 메모리 적절히 설정 (성능 ↑ = 시간 ↓)
2. 불필요한 라이브러리 제거
3. 콜드 스타트 최소화

DynamoDB

On-Demand vs Provisioned:

On-Demand:
- 예측 불가능한 트래픽
- 초기 단계

Provisioned + Auto Scaling:
- 예측 가능한 트래픽
- 최대 70% 절감