처리율 제한 장치 (Rate Limiter) 도입

토큰-버킷 알고리즘을 사용하는 bucket4j 도입기

PreviousI/O와 트랜잭션 분리하기 Next외부 API 의존성을 줄여보자

Last updated 9 days ago

Was this helpful?

처리율 제한 장치 (Rate Limiter) 도입

토큰-버킷 알고리즘을 사용하는 bucket4j 도입기

처리율 제한 장치를 도입하게 된 계기와 어떻게 도입했는지 정리해보고자 한다.

처리율 제한 장치 도입 계기

현재 스부키 프로젝트에서의 책 검색 기능은 카카오의 API를 통해 검색을 진행하고 있다. 이러한 상황에서 카카오는 일간 호출 횟수를 다음과 같이 지정하고 있다.

만약, 악의적인 사용자가 하루에 30,000건을 모두 소진해버린다면 우리의 서비스는 어떤 문제가 발생할까?

사용자가 책을 검색하고 책을 등록하며, 등록된 책에 대한 평가를 내리는 시스템 중 가장 핵심적인 부분이 정상적으로 동작하지 않는다!

일간 호출 횟수가 제한되어 있다.
외부 API가 호출되지 않는다면 시스템적 큰 문제가 일어난다.

이러한 이유로 처리율 제한 장치를 도입하고자 한다. (나중에 캐싱으로도 같은 요청에 대해 외부 API를 안타게 해보는 방법도 고민해보자.)

간단하게 알아보는 처리율 제한 장치

간단하게 장점을 되짚어보자면,

DoS(Denial of Service) 공격에 의한 자원 고갈을 방지할 수 있다.
비용을 절감한다.
서버 과부하를 막는다.

처리율 제한장치는 서버 내부, 미들웨어, API Gateway 에 존재할 수 있다. 현재 진행하는 프로젝트의 규모를 생각하여 별도의 미들웨어나 게이트웨이보다는 서버 내부에 존재하도록 구현을 해보자.

어떤 처리율 제한 알고리즘을 적용해야 할까?

다양한 알고리즘이 존재하지만, 인기 알고리즘인 토큰 버킷 알고리즘 을 통해 구현하고자 한다.

토큰 버킷: 지정된 용량을 갖는 컨테이너
토큰 공급기(refiller): 일정 시간동안 토큰을 추가하며 버킷이 가득찬 경우 토큰은 버려진다.

이에 따라 공급 제한 규칙을 정해야 한다.

어떤 API 엔드포인트에 적용할 것인지
버킷을 지정하는 기준은 무엇인지

버킷의 크기와 토큰 공급률을 적절하게 튜닝하기 어렵다는 단점이 존재한다.

이러한 알고리즘을 기반으로 만들어진 bucket4j 라는 라이브러리가 존재한다.

Bucket4j는 JDBC, 캐시(로컬, 글로벌) 등 다양한 데이터베이스 환경을 지원한다. 또한, 구글링을 많이 하다보면 다양한 예시가 있을 것이니 필요하다면 해당 글들을 참고하면 좋을 것 같다.

과거에 Redis를 사용해서 Bucket4j 테스트를 진행해본 적이 있고, 실제로 사용할 경우 인스턴스가 한 대로 운영만 하는 상황이 아닐 것이라고 가정하였다.

처리율 제한 전략을 어떻게 세워야 할까?

처리율 제한 장치를 도입하는 이유가 뭔지 정확하게 무엇인가?

악의적인 요청으로부터 서비스의 핵심 기능을 보호하고, 가용성과 안정성을 보장하기 위함이다. 외부 API에 의존하는 현재 서비스에서 할당량을 효율적으로 관리하여 서비스 전체에 문제를 방지하는 것은 중요하다고 판단했다.

따라서, 먼저 '카카오 책 검색 API' 에 대해서만 처리율 제한을 두고자 한다.

이러한 상황에서 고려해야할 점은 무엇일까?

제한 범위 설정 (버킷을 누구를 기준으로 생성할 것인가?)

글로벌 제한: 모든 사용자가 공유하는 하나의 버킷으로 API 요청 제한
사용자별 제한: JWT를 활용하여 사용자별 별도의 토큰 버킷 생성

두 방식을 적절히 섞어서 사용하는 방법도 있을 것 같다.

시간 간격 설정 (refill 시간)

기본적으로 일일 제한은 30,000개 요청인 상황
기본 버킷 용량과 리필 속도를 잘 조절해야 할 것으로 예상됨

명확한 기준은 아직 잘 모르겠다. 실제로 구현해보면서 조정이 필요해보인다.

식별 키 선택

ID Address: 구현이 간단하지만, Next.js 서버에서 호출을 하는 것이라서 사용자 파악이 가능할지 의문
X-USER-ID: 커스텀 헤더를 사용하여 식별 가능하지만, 위변조 가능성이 높음
JWT (Access Token): 로그인된 사용자만 사용가능하며, 안전한 사용자 식별 기능

