<Spring> Slack 메세지 전송 에러 처리

📌 에러 상황

현재 수수에서 Slack을 통계, 에러 알림으로 사용하고 있다.

근데 가끔 Slack 메시지 전송이 실패했다고 알림이 온다.

스케줄러 에러인 것을 보아, 통계 알림의 슬랙 메시지 전송 부분에서 발생한 것을 알 수 있다.

근데 위에 보여준 건 여러 개 중 하나이다.

다행히 모두 통계 알림 에러이다.

하지만 이 일이 에러 알림에서 발생해도 이상하지 않다.

센트리로 에러 사항이 기록되긴 하지만 개발자가 이 에러의 발생을 빠르게 파악할 수 있는 것은 슬랙 알람뿐이다.

그러므로 알람이 전달됨을 보장해주는 로직이 필요했다.

 

 

📌 처리에 관한 고민

처음엔 서킷 브레이커를 생각했다.

슬랙이 오랜 기간 동안 장애가 발생한다면 시스템의 안정성을 높이는 부분에 있어서 유용할 것이다.

하지만 슬랙이 오랜 기간 동안 장애가 발생한 것을 보지 못했다는 점, 지금까지 발생한 슬랙 커넥션 관리 문제라는 점을 고려했을 때, 일시적인 오류, 특히 특정 요청에서만 발생하는 상황에서는 서킷 브레이커는 불필요하다고 생각했다.

 

그렇다면 retry 로직은 어떠한가?

일시적인 오류가 자주 발생하는 작업에서 재시도를 통해 성공 가능성을 높일 수 있다.

물론 일시적 오류가 아닌 장기간 장애가 이어질 경우 요청에 대한 접근을 제한하는 대신 재처리만 반복하기 때문에, 시스템의 리소스가 낭비되는 상황이 일어날 수 있다.

그러므로 서킷 브레이커 보다 안정성은 떨어진다.

하지만 위 문제상황과 같은 상황이라면 retry가 더 적합하다 생각한다.

 

위는 모두 요청에 대한 처리이다.

우리는 전달되지못한 요청에 대한 처리도 해줘야 한다.

실패하면, 저장해 두고 따로 처리하면 될 것이라 생각한다.

 

📌 고민에 대한 나의 답

retry와 batch job을 조합해서 처리했다.

retry는 실패 작업을 다시 수행하는 역할이고, batch job은 반복 수행해도 전송 실패한 메시지를 모아서 다시 전송하는 것이다.

 

🔵 Retry

retry는 2가지 방법이 있었다.

1. reactor
2. spring retry

결론부터 말하자면 나는 reactor retry를 사용했다.

spring retry는 어노테이션으로 쉽게 설정하여 사용할 수 있다.

하지만 현재 외부 서비스와 연결이 webclient로 한정되어 있다는 점과 retry를 적용하려 하는 부분이 webclient로 한정되어 있다는 점을 고려했을 때, 추가적인 라이브러리의 설치와 설정이 필요한 spring retry 보단 reactor가 적절하다 판단했다.

class SuspendableSlackClient(
    private val webclient: WebClient,
    private val slackWebhookConfig: SlackConfig.SlackWebhookConfig,
    private val cacheService: CacheService,
) : SlackClient {
    private val logger = KotlinLogging.logger { }

    override suspend fun sendSummary(message: SlackMessageModel): String {
        return sendMessage(message, slackWebhookConfig.summaryToken)
    }

    override suspend fun sendError(message: SlackMessageModel): String {
        return sendMessage(message, slackWebhookConfig.errorToken)
    }

    override suspend fun sendMessage(message: SlackMessageModel, token: String, withRecover: Boolean): String {
        return runCatching {
            withMDCContext(Dispatchers.IO) {
                webclient.post()
                    .uri("/${token}")
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(message)
                    .retrieve()
                    .bodyToMono<String>()
                    .retryWhen(Retry.fixedDelay(2, Duration.ofMillis(500)))
                    .awaitSingleOrThrow()
            }
        }.onFailure {
            if (withRecover) {
                recoverSendMessage(message, token)
            }
        }.getOrThrow()
    }

    private suspend fun recoverSendMessage(message: SlackMessageModel, token: String) {
        val model = FailedSentSlackMessageCache(
            token = token,
            message = message.text
        )

        cacheService.sSet(Cache.getFailedSentSlackMessageCache(LocalDateTime.now()), model)
    }
}

