Serverless #

# ✗ Anti-pattern: inisialisasi mahal di dalam handler
import boto3

def handler(event, context):
    # Ini dieksekusi SETIAP kali handler dipanggil
    # Bahkan untuk warm invocation yang sudah punya connection
    db_client = boto3.client('dynamodb')  # overhead setiap call
    s3_client = boto3.client('s3')
    config = load_config_from_s3()       # network call setiap kali!

    # ... proses event ...

# ✓ Benar: inisialisasi di luar handler (global scope)
# Dieksekusi SEKALI saat cold start, di-reuse di warm invocation
import boto3
import json

# Global scope — dieksekusi saat container diinisialisasi
db_client = boto3.client('dynamodb')
s3_client = boto3.client('s3')
config = None  # lazy load

def get_config():
    global config
    if config is None:
        response = s3_client.get_object(Bucket='config', Key='app.json')
        config = json.loads(response['Body'].read())
    return config

def handler(event, context):
    # Handler dipanggil setiap invocation
    # Tapi db_client, s3_client sudah ada dari global scope
    cfg = get_config()  # hanya load pertama kali per container

    # ... proses event menggunakan client yang sudah ada ...
    return process_event(event, cfg, db_client)

# Teknik mitigasi cold start lainnya:

# 1. Kurangi ukuran deployment package
# Semakin kecil package → download lebih cepat → cold start lebih pendek
# requirements.txt: hanya include yang benar-benar dibutuhkan
# Gunakan Lambda Layers untuk dependency yang jarang berubah

# 2. Provisioned Concurrency (AWS Lambda)
# Pre-warm N container sebelum ada request
# Bayar extra tapi cold start dihilangkan untuk traffic yang bisa diprediksi
# aws lambda put-provisioned-concurrency-config \
#     --function-name my-function \
#     --qualifier LIVE \
#     --provisioned-concurrent-executions 5

# 3. Scheduled ping untuk keep-alive (workaround murah)
# CloudWatch Event setiap 5 menit → invoke Lambda
# → Container tetap warm
# Tapi ini tidak scale untuk fungsi yang butuh banyak concurrent instance

# 4. Pilih runtime dengan cold start lebih cepat
# Cold start ranking (dari tercepat):
# Go/Rust > Node.js ≈ Python > C# > Java
# Untuk fungsi yang latency-sensitive, hindari Java tanpa GraalVM native image

# ✗ Anti-pattern: menyimpan state di memory container

# State ini tidak konsisten antar invocation!
request_counter = 0  # container A: 5, container B: 3, container C: 0

def handler(event, context):
    global request_counter
    request_counter += 1
    # request_counter berbeda di setiap container
    # Tidak bisa diandalkan untuk apapun

# ✓ Benar: semua state di external storage
import boto3
import redis

# Untuk counter yang akurat
dynamodb = boto3.client('dynamodb')

def increment_request_counter(function_name: str) -> int:
    response = dynamodb.update_item(
        TableName='counters',
        Key={'name': {'S': function_name}},
        UpdateExpression='ADD #count :one',
        ExpressionAttributeNames={'#count': 'count'},
        ExpressionAttributeValues={':one': {'N': '1'}},
        ReturnValues='UPDATED_NEW'
    )
    return int(response['Attributes']['count']['N'])

# Untuk session/cache
redis_client = redis.Redis(
    host=os.environ['REDIS_HOST'],
    port=6379,
    decode_responses=True
)

def handler(event, context):
    # State selalu dari external storage
    counter = increment_request_counter('my-function')
    session = redis_client.get(f"session:{event['session_id']}")
    # ... proses ...

# Contoh Lambda function yang menangani berbagai trigger

# 1. HTTP request via API Gateway
def api_handler(event, context):
    """Handler untuk HTTP request."""
    path = event['path']
    method = event['httpMethod']
    body = json.loads(event.get('body', '{}'))

    if method == 'POST' and path == '/orders':
        return create_order(body)

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

# 2. S3 event — ketika file di-upload
def s3_event_handler(event, context):
    """Proses file yang baru di-upload."""
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # Process file: resize gambar, parse CSV, scan virus, dll
        process_uploaded_file(bucket, key)

# 3. SQS queue — proses pesan dari antrian
def sqs_handler(event, context):
    """Proses pesan dari SQS queue."""
    failed_messages = []

    for record in event['Records']:
        message_id = record['messageId']
        body = json.loads(record['body'])

        try:
            process_message(body)
        except Exception as e:
            # Tandai sebagai gagal — akan kembali ke queue atau ke DLQ
            failed_messages.append({'itemIdentifier': message_id})
            logger.error(f"Failed to process message {message_id}: {e}")

    # Return batch item failures — hanya yang gagal yang di-retry
    return {'batchItemFailures': failed_messages}

# 4. CloudWatch Events (scheduled)
def scheduled_handler(event, context):
    """Dijalankan setiap hari jam 08:00."""
    send_daily_report()
    cleanup_expired_sessions()
    return {'status': 'completed'}

# 5. DynamoDB Stream — react terhadap perubahan database
def dynamodb_stream_handler(event, context):
    """Trigger ketika ada record baru atau berubah di DynamoDB."""
    for record in event['Records']:
        if record['eventName'] == 'INSERT':
            new_item = record['dynamodb']['NewImage']
            user_id = new_item['user_id']['S']
            send_welcome_email(user_id)

# Best practices untuk observability di Lambda

import json
import logging
import time
from functools import wraps
import boto3

# Structured logging — lebih mudah di-parse di CloudWatch Logs Insights
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

def log_event(event_type: str, **kwargs):
    """Log structured event untuk observability."""
    logger.info(json.dumps({
        'event_type': event_type,
        'timestamp': time.time(),
        **kwargs
    }))

