[Spring Boot] AWS 분산 시스템에서 Zipkin으로 로그 추적하기: SNS/SQS 통합 실전 적용

AWS 분산 시스템에서 Zipkin으로 로그 추적하기: SNS/SQS 통합 실전 적용

 

Spring Boot

 

이번 글에서는 서버 간 연동 과정을 한눈에 파악할 수 있는 방법에 대해 탐구해보려 한다.

AWS에서는 SNS, SQS, 그리고 각각의 Spring Boot 서비스들을 CloudWatch를 통해 볼 수 있다. 그러나 이러한 서비스들이 서로 어떻게 연동되고 있는지, SNS 메시지가 올바르게 전송되고 SQS에 의해 제대로 처리되는지, 또한 Feign 요청이 성공적으로 수행되는지 등의 상세한 통신 과정을 한눈에 파악하는 것은 쉽지 않다.

이러한 문제의 해결책으로 Zipkin의 도입을 결정했다. Zipkin은 분산 시스템에서 서버 간의 상호작용을 시각화하여 표시한다. 이를 통해, SNS 메시지의 발송부터 SQS에 의한 처리, 그리고 Feign 요청의 수신까지, 전체 통신 흐름을 명확하고 직관적으로 볼 수 있다. 이는 개발자가 시스템의 성능을 모니터링하고, 잠재적인 문제를 신속하게 파악하며, 전체 아키텍처의 이해도를 높이는 데 크게 도움이 된다.

 

이번 글에서 다룰 내용은 닉네임 변경 요청의 로그 추적 방법에 대한 것이다. 하지만 여기에 제시된 코드는 완전한 코드가 아니라, 이전 글에서 다뤘던 코드에 추가되어야 하는 부분이다.
이번 글의 내용을 제대로 활용하려면, 이전 글의 내용을 숙지하는 것이 필수적이다.

 

 

- 전체 프로세스 이해를 위한 링크

[Spring Boot]MSA 환경에서 SNS/SQS를 활용한 이벤트 처리: 이벤트 유실 문제 해결 방안

[Spring Boot]MSA 환경에서 SNS/SQS를 활용한 이벤트 처리: 이벤트 유실 문제 해결 방안

 

- Zipkin 서버 설정을 위한 링크

[AWS] ECS에서 Zipkin을 통한 스프링 부트 서비스 트레이싱 구축하기

[AWS] ECS에서 Zipkin을 통한 스프링 부트 서비스 트레이싱 구축하기

 

이번 글 간략 소개

Zipkin은 HTTP 요청을 통한 서비스 간 통신을 자동으로 추적하는 데 사용되지만, AWS의 SNS와 SQS와 같은 메시징 서비스는 표준 HTTP 요청과 다른 네트워크 통신 방식을 사용한다. 이 때문에 기존의 Zipkin 설정만으로는 SNS와 SQS를 통한 메시지 전달 과정에서의 로그 추적이 자동으로 이루어지지 않는다.

 

이러한 상황에서 우리는 서버 간의 모든 통신 과정을 추적하기 위해 Zipkin의 추적 기능을 확장할 필요가 있었다. 특히, 서비스 간 비동기 메시지 교환을 처리하는 SNS와 SQS에서도 요청의 흐름을 명확하게 추적하고자 했다. 이를 위해, HTTP 요청뿐만 아니라 SQS 리스닝 과정에서도 TraceID를 관리하고 전달하는 사용자 정의 로직을 구현했다.

 

이 구현을 통해, 멤버 서버에서 시작된 닉네임 변경 요청의 흐름을 레시피 서버에서의 처리까지 추적할 수 있게 되었다. 이 과정에서 생성된 TraceID는 요청이 SNS를 거쳐 SQS로 전달될 때도 유지되며, 이를 통해 전체 통신 과정을 하나의 연속된 흐름으로 볼 수 있다. 이러한 방식은 서버 간 통신의 성공 여부, 성능 문제, 잠재적 오류 등을 파악하는 데 매우 중요하다.

 

이번 글에서는 이러한 접근 방식의 구체적인 구현 방법과 함께, 각 단계에서의 코드 설명을 통해 Zipkin을 사용한 분산 서버 간의 로그 추적 방식을 자세히 살펴볼 것이다. 이를 통해 복잡한 분산 시스템에서의 데이터 흐름과 처리 과정을 효과적으로 파악하고, 시스템의 성능과 안정성을 개선하는 방법을 이해할 수 있을 것이다.

 

 

TraceID, Span 소개

traceId와 span은 분산 시스템에서의 로그 추적, 특히 Zipkin과 같은 추적 시스템에서 중요한 개념이다.

 

TraceID란 무엇인가?

• TraceID: 분산 시스템에서 하나의 요청 또는 트랜잭션을 추적할 때 사용하는 유니크한 식별자. 하나의 요청이 시스템의 여러 컴포넌트를 거치며 처리될 때, 이 요청에 관련된 모든 작업들이 같은 TraceID를 공유한다. 이를 통해 개별 요청의 전체 경로를 추적하고 분석할 수 있다.

Span이란 무엇인가?

• Span: 분산 추적 시스템에서, 하나의 특정한 작업 단위 또는 요청 처리 단계를 나타낸다. 예를 들어, 하나의 HTTP 요청, 데이터베이스 쿼리, 메시지 처리 등이 각각 하나의 Span이 될 수 있다.
• Span은 시작과 끝이 있으며, 각 Span은 고유한 Span ID를 가진다. 또한, Span은 해당 Span을 포함하고 있는 요청의 TraceID도 가지고 있어, 전체 요청의 일부로서의 역할을 나타낼 수 있다.

TraceID가 Span에 포함된 이유

• 하나의 요청(Trace)은 여러 개의 Span으로 구성될 수 있다. 예를 들어, 사용자의 요청이 웹 서버를 거쳐 데이터베이스 서버로 이동하고, 응답이 돌아오는 전체 과정은 여러 Span으로 나누어질 수 있다.
• 각 Span은 해당 요청의 일부분을 나타내므로, Span은 해당 요청의 TraceID를 포함한다. 이를 통해 모든 Span을 하나의 요청 흐름에 연결할 수 있고, 전체 요청 경로를 시각화하거나 분석할 수 있다.

