Filebeat -> ElasticSearch, Logstash 로그 수집설정

이부분은 교재를 참고했다.
교재 [기초부터 다지는 ElasticSearch 운영 노하우]

• 설정 파일 경로나 파일명은 OS에 따라 조금씩 차이가 있을 수 있음을 안내드립니다.
• 다른 OS나 다운로드 형태 기준으로 디렉토리 확인은 이 공식 사이트에서 확인 바랍니다.

1. Filebeat 설치하기

brew install elastic/tap/filebeat-full

Filebeat 기본 파일들의 스펙은 다음과 같다.

• fields.yml: Filebeat가 Logstash를 거치지 않고 직접 ElasticSearch에 쓸 때 사용하는 타입과 필드를 정의한다.
• filebeat.reference.yml: filebeat.yml 파일에서 설정할 수 있는 모든 설정들이 예제 형식으로 제공된다. 이 파일의 내용을 참고해서 filebeat.yml 파일을 설정하면 된다.
• filebeat.yml: Filebeat가 실행하면서 읽는 환경 설정 파일이다. 여기에 설정한 내용을 바탕으로 Filebeat가 설정된다.
• modules.d: Filebeat가 직접 Json파싱까지 할 때에 사용하는 환경설정들을 저장하는 디렉토리이다.

2. 설정파일 수정하기

cd /opt/homebrew/etc/filebeat/

으로 올라간다음에 filebeat.yml 파일을 찾는다.

확인 사항

이때, 바로 ES로 보낼 것인지 아니면 Logstash를 거쳐서 보낼 것인지에 따라
설정 파일의 내용이 조금 바뀐다.

두가지 모두 큰 차이는 없으며, 관리 측면에서 본인이 어떤 것을 할 것인지 선택하면 된다. 나는 먼저 샘플 테스트로 2가지 모두 진행해볼 예정이고, 향후에는 Filebeat > Logstash > ElasticSearch 구조로 진행할 것이다.

유지 보수 측면에서 Filebeat는 수집만의 역할을 하고, Logstash는 수집된 데이터를 파싱, ES가 적재 Kibana가 시각화 하는 순으로 역할을 분리하여 의존성을 줄이는게 좋을 것 같기 때문이다.

1) Filebeat > ElasticSearch 설정하기

파일에서 바로 수집해 별도의 파싱과정 없이 엘라스틱 서치로 보내는 경우의 설정이다.

- type: log

  # Change to true to enable this input configuration.
  enabled: false

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - '/Users/munang/files.json' -> ********이부분 
  enabled: true
  #- c:\programdata\elasticsearch\logs\*


output.elasticsearch:
  # The Logstash hosts
  hosts: ["localhost:9200"]

이렇게 설정하고 filebeat를 실행해보면 화면에는 아무것도 나오지 않는다.
기본적으로 background로 실행되기 때문이다. -e 옵션을 붙여 사용하면 foreground로 실행되어 관련된 로그를 확인할 수 있다.

하지만 키바나 관리 화면으로 가서 Index Management 탭을 확인해보면 새로운 인덱스가 생성된 것을 볼 수 있다.

다시 키바나의 Dev tool 로 들어가서, 다음과 같이 search 쿼리를 날려본다. (옆에 작은 플레이 버튼을 누르면 된다. 아니면 커맨드+엔터키)

그러면 다음과 같이 내가 path에 설정한 파일과 관련된 내용이 확인된다.

"log": {
                        "offset": 920,
                        "file": {
                            "path": "/Users/munang/files.log"
                        }
                    },
                    "index": {
                        "_index": "packets-2021-08-15",
                        "_type": "doc"
                    }

나는 로그 파일을 읽어서 ES에 넣어주길 원했는데...?
-> 따로 파싱룰을 설정 안해두면, 이렇게 읽어오는 파일의 정보와 filebeat의 Agent역할을 하는 Host의 정보(이 경우에는 내 로컬 컴퓨터)에 대해 적재가 된다.
-> 자세히 확인해보면 fields.yml 파일의 타입과 필드로 이뤄진 도큐먼트 임을 확인할 수 있다.
-> filebeat에서도 별도의 파싱을 만들 수 있는데 fields.yml 파일을 이용해 만들면 된다.
-> 하지만 본 게시글에서는 Logstash를 이용해 파싱하는 내용을 다루겠다.

2) Filebeat > Logstash > ElasticSearch 설정하기

• Filebeat 설정
  앞에서 ElasticSearch.output 관련된 내용은 앞에 #를 입력해 주석처리를 해준다.

- type: log

  # Change to true to enable this input configuration.
  enabled: false

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - '/Users/munang/files.json' -> ********이부분 
  enabled: true
  #- c:\programdata\elasticsearch\logs\*


output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]

환경 설정은 끝이다. 그리고 나서 Logstash를 켜본다.

• Logstash 설정하기

cd /opt/homebrew/etc/logstash/
vi logstash-sample.conf

로그 스태시의 설정파일을 열고, 다음과 같은 내용을 입력해준다.

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  beats {
    port => 5044 
    # 인풋으로 beats를 통해 받을때 beats가 설정된 포트 번호 (디폴트가 5044임)
  }
}

filter{
  grok{
    match => {
        "message" => "%{NUMBER:number}   %{NUMBER:duration} %{IP:src_ip} → %{IP:desti_ip} %{WORD:protocol}.* %{INT:src_port} → %{INT:desti_port} %{GREEDYDATA:message}"
        # 메세지 내용을 파싱하는 룰입니다. 대부분 GROK 문법을 사용하며 관련된 내용은 자세하게 다시 포스팅 하겠습니다. 
     }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    # 출력으로 나오게될 ElasticSearch의 호스트 명을 적어주면 된다. 
    #user => "elastic"
    #password => "changeme"
  }
}

GROK 문법과 Kibana를 이용해 사용하는 방법을 포스팅 예정이며,
여기서 참조해주시면 됩니다.
이렇게 설정해주셨으면 logstash와 filebeat를 실행합니다.

filebeat -e #e 옵션을 붙여서 사용합니다. 
logstash -f logstash-sample.conf # logstash는 logstash 디렉토리에서 실행해주셔야 합니다. 

3. 실시간 수집 확인

2가지 실행을 모두 실행했고, 실시간으로 수집하려는 것은 확인했으나 제대로 파싱이 되지 않았다. 아무래도 filter 옵션을 잘못설정해준 것처럼 보였다.

이러한 오류가 발생했다.

2021-08-15T18:11:48.415+0900	ERROR	[reader_json]	readjson/json.go:74	Error decoding JSON: json: cannot unmarshal number into Go value of type map[string]interface {}

비슷한 오류를 검색하다가 logstash의 GROK 문법이 문제임을 유추할 수 있었다.