# Decorator untuk tracing dan error handling
def traced_handler(func):
    @wraps(func)
    def wrapper(event, context):
        start_time = time.time()

        # Log invocation metadata
        log_event('invocation_start', {
            'function_name': context.function_name,
            'request_id': context.aws_request_id,
            'remaining_time_ms': context.get_remaining_time_in_millis(),
            'memory_limit_mb': context.memory_limit_in_mb,
        })

        try:
            result = func(event, context)

            duration_ms = (time.time() - start_time) * 1000
            log_event('invocation_success', {
                'duration_ms': round(duration_ms, 2),
                'request_id': context.aws_request_id,
            })

            return result

        except Exception as e:
            duration_ms = (time.time() - start_time) * 1000
            log_event('invocation_error', {
                'error_type': type(e).__name__,
                'error_message': str(e),
                'duration_ms': round(duration_ms, 2),
                'request_id': context.aws_request_id,
            })

            # Re-raise agar Lambda tahu ini adalah error
            raise

    return wrapper

@traced_handler
def handler(event, context):
    # Fungsi handler utama
    pass

# AWS X-Ray untuk distributed tracing

from aws_xray_sdk.core import xray_recorder, patch_all
from aws_xray_sdk.core import xray_recorder

# Patch semua AWS SDK call untuk auto-tracing
patch_all()

@xray_recorder.capture('process_order')
def process_order(order_data: dict):
    """Function ini akan muncul sebagai subsegment di X-Ray trace."""

    with xray_recorder.in_subsegment('validate_order') as subsegment:
        subsegment.put_annotation('order_id', order_data['id'])
        validate_order(order_data)

    with xray_recorder.in_subsegment('save_to_database') as subsegment:
        order = save_order(order_data)
        subsegment.put_metadata('saved_order', order)

    return order

# serverless.yml — menggunakan Serverless Framework

service: my-app

provider:
  name: aws
  runtime: python3.12
  region: ap-southeast-1
  memorySize: 256  # MB, default untuk semua fungsi
  timeout: 30      # detik, default
  environment:
    APP_ENV: ${opt:stage, 'development'}
    DATABASE_URL: ${env:DATABASE_URL}

  # IAM role — least privilege
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:UpdateItem
          Resource: !GetAtt OrdersTable.Arn

functions:
  # HTTP API
  api:
    handler: src/api.handler
    events:
      - httpApi:
          path: /orders
          method: POST
      - httpApi:
          path: /orders/{id}
          method: GET
    memorySize: 512  # override default untuk fungsi yang butuh lebih

  # Scheduled job
  dailyReport:
    handler: src/reports.daily_handler
    timeout: 300  # 5 menit untuk proses report
    events:
      - schedule:
          rate: cron(0 8 * * ? *)  # Setiap hari jam 08:00 UTC
          enabled: true

  # SQS consumer
  orderProcessor:
    handler: src/processor.sqs_handler
    reservedConcurrency: 10  # Limit concurrency untuk melindungi downstream
    events:
      - sqs:
          arn: !GetAtt OrderQueue.Arn
          batchSize: 10
          functionResponseType: ReportBatchItemFailures

resources:
  Resources:
    OrdersTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:service}-${opt:stage}-orders
        BillingMode: PAY_PER_REQUEST  # On-demand — no capacity planning
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH

    OrderQueue:
      Type: AWS::SQS::Queue
      Properties:
        QueueName: ${self:service}-${opt:stage}-orders
        VisibilityTimeout: 60
        RedrivePolicy:
          deadLetterTargetArn: !GetAtt OrderDLQ.Arn
          maxReceiveCount: 3  # Retry 3 kali sebelum ke DLQ

    OrderDLQ:
      Type: AWS::SQS::Queue
      Properties:
        QueueName: ${self:service}-${opt:stage}-orders-dlq
        MessageRetentionPeriod: 1209600  # 14 hari

# ✗ Anti-pattern 1: inisialisasi mahal di dalam handler
def handler(event, context):
    db = connect_database()  # connection baru setiap invocation!
    config = load_from_s3()  # network call setiap invocation!

# ✓ Solusi: global scope untuk inisialisasi yang bisa di-reuse
db = connect_database()  # sekali per container
config = load_from_s3()  # sekali per container

def handler(event, context):
    return process(event, db, config)  # reuse dari global

# ✗ Anti-pattern 2: menyimpan state di /tmp dan mengandalkannya persist
def handler(event, context):
    with open('/tmp/data.json', 'w') as f:
        json.dump(data, f)
    # Invocation berikutnya mungkin container berbeda — file tidak ada!

# ✓ Solusi: gunakan S3 atau database untuk persistent state

# ✗ Anti-pattern 3: timeout yang terlalu pendek tanpa retry logic
# Default Lambda timeout: 3 detik
# Jika downstream service lambat → timeout sebelum selesai
# ✓ Solusi: set timeout yang realistic + implementasi retry di SQS

# ✗ Anti-pattern 4: tidak ada Dead Letter Queue
# Message gagal diproses → hilang begitu saja
# ✓ Solusi: selalu konfigurasi DLQ untuk SQS consumer dan async invocation

# ✗ Anti-pattern 5: concurrency tidak di-limit untuk downstream protection
# Lambda scale ke ribuan concurrent → flood database atau third-party API
# ✓ Solusi: set reservedConcurrency atau gunakan SQS untuk throttling alami

# ✗ Anti-pattern 6: fungsi yang melakukan terlalu banyak (monolithic Lambda)
# Satu fungsi handle semua: auth + business logic + data access + notification
# ✓ Solusi: single responsibility per fungsi, composisi via event/message