說明

filebeat中message要么是一段字符串，要么在日志生成的時(shí)候拼接成json然后在filebeat中指定為json。但是大部分系統(tǒng)日志無法去修改日志格式，filebeat則無法通過正則去匹配出對(duì)應(yīng)的field，這時(shí)需要結(jié)合logstash的grok來過濾，架構(gòu)如下：

實(shí)例說明：

以系統(tǒng)登錄日志格式為例：

登錄成功日志

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

這里需要定義兩個(gè)field，Status和ClientIP來獲取某個(gè)IP登錄服務(wù)器的頻率和狀態(tài)
而單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無法通過正則匹配出想要的數(shù)據(jù)，所以filebeat只負(fù)責(zé)在服務(wù)器上收索轉(zhuǎn)發(fā)日志數(shù)據(jù)，過濾功能則交給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服務(wù)器和logstash監(jiān)聽filebeat的端口，這里為了測試方便，將filebeat和logstash裝在同一臺(tái)機(jī)器
• 更多參數(shù)請(qǐng)查看** filebeat.full.yml **文件

logstash_ssh.conf

• input beats來指定logstash監(jiān)聽filebeat的端口
• filter 過濾插件，詳情查看Grok正則過濾Linux系統(tǒng)登錄日志
• output 這里為了測試輸出到控制臺(tái)，實(shí)際生產(chǎn)中可輸出到elasticserach

輸出結(jié)果

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