filebeat + logstash 对message提取指定

说明

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"
}