[모플 프로젝트] Vector Loki Grafana로 로그 수집 파이프라인 구축하기 !!

 

로그 파이프라인 구축 도구 정하기

로그 수집 도구

`Log Stash` : 많은 레퍼런스와 안정성 보장, JVM 기반으로 리소스 사용량 높음

`Fluentd` : 리소스 사용량 높은 편

`Vector` : `Log` 뿐 아니라 `Metrics`, `Tracing` 지원

`Promtail` : `Loki` 와 통합이 잘 되지만, `Log` 만 지원하고 단순 처리만 가능

따라서 `Vector`

 

로그 저장 도구

`ES` : 리소스 사용량을 감당할 수 없음

`Loki` : 기존 `Grafana` 와 통합 쉬움

따라서 `Loki`

 

이렇게 프로젝트 팀원 분이 결정해 주셨다. 👍👍

(나도 앞으로 기술을 선택할 때, 현재 상황에 맞춰서 장단점을 보고 골라야겠다고 다짐했다.)

 

그럼 이제 `Loki` 로 로그를 저장하고, `Vector` 로 로그를 수집, 전송한 뒤, `Grafana` 로 시각화 하는 로그 파이프라인을 구축해야 한다.

 

Logger 설계

로그 레벨을 설계할 때,

정상 로그는 `info`, 주의 깊게 봐야 하거나 클라이언트 오류는 `warn`, 긴급한 오류나 서버 오류는 `error` 로 설정했다.

 

MDC

mdc 를 이용하여, `userInfo`, `requestId`, `feature` 를 저장했다.

  - `userInfo` : `JwtFilter` 에서 인가된 사용자의 id 를 넣었다.

  - `requestId` : 모든 요청을 처리하는 필터에서 `requestId` 를 생성하여 넣었다.

  - `feature` : 내부 API 응답을 처리하는 인터셉터에서 `@Tag` 애노테이션의 `name` 속성을 넣었다.

중요한 점은 에러가 발생하면 Spring 이 내부적으로 새로운 요청처럼 `/error` 로 포워딩 하는데,
이때 `mdc` 는 `ThreadRocal` 기반이라 `mdc` 안의 내용은 지워진다.
따라서 내부 API 응답을 처리하는 인터셉터에서 `request` 에 값들을 저장하여 `mdc`값을 복원하였다.

 

기본 로그 포맷

<encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
  <providers>
    <mdc>
      <includeMdcKeyName>feature</includeMdcKeyName>
      <includeMdcKeyName>requestId</includeMdcKeyName>
      <includeMdcKeyName>userInfo</includeMdcKeyName>
    </mdc>
    <timestamp>
      <fieldName>@timestamp</fieldName>
    </timestamp>
    <logLevel>
      <fieldName>level</fieldName>
    </logLevel>
    <pattern>
      <pattern>
        { "location": "%file:%line" }
      </pattern>
    </pattern>
    <pattern>
      <pattern>
        { "thread": "%thread" }
      </pattern>
    </pattern>
    <arguments/>
    <message />
  </providers>
</encoder>

`feature`, `requestId`, `userInfo`, `@timestamp`, `level`, `location`, `thread`, `message`, `arguments` 가 json 형식으로 저장된다.

(`json` 형식으로 저장하려면 `Logstash Logback Encoder` 를 `gradle` 의존관계 추가해야 한다.)

여기서 `arguments` 는 각각의 로거들에게 직접 넣은 값들이다. (ex `executionTime`, `errorMessage` 등)

 

Internal

  - 내부 API 요청을 인터셉터로 응답 시간 체크

  - 응답 성공, 응답 지연에는 `executionTime`, `success` 필드

  - 응답 실패에는 `errorMessage`, `success` 필드

  - 응답 성공 `info`, 응답 지연 `warn`, 응답 실패 `warn` (서버 에러, 클라이언트 에러 구분 X)

이때, 인터셉터에 `/error` 경로를 제외해야 로그가 2번 호출되지 않는다. 

 

External

  - 외부 API 요청에 `@ExternalApiLogging` 애노테이션을 붙이면, Aop 로 응답 시간을 체크

  - 응답 성공, 응답 지연에는 `executionTime`, `success` 필드

  - 응답 실패에는 `errorMessage`, `success` 필드

  - 응답 성공 `info` , 응답 지연 `warn`, 응답 실패 `warn` (서버 에러, 클라이언트 에러 구분 X)

 

Bisness

  - 중요 메서드에 `@BisnessLogicLogging` 애노테이션을 붙이면, Aop 로 응답 시간을 체크

  - 응답 성공, 응답 지연에는 `executionTime`, `success` 필드

  - 응답 실패에는 `errorMessage`, `success` 필드

  - 응답 성공 `info` , 응답 지연 `warn`, 응답 실패 `warn` (서버 에러, 클라이언트 에러 구분 X)

 

