[ELK] 9. filebeat - logstash - elasticsearch

beats

2022.06.21 - [ELK stack] - [ELK] 1. Elastic Stack

[ELK] 1. Elastic Stack

 

filebeat 

전 포스팅에서 아파치 웹 로그 데이터를 받았기 때문에, 그것을 활용하기 위해 여러 beats 중 filebeat를 사용하겠습니다.

2022.06.29 - [ELK stack] - [ELK] 8. Logstash 실행 및 filter 설정

[ELK] 8. Logstash 실행 및 filter 설정

버전에 맞는 filebeat를 다운로드하였다고 가정하겠습니다. filebeat를 사용하기 전 filebeat.yml 설정 몇 가지를 해야 합니다.

 

filebeat inputs

type : log

enabled : true

paths : {사용할 아파치 로그 데이터 path}

 

elasticsearch outputs

host : {외부ip주소} (ex. '34.64.211.132:9200' )

username : {유저네임}

password : {패스워드}

 

logstash outputs

host : {logstash 호스트 주소} (ex. ['localhost:5044'] )

 

 

filebeat 실행

실행 - ./filbeat

실행 후 elasticsearch에서 확인 - GET _cat/indices

데이터 확인 : GET {파일 이름}/_search

데이터 개수 : GET {파일 이름}/_count

 

 

filebeat -> logstash -> elasticsearch

주의점 1 - beats는 agent 필드도 저장해서 보내는데 그것은 자신의 localhost 정보인데, filter에서 useragent에 agent 필드가 중복이 되기 때문에 grok 하기 전에 beat의 agent 필드를 지워야 합니다.

ex. beats agent 필드

주의점 2 - filebeat은 지금까지 읽었던 파일은 읽지 않기 때문에 data 디렉터리 삭제해야 합니다. 방법 :  rm -fr data
1. 데이터 출력이 많을 때 stdout 보여주면 너무 많은 데이터를 보여주기 때문에 dots으로 output을 확인합니다. (default - rubydebug |  json, dots 등등)

2. 필요 없는 필드 삭제

- @timestamp가 있기 때문에 기존 timestamp 삭제

- log host가 아닌 elasticsearch가 찍어준 단순 host라서 삭제

- @version 삭제

- agent로 파싱 해줬기 때문에 삭제 -> message는 원본 데이터 성격이기 때문에 그냥 두기
3. elasticsearch의 인덱스 명 바꾸기 -> @timestamp 년월일을 보고 해당 형식에 맞게 저장됨.

input {
	beats {
		port => 5044 # beat -> logstash(filebeat.yml의 logstash output 호스트 번호 확인)
	}
}
filter {
	mutate {
		remove_field => ["agent"]  ----- 주의점
	}
	grok {
		match => { "message" => "%{COMBINEDAPACHELOG}" } 
	}
	geoip {
		source => "clientip"
	}
	useragent {         
		source => "agent"
		target => "useragent"
	}
	mutate {               
		convert => {
			"bytes" => "integer"
		}
	}
	date {        
		match => {"timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
	}
	mutate { -------- 2
		remove_field => ["timestamp", "host", "@version", "agent"]
	}
}
output {
	stdout {  -------- 1
		codec => "dots"
	}
	elasticsearch {
		index => "apachelog-%{+yyyy.MM.dd}" --- 3
		hosts => [{ip주소:9200}]
		user => "elastic"
		password => "cjswnstjr"
	}
}

beat -> logstash
stdout으로 .(dots)로 표시

logstash -> elasticsearch
GET _cat/indices
GET _cat/indices/apachelog-* - 설정한 날짜 별로 저장

 

데이터 확인

GET apachelog-*/_search를 할 경우 value값이 제대로 나오지 않습니다. (성능을 위한 샘플링) -> 10000

GET apachelog-*/_count를 하면 알 수 있습니다.

또는 터미널 데이터 path에서 데이터 개수 확인 가능 - wc -l weblog-sample.log 

 

 

마무리

이번 포스팅은 filebeat - logstash - elasticsearch로 이어지는 파이프라인에 대해 알아보았습니다. 설치에 관한 내용은 포스팅에 포함되지 않으니 공식 홈페이지에서 다운로드하면 되겠습니다.

다음 포스팅은 위 과정을 통해 elasticsearch로 받은 아파치 로그 데이터에 인덱스 재색인에 관한 포스팅을 진행하겠습니다. 감사합니다.