String traceId = tracer.currentSpan().context().traceIdString() 코드의 의미

• tracer.currentSpan(): 현재 실행 중인 Span을 가져온다.
• context(): 해당 Span의 컨텍스트(정보)를 가져옵니다. 컨텍스트에는 Span ID, TraceID와 같은 정보가 포함된다.
• traceIdString(): 해당 Span의 TraceID를 문자열 형태로 추출한다.

결국, 이 코드 라인은 "현재 실행 중인 작업(Span)에 연결된 전체 요청(Trace)의 식별자(TraceID)를 가져와라"라는 의미다. 이렇게 TraceID를 사용함으로써, 서로 다른 서비스나 컴포넌트에 걸친 전체 요청 경로를 추적하고 분석할 수 있다.

 

Span feignRequestSpan = tracer.nextSpan().name("[RECIPE] /feign/member/getNickname request").start() 코드의 의미

• tracer.nextSpan(): 현재 추적 컨텍스트에 기반하여 새로운 Span을 생성한다. 이는 현재 실행 중인 작업 또는 요청과 연관된 새로운 Span을 생성하는 데 사용되며, 생성된 Span은 현재 컨텍스트와 연관된 TraceID를 사용한다. 즉, 새로 생성되는 Span은 현재 추적 중인 전체 요청 또는 트랜잭션과 연관된다.
• .name("processing something"): 이 부분은 생성된 Span에 이름을 지정한다. Span 이름은 해당 Span이 수행하는 작업을 설명한다. 이 이름은 Zipkin UI에서 추적 정보를 나타낼 때 사용되는 문구다.
• .start(): 이 메소드는 Span을 활성화하고, 해당 작업의 시작 시간을 기록한다. 이 메소드 호출로 Span이 실제로 시작되며, 이후에 이루어지는 작업들은 이 Span과 연결된다. (실제로 사용하려면 추가 작업이 필요하다. 아래에서 상세히 설명하겠다.)

결국, 이 코드 라인은 "해당 서비스에서 processing something 요청을 처리하기 위한 새로운 Span을 생성하고 시작하라"는 의미를 가지며, 이를 통해 해당 요청의 추적과 분석을 가능하게 한다. 추적 시스템은 이 Span을 사용하여 요청의 시작, 실행 중인 작업, 그리고 종료 시점을 포함한 전체 요청 경로를 시각화하고 분석할 수 있다.

 

 

 

TraceID가 전달되는 전체 프로세스 간략 설명

1. SNS 발행 단계의 TraceID 관리
   • 멤버 서버에서 닉네임 변경 이벤트가 발생하면, 현재 TraceID를 추출하고, 이를 SNS 메시지에 포함시켜 발행한다.
   • 이 TraceID는 추후의 로그 추적 과정에서 핵심적인 역할을 한다.
2. SQS 메시지 수신 및 처리
   • 레시피 서버의 SQS 리스너(AwsSqsListenerService)가 SNS 메시지를 받는다.
   • 리스너는 메시지에서 TraceID를 추출하고, 이를 기반으로 새로운 Span을 생성하여 처리 과정을 추적한다.
   • 이 단계에서 TraceID의 유지는 메시징 과정에서의 연속성을 보장한다.
3. SQS에서 Feign 요청으로의 TraceID 전달
   • SQS 메시지 처리 후, 레시피 서버는 멤버 서버로 Feign 요청을 보낸다.
   • Feign 클라이언트 설정(FeignClientConfig)에서는 요청 헤더에 TraceID를 추가하여 전달한다.
   • 이 과정은 서버 간 통신에서 TraceID의 연속성을 유지하는 핵심 단계다.
4. 멤버 서버의 응답 처리 및 TraceID 관리
   • 멤버 서버는 Feign 요청을 받고, FeignTraceIdFilter 필터를 통해 Feign 으로 들어온 요청일때 추가 작업을 실행한다.
   • 필터는 요청 헤더에서 TraceID를 추출하고, 이를 기반으로 새로운 Span을 시작하여 처리 과정을 추적한다.

 

 

1. 멤버 서버에서 닉네임 변경 SNS 발행

• 우선, 멤버 서버에서 닉네임 변경 HTTP 요청을 받는다.
• SNS service에서 현재 TraceID를 SNS 메시지에 포함시켜 다른 서버로 전송한다.
• 이 TraceID는 추적의 핵심이 되는데, 이는 전체 프로세스의 연속성을 보장한다.

SNS 발행 전 traceId 추출 로직

1-1. Tracer 객체의 사용

• Tracer 객체: Zipkin 분산 추적 시스템에서 중요한 역할을 한다. 이 객체를 통해 현재 실행 중인 Span에 접근하고, 이 Span의 TraceID를 추출할 수 있다.
• 전역 변수로 선언: Tracer를 전역 변수로 선언함으로써, SnsService 클래스의 모든 메소드에서 추적 컨텍스트에 쉽게 접근할 수 있다.

1-2. TraceID의 추출과 사용

• TraceID 추출: tracer.currentSpan().context().traceIdString() 호출을 통해 현재 실행 중인 Span의 TraceID를 문자열 형태로 추출한다. TraceID는 분산 시스템에서 특정 요청 또는 트랜잭션을 식별하는 데 사용되는 유니크한 식별자다.
• TraceID를 SNS 메시지에 포함: 추출한 TraceID를 SNS 메시지에 포함시키면, 이 메시지를 수신하는 다른 서비스나 컴포넌트에서도 동일한 요청 또는 트랜잭션을 추적할 수 있다. 이는 분산 시스템에서 서로 다른 서비스 간의 작업 연결성을 유지하는 데 중요하다.

1-3. SNS 메시지 발행

