Logstash过滤插件

filter初级

Logstash安装

### 设置YUM源

# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

# tee /etc/yum.repos.d/elastic.repo << EOF

[logstash-5.x]

name=Elastic repository for 5.x packages

baseurl=https://artifacts.elastic.co/packages/5.x/yum

gpgcheck=1

gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch

enabled=1

autorefresh=1

type=rpm-md

EOF

# yum install -y logstash

基本使用

# tee filter.conf << EOF

input {

    stdin {

    }

}

filter {

    mutate {

        split => ["message", "|"]

    }

}

output {

    stdout {

    }

}

EOF

# /usr/share/logstash/bin/logstash -f filter.conf --path.settings /etc/logstash

Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties

The stdin plugin is now waiting for input:

12|fwd|343|dd

2017-09-18T01:35:03.342Z dnode [12, fwd, 343, dd]

ruby语法基本使用

# tee filter.conf << EOF

input {

    stdin {

    }

}

filter {

    mutate {

        split => ["message", "|"]

    }

    ruby {

        code => '

          msgs = event.get("message")

          puts msgs.length

        '

    }

}

output {

    stdout {

      codec => "rubydebug"

    }

}

EOF

# /usr/share/logstash/bin/logstash -f ruby.conf --path.settings /etc/logstash

Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties

The stdin plugin is now waiting for input:

r|g

2

{

      "@version" => "1",

          "host" => "dnode",

    "@timestamp" => 2017-09-18T09:06:12.546Z,

       "message" => [

        [0] "r",

        [1] "g"

    ]

}

filter高级用法

grok插件

自定义正则：将需要提取的正则表达式用()括起来，然后使用?<tag_name>的固定语法格式给匹配项打上标签
内置正则：使用%{WORD:tag_name}内置正则地址

如果想要给一串很长的字符的很多字段都打上标签，即多个自定义组合的情况，那么正则必须能完全匹配整个字符串（可以使用.*的方式跳过不关心的字段）

在线测试地址

ruby插件

### 1. 先实现rb脚本，输入从变量读取，输出也保存到变量

### 2. 脚本的输入由变量改成event.get("name")

### 3. 脚本的输出由变量改成event.set("name", $value)

举例

样例字符串一

[NEW] tcp

使用grok内置正则

自定义正则

样例字符串二

[MAN] name=fwd age=12#[WONMEN]name=xb age=10

将字符串转换成JSON

### 编写rb脚本实现所需功能

# vim ruby.rb

$result = Hash.new

$people = []

begin

  msgs = "[MAN] name=fwd age=12#[WONMEN]name=xb age=10"

  msgs.split("#").each { |msg|

    ret = Hash.new

    item = msg[/(?<=\[)MAN(?=\])|(?<=\[)WONMEN(?=\])/]

    if item.empty?

      raise "Invalid format"

    end

    ret["sex"] = item

    beg = msg.index("name")

    if beg == nil

      raise "Invalid format"

    end

    msg[beg..-1].split().each { |item|

      key, value = item.split("=")

      ret[key] = value

    }

    $people.push(ret)

  }

$result["peoples"] = $people

puts $result

end

# ruby ruby.rb

{"peoples"=>[{"sex"=>"MAN", "name"=>"fwd", "age"=>"12"}, {"sex"=>"WONMEN", "name"=>"xb", "age"=>"10"}]}

将ruby脚本放入Logstash的filter插件中

# vim ruby.conf

input {

    stdin {

    }

}

filter {

    ruby {

        code => '

          $result = Hash.new

          $people = []

          begin

            msgs = event.get("message")

            msgs.split("#").each { |msg|

              # 分割后的字符串样例 => [MAN] name=fwd age=12

              ret = Hash.new

              # 匹配头部的[MAN]或[WONMEN]

              item = msg[/(?<=\[)MAN(?=\])|(?<=\[)WONMEN(?=\])/]

              if item.empty?

                raise "Invalid format"

              end

              ret["sex"] = item

              # 获取从name到结束的字符串 => name=fwd age=12

              beg = msg.index("name")

              if beg == nil

                raise "Invalid format"

              end

              msg[beg..-1].split().each { |item|

                # 分割后的字符串样例 => name=fwd

                key, value = item.split("=")

                ret[key] = value

              }

              $people.push(ret)

            }

            $result["peoples"] = $people

            event.set("message", $result)

            event.set("[@metadata][drop]", false)

          rescue

            puts $!

            event.set("[@metadata][drop]", true)

          end

        '

    }

}

output {

    if ![@metadata][drop] {

        stdout {

          codec => rubydebug

        }

    }

}

# /usr/share/logstash/bin/logstash -f ruby.conf --path.settings /etc/logstash

Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties

The stdin plugin is now waiting for input:

[MAN] name=fwd age=12#[WONMEN]name=xb age=10

{

      "@version" => "1",

          "host" => "dnode",

    "@timestamp" => 2017-09-20T08:40:26.293Z,

       "message" => {

        "peoples" => [

            [0] {

                "name" => "fwd",

                 "age" => "12",

                 "sex" => "MAN"

            },

            [1] {

                "name" => "xb",

                 "age" => "10",

                 "sex" => "WONMEN"

            }

        ]

    }

}

参考文档

Logstash实践

 关于Logstash中grok插件的正则表达式例子

 elastic文档

 elastic插件文档

巴特西

Logstash过滤插件

filter初级

Logstash安装

基本使用

ruby语法基本使用

filter高级用法

grok插件

ruby插件

举例

样例字符串一

样例字符串二

参考文档

最新文章

热门文章