[Zipkin] 분산 추적을 위한 Zipkin, Kafka 통신 시 TraceId 안 찍히는 문제 해결

 

1. Zipkin 이란?

Zipkin은 분산 시스템에서의 요청 추적을 위한 오픈소스 도구이다. 마이크로서비스 아키텍처(MSA)와 같이 여러 서비스가 상호작용하는 환경에서 어떤 요청이 어디서 지연되었는지, 문제가 발생했는지를 추적하고 분석할 수 있게 도와준다.

 

Zipkin으로 추적할 수 있는 분산 트랜잭션은 대표적으로 HTTP, gRPC가 있다. 나중에 다룰 내용이지만, Kafka를 사용한 통신 시 Zipkin에서는 흐름을 추적할 수 없다. 이를 해결하기 위한 내용도 다룰 예정이니 참고바란다. 

 

 

 

 

 

 

 

 

 

 

 

 

2. Zipkin 관련 컴포넌트 용어 정리

Zipkin Client Library

서비스에서 트래이스 정보를 수집하여 Zipkin 서버의 Collector로 데이터를 전송하는 역할을 한다.

 

Collector로 데이터를 전송할 때는 일반적으로 HTTP를 사용하지만, 시스템이 클 경우 Kafka 큐를 통해 전송하기도 한다. 예를들어 어플리케이션에 http 요청 또는 gRPC 요청이 들어오면 Zipkin Client Library에서 로그를 발생시킨뒤 Zipkin 서버의 Colector에게 해당 데이터를 보낸다.

 

 

 

Collector

Zipkin Client Library 로부터 데이터를 받아 트레이스 정보와 유효성을 검증하고 검색이 가능하도록 저장 및 인덱싱한다. 

 

 

 

Storage

Zipkin Collector에서 받은 트레이스 정보를 Storage에 저장한다. 주로 ElasticSearch나 MySQL을 사용하여 저장하는데, 인메모리 또한 지원하기 때문에 로컬에서 테스트하거나 개발할 때는 인메모리 데이터베이스를 사용하는 것도 좋다고 한다.

 

명시적으로 지정하지 않으면 In-Memory 저장소에 저장된다. 

 

 

 

API

저장된 로그 정보를 검색하기 위한 JSON 형식의 API

 

 

 

 

 

 

 

 

 

 

 

3. Zipkin 흐름

 

Trace Id = X 라는 Trace Id 가 service1,2,3,4 까지 연결되어 흐르는 것을 볼 수 있고,

 

Span Id 는 각 service 내에 서로 다르게 정의되어 선언된 것을 볼 수 있다. 

 

즉, MSA 에 첫 entry point service 에서 Trace 를 정의하고, 다른 service 까지 정의한 Trace 를 계속 전파하면서 TraceContext 를 유지시키는 것이다. 

 

이렇게 전파된 TraceContext 는 logging 에 쓰여서 요청에 흐름에 대한 발자취(slueth)를 남길 수 있다. 

 

 

 

 

Trace만 사용해도 추적을 할 수 있지만 Span까지 사용한 이유는 직전 Service간의 연결 관계를 파악하기 위해서이다. Parent Id 는 이전 Span Id 를 가리킴으로써 연결관계를 표현한다. 

 

 

 

 

Span, Trace .. 가 무엇이냐?

 

• Span
  • The Basic Unit Of Work. 즉, communication 방식에 대한 실질적인 작업 단위이다. 예를 들어 API 요청, gRPC 요청이 있다.
• Trace
  • Tree 구조로 구성된 Span 의 모음
• Tracer
  • Span 의 Life Cycle을 관리한다. Span을 시작하거나 멈추거나 전파하거나 등
• Trace Context
  • 분산 tracing을 위해서 network 간의 전파되어야 하는 context(span id, trace id .. etc)
• Log correlation
  • Trace context를 기반으로 로그를 모아 관리하는 역할

 

 

 

 

 

 

 

 

 

4. Tracer Context의 전파 방식

Trace Context를 각 Service간에 서로 다른 규약으로 전송하지 않고, 하나의 표준을 정해 전송하기 위해 b3 방식과 w3c 방식을 채택하고 있다. 

 

B3 propagation은 간단히 말해 'X-B3-'으로 시작하는 X-B3-TraceId와 X-B3-ParentSpanId, X-B3-SpanId, X-B3-Sampled, 이 4개 값을 전달하는 것을 통해서 트레이스 정보를 관리한다.

 

HTTP를 통해 다른 서버로 전달하는 경우에는 HTTP 헤더를 통해서 전달하고, Kafka 메시지를 통해 전달하는 경우에는 Kafka 헤더를 통해서 전달한다.

 

 

 