• SNS 발행 요청: PublishRequest 객체를 생성하여 SNS에 보낼 메시지와 대상 Topic ARN을 지정한다.
• SNS 클라이언트를 통한 발행: snsClient.publish(publishRequest)를 호출하여 구성된 메시지를 SNS Topic으로 발행한다.

 

(전체 코드)

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.recipia.member.config.aws.AwsSnsConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import software.amazon.awssdk.services.sns.SnsClient;
import software.amazon.awssdk.services.sns.model.PublishRequest;
import software.amazon.awssdk.services.sns.model.PublishResponse;
import brave.Tracer;

import java.util.HashMap;
import java.util.Map;

@Slf4j
@RequiredArgsConstructor
@Service
public class SnsService {

    private final SnsClient snsClient;
    private final AwsSnsConfig awsSnsConfig;
    private final ObjectMapper objectMapper;
    private final Tracer tracer;

    public PublishResponse publishNicknameToTopic(String message) {

        // 현재 TraceID 추출
        String traceId = tracer.currentSpan().context().traceIdString(); 

        Map<String, Object> messageMap = new HashMap<>();
        messageMap.put("traceId", traceId); // TraceID를 메시지에 추가
        messageMap.put("message", message);

        // SNS 발행 요청 생성
        PublishRequest publishRequest = PublishRequest.builder()
                .message(convertMapToJson(messageMap))
                .topicArn(awsSnsConfig.getSnsTopicNicknameChangeARN())
                .build();

        // SNS 클라이언트를 통해 메시지 발행
        PublishResponse response = snsClient.publish(publishRequest);

        // messageId 로깅
        log.info("[MEMBER] Published message to SNS with messageId: {}", response.messageId());

        return response;
    }

    private String convertMapToJson(Map<String, Object> messageMap) {
        try {
            return objectMapper.writeValueAsString(messageMap);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Error converting message map to JSON", e);
        }
    }

}

 

 

 

2. 레시피 서버의 SQS 메시지 처리

• 다음으로, 레시피 서버의 AwsSqsListenerService가 SNS 메시지를 받는다.
• 이 리스너는 메시지에서 TraceID를 추출하고, 이를 사용해 새로운 Span을 생성하여 처리 과정을 추적한다.
• 이 단계에서 TraceID를 유지하며, 추적의 연속성을 보장한다.

 

2-1. 메시지 읽고 그 안에 담긴 traceId 추출하기

• SQS로부터 받은 메시지(JSON 형식)를 파싱하여 필요한 정보(예: messageId, messageContent)를 추출한다.

SQS 리스너에서 JSON 파싱

 

JsonNode에서 "traceId" 추출하는 내부 메소드

 

 

2-2. 추출한 TraceId로 컨텍스트 설정

• 메시지에서 traceId를 추출하고, 이를 사용하여 새로운 TraceContext를 생성한다.
• 이 TraceContext는 분산 시스템에서의 요청 추적을 위한 컨텍스트로 사용됩니다.

내부 메소드에 traceId 전달

private TraceContext buildTraceContext(String traceId) {
    TraceContext.Builder contextBuilder = TraceContext.newBuilder();
    if (traceId.length() == 32) {
        long traceIdHigh = Long.parseUnsignedLong(traceId.substring(0, 16), 16);
        long traceIdLow = Long.parseUnsignedLong(traceId.substring(16), 16);
        contextBuilder.traceIdHigh(traceIdHigh).traceId(traceIdLow);
    } else {
        long traceIdLow = Long.parseUnsignedLong(traceId, 16);
        contextBuilder.traceId(traceIdLow);
    }
    // 새로운 Span ID 생성 (tracer 인스턴스를 사용)
    contextBuilder.spanId(tracer.nextSpan().context().spanId());
    return contextBuilder.build();
}

 

• buildTraceContext 메소드는 TraceContext 객체를 생성하는 과정을 담당하는 코드 부분이다.
• TraceContext는 Zipkin에서 로그 추적에 필요한 정보를 담고 있는 객체로, 여기에는 트레이스 ID, 스팬 ID 등 추적에 필요한 중요한 정보들이 포함된다. 코드를 이해하기 위해 단계별로 살펴보자.

 

2-2-1. TraceContext.Builder의 생성

TraceContext.Builder contextBuilder = TraceContext.newBuilder();

• 이 줄은 TraceContext 객체를 생성하기 위한 빌더를 초기화하는 부분이다. 빌더 패턴을 사용하여 객체를 단계적으로 구성할 수 있다.

2-2-2. Trace ID 처리

if (traceId.length() == 32) {
    long traceIdHigh = Long.parseUnsignedLong(traceId.substring(0, 16), 16);
    long traceIdLow = Long.parseUnsignedLong(traceId.substring(16), 16);
    contextBuilder.traceIdHigh(traceIdHigh).traceId(traceIdLow);
} else {
    long traceIdLow = Long.parseUnsignedLong(traceId, 16);
    contextBuilder.traceId(traceIdLow);
}

• 여기서는 traceId 문자열을 분석하여 16진수로 변환하고, TraceContext에 설정한다. Zipkin의 Trace ID는 16자리 또는 32자리 16진수일 수 있다.
  • 32자리 16진수인 경우: 첫 16자리를 traceIdHigh로, 나머지 16자리를 traceIdLow로 분리하여 처리한다.
  • 16자리 16진수인 경우: 전체 문자열을 traceIdLow로 사용한다.

 

💡 traceIdHigh와 traceIdLow는 Zipkin에서 사용되는 Trace ID의 구성 요소들이다. Zipkin의 Trace ID는 분산된 시스템에서 각 요청 또는 이벤트를 추적하기 위한 유니크한 식별자다. 이 Trace ID는 16자리 또는 32자리의 16진수로 표현될 수 있습니다.

32자리 16진수 Trace ID
 - 더 큰 시스템 또는 더 복잡한 트레이싱 요구사항을 지원하기 위해 사용된다. 이 경우, Trace ID는 두 부분으로 나뉜다.
    = traceIdHigh: Trace ID의 앞 16자리를 나타내며, 여기에 해당하는 부분의 값을 16진수로 해석하여 저장한다.
    = traceIdLow: Trace ID의 뒤 16자리를 나타내며, 이 부분의 값을 16진수로 해석하여 저장한다.

