说明
filebeat中message要么是一段字符串，要么在日志生成的时候拼接成json然后在filebeat中指定为json。但是大部分系统日志无法去修改日志格式，filebeat则无法通过正则去匹配出对应的field，这时需要结合logstash的grok来过滤，架构如下：
实例说明：
以系统登录日志格式为例：
登录成功日志
Jan 6 17:11:47 localhost sshd[3324]: Received disconnect from 172.16.0.13: 11: disconnected by user
Jan 6 17:11:47 localhost sshd[3324]: pam_unix(sshd:session): session closed for user root
Jan 6 17:11:48 localhost sshd[3358]: Address 172.16.0.13 maps to localhost, but this does not map back to the address - POSSIBLE BREAK-IN ATTEMPT!
Jan 6 17:11:51 localhost sshd[3358]: Accepted password for root from 172.16.0.13 port 38604 ssh2
Jan 6 17:11:51 localhost sshd[3358]: pam_unix(sshd:session): session opened for user root by (uid=0)
登录失败日志
Jan 6 17:13:10 localhost sshd[3380]: pam_unix(sshd:auth): authentication failure; logname= uid=0 euid=0 tty=ssh ruser= rhost=172.16.0.39 user=root
Jan 6 17:13:12 localhost sshd[3380]: Failed password for root from 172.16.0.39 port 58481 ssh2
这里需要定义两个field，Status和ClientIP来获取某个IP登录服务器的频率和状态
而单filebeat输出信息为：
{"@timestamp":"2017-01-12T03:12:46.772Z","beat":{"hostname":"localhost","name":"localhost","version":"5.1.1"},"input_type":"log","message":"Jan 12 11:11:40 localhost sshd[1564]: Accepted password for root from 172.16.11.239 port 65278 ssh2","offset":8548,"source":"/var/log/secure","type":"log"}
message为字符串，且filebeat无法通过正则匹配出想要的数据，所以filebeat只负责在服务器上收索转发日志数据，过滤功能则交给logstash来处理，配置如下：
filebeat_ssh.yaml
filebeat.prospectors:
- input_type: log
paths: /var/log/secure
include_lines: [".*Failed.*",".*Accepted.*"]
output.logstash:
hosts: ["localhost:5044"]
logstash_ssh.conf
input {
beats {
port => 5044
}
}
filter {
grok {
match => {
"message" => ".* sshd\[\d+\]: (?<status>\S+) .* (?<ClientIP>(?:\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})?) .*"
}
overwrite => ["message"]
}
}
output {
stdout {
codec=>rubydebug
}
}
配置解释
filebeat_ssh.yaml
- include_lines：filebeat只过滤出包含该字符串的行，列表形式
- output.logstash中指定logstash服务器和logstash监听filebeat的端口，这里为了测试方便，将filebeat和logstash装在同一台机器
- 更多参数请查看** filebeat.full.yml **文件
logstash_ssh.conf
- input beats来指定logstash监听filebeat的端口
- filter 过滤插件，详情查看Grok正则过滤Linux系统登录日志
- output 这里为了测试输出到控制台，实际生产中可输出到elasticserach
输出结果
{
"@timestamp" => 2017-01-12T04:00:16.325Z,
"offset" => 9538,
"@version" => "1",
"input_type" => "log",
"beat" => {
"hostname" => "localhost",
"name" => "localhost",
"version" => "5.1.1"
},
"host" => "localhost",
"source" => "/var/log/secure",
"message" => "Jan 12 12:00:08 localhost sshd[2043]: Accepted password for root from 172.16.11.239 port 51763 ssh2",
"type" => "log",
"ClientIP" => "172.16.11.239",
"tags" => [
[0] "beats_input_codec_plain_applied"
],
"status" => "Accepted"
}
