📦ELK 로그 관리

먼저 보면 좋은 로그 관련 포스팅

🚀 도입

로그는 오류 현상 확인과 CS 등을 위해 반드시 관리가 필요합니다.

만약, 운영중인 서비스에, 특정 서비스에 대한 오류 수정 CS 요청이 접수된다면, 바로 해결이 가능하다면 좋겠지만, 로그 분석을 통해 어떤 부분에서 문제가 발생했는지 특정하는 것이 중요합니다.

만약, 특정 API의 요청이 간헐적으로 실패한다고 가정하겠습니다.

웹서버로 앞단에 NginX 가 걸려 있는 상태이고, 클라이언트 측에서는 어떤 오류인지 특정할 수 없는 500 Code 가 떨어지고 있습니다.

가장 간편하게 확인 가능한 방법은, NginX 의 에러 로그와, SpringBoot 등 Application Server 의 로그를 확인하는 것이 가장 빠르게 확인 가능한 상황일 것 입니다.

만약 이때, App Server 가 새로 기동되어 그전의 로그를 기록해두지 않았다면 어떤 상황이 발생하게 될 까요?
로그를 확인할 수 없게 됩니다.

즉, 로그또한 영속성 적으로 관리하는 것이 필요합니다.

ELK 로 로그 관리의 중요성

로그를 저장하여 관리하게 된다면, 파일로 저장하여 (링크) 확인할 수 있습니다.

이 경우, 검색을 통해서 원하는 시간대의 자료를 검색하여 확인할 수 있습니다.

ELK 는, 대시 보드와 최적화된 검색 쿼리 등을 사용하여 빠르게 검색과 문제 특정이 가능하게 됩니다.

ELK 도입하기

Logback 설정 추가
ELK 설치
간단한 쿼리 소개
대시 보드 (visualization) 확인

사용되는 프로덕트
Spring Boot (3.0 ↑)
Elastic Stack (9.2.0)
Docker (docker-compose)

💥 elk 9.2 버젼을 사용하게 된 이유는, 해당 프로젝트에 RAG 시스템을 도입했기 때문에, Vector 관련 업데이트가 이뤄진 최신 버젼을 사용하였습니다.
낮은 버젼을 사용해도 전혀 상관없습니다.

프로세스 확인

전반적인 프로세스는 다음과 같이 진행됩니다.

Server(Spring) 에서 Logback 등의 로깅 프레임워크를 통해 로그를 생성
Logback 의 Appender 를 통해 Logstash 로 Log 전송
Logstash 에서, 로그 가공 후 ElasticSearch 에 저장
Kibana 에서, ElasticSearch 데이터를 읽어 시각화

Logback 설정

Log 관련 포스팅과 유사한 부분이 많습니다. (포스팅)
어느 경로에 xml 을 추가하고, appender 가 어떤 것인지 잘 모른다면 참고!

Logback 에는 다양한 Appender 가 있고, 이중에서는 Logstash 와 관련된 어펜더도 존재합니다.

이를 사용하기 위해 다음과 같이 의존성을 추가해 줍니다.

implementation 'net.logstash.logback:logstash-logback-encoder:9.0'

다음으로, 다음과 같은 xml을 추가해줍니다.