16자리 16진수 Trace ID
 - 더 간단한 시스템 또는 트레이싱 요구사항에 적합하다. 이 경우, 전체 Trace ID가 traceIdLow에 저장된다. traceIdHigh는 사용되지 않거나 0으로 설정된다.

이렇게 Trace ID를 두 부분(traceIdHigh와 traceIdLow)으로 나누는 이유는, 32자리의 긴 Trace ID를 효율적으로 처리하고, 더 큰 시스템에서도 충분한 고유성을 제공하기 위함이다. 이 구조는 Zipkin의 추적 시스템이 큰 규모의 분산 시스템에서도 유니크한 추적 정보를 유지할 수 있도록 돕는다.

 

 

2-2-3. 새로운 Span ID 생성

contextBuilder.spanId(tracer.nextSpan().context().spanId());

• 이 코드는 새로운 Span ID를 생성한다. 이 Span ID는 로그 추적에서 각각의 작업 단위를 구분하는 데 사용된다.

 

2-2-4. TraceContext 반환

return contextBuilder.build();

• 생성된 TraceContext 객체는 호출자에게 반환됩니다.

 

 

2-3. Span 생성 및 시작

• Tracer 객체를 사용하여 새로운 추적 단위인 Span을 생성한다.

컨텍스트를 활용한 Span 객체 생성

 

• TraceContextOrSamplingFlags.create(context)는 context 매개변수로 받은 TraceContext 객체를 기반으로 TraceContextOrSamplingFlags 객체를 생성한다.

💡 TraceContextOrSamplingFlags 객체란?

TraceContextOrSamplingFlags 객체는 분산 추적 시스템에서 중요한 역할을 하는 두 가지 요소를 결합한 것이다:

1. TraceContext: 이것은 특정 추적의 컨텍스트 정보를 포함한다. 예를 들어, 특정 요청이 시스템의 어떤 부분을 통과했는지, 해당 요청에 할당된 고유 식별자(trace ID) 등의 정보를 담고 있다. 이 정보는 요청이 다양한 서비스나 컴포넌트를 거치며 전달될 때 그 경로를 추적하는 데 사용된다.

2. SamplingFlags: 이것은 샘플링에 대한 결정을 나타낸다. 샘플링은 모든 요청을 추적하지 않고, 일정 비율 또는 특정 조건을 만족하는 요청만을 추적하는 방법이다. 이는 데이터 양을 관리 가능한 수준으로 유지하고, 시스템의 성능을 최적화하기 위해 사용된다. SamplingFlags는 추적해야 할 요청을 결정하는 데 사용되는 규칙이나 설정을 포함할 수 있다.

TraceContextOrSamplingFlags.create(context) 메소드는 주어진 TraceContext를 기반으로 새로운 TraceContextOrSamplingFlags 객체를 생성한다. 이 객체는 추적 컨텍스트와 샘플링 결정을 함께 포함하므로, 추적 시스템은 이 정보를 사용하여 어떤 요청을 추적할지, 그리고 추적 정보를 어떻게 처리할지 결정할 수 있다.

요컨대, TraceContextOrSamplingFlags는 분산 추적 시스템에서 요청의 경로를 추적하고, 효율적으로 데이터를 수집 및 처리하는 데 필요한 핵심 정보를 제공한다.

 

• .start() 메소드를 호출하면 Span은 활성화되고, 시간과 데이터를 기록하기 시작한다. 하지만 이 시점에서는 Span이 실제 코드 실행과 연결되지 않는다. 아래 연결 코드가 필수적으로 필요하다.

 

2-4. 생성된 Span을 현재 실행 컨텍스트에 연결

• 아래 코드는 try-with-resources 구문을 사용한 Span 처리다.

새로 생성된 Span을 현재 실행 컨텍스트에 연결

 

닉네임 변경 스프링 이벤트를 발행

 

1. Span 관리
   • Span은 추적 시스템에서 특정 작업을 나타내는 단위다.
   • tracer.withSpanInScope(span)을 통해 생성된 Span을 현재 실행 컨텍스트에 연결한다.
   • 이 연결을 통해 해당 Span이 추적하는 작업 범위 내에서 발생하는 모든 이벤트와 메트릭스가 해당 Span에 기록된다.
2. 자동 리소스 관리
   • try-with-resources 구문은 자동으로 리소스를 관리한다.
   • 이 구문 내에서 열린 리소스는 구문이 종료될 때 자동으로 닫힌다.
   • Tracer.SpanInScope 객체도 이와 같이 try 블록이 종료되면 자동으로 닫히며, 이를 통해 Span이 올바르게 종료된다.
3. 오류 처리와 로깅
   • catch 블록에서 발생한 예외를 처리한다.
   • span.tag("error", e.toString())는 발생한 예외를 추적 Span에 태그로 추가하여, 오류 발생 사실을 추적 시스템에 기록한다. log.error를 통해 로그 시스템에도 오류를 기록한다.
   • catch 블록에서 span.tag("error")를 추가하지 않으면, Zipkin과 같은 추적 시스템에서 오류를 제대로 인식하지 못할 수 있다. 이 태그를 추가하면 Zipkin UI에서도 오류를 확인할 수 있다.
4. Span 완료
   • finally 블록은 try 또는 catch 블록 이후 항상 실행된다.
   • 여기서 span.finish()를 호출하여 Span의 수명 주기를 완료한다.
   • 이 호출은 추적 시스템에 Span이 종료됨을 알리고, 수집된 추적 데이터를 추적 시스템으로 전송하는 데 필요하다.
   • try-with-resources 구문이 SpanInScope를 자동으로 닫아주지만, Span 자체의 종료를 명시적으로 선언하는 것이 좋은 관행이다.
   • 이렇게 하면 추적 데이터의 정확성과 완전성이 보장된다.

