[ELK Stack] logstash.conf, filebeat.yml

(ㄱ) logstash.conf

※ logstash.conf는 파이프라인 구성하는 파일입니다. 이를 통해 데이터 수신, 전처리, 출력 방식을 지정할 수 있습니다.

 

(1) input : 데이터 수신
- beats {} : filebeat와 같은 에이전트로부터 데이터 수신하기 위해 사용하는 플러그인
- port (Default 5044) : 데이터 수신 포트
- host : 네트워크 인터페이스

즉, Beats 에이전트로부터 포트 5044를 통해 데이터를 수신합니다.

 

(2) filter : 데이터 전처리

- date {} : 로그 메시지에서 날짜를 추출하여 Logstash의 @timestamp 필드 설정하는 필터

- match => [ "timestamp", "ISO8601" ] : 로그 이벤트의 timestamp 필드 읽고, ISO8601 형식으로 파싱

- target => "@timestamp" : 파싱된 타임스탬프 값을 Logstash의 표준 타임스탬프 필드인 @timestamp에 저장

- mutate {} : mutate 플러그인, 이벤트 필드 변환

- add_field : 새로운 필드 추가 옵션

- [@metadata][index] : 메타데이터 필드 안에 'index'라는 하위 필드 추가

※ 동적으로 인덱스 생성합니다.

※ 매일 새로운 인덱스를 생성하며, 인덱스 이름은 logs-YYYY.MM.dd 형식을 따릅니다.

- "logs-%{+YYYY.MM.dd}" } : 날짜 포함하는 문자열

즉, 동적으로 날짜 포함하는 문자열을 이름으로 갖는 인덱스를 생성합니다.

※ 이 외의 다양한 옵션이 있습니다.

 

- grok {} : 정규 표현식 사용하여 로그 메시지를 구조화된 데이터로 파싱하는 필터

- match => { "message" => "%{COMBINEDAPACHELOG}" }: 필드 내용 Apache 로그 형식으로 파싱

- %{COMBINEDAPACHELOG} :  Apache 로그 형식 정의 Grok 패턴

- date {} : 로그 메시지에서 날짜를 추출하여 Logstash의 @timestamp 필드 설정하는 필터

- match => [ "timestamp", "ISO8601" ] : 로그 이벤트의 timestamp 필드 읽고, ISO8601 형식으로 파싱

- target => "@timestamp" : 파싱된 타임스탬프 값을 Logstash의 표준 타임스탬프 필드인 @timestamp에 저장

즉, 수신된 로그 데이터를 Grok 패턴을 사용해 Apache 로그 형식으로 파싱하고,  timestamp 필드를 @timestamp 필드로 변환합니다.

(3) output : 데이터 전송

- elasticsearch {} : 데이터를 Elasticsearch로 전송하는 플러그인

- hosts => ["http://localhost:9200"]: Elasticsearch 인스턴스의 주소 지정

- index => "": 데이터 저장할 Elasticsearch 인덱스 이름 지정

- stdout {} : 데이터를 표준 출력(stdout)으로 출력하는 플러그인
- codec => rubydebug: 데이터 읽기 쉽게 출력하기 위한 코덱

즉, 처리된 데이터를 Elasticsearch의 인덱스에 저장하고, 동시에 표준 출력으로 디버그 형식(rubydebug)으로 출력합니다.

 

(+) 인덱스 완전 삭제

# curl -X DELETE "http://localhost:9200/<-인덱스명->"

 

(+) 인덱스 확인

# curl -X GET "http://localhost:9200/_cat/indices?v"

 

# vi /etc/logstash/conf.d/logstash.conf

 

(예시 1) 단일 인덱스

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "test"  ※ 존재하지 않을 경우, 자동으로 test 인덱스 생성됩니다.
  }
  stdout { codec => rubydebug }
}

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

 

(예시 2) 여러 인덱스

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

input {
  beats {
    port => 5044
  }
}

filter {
  date {
    match => [ "timestamp", "ISO8601" ]
    target => "@timestamp"
  }
  mutate {
    add_field => { "[@metadata][index]" => "logs-%{+YYYY.MM.dd}" }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][index]}"
  }
  stdout { codec => rubydebug }
}

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

input {
  beats {
    port => 5044
  }
}

filter {
  if [path] =~ "apache" {
    mutate { add_field => { "[@metadata][index]" => "apache-%{+YYYY.MM.dd}" } }
  } else if [path] =~ "syslog" {
    mutate { add_field => { "[@metadata][index]" => "syslog-%{+YYYY.MM.dd}" } }
  } else {
    mutate { add_field => { "[@metadata][index]" => "default-%{+YYYY.MM.dd}" } }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%+YYYY.MM.dd}"  // "%{[@metadata][index]}"
  }
  stdout { codec => rubydebug }
}

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

input {
  beats {
    port => 5044
  }
}

filter {
  mutate {
    add_field => { "[@metadata][index]" => "logs-%{+YYYY.MM.dd}" }

  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][index]}"
  }
  stdout { codec => rubydebug }
}

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

 

 

(ㄴ) filebeat.yml

# vi /etc/filebeat/filebeat.yml

-----------------------------------------------------------------------------------------------------------------------------------------------------------------

filebeat.inputs: # 로그 파일 읽기
- type: log  # 입력 유형
  enabled: true
  paths:
    - /var/log/*.log  # 읽을 로그 파일 경로

※ filebeat.inputs : Filebeat가 /var/log/*.log 경로의 로그 파일을 읽도록 설정합니다.
output.logstash:   # 로그 파일 내보내기
  hosts: ["192.168.112.202:5044"]

※ output.logstash : 수집한 로그 데이터를 192.168.112.202:5044의 Logstash로 전송합니다.
logging.level: info  # 로깅 수준
logging.to_files: true  # 로깅 기록 로그 파일 여부
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7  # 보관할 로그 파일 최대 개수
  permissions: 0644  # 로그 파일 권한

※ logging : 정보 수준 이상의 로그 메시지를 /var/log/filebeat 디렉토리의 filebeat.log 파일에 기록하며, 최대 7개의 로그 파일을 보관하고 파일 권한을 0644로 설정합니다.

-----------------------------------------------------------------------------------------------------------------------------------------------------------------