[예제][3장][작전2] windows 에서 마지막 예제
미해결
죽음의 Spring Batch: 새벽 3시의 처절한 공포는 이제 끝이다.
... @Slf4j @Configuration @RequiredArgsConstructor public class LogProcessingJobConfig { private final JobRepository jobRepository; private final PlatformTransactionManager transactionManager; private final String BASE_PATH = "C:\\Users\\user\\Desktop\\batch\\section3\\flatfileitemwriter"; @Bean public Job logProcessingJob( Step createDirectoryStep, Step logCollectionStep, Step logProcessingStep ) { return new JobBuilder("logProcessingJob", jobRepository) .start(createDirectoryStep) .next(logCollectionStep) .next(logProcessingStep) .build(); } @Bean public Step createDirectoryStep(SystemCommandTasklet mkdirTasklet) { return new StepBuilder("createDirectoryStep", jobRepository) .tasklet(mkdirTasklet, transactionManager) .build(); } @Bean @StepScope public SystemCommandTasklet mkdirTasklet( @Value("#{jobParameters['date']}") String date ) { SystemCommandTasklet tasklet = new SystemCommandTasklet(); tasklet.setWorkingDirectory(BASE_PATH); String collectedLogsPath = "collected_ecommerce_logs\\" + date; String processedLogsPath = "processed_logs\\" + date; try { tasklet.setCommand("cmd.exe", "/c", "mkdir", collectedLogsPath, processedLogsPath); tasklet.setTimeout(3000); } catch (Exception e) { log.error("mkdirTasklet error: ", e); } return tasklet; } @Bean public Step logCollectionStep(SystemCommandTasklet scpTasklet) { return new StepBuilder("logCollectionStep", jobRepository) .tasklet(scpTasklet, transactionManager) .build(); } @Bean @StepScope public SystemCommandTasklet scpTasklet( @Value("#{jobParameters['date']}") String date ) { SystemCommandTasklet tasklet = new SystemCommandTasklet(); tasklet.setWorkingDirectory(BASE_PATH); String processedLogsPath = "collected_ecommerce_logs\\" + date; StringJoiner commandBuilder = new StringJoiner(" & "); for (String host : List.of("localhost")) { // Windows 환경에서 해당 코드를 적용하려면 ssh 관련 설정들을 찾아야 해서 주석 처리 // String command = String.format("scp %s:~/ecommerce_logs/%s.log %s\\%s.log", host, date, processedLogsPath, host); String sourcePath = BASE_PATH + "\\ecommerce_logs\\" + date + ".log"; String command = String.format("copy %s %s\\%s.log", sourcePath, processedLogsPath, host); commandBuilder.add(command); } tasklet.setCommand("cmd.exe", "/c", commandBuilder.toString()); tasklet.setTimeout(10000); return tasklet; } @Bean public Step logProcessingStep( MultiResourceItemReader<LogEntry> multiResourceItemReader, LogEntryProcessor logEntryProcessor, FlatFileItemWriter<ProcessedLogEntry> processedLogEntryJsonWriter ) { return new StepBuilder("logProcessingStep", jobRepository) .<LogEntry, ProcessedLogEntry>chunk(10, transactionManager) .reader(multiResourceItemReader) .processor(logEntryProcessor) .writer(processedLogEntryJsonWriter) .build(); } @Bean @StepScope public MultiResourceItemReader multiResourceItemReader( @Value("#{jobParameters['date']}") String date ) { MultiResourceItemReader<LogEntry> resourceItemReader = new MultiResourceItemReader<>(); resourceItemReader.setName("multiResourceItemReader"); resourceItemReader.setResources(getResources(date)); resourceItemReader.setDelegate(logFileReader()); return resourceItemReader; } private Resource[] getResources(String date) { try { String formattedBasePath = BASE_PATH.replace("\\", "/"); String location = "file:///" + formattedBasePath + "/collected_ecommerce_logs/" + date + "/*.log"; PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver(); return resolver.getResources(location); } catch (IOException e) { throw new RuntimeException("Failed to resolve log files", e); } } @Bean public FlatFileItemReader<LogEntry> logFileReader() { return new FlatFileItemReaderBuilder<LogEntry>() .name("logFileReader") .delimited() .delimiter(",") .names("dateTime", "level", "message") .targetType(LogEntry.class) .build(); } @Bean public LogEntryProcessor logEntryProcessor() { return new LogEntryProcessor(); } @Bean @StepScope public FlatFileItemWriter<ProcessedLogEntry> processedLogEntryFlatFileItemWriter( @Value("#{jobParameters['date']}") String date ) { String outputPath = Paths.get(BASE_PATH, "processed_logs", date, "processed_logs.jsonl").toString(); ObjectMapper objectMapper = new ObjectMapper(); JavaTimeModule javaTimeModule = new JavaTimeModule(); javaTimeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"))); objectMapper.registerModule(javaTimeModule); return new FlatFileItemWriterBuilder<ProcessedLogEntry>() .name("processedLogEntryJsonWriter") .resource(new FileSystemResource(outputPath)) .lineAggregator(item -> { try { return objectMapper.writeValueAsString(item); } catch (JsonProcessingException e) { throw new RuntimeException("Error Converting item to JSON", e); } }) .build(); } @Data public static class LogEntry { private String dateTime; private String level; private String message; } @Data public static class ProcessedLogEntry { private LocalDateTime dateTime; private LogLevel level; private String message; private String errorCode; } public enum LogLevel { INFO, WARN, ERROR, DEBUG, UNKNOWN; public static LogLevel fromString(String level) { if (level == null || level.trim().isEmpty()) { return UNKNOWN; } try { return valueOf(level.toUpperCase()); } catch (IllegalArgumentException e) { return UNKNOWN; } } } public static class LogEntryProcessor implements ItemProcessor<LogEntry, ProcessedLogEntry> { private static final DateTimeFormatter ISO_FORMATTER = DateTimeFormatter.ISO_DATE_TIME; private static final Pattern ERROR_CODE_PATTERN = Pattern.compile("ERROR_CODE\\[(\\w+)]"); @Override public ProcessedLogEntry process(LogEntry item) throws Exception { ProcessedLogEntry processedLogEntry = new ProcessedLogEntry(); processedLogEntry.setDateTime(parseDateTime(item.getDateTime())); processedLogEntry.setLevel(parseLevel(item.getLevel())); processedLogEntry.setMessage(item.getMessage()); processedLogEntry.setErrorCode(extractErrorCode(item.getMessage())); return processedLogEntry; } private LocalDateTime parseDateTime(String dateTime) { return LocalDateTime.parse(dateTime, ISO_FORMATTER); } private LogLevel parseLevel(String level) { return LogLevel.fromString(level); } private String extractErrorCode(String message) { if (message == null) { return null; } Matcher matcher = ERROR_CODE_PATTERN.matcher(message); if (matcher.find()) { return matcher.group(1); } // ERROR 문자열이 포함되어 있지만 패턴이 일치하지 않는 경우 if (message.contains("ERROR")) { return "UNKNOWN_ERROR"; } return null; } } } 윈도우즈에서는 scp 명령어 관련해서 windows 서비스에서 open-ssh-server 실행 현재 사용하고 있는 사용자(windows 사용자) 가 관리자 권한을 보유 할 경우 programdata/sshd_config 파일 수정 필요 기타 파일 권한 정리 필요 로 인하여 scp 프로그램을 windows 환경의 intellij idea 에서 id, 비번치게끔 ide 콘솔에서 지원하지 않는거 같아. 그래서 제미니가 추천해준대로 실습 환경에서 scp 접속 단을 copy 로 그냥 퉁치는 것으로 바꿔놨어. 나처럼 windows 환경인 사람들이 있으면 혹시나 참고했으면 좋겠네. ObjectMapper, JavaTimeModule 같은 경우는 org.springframework.boot:spring-boot-starter-json 의존성 추가 필요
- java
- spring
- spring-boot
- spring-batch
- backend