전송해 보고 실패하면 값을 저장하도록 했다.

사실 저장하는 로직도 doOnError로 처리할 수 있었다.

하지만 코루틴으로 처리하는 게 가독성을 높일 수 있기 때문에 이는 이용하지 않았다.

 

🔵 Batch Job

batch는 간단하다

저장된 값을 가져와서 병합한 후, 전송하는 게 전부이다.

또한 저장된 값을 재전송할 때는 메시지의 앞에 재처리된 메시지임을 표시했다.

이 작업을 1분에 1번씩, 1분 전까지 기록된 값을 처리해 준다.

(현재가 24분이면 23분 값을 처리한다는 것)

이때, 에러 상황이 길어져서 1분 뒤에도 전송하지 못하는 경우도 대비해야 했다.

이 경우는 1분 뒤에 조회될 값들에 전송되지 못한 값들을 추가하여, 메시지를 1분 뒤에 다시 전송하도록 조치했다.

@Component
class ResendFailedSentSlackMessageJob(
    private val cacheService: CacheService,
    private val slackClient: SlackClient,
) {
    private val logger = KotlinLogging.logger { }

    companion object {
        private const val RESEND_BEFORE_MINUTES = 1L
    }

    suspend fun resendFailedSentSlackMessage() {
        logger.info { "start resend failed sent slack message" }

        // 1분 전에 실패한 것이 타겟 (현재가 24분이면 23분을 말하는 것)
        val targetTime = LocalDateTime.now().minusMinutes(RESEND_BEFORE_MINUTES)

        // 실패 메세지 조회 및 삭제
        val failedMessages = withContext(Dispatchers.IO) {
            cacheService.sGetMembers(Cache.getFailedSentSlackMessageCache(targetTime))
        }

        withContext(Dispatchers.IO) {
            cacheService.sDelete(Cache.getFailedSentSlackMessageCache(targetTime))
        }

        // 다수 메세지 token 별로 하나의 메세지로 병합
        val message = mergeFailedMessage(failedMessages)

        // 재전송
        runCatching {
            coroutineScope {
                val sendDeferreds = message.map { (token, message) ->
                    val slackMessageModel = SlackMessageModel(text = message)

                    async(Dispatchers.IO) {
                        slackClient.sendMessage(
                            message = slackMessageModel,
                            token = token,
                            withRecover = false
                        )
                    }
                }.toTypedArray()

                awaitAll(*sendDeferreds)
            }
        }.onFailure {
            // 재전송 실패시 1분 뒤에 다시 보낼 수 있게, 1분 뒤에 보내는 메세지 목록에 추가
            logger.warn { "postpone resend slack message" }

            postponeResendTimeOfFailedMessage(targetTime, message)
        }

        logger.info { "finish resend failed sent slack message" }
    }

    private suspend fun mergeFailedMessage(failedMessages: List<FailedSentSlackMessageCache>): Map<String, String> {
        val message = mutableMapOf<String, String>()

        failedMessages.forEach { model ->
            val recoverMsg = if (model.isStacked) {
                model.message
            } else {
                "[RECOVER - ${model.failedAt} slack failure] ${model.message}"
            }

            val stackedMessage = message[model.token]

            message[model.token] = if (stackedMessage == null) {
                recoverMsg
            } else {
                "$stackedMessage\n$recoverMsg"
            }
        }

        return message
    }

    private suspend fun postponeResendTimeOfFailedMessage(targetTime: LocalDateTime, message: Map<String, String>) {
        val nextTime = targetTime.plusMinutes(RESEND_BEFORE_MINUTES)

        coroutineScope {
            message.map { (token, message) ->
                val model = FailedSentSlackMessageCache(
                    token = token,
                    message = message,
                    isStacked = true
                )

                async(Dispatchers.IO) {
                    cacheService.sSet(
                        cache = Cache.getFailedSentSlackMessageCache(nextTime),
                        value = model
                    )
                }
            }
        }
    }
}

 

📌 마무리

DLQ를 의도하고 만든 건 아닌데, 만들고 보니까 DLQ랑 비슷한 것 같다.

곧 슬랙에서 디스코드로 이관해야 하는데, 거기도 이런 거 만들어둬야겠다.

만약에 슬랙 쪽에 지속적인 문제가 발생한다면 서킷 브레이커 도입을 고려해 봐야겠다.