회원인 경우에만 사용 가능한 API이므로 JWT를 잘 활용하면 되지 않을까?

제한 초과 시 대응

응답 상태 코드: 429 Too Many Requests
응답 헤더: X-RateLimit-Remaining , Retry-After 등의 헤더를 통해 클라이언트에게 정보 전달

응답 예시

[429 Too Many Requests]

{
  "resultType": "ERROR",
  "data": null,
  "error": {
    "code": "EXTERNAL_API_429",
    "message": "Exceeded the limit. Please try again after 100 seconds.",
    "data": {
      "retryAfter": 100, // 남은 시간초
    }
  }
}

위를 기반으로 처리율 제한 전략을 대략적으로 설계해보고 이를 기반으로 도입해보자.

Bucket4j - Redis(Lettuce)

현재 스부키 프로젝트에서 Redis 클라이언트로 Lettuce를 사용하고 있는 상황이다. 동일하게 Lettuce를 사용해주면서 최신 버전의 라이브러리를 사용하기 위해서 공식문서를 읽어보면서 도입해보고자 한다.

1. 의존성 설정

dependencies {
    ...

    // bucket4j (api 모듈까지 전파하기 위해 api로 설정)
    api 'com.bucket4j:bucket4j_jdk17-core:8.14.0'
    api 'com.bucket4j:bucket4j_jdk17-lettuce:8.14.0'
}

스부키 프로젝트에서 JDK17을 사용하고 있기 때문에 호환되는 버전을 사용한다.

멀티 모듈(api, core) 이므로 core 영역에서 토큰 버킷저장소에 관련된 설정을 진행하고, api 영역에서 클라이언트(사용자별)마다 버킷을 사용할 수 있도록 구성하고자 한다. 이런 방식으로 관심사를 분리하면 좋을 것이라고 판단했다.

moudel core: RateLimiter 전용 저장소(Redis) 설정
module api: RateLimiter 설정 (Bucket, Capacity, Refill ...)

2. Redis 설정 - Core Module

기존에 설정해둔 Redis 설정에 영향을 주지 않도록 별도로 레디스 설정을 한다. Spring Data Redis와 Bucket4j-Lettcue 는 다른 방식으로 Redis에 접근하기 때문에 서로 간섭할 경우 문제가 발생할 수 있다.

RateLimiterRedisConfig.java

@Configuration
@RequiredArgsConstructor
public class RateLimitRedisConfig {

    private final RedisProperties redisProperties;

    @Bean
    public RedisClient redisClient() {
        return RedisClient.create(
                RedisURI.builder()
                        .withHost(redisProperties.getHost())
                        .withPort(redisProperties.getPort())
                        .withPassword(redisProperties.getPassword().toCharArray())
                        .build());
    }
    
    @Bean
    public ProxyManager<String> lettuceBasedProxyManager(RedisClient redisClient) {

        StatefulRedisConnection<String, byte[]> redisConnection =
                redisClient.connect(RedisCodec.of(StringCodec.UTF8, ByteArrayCodec.INSTANCE));

        return Bucket4jLettuce.casBasedBuilder(redisConnection)
                              .expirationAfterWrite(
                                      ExpirationAfterWriteStrategy.basedOnTimeForRefillingBucketUpToMax(
                                              Duration.ofSeconds(10)))
                              .build();
    }
}

RedisClient 설정은 별도로 설명하지는 않고 ProxyManager 설정을 위주로 보자.

Redis Connection Config

StatefulRedisConnection<String, byte[]> redisConnection =
        redisClient.connect(RedisCodec.of(StringCodec.UTF8, ByteArrayCodec.INSTANCE));

StatefulRedisConnection : Redis와 스레드 연결을 생성
RedisCodec.of(StringCodec.UTF8, ByteArrayCodec.INSTANCE) : 키는 UTF-8 문자열로, 값은 Byte Array로 저장하는데 버킷 상태를 효율적으로 직렬화 및 역직렬화 하기 위함이다.

CAS 기반 ProxyManager 생성

Bucket4jLettuce.casBasedBuilder(redisConnection)

CAS(Compare-And-Swap) 작업을 사용하여 동시성 문제 해결
여러 서버가 동시에 같은 버킷을 수정할 때 데이터 일관성을 보장

내부적으로 루아 스크립트(Lua Script)를 사용한다. 해당 내용은 잘 모르지만, Redisson Client에서도 동시성 문제를 해결하기 위해 루아 스크립트를 사용한다고 들었다. (더 공부해볼 내용 🙂)

public class LettuceBasedProxyManager<K> extends AbstractCompareAndSwapBasedProxyManager<K> {
    