<configuration>
    <conversionRule conversionWord="clr" class="org.springframework.boot.logging.logback.ColorConverter"/>
    <conversionRule conversionWord="wex" class="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
    <conversionRule conversionWord="wEx" class="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
    <conversionRule conversionWord="esb" class="org.springframework.boot.logging.logback.EnclosedInSquareBracketsConverter" />

    <property name="LOG_FILE" value="logs/application.log"/>
    <property name="CONSOLE_LOG_PATTERN" value="${CONSOLE_LOG_PATTERN:-%clr(%d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd'T'HH:mm:ss.SSSXXX}}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}){} %clr(${PID:-}){magenta} %clr(--- %esb(){APPLICATION_NAME}%esb{APPLICATION_GROUP}[%15.15t] ${LOG_CORRELATION_PATTERN:-}){faint}%clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
    <property name="CONSOLE_LOG_CHARSET" value="${CONSOLE_LOG_CHARSET:-${file.encoding:-UTF-8}}"/>

    <!-- 신규 추가된 부분 -->
    <!-- Logstash로 전송할 Appender -->
    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>localhost:5044</destination>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <timeZone>Asia/Seoul</timeZone>
        </encoder>
    </appender>

    <!-- 콘솔 출력 -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
            <charset>${CONSOLE_LOG_CHARSET}</charset>
        </encoder>
    </appender>

    <!-- 파일 출력 -->
    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_FILE}</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>logs/application.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss} %-5level [%thread] %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <!-- Logger 설정 -->
    <logger name="baekgwa.blogserver" level="DEBUG" />
    <root level="INFO">
        <appender-ref ref="CONSOLE" />
        <appender-ref ref="FILE" />
        <appender-ref ref="LOGSTASH" />
    </root>
</configuration>

destination : 연결될 Logstash 주소
encoder : 데이터를 전송할 때 사용될 encoder
timezone : 데이터의 날짜 형식을 지정

ELK 설치

Docker-compose 로 ELK 를 설치합니다.

elk-docker-compose.yml

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:9.2.0
    container_name: elasticsearch
    networks:
      - baekgwa-network
    ports:
      - "9200:9200"
    volumes:
      - es-data:/usr/share/elasticsearch/data
    environment:
      - discovery.type=single-node
      - ES_JAVA_OPTS=-Xms1g -Xmx1g
      - xpack.security.enabled=false
      - xpack.security.http.ssl.enabled=false
    restart: always

  kibana:
    image: docker.elastic.co/kibana/kibana:9.2.0
    container_name: kibana
    networks:
      - baekgwa-network
    ports:
      - "5601:5601"
    volumes:
      - kibana-data:/usr/share/kibana/data
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    restart: always
    depends_on:
      - elasticsearch

  logstash:
    image: docker.elastic.co/logstash/logstash:9.2.0
    container_name: logstash
    networks:
      - baekgwa-network
    ports:
      - "5044:5044" #Logback TCP 입력 포트
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline:ro
    environment:
      - LS_JAVA_OPTS=-Xms1g -Xmx1g
      - ELASTIC_PASSWORD=${ELASTIC_PASSWORD}
    restart: always
    depends_on:
      - elasticsearch

volumes:
  es-data:
    driver: local
  kibana-data:
    driver: local

networks:
  baekgwa-network:
    external: true

인증 처리부분은 false 로 두어, 별도의 로그인 없이 사용가능 하도록 구성하였습니다.
운영중인 환경이라 접근 제한이 필요할 경우, 공식문서 등을 참고해 구성하시기 바랍니다.

다음으로 Logstash 에서 데이터를 받아, ElasticSearch 로 전송할 때, 데이터를 가공할 수 있는 config 파일을 작성해줍니다.

/logstash/pipeline/logstash.conf

input {
    tcp {
        port => 5044
        codec => json_lines
    }
}

filter {
    ruby {
        code => "event.set('[@metadata][index_suffix]', event.timestamp.time.localtime('+09:00').strftime('%Y.%m.%d'))"
    }
}

output {
    elasticsearch {
        hosts => ["http://elasticsearch:9200"]
        index => "application-logs-%{[@metadata][index_suffix]}"
    }
}

input 블록은 Logstash가 데이터를 수집하는 “입력(source)” 부분을 정의하는 구문입니다. 즉, Logstash가 어디서, 어떤 방식으로, 어떤 포맷으로 데이터를 받을지 설정하는 영역입니다.