Exception

  - `@RestControllerAdvice` 에서 커스텀 된 예외들을 각각 클라이언트 에러, 서버 에러로 분류하여 로깅

  - 커스텀 되지 않은 예외들은 `DefaultErrorAttribute` 를 커스텀 구현하여 상태 코드를 기반으로 클라이언트 에러, 서버 에러로 분류하여 로깅

  - 클라이언트 에러에는 `status`, `errorMessage` 필드

  - 서버 에러에는 `status`, `errorMessage`, `stackTrace` 필드

  - 클라이언트 에러 `warn` , 서버 에러 `error`

 

SlowQuery

  - 커스텀 `DataSource` 의 프록시인 `LoggingDataSource` 로 `QueryExecutionListener` 를 활용하여 모든 SQL 쿼리의 실행 시간을 측정하고, 임계 시간보다 오래 걸리는 쿼리를 로깅

  - 슬로우 쿼리에는 `query`, `executionTime` 필드

  - 슬로우 쿼리 `warn`

  (`DataSoruce` 프록시를 사용하려면 `DataSource Proxy` 를 `gradle` 에 추가해야 한다.)

 

Logback-spring 설정

전체적인 로그 포맷은 동일하기 때문에 `Internal` 로거의 설정만 보면,

 

<!-- 내부 API 로거 -->
    <logger name="internal-api" level="INFO" additivity="false">
        <appender-ref ref="INTERNAL_API_FILE"/>
    </logger>

<!-- 내부 API 로그 파일 -->
    <appender name="INTERNAL_API_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>logs/internal-api.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>logs/internal-api.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <maxFileSize>100MB</maxFileSize>
            <maxHistory>30</maxHistory>
            <totalSizeCap>2GB</totalSizeCap>
        </rollingPolicy>

        <encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <mdc>
                    <includeMdcKeyName>feature</includeMdcKeyName>
                    <includeMdcKeyName>requestId</includeMdcKeyName>
                    <includeMdcKeyName>userInfo</includeMdcKeyName>
                </mdc>
                <timestamp>
                    <fieldName>@timestamp</fieldName>
                </timestamp>
                <logLevel>
                    <fieldName>level</fieldName>
                </logLevel>
                <pattern>
                    <pattern>
                        { "location": "%file:%line" }
                    </pattern>
                </pattern>
                <pattern>
                    <pattern>
                        { "thread": "%thread" }
                    </pattern>
                </pattern>
                <arguments/>
                <message />
            </providers>
        </encoder>
    </appender>

로거 설정

  : 이름 `internal-api` 를 설정하고, 로거 파일 이름 `INTERNAL_API_FILE` 을 설정

 

로거의 로그 파일

  : 로거 파일 이름 `INTERNAL_API_FILE` 로 로거를 매핑

  : 로그를 저장할 파일 `logs/internal-api.log` 를 설정

  : `SizeAndTimeBasedRollingPolicy` 로 사이즈, 시간 별로 파일을 `logs/internal-api.%d{yyyy-MM-dd}.%i.log` 분기

    (ex `logs/exception.2025-06-13.0.log`)

  : 하나의 로그 파일의 크기 `100MB`, 로그 보관일 `30일`, 로그 파일의 총 합 `2GB` 가 되면 오래된 로그 순으로 삭제

    (위에서 설치한  `Logstash Logback Encoder` 으로 로그를 `json` 형식으로 저장할 수 있게 된다.)

 

Vector 설정

# 내부 API 로그 수집
[sources.internal_api_logs]
type = "file"
include = ["/etc/vector/logs/internal-api*.log"]
read_from = "beginning"

`sources` 로 수집한 로그 이름을 `internal_api_logs` 라고 하고,

`file` 타입으로 `/etc/vector/logs/internal-api*.log` 위치에 저장한다.

그리고 파일 `처음부터` 읽어 기존 로그도 수집하게 한다.

 

# 내부 API 로그 파싱
[transforms.parse_internal_api_logs]
type = "remap"
inputs = ["internal_api_logs"]
source = '''
. = parse_json!(.message)

.requestId = to_string!(.requestId)
.userInfo = to_string!(.userInfo)
.feature = to_string!(.feature)
.level = to_string!(.level)

.log_type = "internal-api"
'''

`transforms` 로 `internal_api_logs` 를 인풋으로 받고, 받은 로그들을 변환하기 위해 타입을 `remap` 으로 설정한다.

(여기서 로그를 `json` 으로 수집했기 때문에 `grafana` 에서 라벨로 찾을 필드만 따로 파싱 하면 된다. 정규식으로 붙잡다 2일을 보낸 슬픈 사연이 있다....)

`.requestId`, `.userInfo`, `.feature`, `.level`, `.log_type` 필드로 값을 따로 저장한다.

 

# Loki로 전송
[sinks.loki_sink]
type = "loki"
inputs = [
    "parse_internal_api_logs",
]
endpoint = "http://loki:3100"
encoding.codec = "json"

