⚙️ 시스템 성능 개선 실전 가이드

import time
import logging
from contextlib import contextmanager

logger = logging.getLogger(__name__)

@contextmanager
def timer(label: str):
    """요청 처리 각 구간 시간을 측정하는 컨텍스트 매니저"""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.info(f"[PERF] {label}: {elapsed_ms:.1f}ms")

# 사용 예시 — FastAPI 핸들러
async def get_dashboard(user_id: int):
    with timer("db.fetch_user"):
        user = await db.get_user(user_id)

    with timer("db.fetch_stats"):
        stats = await db.get_stats(user_id)

    with timer("cache.set"):
        await redis.set(f"dashboard:{user_id}", stats, ex=300)

    return {"user": user, "stats": stats}

# 로그 출력 예시:
# [PERF] db.fetch_user: 3.2ms
# [PERF] db.fetch_stats: 847.5ms  ← 병목!
# [PERF] cache.set: 1.1ms

-- 문제: user_id + created_at 복합 조회가 Full Scan
SELECT * FROM orders
WHERE user_id = 123
ORDER BY created_at DESC
LIMIT 20;

-- 해결: 복합 인덱스 생성 (선택도 높은 컬럼을 앞에)
CREATE INDEX CONCURRENTLY idx_orders_user_created
  ON orders (user_id, created_at DESC);

-- 커버링 인덱스: SELECT 컬럼까지 인덱스에 포함 → 테이블 접근 0회
CREATE INDEX CONCURRENTLY idx_orders_cover
  ON orders (user_id, created_at DESC)
  INCLUDE (status, total_amount);

# 나쁜 예: N+1 쿼리 (게시글 10개 → 쿼리 11회)
posts = await db.query("SELECT * FROM posts LIMIT 10")
for post in posts:
    # 게시글마다 별도 쿼리 실행
    author = await db.query("SELECT * FROM users WHERE id = ?", post.user_id)
    post.author = author

# 좋은 예 1: JOIN으로 단일 쿼리
posts = await db.query("""
    SELECT p.*, u.name AS author_name, u.avatar AS author_avatar
    FROM posts p
    JOIN users u ON u.id = p.user_id
    LIMIT 10
""")

# 좋은 예 2: IN 절로 배치 조회 (ORM 사용 시)
posts = await Post.query.limit(10).all()
user_ids = {p.user_id for p in posts}
users = {u.id: u for u in await User.query.filter(User.id.in_(user_ids)).all()}
for post in posts:
    post.author = users[post.user_id]

-- PostgreSQL: 실제 실행 계획 + 소요 시간 확인
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT o.*, u.name
FROM orders o
JOIN users u ON u.id = o.user_id
WHERE o.status = 'pending'
ORDER BY o.created_at DESC
LIMIT 50;

-- 핵심 키워드 해석
-- Seq Scan     → 풀스캔 (인덱스 없음 또는 미사용)
-- Index Scan   → 인덱스 사용 (양호)
-- Index Only   → 커버링 인덱스 (최적)
-- Hash Join    → 대용량 JOIN (정상)
-- Nested Loop  → 소용량 JOIN (정상), 대용량이면 문제
-- rows=10000 / actual rows=1  → 통계 오래됨 → ANALYZE 실행

-- 통계 갱신 (실행 계획 추정치가 실제와 크게 다를 때)
ANALYZE orders;

-- 인덱스 사용 현황 확인 (사용 안 된 인덱스 제거 대상)
SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
ORDER BY idx_scan ASC;

import json
import hashlib
from functools import wraps
import redis.asyncio as redis

r = redis.from_url("redis://localhost:6379", decode_responses=True)

