过滤插件

json

输入配置

[root@es2 conf.d]# vi test.conf 
input {
  file {
    path => "/var/log/test/*.log"
    exclude => "error.log"
    tags => "web"
    tags => "nginx"
    type => "access"
    add_field => {
      "project" => "microservice"
      "app" => "product"
    }
  }
}
filter {
  json {
    source => "message"
  }
}
output {
  file {
    path => "/tmp/test.log"
  }
}


-------------
模拟数据：
{"remote_addr": "192.168.1.10","url":"/index","status":"200"}

输出结果

{
    "host": "es2",
    "@version": "1",
    "path": "/var/log/test/1.log",
    "project": "microservice",
    "type": "access",
    "app": "product",
    "status": "200",
    "remote_addr": "192.168.1.10",
    "tags": [
        "web",
        "nginx"
    ],
    "message": "{\"remote_addr\": \"192.168.1.10\",
                 \"url\":\"/index\",
                 \"status\":\"200\"}",
    "@timestamp": "2021-11-04T14:15:49.388Z",
    "url": "/index"
}

输出至ES

input {
  file {
    path => "/var/log/test/*.log"
    exclude => "error.log"
    tags => "web"
    tags => "nginx"
    type => "access"
    add_field => {
      "project" => "microservice"
      "app" => "product"
    }
  }
}
filter {
  json {
    source => "message"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.153.25:9200"]
    index => "test-%{+YYYY.MM.dd}"
  }
}

kibana显示

1636036166459.png 1636036254563.png 1636036576720.png

KV

输入配置

input {
  file {
    path => "/var/log/test/*.log"
    exclude => "error.log"
    tags => "web"
    tags => "nginx"
    type => "access"
    add_field => {
      "project" => "microservice"
      "app" => "product"
    }
  }
}
filter {
  kv {
    field_split => "&?"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.153.25:9200"]
    index => "test-%{+YYYY.MM.dd}"
  }
}

-----------------------------------------------------------------------------
模拟数据：
www.ctnrs.com?id=1&name=aliang&age=30

kibana显示

1636036923256.png

Grok

正则表达式

1636082254273.png

Grok Debugger

1636077090199.png

正则匹配模式

#样例数据
192.168.1.10 GET /index.html 15824 0.043

#Grok 模式
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

#结构化数据
{
  "duration": "0.043",
  "request": "/index.html",
  "method": "GET",
  "bytes": "15824",
  "client": "192.168.1.10"
}

#样例数据
192.168.1.10 GET /index.html 15824 0.043

#Grok 模式
(?<ip>\d+\.\d+\.\d+\.\d+) (?<method>\w+) (?<request>/.*) (?<bytes>\d+) (?<duration>\d+\.\d+) 

#结构化数据
{
  "duration": "0.043",
  "request": "/index.html",
  "method": "GET",
  "bytes": "15824",
  "client": "192.168.1.10"
}

自定义模式

#样例数据
192.168.1.10 GET /index.html 15824 0.043 123456

#Grok 模式
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}

#自定义模式
CID [0-9]{5,6}

#结构化数据
{
  "duration": "0.043",
  "request": "/index.html",
  "method": "GET",
  "bytes": "15824",
  "client": "192.168.1.10",
  "cid": "123456"
}

配置文件（自定义）

input {
  file {
    path => "/var/log/test/*.log"
    exclude => "error.log"
    tags => "web"
    tags => "nginx"
    type => "access"
    add_field => {
      "project" => "microservice"
      "app" => "product"
    }
  }
}
filter {
  grok {
    patterns_dir =>"/opt/patterns"
    match => {
      "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}"
    }
  }
}
output {
  elasticsearch {
    hosts => ["192.168.153.25:9200"]
    index => "test-%{+YYYY.MM.dd}"
  }
}

---------------------------------------------------------------------------
[root@localhost conf.d]# vi /opt/patterns
CID [0-9]{5,6}

--------------------------------------------------------------------------------

模拟数据：
192.168.1.10 GET /index.html 15824 0.043 123456

配置文件（多格式匹配）

input {
  file {
    path => "/var/log/test/*.log"
    exclude => "error.log"
    tags => "web"
    tags => "nginx"
    type => "access"
    add_field => {
      "project" => "microservice"
      "app" => "product"
    }
  }
}
filter {
  grok {
    patterns_dir =>"/opt/patterns"
    match => [
         "message", "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}",
         "message", "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{EID:eid} %{TAG:tag}"
        ]
  }
}
output {
  elasticsearch {
    hosts => ["192.168.153.25:9200"]
    index => "test-%{+YYYY.MM.dd}"
  }
}

-------------------------------------------------------------------
[root@localhost conf.d]# vi /opt/patterns
CID [0-9]{5,6}
EID [a-z]{5,6}
TAG \w+

--------------------------------------------------------------------
#192.168.1.10 GET /index.html 15824 0.053 123456
#192.168.1.10 GET /index.html 15824 0.043 abcdef xyz
#两条都能匹配和接收

GeoIP

配置文件

input {
  file {
    path => "/var/log/test/*.log"
    exclude => "error.log"
    tags => "web"
    tags => "nginx"
    type => "access"
    add_field => {
      "project" => "microservice"
      "app" => "product"
    }
  }
}
filter {
  grok {
    patterns_dir =>"/opt/patterns"
    match => [
         "message", "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}",
         "message", "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{EID:eid} %{TAG:tag}"
        ]
  }
  geoip {
    source => "client"
    database => "/opt/GeoLite2-City.mmdb"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.153.25:9200"]
    index => "test-%{+YYYY.MM.dd}"
  }
}

--------------------------------------------------------------------------
source => "client" client是ip字段（%{IP:client}）

测试数据：
8.8.8.8 GET /index.html 15824 0.043 abcdef xyz

kibana显示

1636081310192.png

条件判断

配置文件

input {
  file {
    path => "/var/log/test/test.log"
    add_field => {
    "log_type" => "test"
    }
  }
  file {
    path => "/var/log/test/prod.log"
    add_field => {
    "log_type" => "prod"
    }
  }
}

filter {
  json  {
    source => "message"
  }

  if [log_type] in ["test","dev"] {
    mutate {
      add_field => {
        "[@metadata][target_index]" => "test-%{+YYYY.MM}"
      }
    }
  } else if [log_type] == "prod" {
    mutate {
      add_field => {
        "[@metadata][target_index]" => "prod-%{+YYYY.MM.dd}"
      }
    }
  } else {
    mutate {
      add_field => {
        "[@metadata][target_index]" => "unknown-%{+YYYY}"
      }
    }
  }



}

output {
  elasticsearch {
    hosts => "192.168.153.25:9200"
    index => "%{[@metadata][target_index]}"
  }
}



-------------
模拟数据：
{"remote_addr": "192.168.1.10","url":"/index","status":"789"}

kibana显示

1636098769558.png