# 라벨 설정
labels.log_type = "{{ log_type }}"
labels.feature = "{{ .feature }}"
labels.requestId = "{{ .requestId }}"
labels.level = "{{ .level }}"

이제 `sinks` 로 전송을 하는데 여기에 아까 파싱처리한 `parse_internal_api_logs` 를 인풋으로 받아 전송한다.

`http://loki:3100` 은 `loki` 로 전송된 로그들을 `grafana` 에서 연결할 `endpoint` 이다.

 

그리고 아까 따로 저장한 필드를 라벨로 설정하여 `grafana` 에서 해당 필드로 필터링하기 쉽도록 한다.

 

Loki 설정

auth_enabled: false

server:
  http_listen_port: 3100
  grpc_listen_port: 9096

우선 `loki` 에 접근할 때 인증을 사용하지 않도록 했다.

(로컬, dev 는 괜찮지만 prod 에서는 인증 설정을 고려해야 함)

그리고 HTTP API 요청을 받을 포트 `3100`, 내부 컴포넌트 간 `gRPC` 통신에 사용될 포트 `9096` 포트를 설정한다.

 

common:
  path_prefix: /loki
  storage:
    filesystem:
      chunks_directory: /loki/chunks
      rules_directory: /loki/rules
  replication_factor: 1
  ring:
    instance_addr: 127.0.0.1
    kvstore:
      store: inmemory

`loki` 가 사용할 기본 디렉토리 경로를 `/loki` 로 설정하고,

로그를 `chunk` 단위로 저장하는 경로 `/loki/chunks`,

`Alerting rule` 저장 경로 `/loki/ruels` (프로메테우스 룰 등 저장한다. 아직 어려워서 잘 모르겠다 ...)

`replication_factor` 는 복제본 개수인데 우선 `1`개로 설정하여 싱글 노드로 하였다.

`ring` 은 분산 환경에서 사용하는 토큰링으로, 우선은 `inmemory` 라 단일 노드 용도로 설정하였다.

 

query_range:
  results_cache:
    cache:
      embedded_cache:
        enabled: true
        max_size_mb: 100

쿼리 결과를 캐싱해서 속도를 향상하고, `embedded_chach` 내장 캐시를 사용하도록 했다.

(디스크 or 메모리 기반 아님)

그리고 최대 캐시 용량을 `100MB` 로 설정하였다.

 

schema_config:
  configs:
    - from: 2025-06-11
      store: boltdb-shipper
      object_store: filesystem
      schema: v11
      index:
        prefix: index_
        period: 24h

인덱스 저장 방식 설정이다.

이 설정을 적용할 시작 날짜를 기록하고, 인덱스를 `filesystem` 으로 저장하도록 했다. 

여기서 `boltdb-shipper` 는 로컬 디스크에 인덱스를 저장하며 공유 저장소로도 업로드하는 방식이다.

`loki` 에서 사용하는 인덱스 스키마 버전으로 `v11`, 

`index_` 파일 접두사, `24h` 단위로 인덱스 분리하도록 하였다.

 

storage_config:
  boltdb_shipper:
    active_index_directory: /loki/index
    cache_location: /loki/boltdb-cache
    shared_store: filesystem

  filesystem:
    directory: /loki/chunks

analytics:
  reporting_enabled: false

저장 설정으로,

현재 인덱스를 저장할 경로 `/loki/index` , 로컬 인덱스 캐시 저장 위치 `/loki/boltdb-cache` 를 지정해 준다.

그리고 데이터 공유 저장소 방식으로 `filesystem` (로컬 디스크) 를 선택하고,

`filesystem` 경로를 아까 로그 `chunk` 저장 경로와 동일하게 해 준다.

그리고 마지막으로 `grafana` 로 사용 통계 정보를 전송하지 않도록 했다.

 

Grafana 사용

라벨 기반 필터링

{log_type="internal-api", level="WARN", requestId="bb80cdfd"}

우선 아까 설정한 라벨로 `level`, `feature`, `log_type`, `requestId` 로 로그를 필터링할 수 있다.

 

전체 로그 중 특정 단어 포함

{log_type="exception"} |= `서버 오류`

`서버 오류` 가 포함된 `exception` 로그만 볼 수 있다.

(`|=` : 포함, `!=` 불포함)

 

정규 표현식 필터

{log_type="slow_query"} |~ `"executionTime":([3-9][0-9][0-9][0-9])`

`executionTime` 이 `3000` 이상인 `slow-query` 를 찾을 수도 있다.

 

로그 메시지 포맷 지정

{log_type="business_logic"} | json | line_format `{{ .timestamp }} [{{ .thread }}] {{ .location }} {{ .userInfo }} {{ .message }}`

`2025-06-14 01:36:11.077 [http-nio-8284-exec-6] BusinessLogicLogger.java:34 152 Business logic 응답 실패`

이런 식으로 로그가 아주 예쁘게 포맷되어 나온다 !