springboot log pipeline: EFK 로그 전처리와 opensearch

이전 포스트에서는 k8s 에 fluent-bit, data-prepper, opensearch cluster와 opensearch dashboard를 배포하였다.

https://fullmooney.tistory.com/63

springboot log pipeline: EFK setup (opensearch)

 

이번에는 data-prepper로 logback의 log pattern에 매핑되는 전처리를 통해, opensearch에서 사용될 @timestamp를 포함한 필드를 만들어보자

 

helm으로 data prepper 가 배포된 네임스페이스에 보면 data prepper configmap이 생성되어 있는데 data-prepper-config.yaml 과 pipelines.yaml 을 확인할 수 있다. pipelines.yaml을 수정하고, data-prepper 파드를 재시작하여 설정을 적용해보자.

 

fluent-bit의 OUTPUT 에서 data-prepper 서비스를 지정하였고, URI 를 /log/ingest를 설정하였었다.

[OUTPUT]

      Name http

      Match *

      Host data-prepper-headless

      Port 21890

      URI /log/ingest

      Format json

/log/ingest를 받아 전처리 할 수 있는 log-pipeline을 만든다.

kind: ConfigMap
apiVersion: v1
metadata:
  name: data-prepper-config
data:
  data-prepper-config.yaml: |
    ssl: false
    serverPort: 4900
    peer_forwarder:
      discovery_mode: dns
      domain_name: "data-prepper-headless"
  pipelines.yaml: |
    log-pipeline:
      source:
        http:
          path: "/log/ingest"
          port: 21890
      sink:
        - stdout:

 

log-pipeline은 크게 source, processor, sink로 되어있는데, 데이터소스->전처리->목적지로 생각하면된다.

 

source.http.path와 source.http.port는 fluent-bit OUTPUT과 매칭되도록 설정하고, sink를 -stdout: 만 추가해 보고 배포를 해보면 콘솔로그에 전송값들을 확인할 수 있다. 로그가 안올라온다면 application에서 Rest API호출등 액션을 해보고 다시 로그를 확인해본다.

 

여기까지 이상이 없다면 우선 fluent-bit에서 데이터는 잘 받아오는 것이므로 processor와 sink.opensearch 를 추가해보자.

 

processor는 전처리를 담당하는데  "\\[%{TIMESTAMP_ISO8601:timestamp}:%{INT:relative}\\]\\[%{DATA:thread}\\] %{LOGLEVEL:level} %{DATA:logger} \\[%{INT:L}]-\\[%{DATA:SPRING_APP_NAME},%{DATA:traceId}\\]-%{GREEDYDATA:msg}" 는 logback-spring.xml의 로그파일패턴에 따라 다시 field로 쪼개는 것이다.

나같은 경우 아래 패턴으로 작성했다.

<property name="LOGS_FILE_PATTERN"    value="[%d{yyyy-MM-dd HH:mm:ss}:%-3relative][%thread] %-5level %logger{35} [%L]-[${SPRING_APP_NAME:-},%X{traceId:-}]-%msg%n" />

 

timestamp, relative, thread, level, logger, L, SPINRG_APP_NAME, traceId, msg라는 text 형식의 field가 opensearch 인덱스에 추가된다.

 

시간정보가 지정이 되어 있지않으면 opensearch에서 로그 순서가 뒤죽박죽으로 나오고, discover상단에서 차트나 검색조건 또한 보이지 않으므로 text형식의 timestamp를 다시 @timestamp라는 date형식의 필드에 복제 및 형변환해보자.

 

서버시간에 따라 source와 destination의 timezone을 맞춰주고, output format을 지정해서 @timestamp에 담아준다.

 

sink는 여러개를 해도 상관이 없는데, 추가로 opensearch를 연결해준다. host와 index, index_type, bulk_size를 입력해준다. index에 입력한 값이 opensearch에 자동으로 추가될 index 명이다.

 

최종적으로 변경된 configmap은 다음과 같다.

kind: ConfigMap
apiVersion: v1
metadata:
  name: data-prepper-config
data:
  data-prepper-config.yaml: |
    ssl: false
    serverPort: 4900
    peer_forwarder:
      discovery_mode: dns
      domain_name: "data-prepper-headless"
  pipelines.yaml: |
    log-pipeline:
      source:
        http:
          path: "/log/ingest"
          port: 21890
      processor:
        - grok:
            match:
              log:
                - "\\[%{TIMESTAMP_ISO8601:timestamp}:%{INT:relative}\\]\\[%{DATA:thread}\\] %{LOGLEVEL:level} %{DATA:logger} \\[%{INT:L}]-\\[%{DATA:SPRING_APP_NAME},%{DATA:traceId}\\]-%{GREEDYDATA:msg}"
        - date:
            match:
              - key: timestamp
                patterns: ["yyyy-MM-dd HH:mm:ss"]
            destination: "@timestamp"
            output_format: "yyyy-MM-dd HH:mm:ss"
            source_timezone: "UTC"
            destination_timezone: "Asia/Seoul"
            locale: "ko_KR"
      sink:
        - opensearch:
            hosts: ["http://opensearch:9200"]
            index: app-parse
            index_type: custom
            bulk_size: 200
        - stdout:

 

이번에는 opensearch로 이동한다. 로그가 들어오긴 전  우선 @timestamp를 date 타입으로 업데이트 하기 위해, app-parse라는 인덱스를 생성하고, Management > Dev Tools메뉴로 이동하여 @timestamp필드를 업데이트 한다. 

PUT /app-parse{  # data-prepper에서 지정한 INDEX명
  "mappings": {
    "properties": {
      "@timestamp": {# 대상필드
        "type": "date", #데이터 타입
        "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis“
      }
    }
  }
}

 

이제 data-prepper의 파드를 재시작하여, configmap의 설정을  다시 적용하고, 로그를 발생시켜보자.

app-parse인덱스에 @timestamp라는 하나의 필드만 있었는데, index management로 이동하여 확인하니 grok 에서 쪼갰던 필드들이 생성되어 있음을 확인할 수 있다. (이부분이 어려운 부분이다. timestamp의 타임존이 잘못 설정된 경우 필드 추가가 안되거나 할 수 있다. 그럴땐 깔끔하게 인덱스를 삭제하고 timezone을 다시 맞춰보는것이 이래저래 맘이 편했다)

이제 index pattern을 생성하고, discover를 통해 로그를 조회해보자.

 

Management > Dashboards Management > index patterns로 이동한다.

create index pattern 버튼을 누른다.

다음과 같이 app-* 라는 패턴을 입력하고 next를 누른다.

date필드에 @timestamp를 선택해주고 Create index pattern 버튼을 클릭하여 생성완료한다.

discover로 이동하여 app-*패턴으로 로그를 검색해보자.

좌측 필드 선택창에서 timestamp, SPRING_APP_NAME, traceId, msg만 선택하고 로그를 조회해본다.