[Spring Sleuth] AWS SNS 쓰레드에 Trace ID 넣어주기

이상하다?

AWS SNS & SQS를 도입해서 사용하다가 로그에서 이상한 점 하나를 발견했다.

 

SQS Receiver쪽 Thread에는 Trace ID가 잘 찍혀나오는데,

SNS Publisher쪽 Thread에는 Trace ID가 비어있었다.

 

현재 상황은 이렇다.

A 서비스에서 SNS 메세지 보냄

    =>  SQS로 메세지 이동  

    =>  B 서비스에서 SQS 메세지 받음

    =>  (Optional) B -> A 서비스 API 호출

    =>  (Optional) A 서비스에서 로직 실행

 

 

각 스텝마다 로그를 찍도록 구현했는데, 이 때 Trace ID는 다음과 같다.

A 서비스에서 SNS 메세지 보냄  ( TRACE ID = 비어있음 )

    =>  SQS로 메세지 이동  ( AWS 영역 )

    =>  B 서비스에서 SQS 메세지 받음  ( TRACE ID = A )

    =>  (Optional) B -> A 서비스 API 호출  ( TRACE ID = A )

    =>  (Optional) A 서비스에서 로직 실행  ( TRACE ID = A )

 

 

왜 SNS 메세지를 보낼때만 Trace ID가 비어서 나올까?

 

 

 

라이브러리 스터디

현재 사용하고 있는 AWS 라이브러리 종류는 크게 3가지이다.

1. AWS 공식 SNS 라이브러리

2. AWS 공식 SQS 라이브러리

3. SQS 서드파티 라이브러리 (이하 awspring 라이브러리)

 

1. AWS 공식 SNS 라이브러리

• AWS SNS를 사용할 수 있도록 하는 라이브러리
• Async로 사용할 경우 내부에서 newFixedThreadPool() 을 사용해 멀티스레딩을 진행하고 있다. 

 

2. AWS 공식 SQS 라이브러리

• AWS SQS를 사용할 수 있도록 하는 라이브러리
• Async로 사용할 경우 내부에서 newFixedThreadPool() 을 사용해 멀티스레딩을 진행하고 있다. 

 

3. awspring 라이브러리

• AWS 공식 SQS 라이브러리를 더 쉽게 사용할 수 있도록 구현한, 일종의 Wrapping 라이브러리
• 라이브러리 내부에서 메인으로 사용하는 클래스인 SimpleMessageListenerContainerFactory 에서 ThreadPoolTaskExecutor를 사용해 멀티스레딩을 진행하고 있다.

 

Spring Sleuth 라이브러리

• MSA 구조에서 Trace ID와 Span ID를 사용해 이슈 트래킹을 쉽게 할 수 있도록 돕는다.
• 멀티스레드 상황에서, ThreadPoolTaskExecutor 등을 Bean으로 등록해 사용한다면 Trace ID가 그대로 전파된다.
  
  • 참조 : https://duooo-story.tistory.com/65

@EnableAsync
@Configuration
class SomeConfig {
	@Bean
	public ThreadPoolTaskExecutor testThreadPoolTaskExecutor() {
		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		executor.setThreadNamePrefix("thread-");
		executor.setCorePoolSize(20);
		executor.setMaxPoolSize(20);
		executor.afterPropertiesSet();
		return executor;
	}
}
 
 
@Slf4j
@Service
class SomeService {
	public void helloWorld() {
 		log.info("flag1");
		someMethod();
	}
 
 	@Async("testThreadPoolTaskExecutor")
 	public void someMethod() {
		log.info("flag2");
	}
}

// SomeService.helloWorld() 가 호출됐을 경우
2023-02-22 [INFO] [main-thr] [TraceId_a123,SpanId_a] flag1
2023-02-22 [INFO] [thread-1] [TraceId_a123,SpanId_b] flag2

 

 

 

• 만약 Bean으로 등록되어있지 않은 스레드에서 로그를 발생시키면 Trace ID가 유지되지 않거나 빈칸으로 나오게 된다.

@Slf4j
@Service
class SomeService {
	public void helloWorld2() {
 		log.info("flagA");
		someMethod();
	}
 
 	@Async
 	public void someMethod2() {
		log.info("flagB");
	}
}

// SomeService.helloWorld2() 가 호출됐을 경우
2023-02-22 [INFO] [main-thread] [TraceId_a123,SpanId_a] flagA
2023-02-22 [INFO] [jv-thread-1] [,] flagB

or

2023-02-22 [INFO] [main-thread] [TraceId_a123,SpanId_a] flagA
2023-02-22 [INFO] [jv-thread-1] [TraceId_b456,SpanId_c] flagB