1. A Service가 request를 받아 Trace ID, Span ID를 생성하고 Parent Span ID는 null로 지정된다.
2. B Service가 A서비스로부터 요청을 받으면 X-B3 헤더를 기반으로 Trace ID는 그대로 이어받고 Span ID는 새롭게 생성하며, Parent Span ID는 A service의 Span ID로 지정한다.

 

Request의 전체 흐름은 Trace ID를 기준으로 트래킹 하며 Span ID로는 해당 Request가 속했던 서비스를 식별이 가능하다. Parent ID로는 호출 간의 상관관계를 파악할 수 있다.

 

 

 

 

 

 

 

 

 

 

5. TraceId 안 찍히는 문제

Zipkin 관련 의존성들을 추가하고 요청을 보냈을 때 첫 요청을 받는 Service에서는 TraceId와 SpanId가 잘 찍혔지만, 그 이후에 호출되는 Service에서는 찍히지 않았다.

 

테스트 해보니 OpenFeign을 사용한 HTTP 요청에서는 Trace Context가 잘 전파되었지만, Kafka를 사용한 통신에서는 찍히지 않았다.

 

Kafka Consumer 부분에서 헤더를 까봤을 때도 Trace Context가 포함되지 않았다. 자동으로 Kafka 헤더를 통해 Trace Context가 전달된다고 되어 있지만 실제로는 전파되지 않았다. 그래서 아래와 같이 Trace Context를 직접 지정해주었다.

 

 

Kafka Producer

@Component
@RequiredArgsConstructor
@Slf4j
public class FtpFileProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final Tracer tracer;

    private static final String TOPIC_NAME = "ftp-file-topic";

    public boolean sendFtpFileList(List<FtpFileRequest> request) {
        try {
            String traceId = tracer.currentSpan().context().traceIdString();
            String spanId = tracer.currentSpan().context().spanIdString();

            // Kafka 메시지에 traceId와 spanId를 헤더로 추가
            Message<String> message = MessageBuilder
                .withPayload(JsonUtil.toJson(request))
                .setHeader(KafkaHeaders.TOPIC, TOPIC_NAME)
                .setHeader("X-B3-TraceId", traceId)
                .setHeader("X-B3-SpanId", spanId)
                .build();
            // Kafka 메시지 전송
            kafkaTemplate.send(message);

            log.info("Kafka 메시지 발송 성공 - 파일 개수: {}", request.size());
            return true;
        } catch (Exception e) {
            log.error("Kafka 메시지 발송 중 오류 발생 - 파일 개수: {}, 오류: {}", request.size(), e.getMessage(), e);
            return false;
        }
    }
}

 

 

Kafka Consumer

@Component
@RequiredArgsConstructor
@Slf4j
public class FtpFileSubscriber {

    private final ParsingCsvUseCase parsingCsvUseCase;
    private final Tracer tracer;

    private static final String TOPIC_NAME = "ftp-file-topic";

    /**
     * zipkin 을 위한 traceId, spanId 설정 로직
     */
    @KafkaListener(topics = TOPIC_NAME, groupId = "ftp-file-group")
    public void consume(String message,
        @Header(name = "X-B3-TraceId", required = false) String traceId) { 
        Span span = null;
        try {
            if (traceId != null) {
                TraceContext traceContext = TraceContext.newBuilder()
                    .traceIdHigh(parseHigh(traceId))
                    .traceId(parseLow(traceId))
                    .spanId(tracer.nextSpan().context().spanId()) // 새 spanId 생성
                    .build();

                // 기존 TraceContext를 기반으로 새 Span 생성
                span = tracer.toSpan(traceContext).start();

                try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
                    processMessage(message);
                }
            } else {
                log.warn("TraceId가 없어 새 Trace로 메시지를 처리합니다.");
                processMessage(message);
            }
        } catch (Exception e) {
            log.error("TraceId 설정 오류 - traceId: {}, 오류: {}", traceId, e.getMessage(), e);
            processMessage(message);
        } finally {
            if (span != null) {
                span.finish();
            }
        }
    }

    private long parseHigh(String traceId) {
        // traceId의 앞부분 64비트를 처리
        return traceId.length() > 16 ? Long.parseUnsignedLong(traceId.substring(0, traceId.length() - 16), 16) : 0L;
    }

    private long parseLow(String traceId) {
        // traceId의 뒷부분 64비트를 처리
        return Long.parseUnsignedLong(traceId.substring(Math.max(traceId.length() - 16, 0)), 16);
    }
}

 

 

 

이렇게 설정하니 Producer도 Producer지만 Consumer의 코드가 너무 더러워졌다. 

 

 

KafkaUtil클래스 를 이용하여 Zipkin을 위한 로직은 최대한 따로 빼주었고, 최종 코드는 아래와 같다. 

 

 

 

