프로젝트에서 Batch Job 로그에 MDC를 적용했다.

slf4j, logback, log4j2 등의 로그 라이브러리에서 제공하는 Map 형식으로 클라이언트의 특징적인 데이터를 저장하기 위한 매커니즘이다. MDC는 key / value 저장소를 지원하고 있으며, 이 저장소는 ThreadContext에 의존한다.

 

MDC가 도입된 이유

여러 스레드가 실행되는 서버 환경에서 로그를 남길 때, 순서가 보장되지 않는다. 예를 들어 동시에 3명의 클라이언트가 요청했다고 가정하면, 이러한 스레드의 요청 처리는 순차적으로 이뤄지지 않으며 심지어는 외부 API 호출 같은 비동기 처리를 위해 다른 스레드를 사용하기도 한다. 그래서 특정 요청에 대한 로그를 추적하기가 어렵다.

 

예를 들어 현재 런세권 프로젝트에서 Spring Batch Job에 대한 로그를 남기고 있는데, 이 사이에 클라이언트 요청이 오면 로그 순서가 보장되지 않기 때문에 Spring Batch와 관련된 로그를 수 많은 로그들 사이에서 찾아야한다.

 

@Slf4j  
@Component  
public class CourseSyncJobListener implements JobExecutionListener {  
  
    @Override  
    public void beforeJob(JobExecution jobExecution) {  
        JobParameters params = jobExecution.getJobParameters();  
        String runId = params.getString("run.id");  
  
        log.info("[CourseSyncJob 시작] run-id: {} | startTime: {}", runId, jobExecution.getStartTime());  
    }  
  
    @Override  
    public void afterJob(JobExecution jobExecution) {  
        JobParameters params = jobExecution.getJobParameters();  
        String runId = params.getString("run.id");  
        BatchStatus status = jobExecution.getStatus();  
  
        Duration duration = Duration.between(jobExecution.getStartTime(), jobExecution.getEndTime());  
  
        log.info("[CourseSyncJob 종료] run-id: {} | endTime: {}", runId, jobExecution.getEndTime());  
        log.info("상태: {}", status);  
        log.info("처리 시간: {}시간 {}분 {}초", duration.toHoursPart(), duration.toMinutesPart(), duration.toSecondsPart());  
        jobExecution.getStepExecutions().forEach(stepExecution ->  
                log.info("Step [{}]: Read={}, Write={}, Skip={}",  
                stepExecution.getStepName(),  
                stepExecution.getReadCount(),  
                stepExecution.getWriteCount(),  
                stepExecution.getSkipCount()  
        ));  
    }  
}

그래서 MDC라는 개념이 나왔고, 요청마다 고유 값을 할당하고, 이를 통해 로그를 관리하고자 등장한 것이다. 즉, 로그에 맥락을 부여하기 위함이다!

 

 

그렇다면, 프로젝트에서는?

위의 런세권 코드 예시에서는 MDC를 사용하는 이유가 조금 흐려진다. 왜냐면, 해당 Batch Job은 단일 스레드로 실행되고 매일 새벽 2시에 딱 한 번 실행된다. 더군다나 `CourseSyncJobListener`를 상속하는 `CourseSyncJobListener`가 내부적으로 Batch Job Id를 가져올 수 있기 때문이다.

 

@Slf4j  
@Component  
public class CourseSyncJobListener implements JobExecutionListener {  
  
    @Override  
    public void beforeJob(JobExecution jobExecution) {  
        JobParameters params = jobExecution.getJobParameters();  
        String runId = params.getString("run.id");  
  
        log.info("[CourseSyncJob 시작] run-id: {} | startTime: {}", runId, jobExecution.getStartTime());  
    }  
  