def cache(ttl: int = 300, key_prefix: str = ""):
    """결과를 Redis에 캐시하는 데코레이터"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 캐시 키: 함수명 + 인자 해시
            raw_key = f"{key_prefix or func.__name__}:{args}:{sorted(kwargs.items())}"
            cache_key = hashlib.md5(raw_key.encode()).hexdigest()

            cached = await r.get(cache_key)
            if cached:
                return json.loads(cached)  # 캐시 히트

            result = await func(*args, **kwargs)
            await r.set(cache_key, json.dumps(result), ex=ttl)
            return result
        return wrapper
    return decorator

# 사용 예시
@cache(ttl=60, key_prefix="user_stats")
async def get_user_stats(user_id: int) -> dict:
    """DB 집계 쿼리 — 60초 캐시"""
    return await db.query(
        "SELECT COUNT(*) orders, SUM(amount) total FROM orders WHERE user_id = ?",
        user_id,
    )

// Next.js App Router — 응답별 캐시 헤더 설정
import { NextResponse } from 'next/server';

// 공개 데이터: CDN이 1시간 캐시, 브라우저도 5분 캐시
export async function GET() {
  const data = await fetchPublicData();
  return NextResponse.json(data, {
    headers: {
      'Cache-Control': 'public, s-maxage=3600, stale-while-revalidate=86400',
    },
  });
}

// 사용자별 데이터: CDN 캐시 금지, 브라우저 1분 캐시
export async function GET(req: Request) {
  const data = await fetchUserData();
  return NextResponse.json(data, {
    headers: {
      'Cache-Control': 'private, max-age=60',
    },
  });
}

# Cloudflare Cache Rules 예시 (wrangler.jsonc)
# 정적 에셋: 1년 캐시
# /api/public/*: Edge 30분 캐시
# /api/user/*: 캐시 없음 (bypass)

# Cache-Control 헤더 가이드
# public, s-maxage=N      → CDN N초 캐시
# stale-while-revalidate  → 만료 후에도 구버전 제공하며 백그라운드 갱신
# no-store                → 캐시 완전 금지 (민감 데이터)
# must-revalidate         → 만료 즉시 서버 재검증 필수

# SQLAlchemy 비동기 커넥션 풀 설정
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,          # 상시 유지 커넥션 수 (CPU 코어 × 2~4 기준)
    max_overflow=10,       # 피크 시 추가 허용 커넥션 (pool_size + max_overflow = 최대)
    pool_timeout=30,       # 풀에서 커넥션 대기 최대 시간(초)
    pool_recycle=1800,     # 30분 후 커넥션 재생성 (DB 타임아웃 방지)
    pool_pre_ping=True,    # 사용 전 커넥션 유효성 검사
)

# PostgreSQL max_connections 설정 기준
# max_connections = (CPU 코어 수 × 4) + 여유분
# 예: 8코어 서버 → max_connections = 200

# PgBouncer (커넥션 풀러) — 수백 개 앱 커넥션을 DB 10개로 압축
[databases]
mydb = host=127.0.0.1 port=5432 dbname=mydb

[pgbouncer]
pool_mode = transaction    # 트랜잭션 단위 풀링 (가장 효율적)
max_client_conn = 1000    # 앱에서 받는 최대 커넥션
default_pool_size = 20    # DB로 내보내는 실제 커넥션

import asyncio

# 나쁜 예: 순차 실행 (합계 시간 = 각 작업 시간의 합)
async def get_dashboard_slow(user_id: int):
    profile = await fetch_profile(user_id)    # 50ms
    orders  = await fetch_orders(user_id)     # 80ms
    stats   = await fetch_stats(user_id)      # 60ms
    # 총 190ms

# 좋은 예: 병렬 실행 (합계 시간 ≈ 가장 느린 작업 시간)
async def get_dashboard_fast(user_id: int):
    profile, orders, stats = await asyncio.gather(
        fetch_profile(user_id),   # ┐
        fetch_orders(user_id),    # ├─ 동시 실행
        fetch_stats(user_id),     # ┘
    )
    # 총 ~80ms (가장 느린 orders 기준)

# 커서 기반 페이지네이션 — OFFSET 방식보다 대용량에서 일관된 성능
async def get_orders_cursor(user_id: int, cursor: int | None, limit: int = 20):
    """
    OFFSET 방식은 10만 번째 페이지에서 풀스캔에 가까워짐.
    커서(마지막 id) 방식은 항상 인덱스만 사용.
    """
    query = """
        SELECT id, status, total_amount, created_at
        FROM orders
        WHERE user_id = :user_id
          AND (:cursor IS NULL OR id < :cursor)
        ORDER BY id DESC
        LIMIT :limit
    """
    rows = await db.fetch_all(query, {"user_id": user_id, "cursor": cursor, "limit": limit})
    next_cursor = rows[-1]["id"] if len(rows) == limit else None
    return {"items": rows, "next_cursor": next_cursor}

import http from 'k6/http';
import { check, sleep } from 'k6';
import { Trend } from 'k6/metrics';

const dbQueryTime = new Trend('db_query_time', true);

export const options = {
  stages: [
    { duration: '1m', target: 50  }, // 워밍업
    { duration: '3m', target: 200 }, // 정상 부하
    { duration: '2m', target: 500 }, // 피크 부하
    { duration: '1m', target: 0   }, // 쿨다운
  ],
  thresholds: {
    'http_req_duration':          ['p(95)<500', 'p(99)<1000'],
    'http_req_failed':            ['rate<0.001'],
    'db_query_time':              ['p(95)<100'],
  },
};

export default function () {
  // 1. 목록 조회 (DB + 캐시 검증)
  const listRes = http.get('http://localhost:3000/api/orders?limit=20', {
    tags: { name: 'list' },
  });
  check(listRes, {
    'list status 200': (r) => r.status === 200,
    'list has items':  (r) => r.json()?.items?.length > 0,
  });

  // 서버가 응답 헤더에 X-DB-Time을 추가하면 측정 가능
  const dbTime = listRes.headers['X-DB-Time'];
  if (dbTime) dbQueryTime.add(Number(dbTime));

  sleep(0.5);

  // 2. 단건 조회 (캐시 히트율 확인)
  const id = Math.floor(Math.random() * 1000) + 1;
  const detailRes = http.get(`http://localhost:3000/api/orders/${id}`, {
    tags: { name: 'detail' },
  });
  check(detailRes, {
    'detail status 200 or 404': (r) => [200, 404].includes(r.status),
  });

  sleep(Math.random() + 0.5);
}