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

Logstash

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

[ELK] 1. Elastic Stack

커맨더 라인으로 logstash 실행

• Mac, Unix & Linux 

• Windows

 

파이프라인 구성

 

logstash 실행을 위한 2가지 방법 (with configurations)

• -e : Set configurations in command line

• -f : If configurations are set in a file (ex. pipeline.conf)

 

 

시작하기 전

• 로컬에서 elasticsearch와 kibana 실행
• config/logstash.yml 편집 - config.reload.automatic: true
• [파일 이름].conf 만들고, input과 output 설정

 

 

logstash 실행

 

 

logstash로 간단한 메시지 보내서 확인

• tcp 9900 포트를 사용해 'hello logstash' 텍스트 보내기

• 출력 로그를 확인

 

 

weblog 데이터를 다운로드하고 logstash로 보내기

• weblog-sample.log 다운로드 : https://drive.google.com/file/d/1anN2G31WlQxzCM39RHraYgoLHXbAibWM/view
• 파일의 첫 번째 라인을 읽고 logstash로 보내기

• 출력 로그 확인

 

 

logstash tcp -> stdout

input {
	tcp {
		port => 9900
	}
}

output {
	stdout { }
}

 

logstash tcp -> elasticsearch

input {
	tcp {
		port => 9900
	}
}

output {
	elasticsearch {
		hosts => [{ip주소:9200}]
		user => "elastic"
		password => "cjswnstjr"
	}
}

• elasticsearch에서 확인 -> GET _cat/indices
• elasticsearch에서 데이터 검색 -> GET logstash-*/_search
• logstash 기본 tcp -> filter -> stdout

 

logstash tcp -> filter -> elasticsearch

1. grok 패턴으로 파싱(예제 아파치 로그 데이터 - https://github.com/logstash-plugins/logstash-patterns-core/tree/main/patterns - COMBINEDAPACHELOG) - 정규식을 이용하여 정형화 -> 여러 개의 field 생성
2. geoip 파싱 -> geoip 필드에 여러 geo 정보 확인
3. useragent 파싱 -> user os, 환경 등 여러 개의 필드 생성
4. mutate : convert -> grok 필드 결과에서 bytes 필드가 텍스트라서 integer로 변경해줘야 함
5. @timestamp는 현재 logstash가 실행된 시간, 우리는 log가 만든 시간에 대한 처리가 더 필요함. timestamp 필드에 있지만 텍스트 타입이 아닌 날짜 타입으로 바꿔줘야 함. -> date 필터
6. output을 elasticsearch로 변경, stdout도 같이 해도 됨.

input {
	tcp {
		port => 9900
	}
}
filter {
	grok {
		match => { "message" => "%{COMBINEDAPACHELOG}" }  -- 1
	}
	geoip {
		source => "clientip" -- 2
	}
	useragent {           -- 3
		source => "agent"
		target => "useragent"
	}
	mutate {                --- 4
		convert => {
			"bytes" => "integer"
		}
	}
	date {         --- 5
		match => {"timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
		target => "logdate"
	}
}
output {    ----- 6
	stdout { }
	elasticsearch {
		hosts => [{ip주소:9200}]
		user => "elastic"
		password => "cjswnstjr"
	}
}

• 기존 로그 삭제 : DELETE logstash*
• logstash 확인 : GET _cat/indices
• 검색 : GET logstash*/_search

elasticsearch output

 

 

마무리

이번 포스팅은 logstash 실행과 filter에 대해서 알아보았습니다. filter 옵션은 위에서 설명한 거보다 많은 옵션들이 있어서 공식 문서를 참고하시면 되겠습니다.

다음 포스팅은 filebeat - logstash - elasticsearch로 이어지는 파이프라인에 대해 기록하겠습니다. 감사합니다.

https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

Filter plugins | Logstash Reference [8.3] | Elastic