    @Override  
    public void afterJob(JobExecution jobExecution) {  
        JobParameters params = jobExecution.getJobParameters();  
        String runId = params.getString("run.id");  
        BatchStatus status = jobExecution.getStatus();  
  
        Duration duration = Duration.between(jobExecution.getStartTime(), jobExecution.getEndTime());  
  
        log.info("[CourseSyncJob 종료] run-id: {} | endTime: {}", runId, jobExecution.getEndTime());  
        log.info("[run-id: {}] 상태: {}", runId, status);  
        log.info("[run-id: {}] 처리 시간: {}시간 {}분 {}초", runId, duration.toHoursPart(), duration.toMinutesPart(), duration.toSecondsPart());  
        jobExecution.getStepExecutions().forEach(stepExecution ->  
                log.info("[run-id: {}] Step [{}]: Read={}, Write={}, Skip={}",
                runId,
                stepExecution.getStepName(),  
                stepExecution.getReadCount(),  
                stepExecution.getWriteCount(),  
                stepExecution.getSkipCount()  
        ));  
    }  
}

코드로 나타내면 이런 방식으로 로그에 대한 고유값을 관리할 수 있는 것 아닌가? 하지만, 여기서 내가 간과한게 있었다. 현재 프로젝트에서는 Batch Job을 자동 실행하는 방법뿐 아니라 수동 호출 할 수 있는 API가 있다는 점이다. 그리고 해당 API는 내부적으로 `CourseSyncService`를 호출한다.

 

@Slf4j  
@Component  
@RequiredArgsConstructor  
public class CourseSyncService {  
  
    private final JobLauncher jobLauncher;  
    private final Job courseSyncJob;  
  
    @Scheduled(cron = "0 0 2 * * *", zone = "Asia/Seoul")  
    public void runScheduledCourseSyncJob() {  
        try {  
            log.info("CourseSyncJob 자동 시작");  
            JobParameters jobParameters = new JobParametersBuilder()  
                    .addString("run.id", "scheduled-" + System.currentTimeMillis())  
                    .toJobParameters();  
            jobLauncher.run(courseSyncJob, jobParameters);  
        } catch (Exception e) {  
            log.warn("[EXCEPTION] CourseSyncJob 자동 실행 중 예외 발생", LogContent.exception(e));  
        }
    }  
  
    @Async("asyncExecutor")  
    public void runCourseSyncJob() {  
        try {  
            log.info("CourseSyncJob 수동 시작");  
            JobParameters jobParameters = new JobParametersBuilder()  
                    .addString("run.id", "manual-" + System.currentTimeMillis())  
                    .toJobParameters();  
            jobLauncher.run(courseSyncJob, jobParameters);  
        } catch (Exception e) {  
            log.warn("[EXCEPTION] CourseSyncJob 수동 실행 중 예외 발생", LogContent.exception(e));  
        }
    }  
}

 

 

여기서 수동 시작과 자동 시작을 구분하기 위해 진입 로그를 남기고 있다. 하지만, `CourseSyncJobListener`가 진입 로그를 다시 남기고 있어 중복된 로그일 뿐 아니라 수동인지 자동인지 분간할 수가 없다. 그래서 기존 Listener의 진입 로그를 지우고, `CourseSyncService`에서 run.id와 함께 남길 필요가 있었다. 이를 위해 run.id에 들어가는 내용을 변수로 분리하고, Job 파라미터로 넘겨도 되지만 나중에 다른 연동되는 부분들이 생기면 똑같이 파라미터로 넘겨줘야 한다! 심지어 매번 로그 앞에 run.id를 남겨야 하는 귀찮음까지 있다. 그래서 차라리 MDC를 이용하는게 훨씬 좋다고 생각했다.

 

@Slf4j  
@Component  
@RequiredArgsConstructor  
public class CourseSyncService {  
  
    private final JobLauncher jobLauncher;  
    private final Job courseSyncJob;  
  
