记录日常工作关于系统运维,虚拟化云计算,数据库,网络安全等各方面问题。
 
0

filebeat + logstash 对message提取指定字段
发表者：admin分类：大数据2021-04-28 09:13:39 阅读[2218]

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 userJan  6 17:11:47 localhost sshd[3324]: pam_unix(sshd:session): session closed for user rootJan  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 ssh2Jan  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=rootJan  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"}



作者：梦想做小猿
链接：https://www.jianshu.com/p/cbb708b2e464
 




转载请标明出处【filebeat + logstash 对message提取指定字段】。

《www.micoder.cc》 虚拟化云计算,系统运维,安全技术服务.

Tags： [阅读全文...]
网站已经关闭评论