[ELK] 파일비트(Filebeat)와 로그스태시(Logstash)에 대한 설정 예시

아래의 내용에서는 ELK(Elasticsearch, Logstash, Kibana) 환경 구성에서 파일비트(Filebeat)와 로그스태시(Logstash)에 대한 설정 예시를 다룹니다. 개인적으로 나중에 참고하려고 작성한 내용입니다. 혹시 파일 비트와 로그스태시에 대한 개념이 부족하다면 다음 포스팅을 참고해주세요. 

 

 

 

1. 파일비트(Filebeat)를 이용한 카프카(Kafka)로의 로그 전송 예시

---

Filebeat는 로그를 전송해주는 경량화된 프로세스이다. 현재 서버 구성으로는 Nginx와 Tomcat 로그가 모두 있어서 두 가지를 같이 남기고 있다. 파일비트의 yml 파일은 다음과 같은데, 각각 살펴보도록 하자.

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /home/mangkyu/logs/tomcat/mangkyu*.log
    fields:
      topic: works_mangkyu
      region: "kr1"
      project: "mangkyu"
      log_type: "tomcat"

    multiline.pattern: ^\[[0-9a-fA-F]{32}\][[:space:]][0-9]{4}-[0-9]{2}-[0-9]{2}[[:space:]][0-9]{2}:[0-9]{2}:[0-9]{2}[[:space:]]\[
    multiline.negate: true
    multiline.match: after

  - type: log
    enabled: true
    paths:
      - /home/mangkyu/logs/nginx/access_ssl_*.log
    fields:
      topic: works_mangkyu
      region: "kr1"
      project: "mangkyu"
      log_type: "nginx"

    multiline.pattern: ^[0-9a-fA-F]{32}[[:space:]]
    multiline.negate: true
    multiline.match: after

output.kafka:
  enabled: true
  hosts: ["kafka-mangkyu.mangkyu.com:9092"]
  topic: '%{[fields.topic]}'

 

 

 

 

[ 입력(Input) ]

먼저 파일로 작성된 로그를 입력으로 받아서 전송하고 있다. 그래서 로그 경로(paths)와 추가 정보를 작성해주었다. 그 다음으로는 fields인데, 이 부분은 임의로 추가할 수 있는 영역이다. 위의 예시에는 4가지 항목을 추가해주었고, 각각의 용도는 다음과 같다.

• topic
  • 카프카 메세지로 사용될 값
  • 해당 토픽을 통해 카프카 메세지를 컨슘하기 위함
• region
  • 해당 서버가 존재하는 지역(kr0, kr1, jp1, jp2) 등을 나타내는 값
  • 어느 지역에서 처리되는 메세지인지를 구분하기 위함
  • (엘라스틱 서버가 크게 KR(kr0, kr1), JP(jp1, jp2)으로 구성되어 내부적으로 인덱스 패턴을 구분하기 위함)
• project
  • 전송되는 로그가 어느 프로젝트의 로그인지 나타내는 값
  • (엘라스틱 서버가 크게 KR(kr0, kr1), JP(jp1, jp2)으로 구성되어 내부적으로 인덱스 패턴을 구분하기 위함)
• log_type
  • 전송되는 로그가 어느 로그인지(tomcat, nginx …) 구분하기 위함
  • 동일한 로그스태시(Logstash) 프로세스에서 Nginx와 Tomcat을 모두 처리하기 위함

 

 

중요한 부분은 Multiline 설정 부분이다. 파일 비트는 기본적으로 한 줄을 하나의 로그로 인식하는데, Multiline 설정을 적용하면 특정 패턴에 맞지 않는 로그는 하나의 로그로 인식하도록 이어붙일 수 있다. 파일 비트에서 처리할 톰캣 로그 패턴은 다음과 같다. 참고로 톰캣 로그 패턴은 Log4J2를 기반으로 하는 패턴이다.

• 톰캣 로그:
  • 패턴: "[%equals{%X{works_rid}}{}{startup}] %d{yyyy-MM-dd HH:mm:ss} [%-5p](%-35c{1}:%-3L) %m%n"
  • 예시: [42744e73aebe5366cf627e8cf71c3da6] 2022-12-29 17:31:43 [DEBUG](GetProductService :118) 존재하지 않는 Product 입니다.

 

 

위와 같이 Multiline을 설정하면 패턴에 맞는 부분을 새로운 line의 시작으로 하여 매칭되지 않는 line은 append를 한다. 만약 신규로 작성한 부분이 매칭되는지를 확인하려면 아래의 코드를 참고해서 실행하면 된다. 다른 곳에서 참고한 기반 코드를 수정한 것으로, Go 언어로 되어있다. 웹에서 실행기를 찾아서 실행하면 된다. 추가적으로 수정할 부분이 있다면 공식 문서를 통해 살펴보도록 하자.

• 코드: https://gist.github.com/MangKyu/8205f167f811249388ce42ea402afefd
• 실행기: https://go.dev/play/p/gej9gMxHwWb

 

 

참고로 Nginx에서 남기는 로그 포맷은 다음과 같다.

log_format  main  '$request_id $remote_addr - $remote_user [$time_local] "$request" '
                   '$status $body_bytes_sent "$http_referer" "$request_time" '
                   '"$http_user_agent" "$http_x_forwarded_for" '
                   '"$ssl_protocol/$ssl_cipher" "$content_length" '
                   '"$request_length" ' ;

// 13ceadb76500f7c1cca856bb1022bcf4 127.0.0.1 - - [27/Dec/2022:15:53:42 +0900] "GET /kr/favicon.ico?20221226 HTTP/2.0" 200 5430 "http://localhost:8080/test" "0.008" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36" "-" "TLSv1.2/ECDHE-RSA-AES128-GCM-SHA256" "-" "148"

 

 

 

 

 

[ 출력(Output) ]

카프카는 로그의 유실을 최대한 방지하고자 사용되었다. topic으로는 위에서 설정한 값을 그대로 사용하도록 하였다. 이 외에 혹시나 Kafka가 아닌 다른 모듈로 설정이 필요하다면 공식 문서를 통해 참고하도록 하자.

 

 

 

 

 

 

2. 톰캣(Tomcat)과 엔진엑스(Nginx) 로그에 대한 Logstash 파이프라인 예시

---

Logstash는 데이터 처리를 위한 파이프라인으로, 카프카로부터 컨슘해온 데이터를 처리한다. 단계는 크게 입력, 필터, 출력으로 구성되는데 각각에 대해 살펴보도록 하자. 코드는 Gist에 업로드되어 있다.

1. 카프카로부터 json 형태로 메세지를 consume함
2. 필터링 과정을 진행함
   1. 먼저 l7check 요청은 drop 시킴
   2. 로그 타입에 맞게 처리함
3. 엘라스틱서치로 로그를 전송함

input {
  kafka {
    bootstrap_servers => "kafka-mangkyu.mangkyu.com:9092"
    topics => ["works_mangkyu"]
    group_id => "works-mangkyu-log"
    client_id => "mangkyu-logstash"
    decorate_events => true
    consumer_threads => 10
    codec => json
  }
}

filter {
  if [message] =~ "l7check" or [message] =~ "jquery" or [message] =~ ".js?"{
      drop {}
  }

  if [fields][log_type] == "nginx" {
    grok {
      match => { "message" => '%{NOTSPACE:requestId} %{IP:ip} - - \[%{HTTPDATE:requestDate}\] "(?:%{WORD:httpMethod} %{NOTSPACE:requestURL}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawRequest})" %{NUMBER:httpStatus} %{NUMBER:bodyBytes} "%{DATA:referer}" "%{BASE16FLOAT:responseTime}" "%{DATA:userAgent}" "%{DATA:forwardedFor}" "%{DATA:tlsVersion}" "%{DATA:contentLength}" "%{GREEDYDATA:etc}"'}
    }

    date {
          match => [ "[requestDate]", "dd/MMM/yyyy:HH:mm:ss +0900" ]
          timezone => "Asia/Seoul"
          target => "@timestamp"
    }

    ruby {
       code => "event.set('date', event.timestamp.time.localtime.strftime('%Y.%m.%d'))"
    }
  }


  if [fields][log_type] == "tomcat" {
    grok {
      match => { "message" => '\[%{NOTSPACE:requestId}\] %{TIMESTAMP_ISO8601:requestDate} \[%{LOGLEVEL:log_level}%{SPACE}*\]'}
    }

    date {
          match => [ "[requestDate]", "yyyy-MM-dd HH:mm:ss.SSS" ]
          timezone => "Asia/Seoul"
          target => "@timestamp"
    }

    ruby {
      code => "event.set('date', event.timestamp.time.localtime.strftime('%Y.%m.%d'))"
    }
  }
}

output {
  elasticsearch {
        hosts => ["http://elastic-mangkyu.com:10200"]
        index => "%{[fields][log_type]}-%{[fields][project]}-%{[fields][region]}-%{date}"
        user => "mangkyu"
        password => "naver!23"
  }
}

 

 

 

 

[ 입력(Input) ]

Input 플러그인에는 로그를 받아올 정보가 필요하다. 앞에서 Filebeat를 이용해 Kafka로 로그를 전송했기 때문에, Kafka로부터 메세지를 컨슘해오도록 작성하였다. 또한 전달받은 메세지를 json으로 파싱해주도록 codec을 사용하였다. 그 외에 자세한 부분은 Kafka 입력 플러그인 문서를 살펴보도록 하자.

 

 

 

[ 필터(Filter) ]

Filter 플러그인에서는 처리를 수행하며, 이벤트 특성에 따라 조건부로 적용되는 경우가 많다. 예를 들어 입력 로그에는 불필요한 부분도 있을 수 있다. 그래서 이러한 부분을 처리하고 있는데, 이를 순서대로 정리하면 다음과 같다.

1. 불필요한 메세지는 버림
2. 로그 메세지를 바탕으로 내용을 파싱함
3. 위에서 파싱된 requestDate 해당하는 값을 뽑아서 @timestamp로 설정함(https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html#plugins-filters-date-match)
4. 현재 날짜를 기준으로 date 필드를 세팅함 ex) 2023.01.02

 

 

 

[ 출력(Output) ]

Output 플러그인에서는 엘라스틱 서치 서버로 파싱된 결과를 전송해준다. 중요한 부분은 index 부분으로 프로젝트와 지역 별로 인덱스 패턴을 구분할 수 있도록 처리하였다. 위와 같이 만들어지는 index 예시는 다음과 같다.

• nginx-mangkyu-jp1-2023.01.05
• tomcat-mangkyu-jp1-2023.01.05

 

 

 

 

 

위의 내용은 개인적으로 나중에 참고하려고 작성한 내용입니다. 혹시 파일 비트와 로그스태시에 대한 개념이 부족하다면 다음 포스팅을 참고해주세요. 추가로 혹시 개선할 부분이 있으면 댓글 남겨주세요! 감사합니다:)