요약하자면, try-with-resources 구문은 리소스 관리와 예외 처리를 안전하고 효율적으로 수행하게 해준다. 이 방식을 사용하면 리소스 누수를 방지하고, 리소스가 적절하게 닫히도록 보장할 수 있다. 또한 Span 추적과 오류 처리에서도 중요한 역할을 한다. 이러한 점들은 특히 리소스 관리와 예외 처리가 중요한 분산 추적 시스템에서 매우 유용하다.

 

 

(전체 코드)

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.recipia.recipe.event.springevent.NicknameChangeEvent;
import io.awspring.cloud.sqs.annotation.SqsListener;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import brave.Span;
import brave.Tracer;
import brave.propagation.TraceContext;
import brave.propagation.TraceContextOrSamplingFlags;

@Slf4j
@RequiredArgsConstructor
@Service
public class AwsSqsListenerService {

    private final ObjectMapper objectMapper;
    private final ApplicationEventPublisher eventPublisher;
    private final Tracer tracer;

    @SqsListener(value = "${spring.cloud.aws.sqs.nickname-sqs-name}")
    public void receiveMessage(String messageJson) throws JsonProcessingException {

        JsonNode messageNode = objectMapper.readTree(messageJson);
        String messageId = messageNode.get("MessageId").asText();
        String messageContent = messageNode.get("Message").asText();

        log.info("[RECIPE] Received message from SQS with messageId: {}", messageId);

        JsonNode message = objectMapper.readTree(messageContent);
        String traceId = extractTraceIdFromMessage(message);

        TraceContext context = buildTraceContext(traceId);

        Span span = tracer.nextSpan(TraceContextOrSamplingFlags.create(context))
                .name("[RECIPE] nickname-change SQS Received") // Span 이름 지정
                .start();

        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            processNicknameMessage(message);
        } catch (Exception e) {
            span.tag("error", e.toString());
            log.error("Error processing SQS message: ", e);
        } finally {
            span.finish();
        }
    }

    private String extractTraceIdFromMessage(JsonNode message) {
        return message.get("traceId").asText();
    }

    private TraceContext buildTraceContext(String traceId) {

        TraceContext.Builder contextBuilder = TraceContext.newBuilder();
        if (traceId.length() == 32) {
            long traceIdHigh = Long.parseUnsignedLong(traceId.substring(0, 16), 16);
            long traceIdLow = Long.parseUnsignedLong(traceId.substring(16), 16);
            contextBuilder.traceIdHigh(traceIdHigh).traceId(traceIdLow);
        } else {
            long traceIdLow = Long.parseUnsignedLong(traceId, 16);
            contextBuilder.traceId(traceIdLow);
        }
        // 새로운 Span ID 생성 (tracer 인스턴스를 사용)
        contextBuilder.spanId(tracer.nextSpan().context().spanId());

        return contextBuilder.build();

    }

    private void processNicknameMessage(JsonNode message) throws JsonProcessingException {
        // 'message' 필드 내의 JSON 문자열을 추출
        String messageContent = message.get("message").asText();

        // 추출된 JSON 문자열을 파싱하여 memberId를 얻음
        JsonNode innerMessageNode = objectMapper.readTree(messageContent);
        Long memberId = Long.valueOf(innerMessageNode.get("memberId").asText());

        // 추출된 memberId로 이벤트 발행 및 로깅
        eventPublisher.publishEvent(new NicknameChangeEvent(memberId));
        log.info("Processed NicknameChangeEvent for memberId: {}", memberId);
    }
}

 

 

3. 레시피 서버에서 멤버 서버로 Feign 요청 및 FeignClientConfig 설정

3-1. 멤버 서버로 Feign 요청

• 닉네임 변경 이벤트를 리스닝하던 메소드가 실행된다.
• 이 메소드는 멤버 서버에 Feign 요청을 보낸다.

닉네임 변경 이벤트 리스너에서 멤버 서버로 Feign 클라이언트 요청

 

1. 새로운 Span 생성의 목적
   • 이전 코드에서 SQS 리스너 메소드에 해당하는 Span은 그 메소드의 종료와 함께 finish()되었다.
   • 그러나 requestMemberChangedNickname 메소드에서는 멤버 서버에 대한 Feign 요청을 보내는 작업에 집중하기 위해 새로운 Span을 생성했다.
   • 이는 Zipkin과 같은 추적 시스템에서 이 요청의 세부 사항을 보다 명확하게 추적하고자 하는 의도에서 비롯된 것이다.
2. Trace ID의 연속성
   • 새로운 Span을 생성하더라도, Tracer 인스턴스는 기존의 추적(Trace) 컨텍스트와 연관된 Trace ID를 유지한다.
   • 즉, 이전 Span에서 설정된 Trace ID가 새로 생성된 Span에도 계승되어, 여러 Span이 동일한 요청 또는 트랜잭션에 속하는 것으로 추적 시스템에 나타난다.
3. 새로운 Span의 역할
   • requestMemberChangedNickname 메소드에서 새로 생성된 Span은 Feign 요청의 전송과 응답 처리 과정을 추적한다.
   • 이 Span은 해당 요청의 시작부터 완료까지의 모든 세부 사항을 기록하여, 추적 시스템에서 요청의 성공, 실패, 지연 등을 분석할 수 있도록 한다.
4. 기존 추적 컨텍스트와의 연동
   • 새로운 Span이 생성되더라도, 이는 기존 추적 컨텍스트에 연결된다.
   • 이를 통해 추적 시스템에서는 여러 Span들이 하나의 큰 트랜잭션 또는 요청 흐름의 일부분으로 보여질 수 있다.
   • 이러한 연동은 분산 시스템에서 서비스 간의 상호작용을 보다 명확하게 이해하는 데 도움을 준다.

 

요약하자면, requestMemberChangedNickname 메소드에서 새로운 Span을 생성하는 것은 Feign 요청의 세부적인 추적을 목적으로한다. 이 새로운 Span은 기존의 Trace 컨텍스트와 연결되어 있으며, 이를 통해 서비스 간의 상호작용과 전체 트랜잭션의 흐름을 추적 시스템에서 명확하게 파악할 수 있다. 이러한 접근 방식은 분산 시스템에서 세부적인 성능 모니터링과 오류 추적을 가능하게 한다.

 

