- 기초부터 다지는 ElasticSearch 운영 노하우 교재를 참고합니다.
파일비트를 통해 로그를 받아오려는 대상 파일을 지목했다. 이제 받아온 파일 내용을 로그 스태시로 받아 ES로 넘겨주면 된다.
brew install elastic/tap/logstash-full이후에 설정파일로 들어가본다.
cd /opt/homebrew/etc/logstash보이는 파일 항목은 다음과 같다.
- logstash.yml: Logstash와 관련된 설정을 할 수 있는 파일이다. 워커의 수는 몇 개로 할 것인지, 한번에 얼만큼의 배치를 처리할 것인지에 대한 것을 설정한다.
- logstash-sample.conf: 파싱에 사용할 플러그인, 다른 파싱 룰을 정의하는 것이다.(인덱스 파싱)
- jvm.options: Logstash를 실행할 때 함께 설정할 JVM 옵션들을 설정하는 파일이다.
- log4j2.properties: 로그 파일 기록과 관련된 설정에 사용하는 파일이다.
여기서 나는 logstash-sample.conf 를 수정한다. 이전에 tshark로 캡쳐한 데이터들에서 인덱스 타입이 명시되어 있던 부분도 이 파일에서 수정하도록 한다.
vi logstash-sample.conf기존에 파일 내용은 이렇게 되어있다.
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.
input {
beats {
port => 5044 # 비트를 통해 입력받는 포트는 5044번이다.
}
}
output {
elasticsearch { => ES로 보내는 호스트명, 대상 인덱스의 템플릿에 대해 나와있다.
hosts => ["http://localhost:9200"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
#user => "elastic"
#password => "changeme"
}
}여기에 filter 필드를 추가한다. 교재에는 다른 로그에 관련한 예제가 나와있어서 공식 홈페이지 내용을 참조해서 filter속성을 추가했다.
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
}
여기서 grok 패턴 매칭 방식을 이용해서 입력된 파일에 대한 파싱 룰을 정의해야 한다.
근데, GROK 문법이 좀 생소하고 어떤걸 인자로 입력해야 할지 어려웠다.
이럴 경우를 대비해 Kibana에서 좋은 기능을 제공한다.
Dev tools를 클릭 후, Grok Debugger을 클릭한다.
여기서 샘플 데이터에 내가 파싱 룰을 적용할 파일의 내용을 기재하고,
패턴을 기재하면 결과가 나오게 된다. (이런식으로 가능하다!)
너무너무 편리하다! 이렇게 보면 굳이 tshark 명령어로 ek 옵션을 주지 않아도 괜찮을 것 같아 옵션을 주지 않고, 파싱하는 테스트를 해봤다.
테스트 데이터는 다음과 같이 했고
1 0.000000 X.X.X.X → X.X.X.X TLSv1.2 114 Application Data
2 0.000373 X.X.X.X → X.X.X.X TCP 54 58355 → 443 [ACK] Seq=1 Ack=61 Win=4096 Len=0파싱하는 Grok Pattern 은 아래와 같이 했다.
%{NUMBER:number} %{NUMBER:duration} %{IP:src_ip} → %{IP:desti_ip} %{WORD:protocol}.* %{INT:src_port} → %{INT:desti_port} %{GREEDYDATA:message}결과는 패턴에 맞지않는 TLSv1.2 프로토콜 유형을 제외하고는 모두 파싱이 되는 것 같다.
일단 파싱이 되는 프로토콜 위주로 실시간 수집을 실행한 후에, 어떤 프로토콜을 수집할지 정해야겠다.
지금은 실시간 수집이 먼저다!
conf 파일을 다음과 같이 수정했다.
filter{
grok{
match => {
"message" => "%{NUMBER:number} %{NUMBER:duration} %{IP:src_ip} → %{IP:desti_ip} %{WORD:protocol}.* %{INT:src_port} → %{INT:desti_port} %{GREEDYDATA:message}"
}
}
}이후 logstash를 재시작 해주었다.
logstash -f logstash-sample.conf로그가 바로 확인된다.
action=>["index", {:_id=>nil, :_index=>"filebeat-7.14.0-2021.08.12", :routing=>nil} ...postman이랑 kibana에 로그가 제대로 수집되었는지 확인했다.
둘다 정상적으로 수집됐다.
다음 포스팅에는 실시간 패킷 수집 내용을 기재해볼 예정이다.
