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插件文档