스케일 아웃 환경에서의 서킷브레이커 동기화

서킷브레이커 패턴?

서킷브레이커 패턴은 간단히 말하면 회로 차단기 패턴이다. 관련해 다른 좋은 글들이 많기 때문에 서킷브레이커에 대한 설명은 생략하겠다.
쨋든 회로 차단기 패턴을 구현한 라이브러리가 resilience4j 이다.

 

resilience4j docs: https://resilience4j.readme.io/docs/circuitbreaker

CircuitBreaker

 

서킷브레이커를 어디에서 사용하는가?

우리는 자체 검색 엔진이 없다. 그래서 검색이 AI 기반으로 돌아가는데 이를 제공해주는 외부 서비스사의 API를 활용한다.
즉, 검색 기능은 외부 API에 의존하고 있는 상태이다.

다시 정리하면 기능이 의존한다는 것은, 외부 서비스사가 잘 동작해야 우리도 잘 동작한다는 뜻이다.

 

일단 서킷브레이커 도입 배경은 아래와 같다.

과거 해당 서비스가 DDos 공격을 받은적이 있었다. 즉, 서비스 마비가 된적이 있었는데 우리는 해당 기능에만 의존하고 있었기 때문에 우리 쪽까지 장애가 전파되는 문제가 있었다.

장애를 경험한 후, 서킷브레이커 패턴을 공부하게 되었고 적용까지 빠르게 하였다. fallback 로직은 우리 DB를 검색하도록 정의하였다.

그 후, 또 다시 외부 서비스가 24시간동안 장애가 난 적이 있었는데 도입한 서킷브레이커를 통해 장애 전파를 잘 막았다.

 

정리하자면 서킷브레이커는 장애 전파를 막을 목적으로 사용한다.

현재 구성도는 아래와 같다.

 

API Server는 스케일 아웃 환경이라 위 처럼 표현하였다.

서킷브레이커는 로컬에서 관리된다. 즉, 스케일 아웃 환경이라면 각 서비스는 각각의 서킷브레이커 상태를 관리하게 된다.

어쨋든, 각 로컬에서 실패율을 판단해서, OPEN-CLOSE를 하게 된다는 것이다.

 

이 구조의 문제점은 각 로컬에서 상태를 관리하기 때문에, 불필요한 요청이 발생할 수 있다는 것이다.

즉, 서버 1에서 서킷이 OPEN 되었는데, 서버 2는 아직 CLOSE 상태여서 어차피 실패할 요청을 계속 한다는 것이다.

예시에서 서버가 3개라서 공감이 안 갈 수도 있는데, 서버들이 더 많아진다면 더욱 불필요한 요청이 발생하게 된다.

 

해결 방법은 서킷브레이커 상태를 동기화하면 된다.

서킷브레이커 상태를 동기화 시키면, 불필요한 요청이 적어질 것이다.

 

모듈 간 상태를 동기화하자

가장 깔끔하고 정확하게 관리할 수 있는 방법은 코디네이터라는 별도의 서버를 만드는 것이다.

즉, 코디네이터를 통해 서킷브레이커의 상태를 관리하는 것이다.

 

대충 이런 느낌인데, 화살표는 해당 방향으로 요청을 보낸다는 것이다.

1번 화살표는 서버 to 코디네이터이다.

발생할 수 있는 요청은 다음과 같다.

1. 코디네이터에 관리해달라고 등록

2. 서킷브레이커 OPEN 알림

 

2번 화살표는 코디네이터 to 서버이다.

발생할 수 있는 요청은 다음과 같다.

1. 서버 헬스 체크(해당 서버가 관리해달라고 등록한 서버인지 확인)

2. 서킷브레이커 상태 동기화

 

즉, 4개의 발생할 수 있는 요청을 해결하면 동기화가 가능하다.

하지만 해당 방식을 선택하진 않았다.

 

우선 API Server는 N개이고, 코디네이터는 1개이다. API -> 코디네이터로 보내는 요청은 각각 1개씩 보내면된다. 하지만 코디네이터 -> API 로 요청은 N개를 보내야한다.

요청이 누락되지 않고, N개의 이벤트를 전달할 방법의 가장 베스트 방법은 카프카인데, 우리 팀은 카프카를 아직 운용하지 않는다.

+ 별도 서버를 띄우지 않고 해결하는게 더 빠르게 도입할 수 있다고 생각해서 다른 방법을 선택하였다.

 

최종 구현 방식

레디스 pub/sub을 활용하여서 최종적으로 구현하였다.

 

1-1. 코디네이터에 관리해달라고 등록: 애플리케이션 시작 시 토픽 sub

1-2. 서킷브레이커 OPEN 알림: 특정 키의 카운트 증가

2-1. 서버 헬스 체크: 라이브러리 단에서 지원(애플리케이션 시작 시 sub 정보 등록/해제)