    // ...
    private RedisFuture<Boolean> compareAndSwapFuture(K[] keys, byte[] originalData, byte[] newData, RemoteBucketState newState) {
        long ttlMillis = this.expirationStrategy.calculateTimeToLiveMillis(newState, this.currentTimeNanos());
        byte[][] params;
        if (ttlMillis > 0L) {
            if (originalData == null) {
                params = new byte[][]{newData, this.encodeLong(ttlMillis)};
                return this.redisApi.eval("if redis.call('set', KEYS[1], ARGV[1], 'nx', 'px', ARGV[2]) then return 1; else return 0; end", ScriptOutputType.BOOLEAN, keys, params);
            } else {
                params = new byte[][]{originalData, newData, this.encodeLong(ttlMillis)};
                return this.redisApi.eval("if redis.call('get', KEYS[1]) == ARGV[1] then redis.call('psetex', KEYS[1], ARGV[3], ARGV[2]); return 1; else return 0; end", ScriptOutputType.BOOLEAN, keys, params);
            }
        } else if (originalData == null) {
            params = new byte[][]{newData};
            return this.redisApi.eval("if redis.call('set', KEYS[1], ARGV[1], 'nx') then return 1; else return 0; end", ScriptOutputType.BOOLEAN, keys, params);
        } else {
            params = new byte[][]{originalData, newData};
            return this.redisApi.eval("if redis.call('get', KEYS[1]) == ARGV[1] then redis.call('set', KEYS[1], ARGV[2]); return 1; else return 0; end", ScriptOutputType.BOOLEAN, keys, params);
        }
    }
}

만료 전략 설정

.expirationAfterWrite(
        ExpirationAfterWriteStrategy.basedOnTimeForRefillingBucketUpToMax(
                Duration.ofSeconds(10)))

버킷이 일정 시간동안 사용되지 않으면 Redis에서 자동으로 제거
basedOnTimeForRefillingBucketUpToMax(Duration.ofSeconds(10)): 버킷이 최대 용량까지 리필되는 데 필요한 시간에 10초를 더한 시간 후에 만료되도록 설정
10초의 jitter 값은 버킷 재생성 빈도를 줄이기 위해 설정됨. 너무 작은 지터 값은 버킷이 자주 재생성되어 문제가 있을 수 있음.
- jitter 값을 튜닝하는 것이 어렵다고 들었는데 구현을 해보고 어떻게 변경하면 좋을지 고민해보고자 한다.

JPA - EntityManager를 사용하는 방식처럼 버킷의 상태를 추적하고 관리하기 위한 목적인 것이다.

위와 같은 방식으로 Redis를 통해 토큰을 관리하도록 설정해주었다. 이제 버킷을 어떻게 생성하고 관리할 것인지 api 모듈에서 설정해보자.

3. Rate Limiter 설정 - API Module

Rate Limiter Policy

위에서 세웠던 전략을 기반으로 처리율 제한 정책을 세워야 한다.

이 정책을 세우는건 많이 어려운 부분인 것 같다. (현업에서는 어떻게하는지 정말 궁금한 부분..) 실제로는 어떻게 구현하는지 잘 모르겠다. 따라서 시나리오를 생각해보고 테스트해보면서 다음과 같은 판단을 내렸다.

20초 안에 30번 이상의 호출이 들어오는건 악의적인 호출이다. (단기 제한: 사용자별)
하루에 30,000건 이상의 트래픽은 금지한다. (장기 제한: 글로벌)

이를 기반으로 다층적 제한을 적용하는 방법도 있을 것 같다. 장기 제한(글로벌)을 조금 더 생각해보면 FeignClient에서 429 Too Many Requests 응답이 올 때 그 값을 잡아서 ControllerAdvice에서 던져주는 방식도 존재할 것이다.

이 부분에 대해서는 별도로 고민이 필요하다고 판단하여 단기 제한만 적용해보고자 한다. 필요한 경우 도입하는 것은 어렵지 않을 것으로 예상된다.

토큰 리필 방식은 어떻게 해야하지?

토큰을 리필하는 방식은 크게 Greedy , Intervally 두 가지로 나뉜다.

Greedy: 토큰을 지속적으로 작은 단위로 나누어 최대한 빠르게 리필
- "20초마다 5개 토큰" -> 4초마다 1개씩 토큰 추가
- 토큰이 더 균등하게 분배되어 버킷이 빠르게 채워짐
- ex. refillGreedy(5, Duration.ofSeconds(20))
Intervally: 지정된 전체 시간 간격이 지난 후에 모든 토큰 리필
- "20초마다 5개 토큰" -> 정확히 20초가 지난 후 5개의 토큰이 한꺼번에 추가
- ex. refillIntervally(5, Duration.ofSeconds(20))

최종적으로 채워지는 토큰의 개수는 동일하지만, 리필 패턴이 다르기 때문에 적절하게 사용하면 된다.