1. tcp { ... }
입력 플러그인(plugin) 중 하나로, Logstash가 TCP 프로토콜을 통해 데이터를 수신하도록 설정합니다.
예를 들어, 애플리케이션이 로그를 TCP 소켓으로 전송하면 Logstash가 이 포트를 통해 수신합니다.
2. port => 5044
Logstash가 수신 대기(listen) 하는 포트 번호입니다.
즉, 클라이언트(예: Filebeat, 애플리케이션 로그 전송기 등)는 5044 포트로 데이터를 보냅니다.
Logstash는 해당 포트에서 TCP 연결을 열고 로그를 읽습니다.
3. codec => json_lines
수신한 데이터의 포맷(형식) 을 정의합니다.
json_lines 코덱은 각 줄(line)이 하나의 JSON 객체로 구성된 로그를 의미합니다.

다음으로 docker-compose 를 up 해서, 컨테이너가 정상적으로 기동되는지 확인 합니다.

Spring 실행 후, 로그 전송 확인

스프링부트를 실행 해서, 로그가 정상적으로 전송되는지 확인합니다.

현재, logback.xml 에서, 로그 레벨을 INFO 레벨로 지정해놓았기 때문에, Spring 에서 발생하는 로그는 INFO 까지 나오게 됩니다.

기동되었다면, http://localhost:9200/_cat/indices?v 에 접속해, 생성된 인덱스를 확인할 수 있습니다.

현재 일자 기준으로 application-logs-YYYY.mm.DD 이 생성되었다면 정상적으로 생성된 것 입니다.

왜? health 가 yellow 인가요? green 이 아니라?
기본적으로 Elasticsearch의 인덱스는 내부적으로 샤드(shard) 라는 단위로 나뉩니다.
인덱스 1개당 총 2개의 shard(1 primary + 1 replica) 가 필요합니다.
하지만 현재는, SingleNode 로 구성되어 있어, 정상적이지만, replica shard 에 미 할당 된 상태라 그렇습니다.

다음으로, Kibana 에 접속해서, 정상적으로 데이터가 들어왔는지 확인해 보겠습니다.

http://localhost:5601/ 에 접속합니다.
http://localhost:5601/app/discover 에 접속해서, Dataview 를 생성해 줍니다.

Create a data View

이전 logstash.conf 파일로, output 블럭에, 다음과 같이 application-logs-날자 로 인덱스를 생성하게 했으니, index pattern 에 다음과 같이 작성해 매칭 시켜 줍니다.

이후, 타임라인을 조정하며, 로그들을 확인합니다.

좌측에 로그들에 사용 가능한 fileds 들이 확인 됩니다.

간단한 쿼리 소개

ELK 는 기본적으로 강력한 검색 기능을 제공하는 솔루션 입니다.

그에 따라 제공되는 다양한 쿼리들을 제공합니다.

🔎 KQL 기본 문법 요약
종류
예시
설명
기본 매칭
status:200
status 필드가 200인 문서 검색
부분 일치
message:*error*
message 필드에 "error"가 포함된 문서 검색
정확한 문자열
user.keyword:"홍길동"
공백이나 대소문자 포함한 정확한 일치 검색
여러 값 중 하나
status:(200 or 404)
상태 코드가 200 또는 404인 문서
부정 (NOT)
NOT status:200 또는 status != 200
200이 아닌 문서
비교 연산자
bytes > 1000 / response_time <= 500
숫자/날짜 비교 가능
날짜 범위
@timestamp >= "now-1d/d"
최근 하루치 로그
문자열 포함 여부
message : "login"
문자열 포함 검색 (*login* 과 동일)
필드 존재 여부
_exists_:user
user 필드가 존재하는 문서
필드 없음
NOT _exists_:user
user 필드가 없는 문서

🧭 요약 정리
구분
연산자
예시
일치
:
status:200
부정
!=, NOT
status != 404
논리
AND, OR, ()
(status:500 OR status:404) AND method:GET
범위
>, <, >=, <=, [TO]
bytes:[1000 TO 10000]
포함
*
message:*fail*
존재 여부
_exists_:
_exists_:user