테스트 결과를 ELK 사용하여 기록 남기기(Logstash Json parsing)

개요 목적

무식한 영어 프로젝트는 영어 식 통 문장 암기를 도와주는 서비스이다. 테스트 페이지를 통해서 번역본을 보고 영어식 표현을 만들 수 있는 지 확인 가능하다.

테스트 페이지 버튼(맞음 틀림 힘트 정답)을 누르면 해당 유저와 해당 문장의 테스트 결과를 ElasticSearch로 모아서 유의미한 데이터를 수집하려 한다.

각 버튼을 눌렀을 때 테스트 결과 데이터를 어떻게 ElasticSearch에 저장을 하고 Kibana 검색을 통해서 유의미한 정보를 화면에 출력할 수 있는 지 알아보자.

ElasticSearch로 Json 직렬화 하여 보낼 데이터는 아래와 같다.

public class TestResultDto {

    private Long userId;
    private Long sentenceId;
    private String check;
    private LocalTime testTime;
}

public enum TestResultEnum {
    HINT("HINT"),
    CORRECT("CORRECT"),
    WRONG("WRONG"),
    TEST_TIME("TEST_TIME");
    private final String stringTestResult;
}

SpringBoot - Logstash 연결 및 설정

[리눅스 인프라/ElasticStack] - ELK 통해서 Spring Server Log 모으기, ELK 기본 설정

기본적인 ELK 설치 방법과 SpringBoot - Logstash 연결 설정 밥법은 전에 작성한 블로그 내용과 일치한다.

유일하게 다른 점은 Logstash에서 Elasticsearch로 데이터를 전달할 때의 설정 값이다.

/etc/logstash/conf.d/spring.conf 값을 아래로 설정하여, message 필드 안의 데이터 중에 testTime이라는 데이터가 존재할 때만 elasticsearch로 전달할 수 있게 설정했다.

input {
   tcp {
   port => 4560
   codec => json
   type => logstash
   }
}

output {
    if [message]=~ "testTime" {
        elasticsearch {
            hosts => "10.0.2.7:9200"
            index => "%{type}-%{+YYYY.MM.DD}"
        }
        stdout {}
    }
}

테스트 결과 데이터 전달 기능 구현하기

정답, 힌트, 틀림 데이터 전달하기

해당 유저 해당 문장 테스트 결과 중 정답, 힌트, 틀림 데이터를 로그로 남겨 → Logstash → ElasticSearch에 데이터 저장하는 로직은 동일하기 때문에 한번에 설명을 한다.

• 먼저 각 버튼을 클릭하면 해당 컨트롤러에 요청을 보낸다.

@RestController
@RequestMapping("/api/test/result")
@RequiredArgsConstructor
@Slf4j
public class TestResultController {

    private final TestResultService testResultService;

    @PostMapping("/correct")
    public ResponseEntity<Void> correctResult(@RequestBody TestResultRequest request)
        throws JsonProcessingException {
        testResultService.correctMessage(request);
        return new ResponseEntity<>(HttpStatus.OK);
    }
}

• 그리고 해당 컨트롤러에 있는 TestResultService가 작동하여 테스트 결과 정보를 담은 데이터를 json으로 직렬화하여 로그를 발생시켜 Logstash에게 전달한다.

@Service
@RequiredArgsConstructor
@Slf4j
public class TestResultService {

    private final RedisTemplate<String,String> redisTemplate;
    private final ObjectMapper objectMapper;

    public void correctMessage(TestResultRequest request) throws JsonProcessingException {
        TestResultDto testResult = new TestResultDto(request.getUserId(), request.getSentenceId(),
            TestResultEnum.CORRECT.getStringTestResult(), null);
        String serializedResult = objectMapper.writeValueAsString(testResult);
        log.info(serializedResult);
    }
}

문제 푼 시간 계산하여 데이터 전달하기

테스트 페이지에서 맞음 버튼을 클릭한 문장이 있을 때 해당 문장을 풀이 시간을 계산하여 ElasticSearch로 해당 정보를 보내는 방법에 대해서 알아본다.

