FastAPI Throtting(요청속도 제한) 처리(feat. Redis helm chart)

API를 개발할 때 트래픽 제한(Throttling)은 필수적인 기능 중 하나입니다. 사용자가 과도한 요청을 보내는 것을 방지하여 서비스 안정성을 유지하고, 악의적인 트래픽을 차단하는 데 도움을 줍니다. FastAPI에서 사용할 수 있는 여러 Throttling 라이브러리가 있지만, 이번 글에서는 SlowAPI를 선택하여 적용하는 과정을 소개합니다

Too many request

 

Throttling 라이브러리 비교

Throttling을 구현할 수 있는 라이브러리로 여러 가지가 있지만, 주요 라이브러리들을 비교해 본 후 SlowAPI를 선택하게 되었습니다.

라이브러리	start	특징	단점
fastapi-limiter	약 1.6k	Redis 기반, FastAPI 전용	설정이 복잡하고 유연성이 부족
slowapi	약 2.3k	Starlette/FastAPI와 호환, 단순한 설정	공식 문서가 부족

SlowAPI를 선정한 이유는 다음과 같습니다:

1. FastAPI와의 높은 호환성: Starlette 기반이라 FastAPI 프로젝트와 잘 어울립니다.
2. GitHub Star 수가 높음: 유지보수 가능성이 높다고 판단했습니다.
3. 유연한 요구사항 반영 가능: 사용자 인증 기반의 요청 제한을 쉽게 구현할 수 있습니다.

---

SlowAPI를 이용한 Throttling 구현

SlowAPI를 이용하여 요청을 제한하는 기능을 추가해 보겠습니다.

from core import settings
from core.exception import TooManyRequestsHTTPException
from database.session import redis_client
from fastapi import Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_ipaddr

# 요청에서 이메일을 가져오는 함수
def get_email(request: Request):
    try:
        return request.state.email
    except Exception:
        return None

# 특정 클라이언트를 차단하는 함수
def block_client(client: str, time=3600):
    try:
        block_key = f"blocked:{client}"
        redis_client.setex(block_key, time, "blocked")
    except Exception:
        pass

# 특정 클라이언트가 차단되었는지 확인하는 함수
def is_blocked(client: str) -> bool:
    try:
        block_key = f"blocked:{client}"
        return redis_client.get(block_key) is not None
    except Exception:
        return False

# Rate Limit 초과 시 block 처리 핸들러
def custom_rate_limit_exceeded_handler(
    request: Request,
    exc: RateLimitExceeded,
):
    if email := get_email(request):
        block_client(email)
    else:
        block_client(get_ipaddr(request))
    return _rate_limit_exceeded_handler(request, exc)

# IP 기반 요청 차단 확인 Depends
async def check_ip_blocked(request: Request) -> Request:
    if is_blocked(get_ipaddr(request)):
        raise TooManyRequestsHTTPException()
    return request

# 이메일 기반 요청 차단 확인 Depends
async def check_email_blocked(request: Request) -> Request:
    body = await request.json()
    client = get_ipaddr(request)
    
    if email := body.get("email", None):
        client = email
        request.state.email = body.get("email", None)
    
    if is_blocked(client):
        raise TooManyRequestsHTTPException()
    
    return request

# IP 기반 제한 설정
ip_limiter = Limiter(
    key_func=get_ipaddr,
    storage_uri=f"redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:{settings.REDIS_PORT}",
    in_memory_fallback_enabled=True, # Redis 미동작 시 제한 미처리
)

# 이메일 기반 제한 설정
email_limiter = Limiter(
    key_func=get_email,
    storage_uri=f"redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:{settings.REDIS_PORT}",
    in_memory_fallback_enabled=True, # Redis 미동작 시 제한 미처리
)

코드 설명

• ip_limiter, check_ip_blocked를 사용해서 IP 요청 제한을 설정합니다.
• email_limiter, check_email_blocked를 사용해서 IP 요청 제한을 설정합니다.
• block을 따로 관리하는 이유는 slowAPI는 TTL 기준으로 차단이 되기 때문에 속도 제한 기준과 제한 시간을 따로 설정하기 위함입니다.
  • 예를 들면 분당 100회 초과 시 1시간 제한 을 가능하게 하기 위함.
• request.state.email을 설정하는 이유:
  • slowapi의 key_func 인자는 동기 함수여야 합니다.
  • request.json()은 비동기 함수이므로, get_email()에서 직접 사용할 수 없습니다.
  • 따라서 요청 본문에서 이메일을 추출하여 request.state.email에 저장하고 동기적으로 가져오도록 했습니다.
• in_memory_fallback_enabled=True 설정 이유:
  • Redis 장애 발생 시 Throttling을 비활성화하기 위함입니다.
  • API 가용성을 보장해야 하므로, Redis가 다운되더라도 API가 정상 동작하도록 구성했습니다.

---

SlowAPI 적용 방법

1. slowapi 라이브러리 설치

 pip install slowapi

2. FastAPI 프로젝트에 적용

from fastapi import FastAPI, Request, Depends
from throttling import ip_limiter, email_limiter, check_email_blocked

app = FastAPI()

@app.post(
    "/test",
    summary="이메일 인증번호 검증",
    dependencies=[Depends(check_email_blocked)],
)
@email_limiter.limit("10/minute")
@ip_limiter.limit("100/minute")
async def test(
    request: Request,
    body: schema_auths.SendEmailAuthNumberRequest,
) -> Any:
    return None

 

4. 실행 후 테스트

• 1분에 10번 초과 요청하면 429 Too Many Requests 응답을 받게 됩니다.

curl -X POST http://127.0.0.1:8000/test -H "Content-Type: application/json" -d '{"email": "test@example.com"}'

5. 테스트 결과

---

k8s 환경에서 redis를 pod로 띄워 테스트하기

bitnami의 redis 를 사용하여 helm 차트를 구성하였습니다.

구성

1. values.yaml

redis:
  namespaceOverride: "dev"
  architecture: replication
  registry: public.ecr.aws/bitnami/redis
  image:
    registry: "public.ecr.aws/bitnami"
    repository: "redis"
    tag: "7.4.2-debian-12-r6"
    pullPolicy: IfNotPresent
  service:
    type: ClusterIP
    port: 6379
  auth:
    enabled: true
    password: "redis-password"
  ingress:
    enabled: false
  master:
    count: 1
    port: 6379
    resources:
      requests:
        memory: "128Mi"
        cpu: "50m"
      limits:
        memory: "256Mi"
        cpu: "100m"
  replica:
    replicaCount: 2
    port: 6379
    resources:
      requests:
        memory: "128Mi"
        cpu: "50m"
      limits:
        memory: "256Mi"
        cpu: "100m"
  global:
    storageClass: ""
    imageRegistry: ""
    imagePullSecrets: []
    nameOverride: "redis"
    fullnameOverride: "redis"

 

2. Chart.yaml

apiVersion: v2
name: redis
description: A Helm chart for deploying Redis with Master-Replica setup

type: application

version: 0.1.0

appVersion: "7.4.2"

dependencies:
  - name: redis
    version: "20.4.0"
    repository: "oci://registry-1.docker.io/bitnamicharts"

동작 

helm dependency update
helm upgrade --install redis . -n dev