3-2. FeignClientConfig 설정 추가

이 코드는 FeignClientConfig 클래스를 통해 Feign 클라이언트의 요청 인터셉터를 설정하는 부분이다.

주된 목적은 두 가지: 하나는 레시피 서버에서 멤버 서버로의 Feign 요청을 추적할 수 있도록 추적 ID(Trace ID)를 전달하는 것이고, 다른 하나는 Feign 요청임을 식별하는 특별한 헤더를 추가하는 것이다. 

import brave.Span;
import feign.RequestInterceptor;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import brave.Tracer;

@RequiredArgsConstructor
@Configuration
public class FeignClientConfig {

    private final Tracer tracer;

    @Bean
    public RequestInterceptor requestInterceptor() {
        return template -> {
            Span currentSpan = tracer.currentSpan();
            String traceId = (currentSpan != null) ? currentSpan.context().traceIdString() : null;

            // TraceID를 HTTP 요청 헤더에 추가 (null이 아닐 경우만)
            if (traceId != null) {
                template.header("X-Trace-Id", traceId);
            }

            // Feign 요청임을 나타내는 헤더 추가
            template.header("X-Feign-Client", "true");
        };
    }
}

 

1. Trace ID 전달:
   • Span currentSpan = tracer.currentSpan();
     • 현재 컨텍스트의 Span을 가져온다. 이 Span은 현재 진행 중인 요청의 추적 정보를 가지고 있다.
   • String traceId = (currentSpan != null) ? currentSpan.context().traceIdString() : null;
     • 현재 Span이 존재한다면 그로부터 Trace ID를 추출한다.
     • Trace ID는 분산 추적 시스템에서 요청의 전체 경로를 식별하는 데 사용되는 고유 식별자이다.
   • if (traceId != null) { template.header("X-Trace-Id", traceId); }
     • 추출한 Trace ID가 있을 경우, 이를 Feign 요청의 HTTP 헤더에 추가한다.
     • 이를 통해 멤버 서버는 레시피 서버에서 시작된 추적을 계속할 수 있습니다.
2. Feign 요청 식별:
   • template.header("X-Feign-Client", "true");
     • 모든 Feign 요청에 X-Feign-Client 헤더를 추가하고, 그 값을 true로 설정한다.
     • 이는 멤버 서버에서 HTTP 요청을 받았을 때, 해당 요청이 Feign 클라이언트를 통해 온 것임을 식별하기 위한 목적이다.
3. 효율적인 추적 처리를 위한 로직:
   • 멤버 서버에서는 X-Feign-Client 헤더의 존재와 그 값이 true일 때만 X-Trace-Id 헤더를 확인하고, 현재 Trace ID로 세팅한다. 이는 멤버 서버가 모든 HTTP 요청에 대해 X-Trace-Id를 검사하는 것이 아닌, Feign 요청에 한정하여 추적을 진행함으로써 불필요한 리소스 낭비를 줄이기 위함이다.

이 설정은 레시피 서버와 멤버 서버 간의 통신을 추적하는 데 있어서 일관된 Trace ID를 유지하면서도, 효율적으로 리소스를 사용하도록 설계되었다. Feign 요청을 명확하게 식별함으로써, 멤버 서버는 이 요청들에 대해서만 추적 로직을 수행하게 되어, 시스템의 부하를 최적화할 수 있다.

 

 

 

4. 멤버 서버의 응답 처리

• 마지막으로, 멤버 서버는 Feign 요청을 받고, FeignTraceIdFilter 필터를 통해 헤더에 X-Feign-Client값이 true일때 추가 작업을 진행한다.
• Feign 요청으로 들어온 HTTP 요청은 요청 헤더에서 TraceID를 추출하고, 이를 사용해 새로운 Span을 시작하여 처리 과정을 추적한다.
• 이 필터의 주 목적은 Feign 클라이언트 요청에서 Trace ID를 추출하고, 이를 사용하여 새로운 추적 Span을 시작하는 것이다.

import brave.Span;
import brave.propagation.TraceContext;
import jakarta.servlet.*;
import brave.Tracer;
import jakarta.servlet.http.HttpServletRequest;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;

import java.io.IOException;

@RequiredArgsConstructor
@Component
public class FeignTraceIdFilter implements Filter {

    private final Tracer tracer;
    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
        HttpServletRequest httpRequest = (HttpServletRequest) request;

        // Feign 클라이언트가 아닌 경우, 다음 필터로 이동
        if (!"true".equals(httpRequest.getHeader("X-Feign-Client"))) {
            chain.doFilter(request, response);
            return;
        }

        String incomingTraceId = httpRequest.getHeader("X-Trace-Id");
        if (incomingTraceId != null && !incomingTraceId.isEmpty()) {
            // 추출한 TraceID로 새로운 Span 시작
            TraceContext incomingContext = buildTraceContext(incomingTraceId);
            Span newSpan = tracer.newChild(incomingContext).start();
            try (Tracer.SpanInScope ws = tracer.withSpanInScope(newSpan)) {
                chain.doFilter(request, response);
            } finally {
                newSpan.finish();
            }
        } else {
            chain.doFilter(request, response);
        }
    }

    private TraceContext buildTraceContext(String traceId) {
        TraceContext.Builder contextBuilder = TraceContext.newBuilder();
        if (traceId.length() == 32) {
            long traceIdHigh = Long.parseUnsignedLong(traceId.substring(0, 16), 16);
            long traceIdLow = Long.parseUnsignedLong(traceId.substring(16), 16);
            contextBuilder.traceIdHigh(traceIdHigh).traceId(traceIdLow);
        } else {
            long traceIdLow = Long.parseUnsignedLong(traceId, 16);
            contextBuilder.traceId(traceIdLow);
        }
        contextBuilder.spanId(tracer.nextSpan().context().spanId());
        return contextBuilder.build();
    }
}

 