책 검색을 하면서 스크롤을 내리면서 토큰을 사용하기 때문에 refillGreedy 방식이 더 적합하다고 판단했다. 사용자 경험과 지속적 요청 처리 관점에서 봤을 때 한꺼번에 주는 것보단 그리디하게 주는게 더 좋다고 보여진다.

@Component
@RequiredArgsConstructor
public class RateLimitPolicy {

    private static final int INTERVAL_CAPACITY = 30;
    private static final int INTERVAL_REFILL = 5;
    private static final int INTERVAL_DURATION_SECONDS = 20;

    /**
     * 기본 30개의 토큰을 20초마다 5개씩 그리디하게 리필하는 RateLimit 정책을 설정합니다.
     */
    public BucketConfiguration createBucketConfig() {
        return BucketConfiguration
                .builder()
                .addLimit(limit -> limit
                        .capacity(INTERVAL_CAPACITY)
                        .refillGreedy(INTERVAL_REFILL, Duration.ofSeconds(INTERVAL_DURATION_SECONDS)))
                .build();
    }
}

Rate Limiter

이제 위에서 만들어 둔 전략을 기반으로 처리율 제한장치를 구현하면 된다.

@Component
@RequiredArgsConstructor
public class RateLimiter {

    private final ProxyManager<String> proxyManager;
    private final RateLimitPolicy rateLimitPolicy;


    public ConsumptionProbe checkRateLimit(String clientKey) {

        Bucket bucket = proxyManager.builder()
                    .build(clientKey, rateLimitPolicy::createBucketConfig);
        
        return bucket.tryConsumeAndReturnRemaining(1);
    }
}

클라이언트 키를 기반으로 버킷을 식별한다.

이미 해당 키로 버킷이 존재하면 기존 버킷을 반환한다.
버킷이 존재하지 않는 경우, rateLimitPolicy::createBucketConfig 가 호출되어 새로운 버킷을 생성

만들어진 처리율 제한장치를 이제 적용해보자. Interceptor, AOP(Annotation) 등 다양한 방법으로 적용 가능하겠지만, 일단 기본적인 Inteceptor로 적용하여 도입하고자 한다.

Interceptor로 구현하기 때문에 이전에 Security Filter Chain을 통해 SecurityContextHolder에 사용자 정보가 담겨져 있을 것이다. 따라서 다음과 같이 사용자 정보를 얻어온 후 해당 값을 기반으로 clientKey를 만들고자 한다.

@Slf4j
@Component
@RequiredArgsConstructor
public class RateLimitInterceptor implements HandlerInterceptor {

    private static final String CLIENT_KEY_PREFIX = "rate-limit:";

    private final RateLimiter rateLimiter;
    private final ObjectMapper objectMapper;

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
            throws Exception {

        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
        log.info("Authentication: {}", authentication);
        if (authentication == null || authentication instanceof AnonymousAuthenticationToken) {
            handleUnauthorized(response);
            return false;
        }

        String clientKey = CLIENT_KEY_PREFIX + authentication.getName();
        ConsumptionProbe probe = rateLimiter.checkRateLimit(clientKey);

        if (probe.isConsumed()) {
            handleAllowedRequest(response, probe);
            return true;
        }

        handleRateLimitedRequest(response, probe);
        return false;
    }

    private void handleUnauthorized(HttpServletResponse response) throws IOException {

        response.setStatus(ErrorType.AUTHENTICATION_FAILED.getStatus().value());
        response.setContentType(MediaType.APPLICATION_JSON_VALUE);

        objectMapper.writeValue(response.getWriter(), ApiResponse.error(ErrorType.AUTHENTICATION_FAILED));
        log.info("Unauthorized request! Authentication is null or anonymous.");
    }

    private void handleAllowedRequest(HttpServletResponse response, ConsumptionProbe probe) {
        long remainingTokens = probe.getRemainingTokens();
        response.addHeader("X-Rate-Limit-Remaining", Long.toString(remainingTokens));
        log.info("Success to consume 1 token! Remaining tokens: {}", remainingTokens);
    }

    private void handleRateLimitedRequest(HttpServletResponse response, ConsumptionProbe probe)
            throws IOException {

        long retryAfterSeconds = probe.getNanosToWaitForRefill() / 1_000_000_000;
        Map<String, Long> errorData = new HashMap<>();
        errorData.put("retryAfter", retryAfterSeconds);

        response.setStatus(ErrorType.RATE_LIMIT_EXCEEDED.getStatus().value());
        response.setContentType(MediaType.APPLICATION_JSON_VALUE);

        objectMapper.writeValue(response.getWriter(), ApiResponse.error(ErrorType.RATE_LIMIT_EXCEEDED, errorData));
        log.info("Rate limit exceeded! Retry after: {} seconds", retryAfterSeconds);
    }
}