Logstash 在 ELK 架構中,是負責把收到的純文字資料,做特定的規則處理,就可以變成指定的欄位。
建立欄位的好處是可以方便搜尋,而且也能做到比全文檢索更好的分析,可說是欄位切的好,查詢沒煩惱。
我個人認為 Logstash 中最精華的部分就屬 Grok Filter。
本篇將簡單教學如何透過 Logstash Grok Filter 建立 Elasticsearch 欄位。
先去下載ELK + F 檔案, 記住版本上的對應.
filebeat.yml
name: localhost
output:
logstash:
enabled: true
hosts:
- localhost:5044
index: "localhost"
filebeat.inputs:
- type: log
paths:
- /Users/xxxxx/Desktop/log/xxxSystem/xxx.log
multiline:
pattern: '^\['
negate: true
match: after
tags: ["restapi"]
#開啟debug模式
logging.level: debug
logging.selectors: [publish]
logging.to_files: true
logging.files:
name: filebeat-localhost
logstash.conf
input {
beats {
port => 5044
}
}
filter {
grok {
match => {"message" => [
"\[(?<XXOrderNo>[\w\d]*),(?<orderNo>[\w\d]*)\] %{TIMESTAMP_ISO8601:logDate} \[(?<threadName>[\w\d\s-]+)\] %{LOGLEVEL:logLevel}(?<emptySpace>\s+)\[(?<className>[\w.]+)\] - %{GREEDYDATA:message}"
]
}
overwrite => [ "message" ]
}
# customize timestamp
date {
timezone => "Asia/Taipei"
match => ["logDate", "ISO8601"]
target => "@timestamp"
}
if "restapi" in [tags]{
mutate {
add_field => { "c_file_name" => "restapi" }
}
}
mutate {
remove_tag => [ "beats_input_codec_plain_applied" ]
remove_field => [ "emptySpace" ]
}
}
output {
stdout {
codec => rubydebug
}
# Sending properly parsed log events to elasticsearch
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-%{c_file_name}-%{+YYYY.MM.dd}"
}
}
// 可使用default setting
// 可使用default setting
// 亦可修改 memory 設定等