Producer & Consumer & Producer KafkaUtil

@Component
@RequiredArgsConstructor
@Slf4j
public class FtpFileProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final Tracer tracer;

    private static final String TOPIC_NAME = "ftp-file-topic";

    public boolean sendFtpFileList(List<FtpFileRequest> request) {
        try {
            // Kafka 메시지 생성 (Util 클래스 사용)
            Message<String> message = KafkaUtil.createKafkaMessage(tracer, TOPIC_NAME, request);

            // Kafka 메시지 전송
            kafkaTemplate.send(message);

            log.info("Kafka 메시지 발송 성공 - 파일 개수: {}", request.size());
            return true;
        } catch (Exception e) {
            log.error("Kafka 메시지 발송 중 오류 발생 - 파일 개수: {}, 오류: {}", request.size(), e.getMessage(), e);
            return false;
        }
    }
}

 

@Component
@RequiredArgsConstructor
@Slf4j
public class FtpFileSubscriber {

    private final ParsingCsvUseCase parsingCsvUseCase;
    private final Tracer tracer;

    private static final String TOPIC_NAME = "ftp-file-topic";

    /**
     * zipkin 을 위한 traceId, spanId 설정 로직
     */
    @KafkaListener(topics = TOPIC_NAME, groupId = "ftp-file-group")
    public void consume(String message, @Header(name = "X-B3-TraceId", required = false) String traceId) {
        Span span = KafkaUtil.startTraceIfExists(tracer, traceId);
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            List<FtpFileRequest> fileRequests = KafkaUtil.toClassList(message, FtpFileRequest.class);
            parsingCsvUseCase.parsingCsv(fileRequests);
            log.info("Kafka 메시지 처리 성공 - 처리된 파일 개수: {}", fileRequests.size());
        } catch (Exception e) {
            log.error("Kafka 메시지 처리 중 오류 발생 - traceId: {}, 오류: {}", traceId, e.getMessage(), e);
        } finally {
            if (span != null) {
                span.finish(); // Span 종료
            }
        }
    }
}

 

@Slf4j
public class KafkaUtil {

    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static <T> T toClass(String json, Class<T> clazz) {
        try {
            return objectMapper.readValue(json, clazz);
        } catch (Exception e) {
            throw new IllegalArgumentException("JSON 변환 중 오류 발생: " + e.getMessage(), e);
        }
    }

    public static <T> List<T> toClassList(String json, Class<T> clazz) {
        try {
            return objectMapper.readValue(json, objectMapper.getTypeFactory().constructCollectionType(List.class, clazz));
        } catch (Exception e) {
            throw new IllegalArgumentException("JSON 리스트 변환 중 오류 발생: " + e.getMessage(), e);
        }
    }

    public static String toJson(Object object) {
        try {
            return objectMapper.writeValueAsString(object);
        } catch (Exception e) {
            log.error("객체를 JSON으로 변환 중 오류 발생: {}", e.getMessage(), e);
            throw new IllegalArgumentException("JSON 변환 오류", e);
        }
    }
    
	public static <T> Message<String> createKafkaMessage(Tracer tracer, String topicName, T payload) {
        // traceId와 spanId 가져오기
        String traceId = tracer.currentSpan().context().traceIdString();
        String spanId = tracer.currentSpan().context().spanIdString();

        // Kafka 메시지 생성
        return MessageBuilder
            .withPayload(toJson(payload))
            .setHeader(KafkaHeaders.TOPIC, topicName)
            .setHeader("X-B3-TraceId", traceId)
            .setHeader("X-B3-SpanId", spanId)
            .build();
    }

    public static Span startTraceIfExists(Tracer tracer, String traceId) {
        if (traceId != null) {
            TraceContext traceContext = TraceContext.newBuilder()
                .traceIdHigh(parseHigh(traceId))
                .traceId(parseLow(traceId))
                .spanId(tracer.nextSpan().context().spanId()) // 새 spanId 생성
                .build();
            return tracer.toSpan(traceContext).start();
        }
        log.warn("TraceId가 없어 새 Trace로 메시지를 처리합니다.");
        return tracer.nextSpan().start();
    }

    private static long parseHigh(String traceId) {
        return traceId.length() > 16 ? Long.parseUnsignedLong(traceId.substring(0, traceId.length() - 16), 16) : 0L;
    }

    private static long parseLow(String traceId) {
        return Long.parseUnsignedLong(traceId.substring(Math.max(traceId.length() - 16, 0)), 16);
    }
}

 

 

 

 

이렇게 직접 TraceId를 작성하는게 근본적인 해결책은 아닐 수 있지만, 찾아봤을 땐 마땅히 방법이 없어 보였고, 직접 Kafka 헤더를 통해 TraceId를 전달하는 방식으로 해결하였다.