1. Feign 클라이언트 요청 식별
   • HttpServletRequest를 사용하여 요청을 검사하고, X-Feign-Client 헤더가 "true"로 설정되어 있는지 확인한다.
   • 이 헤더가 "true"이면 요청이 Feign 클라이언트를 통해 왔다고 간주한다.
2. Trace ID 추출 및 Span 생성
   • X-Trace-Id 헤더에서 Trace ID를 추출한다.
   • 추출한 Trace ID가 유효하면, 이를 사용하여 새로운 추적 Span을 시작한다. 이 과정에서 buildTraceContext 메소드를 사용하여 TraceContext를 구성한다.
   • TraceContext 구성 로직은 Trace ID의 길이에 따라 분기하여, 16진수 값으로 traceIdHigh와 traceIdLow를 파싱하고, 새로운 Span ID를 생성한다.
3. Span을 실행 컨텍스트에 적용
   • 새로운 Span을 현재 쓰레드의 실행 컨텍스트에 적용하기 위해 Tracer.SpanInScope를 사용한다.
   • 이를 통해 이 필터에서 실행되는 모든 코드는 새로운 Span에 의해 추적된다.
4. 요청 및 응답 처리
   • chain.doFilter를 호출하여 요청을 다음 필터나 서블릿으로 전달한다.
   • 이는 필터 체인에서의 다음 단계로 요청을 이동시키는 표준 방법이다.
5. Span 완료
   • finally 블록에서 newSpan.finish()를 호출하여 새로운 Span의 수명 주기를 종료한다.
   • 이는 추적 시스템에 이 Span에 대한 추적이 완료되었음을 알리고, 수집된 데이터를 전송하는 데 필요하다.

이 필터를 통해 멤버 서버는 Feign 클라이언트를 통해 들어오는 요청을 효과적으로 추적할 수 있다. 또한, 이 방식은 분산 추적 시스템에서 요청의 전체 경로를 명확하게 파악하고, 서비스 간의 상호작용을 더 잘 이해하는 데 도움을 준다.

 

💡 Feign 클라이언트를 통한 HTTP 요청의 추적(Trace) 처리는 미묘하게 다루어져야 한다. 중요한 것은 요청에 Trace ID가 포함되어 있는지, 그리고 이 ID를 어떻게 활용할 것인지이다:

Feign 요청에 Trace ID가 포함된 경우
 - Feign 클라이언트를 통한 HTTP 요청에는 Trace ID가 헤더에 포함될 수 있다. 멤버 서버는 이 헤더에서 Trace ID를 추출하여 사용할 수 있다.
 - 이렇게 추출한 Trace ID로 새로운 Span을 생성하면, 요청은 원래의 추적 경로를 계속 따르게 된다. 다시 말해, Feign 요청으로 수행하는 멤버 서버에서의 작업은 최초의 HTTP 요청 처리 과정의 일부로 통합된다.

Feign 요청에서 새로운 Trace ID를 생성하는 경우
 - 만약 Feign 요청에 Trace ID가 포함되어 있지 않다면, 멤버 서버는 새로운 Trace ID를 생성하여 독립적인 추적을 시작할 수 있다. 이 경우, 멤버 서버에서의 작업은 원래 서비스의 추적 경로와 분리된다.

결론
 - Feign 클라이언트를 통해 들어오는 요청이 원래 서비스의 추적 경로를 계속 따를 것인지, 아니면 멤버 서버에서 새로운 추적 경로를 시작할 것인지는 요청의 Trace ID와 멤버 서버의 추적 전략에 따라 결정된다. 이는 서비스 간 상호작용을 명확하게 추적하고 이해하는 데 핵심적인 부분이다.

내가 Feign 요청까지 최초 HTTP 요청의 추적 범위에 포함시키고자 하였기 때문에, Trace ID를 헤더에 담아 보내는 방식을 선택했다. 이 방법을 통해, 멤버 서버에서의 처리 과정도 초기 HTTP 요청의 추적 경로에 연결되어, 전체 트랜잭션의 일관된 추적이 가능하게 된다.

 

 

💡 여기서 알아차린 사람이 있겠지만 지금까지는 Span을 새로 생성할때 tracer.nextSpan()을 사용했고, Filter에서는
tracer.newChild()를 사용했다.

분산 추적 시스템에서의 Span 생성 방식에는 미묘한 차이가 있다. tracer.nextSpan()과 tracer.newChild() 모두 새로운 Span을 생성하지만, 그들이 참조하는 추적 컨텍스트의 차이에 주목해야 한다.

tracer.nextSpan()의 사용
 - tracer.nextSpan()은 현재 추적 컨텍스트(가 있다면)를 기반으로 새로운 Span을 생성한다. 현재 컨텍스트가 없다면 새로운 추적을 시작한다. 이 방식은 주로 현재 서비스 내에서 시작되는 새로운 작업 단위에 적합하다.
 - 예를 들어, 서비스 내의 요청 처리, 내부 로직 실행 등에 주로 사용된다.

tracer.newChild()의 사용
 - tracer.newChild()는 명시적으로 제공된 TraceContext를 부모로 사용하여 새로운 Span을 생성한다. 이는 외부 시스템 또는 서비스로부터 전달받은 추적 컨텍스트를 이어받을 때 유용하다.
 - 예를 들어, 외부 서비스로부터 온 요청을 처리하면서 해당 요청의 추적 경로를 유지하고자 할 때 사용된다.

 결론적으로, 각 메소드의 사용은 추적하고자 하는 작업의 성격과 시작점에 따라 달라진다. 나는 모든 작업을 하나의 Trace ID로 추적하면서, 각각의 Span이 독립적인 작업을 표현하도록 하고자 했다. 이러한 맥락에서, Feign 클라이언트 필터에서 newChild를 사용한 것은 다양한 추적 방법을 실험하고 보여주기 위한 선택이었다. 일반적으로는 nextSpan을 사용했지만, 필터에서 newChild를 사용함으로써 추적 컨텍스트의 다양한 활용 방법을 탐색해볼 수 있었다.

 이걸 보고 "어? 그럼 맨처음 SQSListener에서도 tracer.newChild()를 사용했었어도 됐었네?"라는 생각이 들었다면 아주 잘 이해한거다.

 이러한 접근 방식은 분산 추적 시스템의 유연성을 보여주며, 추적 데이터의 구조와 해석에 있어서 다양한 가능성을 탐구할 수 있도록 한다.

 

