安装Logstash
环境
Java8
CENTOS 7
$ wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.2.tar.gz
$ tar -zxvf logstash-6.2.2.tar.gz -C/opt
$ ln -s /opt/logstash-6.2.2/ /opt/logstash
#验证安装
$ /opt/logstash/bin/logstash --version
配置语法
配置文件结构
输入
input {
…
}
过滤器
filter {
…
}
输出
output {
…
}
验证文件
logstash -f CONFIG_PATH -t
Logstash数据类型
输入插件
file：从文件中获得时间流
input{
file{
path => “var/log/message”
start_position =>“beginning”
sincedb_path =>"/dev/null"
}
}
filter{}
output{stdout{}}
多文件输入时采用数组形式
path =>["/var/log/messages","/var/log/messages-2019"]*
编解码器
json
rubydebug
input {
file {
path =>"/data/stu.json"
start_position =>“beginning”
sincedb_path => “/dev/null”
codec => “json”
}
}
filter {}
output {
stdout { codec => “rubydebug”}
}
过滤器
grok：使用正则表达式解析任意文本和结构
100 10.10.0.1—->%{NUMBER:duration} %{IP:client}
input{stdin{}}
filter{
grok{
match=>{“message”=>"%{NUMBER:duration} %{IP:client}"}
}
}
output{ stdout{codec=>rubydebug}}
grok自定义模式
(?<field_name>the pattern here)
2019-01-01 0:0:0 100 10.10.0.1
(?\d[./-]\d[./-]\d* \d*:\d*:\d*) %{NUMBER:duration} %{IP:client}**
条件判断
控制filter和output处理特定事件
if EXPRESSION {
…
} else if EXPRESSION {
…
} else {
…
}
比较操作
相等： ==, !=, <, >, <=, >=
正则： =~(匹配正则), !~(不匹配正则)
包含：in(包含), not in(不包含)
布尔：and(与), or(或), nand(与非), xor(异或)
取反：!(表达式)
控制duration<100的行输出至stdout
input{stdin{}}
filter{
grok{
match=>{“message”=>"%{NUMBER:duration} %{IP:client}"}
}
mutate{convert=>[“duration”,“integer”]}
}
output{
if [duration]<100{
stdout{codec=>rubydebug}
}
}