새로운 로그를 위해 Filebeat 도입하기

Filebeat를 도입하기 전에 logstash.conf 파일의 input 섹션은은 다음과 같습니다.

input {
  file {
    mode => "read"
    path => "/usr/share/logstash/ingest_data/**/*.log"
    exclude => [ "/usr/share/logstash/ingest_data/logstash_completed.log" ]
    exit_after_read => true
    file_completed_action => "log"
    file_completed_log_path => "/usr/share/logstash/ingest_data/logstash_completed.log"
    start_position => "beginning" # 처음 파일을 읽을 때 어디부터 시작할지 지정
    sincedb_path => "/dev/null" # 오프셋 개념과 비슷하게 이전에 처리한 부분 기록 X
    codec => multiline {
      pattern => "^\[\d{4}-\d{2}-\d{2}.*$" # 멀티 라인 인식을 위한 정규표현식
      negate => true
      what => "previous"
    }
  }
}

 

저는 logstash 컨테이너가 지속적으로 `"/usr/share/logstash/ingest_data/**/*.log"` 경로를 확인하고, 새로운 파일이 생기면 자동으로 ES에 적재하기를 바랬기 때문에 다음과 같은 코드를 추가했습니다.

mode => "tail"
sincedb_path => "/usr/share/logstash/.sincedb"
discover_interval => 10  # 10초마다 새로운 파일 탐색

하지만 logstash는 처음 시작 이래로 아무런 액션이 없었고, 이를 해결하기 위해 안정성과 확장성이 높은 Filebeat를 도입하기로 했습니다.

 

Filebeat을 도입하면서 지금까지 작성한 logstash.conf 파일의 `input` 섹션과 `output` 섹션이 수정됐습니다.

input {
  beats {
    port => 5044  # Filebeat가 데이터를 전송할 포트
  }
}

filter {
  ...
}

output {
  if [application] == "hs" {
    elasticsearch {
      index => "hs-%{[@metadata][kst_date]}"  # KST 기준 날짜 사용
      hosts=> "${ELASTIC_HOSTS}"
      user=> "${ELASTIC_USER}"
      password=> "${ELASTIC_PASSWORD}"
      cacert=> "certs/ca/ca.crt"
    }
  } else if [application] == "lsi" {
    elasticsearch {
      index => "lsi-%{[@metadata][kst_date]}"
      hosts=> "${ELASTIC_HOSTS}"
      user=> "${ELASTIC_USER}"
      password=> "${ELASTIC_PASSWORD}"
      cacert=> "certs/ca/ca.crt"
    }
  } else if [application] == "tsI" {
    elasticsearch {
      index => "tsi-%{[@metadata][kst_date]}"
      hosts=> "${ELASTIC_HOSTS}"
      user=> "${ELASTIC_USER}"
      password=> "${ELASTIC_PASSWORD}"
      cacert=> "certs/ca/ca.crt"
    }
  }
}

그 이유는 더 이상 logstash가 로그 파일을 모니터링하고 수집하지 않기 때문입니다.

 

대신 다음과 같은 filebeat.yml 파일을 통해 3개의 경로(hs, lsi, tsi)를 30초 단위로 모니터링하면서 로그 데이터를 수집할 수 있습니다.

filebeat.inputs:
- type: filestream
  id: hs-logs
  paths:
    - /usr/share/filebeat/ingest_data/hs/*.log
  start_position: beginning
  fields:
    application: "hs"
  fields_under_root: true
  parsers:
    - multiline:
        type: pattern
        pattern: '^\[\d{4}-\d{2}-\d{2}.*$'
        negate: true
        match: after

- type: filestream
  id: lsi-logs
  paths:
    - /usr/share/filebeat/ingest_data/lsi/*.log
  start_position: beginning
  fields:
    application: "lsi"
  fields_under_root: true
  parsers:
    - multiline:
        type: pattern
        pattern: '^\[\d{4}-\d{2}-\d{2}.*$'
        negate: true
        match: after

- type: filestream
  id: tsI-logs
  paths:
    - /usr/share/filebeat/ingest_data/tsI/*.log
  start_position: beginning
  fields:
    application: "tsI"
  fields_under_root: true
  parsers:
    - multiline:
        type: pattern
        pattern: '^\[\d{4}-\d{2}-\d{2}.*$'
        negate: true
        match: after

output.logstash:
  hosts: ["logstash01:5044"]

logstash와 달리 로그 파일에 대한 오프셋 설정을 따로 해줄 필요가 없습니다.

 

이로써 `/elastic-stack/filebeat_ingest_data/` 경로에 들어오는 log 확장자를 가진 모든 파일은 자동으로 ES에 적재되고, 키바나에서 확인할 수 있습니다.

 

최종 ELK Stack 아키텍처는 다음과 같습니다.