(어떤 경우에 비어있게 나오는지는 더 확인해봐야할 것 같다.)

 

 

왜 SQS는 Trace ID가 있고 SNS는 없나?

AWS SQS 받을때

• AWS 공식 SQS 라이브러리 대신 awspring 라이브러리를 사용하고 있다.
• 하단은 Configuration Bean 예시이다.

@RequiredArgsConstructor
@Configuration
@EnableSqs
public class AwsSQSConfigure {
    @Bean
    public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(AmazonSQSAsync amazonSQSAsync) {
        SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();
        factory.setAmazonSqs(amazonSQSAsync);
        factory.setTaskExecutor(threadPoolTaskExecutor());
        factory.setAutoStartup(true);
        return factory;
    }

    @Bean(destroyMethod = "shutdown")
    @Primary
    public AmazonSQSAsync amazonSQSAsync() {
        return AmazonSQSAsyncClientBuilder
                    .standard()
                    .withRegion("some_region")
                    .withCredentials(InstanceProfileCredentialsProvider.getInstance())
                    .build();
    }

    @Bean
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setThreadNamePrefix("sqs-thread-");
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(20);
        executor.afterPropertiesSet();
        return executor;
    }
}

 

• SimpleMessageListenerContainerFactory를 Bean으로 등록했다.
• ThreadPoolTaskExecutor를 Bean으로 등록했다.
• 그렇기 때문에 Sleuth가 자동으로 Trace ID를 주입해줄 수 있게 된다.

 

AWS SNS 보낼때

• AWS 공식 SNS 라이브러리를 사용한다.
• 하단은 Configuration Bean 예시이다.

@RequiredArgsConstructor
@Configuration
public class AwsSNSConfigure {
    @Bean(destroyMethod = "shutdown")
    public AmazonSNSAsyncClient amazonSNSClient() {
        return (AmazonSNSAsyncClient) AmazonSNSAsyncClientBuilder
                    .standard()
                    .withRegion("some_region")
                    .withCredentials(InstanceProfileCredentialsProvider.getInstance())
                    .build();
    }
}

 

• AmazonSNSAsyncClient 만을 Bean으로 등록했다.
• 그렇기 때문에 Sleuth가 자동으로 Trace ID를 주입할 수 없다.
• 위에 설명에 기술했듯, 라이브러리 자체적으로 newFixedThreadPool()을 선언해 사용하기 때문이다.

 

 

아하!

그러면 newFixedThreadPool() 를 Bean으로 등록해 사용하도록 해야겠구나!!

 

더 자세히 뜯어보니 AmazonSNSAsyncClientBuilder 에 withExecutorFactory() 옵션이 존재했다.

 

ExecutorFactory 안에 ExecutorService를 Bean으로 등록해서 사용해봤다.

 

@RequiredArgsConstructor
@Configuration
public class AwsSNSConfigure {
    @Bean(destroyMethod = "shutdown")
    public AmazonSNSAsyncClient amazonSNSClient() {
        return (AmazonSNSAsyncClient) AmazonSNSAsyncClientBuilder
                    .standard()
                    .withRegion("some_region")
                    .withCredentials(InstanceProfileCredentialsProvider.getInstance())
                    .withExecutorFactory(executorFactory())
                    .build();
    }

    @Bean
    public ExecutorFactory executorFactory() {
        return new ExecutorFactory() {
            @Override
            public ExecutorService newExecutor() {
                return executorService();
            }
        };
    }

    @Bean
    public ExecutorService executorService() {
        ThreadFactory factory = new UserDefaultThreadFactory("sns-thread-");
        return Executors.newFixedThreadPool(50, factory);
    }

    private static class UserDefaultThreadFactory implements ThreadFactory {
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private String namePrefix = "";

        UserDefaultThreadFactory(String namePrefix) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                    Thread.currentThread().getThreadGroup();
            this.namePrefix = namePrefix;
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                    namePrefix + threadNumber.getAndIncrement(),
                    0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }
}

 

프로세스를 실행시켜보니 다음과 같이 변했다.

 

A 서비스에서 SNS 메세지 보냄  ( TRACE ID = B )

    =>  SQS로 메세지 이동  ( AWS 영역 )

    =>  B 서비스에서 SQS 메세지 받음  ( TRACE ID = A )

    =>  (Optional) B -> A 서비스 API 호출  ( TRACE ID = A )

    =>  (Optional) A 서비스에서 로직 실행  ( TRACE ID = A )

 

 

빙고! 문제가 해결됐다!!

그런데... Trace ID가 두 개니까 트래킹할 때 뭔가 불편하다.

하나로 통일할 방법이 없을까?

 

에 대해서는 다음 포스팅에서... 다루도록 한다.