2-2. 서킷브레이커 상태 동기화: 특정 키의 카운트가 임계점이 넘으면 pub

 

특정 키의 카운트 증가라는 것은 서킷브레이커 타입별로 OPEN 개수를 센다는 것이다.

카운트를 세는 이유는 한 개의 서버만 OPEN이 된다고 바로 발행시키지 않기 때문이다. 이는 한 개의 서버만 모종의 이유로 외부 API 연결이 안 될 수 있기 때문에, 일정 개수의 서버가 OPEN 된다면 동기화 이벤트를 발행한다.

해당 키의 TTL은 10분으로 잡았다.

 

일정 개수의 서버가 OPEN이 되었다면, 이벤트를 발행한다.

이벤트를 발행하기 전에 레디스 락을 획득한다. 락을 획득하는 이유는 카운트 기반으로 이벤트를 발행하기 때문에 중복 이벤트를 발행시키지 않으려고 하는 것이다.

대략적인 코드는 아래와 같다.

// 카운터 + redis pub
fun checkAndPropagate(circuitBreakerName: String) {
        val key = KeyManager.getKeywordSearchCircuitBreakerCounterKey(circuitBreakerName)
        val openCount = redisTemplate.opsForValue().get(key)?.toInt() ?: 0

        val thresholdKey = KeyManager.getKeywordSearchCircuitBreakerThresholdKey(circuitBreakerName)
        val threshold = redisTemplate.opsForValue().get(thresholdKey)?.toInt() ?: DEFAULT_THRESHOLD

        if (openCount >= threshold) {
            val lockKey = KeyManager.circuitBreakerPropagationLockKey(circuitBreakerName)
            val lockAcquired =
                redisTemplate.opsForValue().setIfAbsent(
                    lockKey,
                    "true",
                    10.minutes.toJavaDuration(),
                )

            if (lockAcquired == true) {
                propagateOpen(circuitBreakerName)
            }
        }
}

private fun propagateOpen(circuitBreakerName: String) {
        val topic = CircuitBreakerType.getName(circuitBreakerName).topicName
        redisTemplate.convertAndSend(topic, "OPEN")
}

// redis sub
override fun onMessage(
        message: Message,
        pattern: ByteArray?,
) {
	val type = CircuitBreakerType.getByTopicName(String(message.channel)) ?: return
	when (type) {
		CircuitBreakerType.KEYWORD_SEARCH -> openKeywordSearch()
        else -> return
	}
}

private fun openKeywordSearch() {
    keywordSearchCircuitBreaker.transitionToOpenState()    
}

 

 

카운터 키 TTL이 만료된다면, 키 만료 이벤트를 수신할 수 있다.

즉, 키가 만료된다면 현재 서버의 로컬 서킷브레이커 상태에 따라 다시 카운터를 세도록 구현하였다.

 

애플리케이션이 재시작 되었을 때도 상태를 유지하기 위해 이벤트 발행 전 락 key가 존재하는지 확인한다. 존재한다면 최근에 서킷이 OPEN된 적이 있는 것이기 때문에, 이 때 서킷을 OPEN하도록 하였다.

@EventListener(ApplicationStartedEvent::class)
fun startEvent(event: ApplicationStartedEvent) {
	val lockKey =
		KeyManager.circuitBreakerPropagationLockKey(
			CircuitBreakerType.KEYWORD_SEARCH.circuitBreakerName,
        )
    if (redisTemplate.opsForValue().get(lockKey) != null) {
		keywordSearchCircuitBreaker.transitionToOpenState()
    }
}

 

이렇게 TTL + Redis Pub/Sub을 활용하여 코디네이터를 대체할 수 있게 되었다.

만약 불일치가 발생한다고 해도, 서킷브레이커는 최종적으로 로컬에서 관리되므로 동작은 올바르게 된다.

 

정리

1. 외부 API에 대해 서킷브레이커가 구현되어 있었다.

2. 서킷브레이커 상태는 로컬에서만 관리되고 있었어서, 불필요한 요청이 추가적으로 발생하는 문제가 있었다.

3. 이를 해결하기 위해 서비스 간 서킷브레이커를 동기화하도록 구현하였다.

4. 코디네이터를 통해 동기화가 가능하다.

5. Redis Pub/Sub을 활용하여 코디네이터 역할을 대체하였다.

 

현재 상황을 정리하면 A 서비스 <-> B 서비스 상황이다. 그렇기에 코디네이터 방식이 아닌, TTL + Redis Pub/Sub 방식을 택하였다.

만약 MSA 환경, 즉 A서비스를 여러 서비스가 보고 있고, A 서비스에 대해 장애 전파 및 동기화 작업을 하려면 이 작업을 매 서비스마다 반복해야 한다. 그런 상황에서는 별도의 코디네이터 서버를 만드는게 더 좋을 것 같다.