• 테스트 페이지의 1번 문제는 테스트 입장할 때, 2번 문제부터는 이전 문제의 정답이나 버튼 클릭할 때 해당 컨트롤러로 요청을 보낸다. 해당 컨트롤러 내 서비스 코드에서 Redis에 키와 userId sentenceId를 조합한 key와 시작 시간을 담은 value를 redis에 저장한다. 해당 키 만료 시간은 5분으로 설정한다.

@PostMapping("/time/start")
    public ResponseEntity<Void> startResult(@RequestBody TestResultRequest request) {
        testResultService.sendStartTimeToRedis(request);
        return new ResponseEntity<>(HttpStatus.OK);
    }

public void sendStartTimeToRedis(TestResultRequest request) {
        String redisKey = "U" + request.getUserId() + "S"
            + request.getSentenceId();
        String nowDateTime = LocalDateTime.now()
            .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        redisTemplate.opsForValue().set(redisKey, nowDateTime);
        redisTemplate.expire(redisKey, 300, TimeUnit.SECONDS);
    }

• 문제 정답 버튼 클릭했다면, Redis에서 값을 가져와 현재 시간과 값 차이를 비교해서 그 값을 dto에 담아 Json 직렬화하고 로그로 출력하여 Logstash에 전달한다.

@PostMapping("/time/end")
    public ResponseEntity<Void> endResult(@RequestBody TestResultRequest request)
        throws JsonProcessingException {
        testResultService.TestTimeMessage(request);
        return new ResponseEntity<>(HttpStatus.OK);
    }

@Service
@RequiredArgsConstructor
@Slf4j
public class TestResultService {

    private final RedisTemplate<String,String> redisTemplate;
    private final ObjectMapper objectMapper;

    public void TestTimeMessage(TestResultRequest request) throws JsonProcessingException {
        LocalTime testTime = getTestTime(request);
        TestResultDto testResult = new TestResultDto(request.getUserId(), request.getSentenceId(),
            TestResultEnum.TEST_TIME.getStringTestResult(), testTime);
        String serializedResult = objectMapper.writeValueAsString(testResult);
        log.info(serializedResult);
    }

    private LocalTime getTestTime(TestResultRequest request) {
        String redisKey = "U" + request.getUserId() + "S"
            + request.getSentenceId();
        String getDateTime = redisTemplate.opsForValue().get(redisKey);
        assert getDateTime != null;
        LocalDateTime parseLocalDateTime = LocalDateTime.parse(getDateTime,
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Duration diff = Duration.between(parseLocalDateTime, LocalDateTime.now());

        LocalTime testTime = LocalTime.of(diff.toHoursPart(), diff.toMinutesPart(),
            diff.toSecondsPart(), diff.getNano());
        return testTime;
    }
}

ElasticSearch로 들어온 데이터 확인하기(중첩된 json 파싱하기)

해당 로직으로 ElasticSearch에 저장된 데이터를 확인해보면 테스트 결과가 message 필드 안에 중첩된 json 형시으로 저장되어 있는 것을 확인할 수 있다.

루트 필드가 아닌 message 내부에 정보들이 들어 있기 때문에 kibana에서 복잡한 검색 처리나 시각화하기가 어렵다.

그래서 logstash에서 message 안에 들어 있는 테스트 결과 json 정보를 root filed로 파싱한 후 ElasticSearch에 전송하도록 설정 방법을 알아보자.

Logstash 설정 값에 json filter plugin 사용하기

기존 /etc/logstash/conf.d/spring.conf에 아래 필터를 추가한다.

그러면 message 안에 있는 json 값들이 root filed로 설정된다.

filter {
 json {
   source => "message"
 }
}

파싱된 json 데이터 필드로 다양한 검색 조건 출력 가능

테스트 결과 정보가 필드로 올라가서 보다 쉽게 kibana 검색 조건에 맞는 데이터를 가져올 수 있다.

예시를 통해 편리해진 필드 검색 방법을 알아보자.

• 문제 풀이 시간이 일정 시간 이상인 데이터만 출력하기

• 특정 유저가 틀린 문제들 정보만 출력하기

---

Reference

https://www.elastic.co/guide/en/kibana/master/lucene-query.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/nested.html