    @Scheduled(cron = "0 0 2 * * *", zone = "Asia/Seoul")  
    public void runScheduledCourseSyncJob() {  
        String runId = "scheduled-" + System.currentTimeMillis();  
        MDC.put("run.id", runId);  
        try {  
            log.info("CourseSyncJob 자동 시작");  
            JobParameters jobParameters = new JobParametersBuilder()  
                    .addString("run.id", runId)  
                    .toJobParameters();  
            jobLauncher.run(courseSyncJob, jobParameters);  
        } catch (Exception e) {  
            log.warn("[EXCEPTION] CourseSyncJob 자동 실행 중 예외 발생", LogContent.exception(e));  
        } finally {  
            MDC.remove("run.id");  
        }  
    }  
  
    @Async("asyncExecutor")  
    public void runCourseSyncJob() {  
        String runId = "manual-" + System.currentTimeMillis();  
        MDC.put("run.id", runId);  
        try {  
            log.info("CourseSyncJob 수동 시작");  
            JobParameters jobParameters = new JobParametersBuilder()  
                    .addString("run.id", runId)  
                    .toJobParameters();  
            jobLauncher.run(courseSyncJob, jobParameters);  
        } catch (Exception e) {  
            log.warn("[EXCEPTION] CourseSyncJob 수동 실행 중 예외 발생", LogContent.exception(e));  
        } finally {  
            MDC.remove("run.id");  
        }  
    }  
}

 

그래서 CourseSyncService를 이렇게 개선했다. 여기서는 runId를 생성하고 MDC에 넣는다. 그럼 해당 스레드가 실행될 때 MDC가 생성되고 이에 run.id를 key로 가지는 데이터를 넣는다. 하지만, CourseSyncService의 스레드와 Batch Job 실행 스레드는 다르기 때문에 스레드 로컬한 MDC를 사용할 수가 없었다!

이럴 때는 스레드가 전환될 때 MDC를 전환되는 스레드에 넘겨주어야 한다. 이는 Spring에서 지원하고 있는 `TaskDecorator`를 상속하여 해결할 수 있었다.

 

private static class MdcTaskDecorator implements TaskDecorator {  
  
    @Override  
    public Runnable decorate(Runnable runnable) {  
        Map<String, String> contextMap = MDC.getCopyOfContextMap();  
  
        return () -> {  
            Map<String, String> previous = MDC.getCopyOfContextMap();  
            try {  
                if (contextMap != null) {  
                    MDC.setContextMap(contextMap);  
                } else {  
                    MDC.clear();  
                }  
                runnable.run();  
            } finally {  
                if (previous != null) {  
                    MDC.setContextMap(previous);  
                } else {  
                    MDC.clear();  
                }  
            }  
        };  
    }  
}

@Slf4j  
@Component  
public class CourseSyncJobListener implements JobExecutionListener {  
  
    @Override  
    public void afterJob(JobExecution jobExecution) {  
        BatchStatus status = jobExecution.getStatus();  
        Duration duration = Duration.between(jobExecution.getStartTime(), jobExecution.getEndTime());  
  
        StringBuilder stepInfo = new StringBuilder();  
        jobExecution.getStepExecutions().forEach(stepExecution ->  
                stepInfo.append(String.format(" | Step [%s]: Read=%d, Write=%d, Skip=%d",  
                        stepExecution.getStepName(),  
                        stepExecution.getReadCount(),  
                        stepExecution.getWriteCount(),  
                        stepExecution.getSkipCount()  
                ))  
        );  
  
        log.info("[CourseSyncJob 종료] 상태: {} | 처리 시간: {}시간 {}분 {}초{}",  
                status,  
                duration.toHoursPart(),  
                duration.toMinutesPart(),  
                duration.toSecondsPart(),  
                stepInfo);  
    }  
}

이를 통해서 CourseSyncJobListener를 개선할 수 있었다. MDC를 통한 개선을 하면서 여러 줄로 나뉜 로그를 하나의 로그로 합치기도 했다. 좀 더 쉽게 추적하기 위함이었다. 그리고, 이러한 목적을 좀 더 잘 나타내기 위해서 Logback 설정을 통해 Batch Job의 경우 기존 로그와 함께 파일로 저장하는 것이 아니라 따로 batch.log라는 파일에 기록하도록 변경했다~!