5. 로그 확인

• 이제 실제로 닉네임 변경 요청을 날려보자.
• postman으로 닉네임 변경 요청을 보내고 "success"를 응답받았다.

postman 요청

 

5-1. 멤버 서버 로그

• 초기 POST 요청
  • 멤버 서버에서 nicknameChange 경로로 POST 요청을 받았다.
  • 이때 추적 정보 Trace ID (655a4f2e2951f25a0e6632b694b088ad)와 Span ID (9a4fec2047ea162a)가 이미 설정되어 있다.
  • 이는 요청이 추적 시스템에 의해 이미 추적되고 있음을 나타낸다.
• (캡쳐에서는 생략된) 데이터베이스 작업과 SNS 메시지 발행
  • 이후 데이터베이스 쿼리와 SNS 메시지 발행 과정에서 동일한 Trace ID와 Span ID가 유지된다.
  • 이는 멤버 서버 내의 작업이 동일한 추적 컨텍스트 아래에서 진행되고 있음을 보여준다.

초기 POST 요청 로그

DB 업데이트 후 SNS 메시지 발행 로그

 

• Feign 요청 처리
  • 멤버 서버가 Feign 클라이언트를 통해 "/feign/member/getNickname?memberId=2" 경로로 요청을 받는다.
  • 이때 새로운 Span ID (9b072eb69599cd2c)가 생성되었지만, Trace ID는 이전과 동일하다.
  • 이는 새로운 작업 단위(Feign 요청 처리)가 시작됐지만, 전체 추적 경로는 동일하게 유지되고 있음을 나타낸다.

Feign 요청 로그

 

5-2. 레시피 서버 로그

• SQS 메시지 수신
  • 레시피 서버가 SQS로부터 메시지를 수신한다.
  • 이 시점의 로그에서는 Trace ID가 없다.
  • 이는 아직 추적 컨텍스트가 설정되지 않았음을 의미한다.
• 추적 정보 설정 후 작업
  • 레시피 서버에서 SQS 메시지 처리를 시작하면서 메시지에 담겨있던 Trace ID (655a4f2e2951f25a0e6632b694b088ad)와 Span ID (f9834b3e23f2586f)가 로그에 나타난다.
  • 이는 레시피 서버에서 새로운 추적 컨텍스트가 설정되었음을 보여줍니다.
• 추적 정보의 일관성
  • 레시피 서버의 후속 로그에서 Trace ID는 일관되게 유지되며, 각 작업 단위에 대해 새로운 Span ID가 생성된다.
  • 이는 레시피 서버 내의 각 작업이 동일한 추적 경로의 일부로 처리되고 있음을 나타낸다.

레시피 서버 로그

 

 

6. Zipkin UI 확인

• 해당 Trace ID로 Zipkin에 검색한다.
• 그럼 아래와 같이 잘 나오는 모습을 확인할 수 있다.

Zipkin UI 정상 작동 화면

 

• 만약에 Feign 요청을 할때 memberId값에 null을 보내게 되면 아래와 같이 error가 발생한걸 확인할 수 있다.

Zipkin UI 에러 발생 화면

 

 

💡 로그의 "member: unknown" 부분은 FeignTraceIdFilter 클래스에서 tracer.newChild(incomingContext).start();를 사용하여 Span을 생성했을 때 이름을 지정하지 않았기 때문에 나타난 것으로 보인다.

Span에 이름을 지정하지 않는 것이 필터에서 직접적인 에러를 발생시키는 원인은 아니다. 그러나 분산 추적 데이터의 분석과 가독성을 향상시키기 위해서는 Span에 명확한 이름을 지정하는 것이 좋다. 이렇게 하면 추적 로그를 통해 어떤 작업이 진행되고 있는지 더 쉽게 파악할 수 있다.

 

7. 마무리

분산 시스템에서 로깅을 구현하는 과정이 이처럼 시간과 노력이 많이 소요될 줄은 예상하지 못했다. 처음에는 Zipkin이 HTTP 요청만 추적한다는 사실을 알고 나서, 간단히 몇 가지 설정만으로 해결될 것으로 생각했다. 그러나 이 과정이 예상보다 훨씬 복잡하고 긴 여정이 될 것이라곤 상상도 못했다.

 

특히, SNS와 SQS 같은 비HTTP 요청을 통한 서버 간 연동에서 데이터베이스의 일관성을 유지하는 과정에서 로그 관리의 중요성이 두드러졌다. 이에 따라, 우리 프로젝트에서는 Zipkin을 단순히 성능 관리 목적이 아닌, 보다 상세한 로그 관리 도구로 활용하기로 결정했다. 이런 세세한 로그 분할이 처음에는 과도해 보일 수도 있지만, SNS와 SQS를 통한 서버 간 통신이 주된 관심사인 우리의 상황에서는 필수적인 접근 방식이었다.

 

분산 시스템에서의 Zipkin 활용에 대한 정보를 찾는 것이 쉽지 않아, 이러한 경험을 다른 개발자들과 공유하고자 이 글을 작성했다. 나의 경험이 SNS와 SQS를 활용하여 Zipkin으로 서버 간 통신을 추적하려는 다른 개발자들에게 실질적인 도움과 영감을 제공하기를 바란다. 이 글이 당신의 프로젝트에 적용할 수 있는 실질적인 아이디어와 동기부여가 되길 희망한다.

 

 

 

 

이 포스트는 Team chillwave에서 사이드 프로젝트 중 적용했던 부분을 다시 공부하며 기록한 것입니다.

 

시간이 괜찮다면 팀원 '개발자의 서랍'님의 블로그도 한번 봐주세요 :)

 

개발자의 서랍