Logstash 过滤 Filter 插件

520

数据从源传输到存储库的过程中，Logstash 过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便进行更强大的分析和实现商业价值。

Logstash 能够动态地转换和解析数据，不受格式或复杂度的影响。

常见的 Filter 插件：

  • 利用 Grok 从非结构化数据中派生出结构
  • 从 IP 地址破译出地理坐标
  • 利用 useragent 从请求中分析操作系统、设备类型
  • 简化整体处理，不受数据源、格式或架构的影响
图片[1]-Logstash 过滤 Filter 插件-李佳程的个人主页
https://www.elastic.co/guide/en/logstash/7.6/filter-plugins.html

1、Grok 插件

Grok 是一个过滤器插件，可帮助您描述日志格式的结构。有超过200种 grok模式抽象概念，如IPv6地址，UNIX路径和月份名称。为了将行与格式匹配

生产环境常需要将如下非结构化的数据解析成 json 结构化数据格式

2022-09-19T18:19:00 [8.8.8.8:prd] DEBUG this is an example log message

使用 Grok 插件可以基于正则表达式技术用内置的正则表达式的别名来表示和匹配上面的日志

%{TIMESTAMP_ISO8601:timestamp} \[%{IPV4:ip};%{WORD:environment}\] %
{LOGLEVEL:log_level} %{GREEDYDATA:message}

最终转换为以下格式

{
  "timestamp": "2022-09-19T18:19:00",
  "ip": "8.8.8.8",
  "environment": "prd",
  "log_level": "DEBUG",
  "message": "this is an example log message"
}

# 参考网站
https://www.elastic.co/cn/blog/do-you-grok-grok

范例：使用 grok pattern 将 Nginx 日志格式化为 json 格式

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_stdout.conf

input {
   http {
       port =>6666
   }
}
filter {
    #将nginx日志格式化为json格式
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"  #将message字段转化为指定的Json格式
       }
   }
}
output {
   stdout {
   codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_stdout.conf -r
{
           "auth" => "-",
        "request" => "/wp-admin/admin-ajax.php",
     "@timestamp" => 2023-01-04T03:04:43.364Z,
           "verb" => "POST",
          "agent" => "\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
        "headers" => {
         "request_method" => "POST",
           "request_path" => "/",
           "http_version" => "HTTP/1.1",
            "http_accept" => "*/*",
              "http_host" => "192.168.1.108:6666",
        "http_user_agent" => "curl/7.68.0",
         "content_length" => "292",
           "content_type" => "application/x-www-form-urlencoded"
    },
       "@version" => "1",
    "httpversion" => "1.1",
       "clientip" => "121.231.7.174",
      "timestamp" => "04/Jan/2023:11:03:05 +0800",
           "host" => "192.168.1.105",
       "response" => "200",
       "referrer" => "\"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\"",
        "message" => "121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] \"POST /wp-admin/admin-ajax.php HTTP/1.1\" 200 109 \"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
          "ident" => "-",
          "bytes" => "109"
}


# 使用curl 或 insomnia 等工具将nginx日志发送http请求至logstash,可以看到上面的输出信息
root@web01:~# curl -XPOST -d'121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] "POST /wp-admin/admin-ajax.php HTTP/1.1" 200 109 "http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"' http://192.168.1.108:6666

范例：直接将nginx的访问日志转化为Json格式

root@logstash01:~# vim /etc/logstash/conf.d/nginx_grok_stdout.conf
input {
   file {
       path => "/var/log/nginx/access.log"
       type => "nginx-accesslog"
       start_position => "beginning"
       stat_interval => "3"
   }
}
filter {
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"
       }
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@web01:~# curl http://192.168.1.108:6666

root@logstash01:~# logstash -f /etc/logstash/conf.d/nginx_grok_stdout.conf
{
           "verb" => "GET",
     "@timestamp" => 2023-01-04T03:07:44.986Z,
           "type" => "nginx-accesslog",
          "agent" => "\"curl/7.68.0\"",
        "request" => "/",
          "bytes" => "12",
          "ident" => "-",
       "referrer" => "\"-\"",
           "host" => "logstash01.test.com",
    "httpversion" => "1.1",
           "path" => "/var/log/nginx/access.log",
           "auth" => "-",
       "response" => "200",
        "message" => "192.168.1.105 - - [04/Jan/2023:03:07:39 +0000] \"GET / HTTP/1.1\" 200 12 \"-\" \"curl/7.68.0\"",
       "clientip" => "192.168.1.105",
      "timestamp" => "04/Jan/2023:03:07:39 +0000",
       "@version" => "1"
}

2、Geoip 插件

geoip 根据 ip 地址提供的对应地域信息，比如:经纬度,国家,城市名等,以方便进行地理数据分析

范例：

root@logstash01:~# vim /etc/logstash/conf.d/http_geoip_stdout.conf

input {
   http {
       port =>6666
   }
}
filter {
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"
       }
   }
   geoip {
        source => "clientip"
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_geoip_stdout.conf -r
{
          "bytes" => "109",
      "timestamp" => "04/Jan/2023:11:03:05 +0800",
       "@version" => "1",
           "verb" => "POST",
          "geoip" => {
           "region_code" => "JS",
              "latitude" => 31.7685,
             "city_name" => "Changzhou",
              "timezone" => "Asia/Shanghai",
                    "ip" => "121.231.7.174",
         "country_code2" => "CN",
           "region_name" => "Jiangsu",
              "location" => {
            "lon" => 119.9527,
            "lat" => 31.7685
        },
             "longitude" => 119.9527,
        "continent_code" => "AS",
         "country_code3" => "CN",
          "country_name" => "China"
    },
        "request" => "/wp-admin/admin-ajax.php",
          "ident" => "-",
          "agent" => "\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
       "clientip" => "121.231.7.174",
       "response" => "200",
     "@timestamp" => 2023-01-04T06:44:47.899Z,
        "message" => "121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] \"POST /wp-admin/admin-ajax.php HTTP/1.1\" 200 109 \"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
           "host" => "192.168.1.105",
           "auth" => "-",
       "referrer" => "\"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\"",
        "headers" => {
           "request_path" => "/",
              "http_host" => "192.168.1.108:6666",
            "http_accept" => "*/*",
         "request_method" => "POST",
         "content_length" => "292",
           "content_type" => "application/x-www-form-urlencoded",
           "http_version" => "HTTP/1.1",
        "http_user_agent" => "curl/7.68.0"
    },
    "httpversion" => "1.1"
}

# 使用curl命令或insomnia等发送nginx日志给logstash可以看到上面信息
root@web01:~# curl -XPOST -d'121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] "POST /wp-admin/admin-ajax.php HTTP/1.1" 200 109 "http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"' http://192.168.1.108:6666

范例：只显示指定的geoip的字段信息

root@logstash01:~# vim /etc/logstash/conf.d/http_geoip_field_stdout.conf

input {
   http {
       port =>6666
   }
}
filter {
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"
       }
   }
   geoip {
       source => "clientip"
       fields => ["continent_code","country_name","country_code2","timezone","longitude","latitude"]
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_geoip_field_stdout.conf -r
{
           "verb" => "POST",
       "response" => "200",
        "request" => "/wp-admin/admin-ajax.php",
          "bytes" => "109",
           "host" => "192.168.1.105",
       "clientip" => "121.231.7.174",
          "geoip" => {
         "country_code2" => "CN",
        "continent_code" => "AS",
          "country_name" => "China",
              "timezone" => "Asia/Shanghai",
             "longitude" => 119.9527,
              "latitude" => 31.7685
    },
       "@version" => "1",
           "auth" => "-",
    "httpversion" => "1.1",
          "agent" => "\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
          "ident" => "-",
        "message" => "121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] \"POST /wp-admin/admin-ajax.php HTTP/1.1\" 200 109 \"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
       "referrer" => "\"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\"",
     "@timestamp" => 2023-01-04T06:48:07.147Z,
        "headers" => {
              "http_host" => "192.168.1.108:6666",
        "http_user_agent" => "curl/7.68.0",
           "http_version" => "HTTP/1.1",
         "content_length" => "292",
            "http_accept" => "*/*",
           "content_type" => "application/x-www-form-urlencoded",
           "request_path" => "/",
         "request_method" => "POST"
    },
      "timestamp" => "04/Jan/2023:11:03:05 +0800"
}

root@web01:~curl -XPOST -d'121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] "POST /wp-admin/admin-ajax.php HTTP/1.1" 200 109 "http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"' http://192.168.1.108:6666

3、Date 插件

date插件可以将日志中的日期字符串解析为日志类型。然后替换@timestamp 字段(此字段默认为当前写放logstash的时间而非日志本身的时间)或指定的其他字段

match    类型为数组，用于指定日期匹配的格式，可以以此指定多种日期格式
target   类型为字符串，用于指定赋值的字段名，默认是 @timestamp
timezone 类型为字符串，用于指定时区域

https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html
http://joda-time.sourceforge.net/timezones.html

范例：

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_date_stdout.conf

input {
   http {
       port =>6666
   }
}
filter {
    #将nginx日志格式化为json格式
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"
       }
   }
    #解析date日期格式为: 14/Jul/2020:15:07:27 +0800,
   date {
       match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
       target => "access_time"       #将时间写入新生成的access_time字段
       #target => "@timestamp"       #将时间覆盖原有的@timestamp字段
       timezone => "Asia/Shanghai"
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_date_stdout.conf -r
{
    "httpversion" => "1.1",
     "@timestamp" => 2023-01-04T06:54:23.239Z,
       "response" => "200",
       "referrer" => "\"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\"",
           "host" => "192.168.1.105",
        "headers" => {
         "content_length" => "292",
        "http_user_agent" => "curl/7.68.0",
            "http_accept" => "*/*",
           "content_type" => "application/x-www-form-urlencoded",
              "http_host" => "192.168.1.108:6666",
           "request_path" => "/",
         "request_method" => "POST",
           "http_version" => "HTTP/1.1"
    },
           "auth" => "-",
       "clientip" => "121.231.7.174",
          "ident" => "-",
      "timestamp" => "04/Jan/2023:11:03:05 +0800",
           "verb" => "POST",
       "@version" => "1",
        "message" => "121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] \"POST /wp-admin/admin-ajax.php HTTP/1.1\" 200 109 \"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
        "request" => "/wp-admin/admin-ajax.php",
          "agent" => "\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
    "access_time" => 2023-01-04T03:03:05.000Z,
          "bytes" => "109"
}

# 用curl提交日志,可以看到上面输出信息
root@web01:~# curl -XPOST -d'121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] "POST /wp-admin/admin-ajax.php HTTP/1.1" 200 109 "http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"' http://192.168.1.108:6666

4、Useragent 插件

useragent 插件可以根据请求中的 user-agent 字段，解析出浏览器设备、操作系统等信息

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_useragent_stdout.conf
input {
   http {
       port =>6666
   }
}
filter {
    #将nginx日志格式化为json格式
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"
       }
   }
    #解析date日期如: 10/Dec/2020:10:40:10 +0800
   date {
       match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
       target => "@timestamp"
       #target => "access_time"
       timezone => "Asia/Shanghai"
   }
    #提取agent字段，进行解析
   useragent {
       source => "agent"     #指定从哪个字段获取数据
       target => "useragent" #转换后的新字段
   }
}
output {
   stdout {
       codec => rubydebug
   }
}


root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_useragent_stdout.conf -r
{
       "referrer" => "\"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\"",
      "useragent" => {
           "version" => "108.0.1462.42",
             "patch" => "1462",
           "os_full" => "Windows 10",
            "device" => "Other",
        "os_version" => "10",
          "os_major" => "10",
              "name" => "Edge",
                "os" => "Windows",
             "minor" => "0",
           "os_name" => "Windows",
             "major" => "108"
    },
       "clientip" => "121.231.7.174",
          "ident" => "-",
          "agent" => "\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
       "@version" => "1",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2023:11:03:05 +0800",
     "@timestamp" => 2023-01-04T03:03:05.000Z,
           "auth" => "-",
           "verb" => "POST",
        "message" => "121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] \"POST /wp-admin/admin-ajax.php HTTP/1.1\" 200 109 \"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42\"",
          "bytes" => "109",
        "headers" => {
         "content_length" => "292",
           "http_version" => "HTTP/1.1",
            "http_accept" => "*/*",
         "request_method" => "POST",
           "request_path" => "/",
              "http_host" => "192.168.1.108:6666",
           "content_type" => "application/x-www-form-urlencoded",
        "http_user_agent" => "curl/7.68.0"
    },
           "host" => "192.168.1.105",
        "request" => "/wp-admin/admin-ajax.php",
       "response" => "200"
}



# 用curl提交日志,可以看到上面输出信息
root@web01:~# curl -XPOST -d'121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] "POST /wp-admin/admin-ajax.php HTTP/1.1" 200 109 "http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"' http://192.168.1.108:6666

5、Mutate 插件

Mutate 插件主要是对字段进行、类型转换、删除、替换、更新等操作，可以使用以下函数

remove_field  删除字段
split         字符串切割,相当于awk取列
add_field     添加字段
convert       类型转换
gsub          字符串替换

5.1、remove_field 删除字段

范例：

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_mutate_remove_field_stdout.conf

input {
   http {
       port =>6666
   }
}
filter {
    #将nginx日志格式化为json格式
   grok {
       match => {
            "message" => "%{COMBINEDAPACHELOG}"
       }
   }
    #解析date日期如: 10/Dec/2020:10:40:10 +0800
   date {
       match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
       target => "@timestamp"
        #target => "access_time"
       timezone => "Asia/Shanghai"
   }
    #mutate 删除操作
   mutate {
       remove_field => ["headers","message", "agent"]
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_mutate_remove_field_stdout.conf -r
{
       "response" => "200",
          "bytes" => "109",
    "httpversion" => "1.1",
           "host" => "192.168.1.105",
          "ident" => "-",
        "request" => "/wp-admin/admin-ajax.php",
           "auth" => "-",
      "timestamp" => "04/Jan/2023:11:03:05 +0800",
           "verb" => "POST",
       "referrer" => "\"http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit\"",
       "clientip" => "121.231.7.174",
       "@version" => "1",
     "@timestamp" => 2023-01-04T03:03:05.000Z
}

# 用curl提交日志,可以看到上面输出信息
root@web01:~# curl -XPOST -d'121.231.7.174 - - [04/Jan/2023:11:03:05 +0800] "POST /wp-admin/admin-ajax.php HTTP/1.1" 200 109 "http://www.lijiach.com/wp-admin/post.php?post=1321&action=edit" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.42"' http://192.168.1.108:6666

5.2、Split 切割

mutate 中的 split 字符切割， 指定字段做为分隔符,生成新的字段名

示例：1000|提交订单|2022-01-08 09:10:21

范例：split 切割字符串取列

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_mutate_split_stdout.conf

input {
   http {
       port =>6666
   }
}
filter {
    #mutate 切割操作
   mutate {
        #字段分隔符
 split => { "message" => "|" } #将message字段按 | 分割成多个列表元素
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_mutate_split_stdout.conf
{
       "message" => [
        [0] "1000",
        [1] "提交订单",
        [2] "2022-01-08 09:10:21"
    ],
          "host" => "192.168.1.105",
       "headers" => {
         "request_method" => "POST",
        "http_user_agent" => "curl/7.68.0",
         "content_length" => "37",
              "http_host" => "192.168.1.108:6666",
           "request_path" => "/",
            "http_accept" => "*/*",
           "http_version" => "HTTP/1.1",
           "content_type" => "application/x-www-form-urlencoded"
    },
    "@timestamp" => 2023-01-04T07:07:17.833Z,
      "@version" => "1"
}

# 用curl提交日志,可以看到上面输出信息
root@web01:~# curl -XPOST -d '1000|提交订单|2022-01-08 09:10:21' http://192.168.1.108:6666/

5.3、add_field 添加字段

范例：

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_mutate_add_field_stdout.conf
input {
   http {
       port =>6666
   }
}
filter {
    #mutate 切割操作
   mutate {
        #字段分隔符
 split => { "message" => "|" }
        #添加字段,将message的列表的第0个元素添加字段名user_id
       add_field => {
            "user_id" => "%{[message][0]}"
            "action" => "%{[message][1]}"
            "time" => "%{[message][2]}"
       }
        #add_field => {"[@metadata][target_index]" => "app-%{+YYY.MM.dd}"}#添加字
段做索引名
        #删除无用字段
       remove_field => ["headers","message"]
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_mutate_add_field_stdout.conf
{
    "@timestamp" => 2023-01-04T07:11:33.758Z,
          "host" => "192.168.1.105",
       "user_id" => "1000",
      "@version" => "1",
        "action" => "提交订单",
          "time" => "2022-01-08 09:10:21"
}

root@web01:~# curl -XPOST -d '1000|提交订单|2022-01-08 09:10:21' http://192.168.1.108:6666/

5.4、convert 转换

mutate 中的 convert 可以实现数据类型的转换。 支持转换integer、float、string等类型

范例：

root@logstash01:~# vim /etc/logstash/conf.d/http_grok_mutate_convert_stdout.conf
input {
   http {
       port =>6666
   }
}
filter {
    #mutate 切割操作
   mutate {
        #字段分隔符
 split => { "message" => "|" }
        #添加字段
       add_field => {
            "user_id" => "%{[message][0]}"
            "action" => "%{[message][1]}"
            "time" => "%{[message][2]}"
       }
        #删除无用字段
       remove_field => ["headers","message"]
        #对新添加字段进行格式转换
       convert => {
            "user_id" => "integer"
            "action" => "string"
            "time" => "string"
       }
        #convert => ["excute_time","float] #此格式也可以支持
        #convert => ["time","string" ]
   }
}
output {
   stdout {
       codec => rubydebug
   }
}

root@logstash01:~# logstash -f /etc/logstash/conf.d/http_grok_mutate_convert_stdout.conf -r
{
          "host" => "192.168.1.105",
    "@timestamp" => 2023-01-04T07:14:41.007Z,
          "time" => "2022-01-08 09:10:21",
      "@version" => "1",
       "user_id" => "1000",
        "action" => "提交订单"
}


root@web01:~# curl -XPOST -d '1000|提交订单|2022-01-08 09:10:21' http://192.168.1.108:6666/

5.5、gsub 替换

gsub 实现字符串的替换

filter {
    mutate {
        gsub=>["message","\n", " "] #将message字段中的换行替换为空格
    }
}

6、条件判断

范例：

# vim /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
  - /var/log/nginx/access.log
  tags: ["access"]
- type: log
  enabled: true
  paths:
  - /var/log/nginx/error.log
  tags: ["error"]
output.logstash:
  hosts: ["192.168.1.108:5044","192.168.1.109:5044",]
  #loadbalance: true       #负载均衡
  #worker: 2 #number of hosts * workers #开启多进程

# vim /etc/logstash/conf.d/filebeat_logstash_es.conf
input {
   beats {
       port => 5044
   }
}
filter {
    if "access" in [tags][0] {
       mutate {
           add_field => { "target_index" => "access-%{+YYYY.MM.dd}"}
       }
   }
    else if "error" in [tags][0] {
       mutate {
           add_field => { "target_index" => "error-%{+YYYY.MM.dd}"}
       }
   }
    else if "system" in [tags][0] {
       mutate {
           add_field => { "target_index" => "system-%{+YYYY.MM.dd}"}
       }
 }
}
output {
   elasticsearch {
       hosts =>["192.168.1.101:9200","192.168.1.102:9200","192.168.1.103:9200"]  # 一般写data地址
       index => "%{[target_index]}"    #使用字段target_index值做为索引名
       template_overwrite => true      #覆盖索引模板
   }
}

范例：

# vim /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
  - /var/log/nginx/access.log
  fields:
    project: test-access
    env: test
output.logstash:

  hosts: ["192.168.1.108:5044","192.168.1.109:5044",]

# vim /etc/logstash/conf.d/filebeat_logstash_es.conf
input {
   beats {
       port => 5044
   }
   file {
       path => "/tmp/wang.log"
       type => wanglog     #自定义的类型,可以用于条件判断
       start_position => "beginning"
       stat_interval => "3"
   }
}
output {
    if [fields][env] == "test" {
        elasticsearch {
            hosts =>["10.0.0.101:9200","10.0.0.102:9200","10.0.0.103:9200"]
            index => "test-nginx-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "wanglog" {
        stdout {
            codec => rubydebug
        }
    }
}

© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
ELK
喜欢就支持一下吧
点赞0 分享
Lijiach的头像-李佳程的个人主页
使用 pg_ctl 启动关闭以及管理 PostgreSQL 数据库-李佳程的个人主页
使用 pg_ctl 启动关闭以及管理 PostgreSQL 数据库
使用 pg_ctl 启动关闭以及管理 PostgreSQL 数据库
9个月前 633
Ceph-FS 文件存储-李佳程的个人主页
Ceph-FS 文件存储
Ceph-FS 文件存储
2年前 336
部署OpenVPN-李佳程的个人主页
部署OpenVPN
部署OpenVPN
3年前 293
OpenVPN 管理-李佳程的个人主页
OpenVPN 管理
OpenVPN 管理
3年前 244
PostgreSQL 关闭数据库的三种方法-李佳程的个人主页
PostgreSQL 关闭数据库的三种方法
PostgreSQL 关闭数据库的三种方法
10个月前 218
mysql8 一键安装脚本-李佳程的个人主页
mysql8 一键安装脚本
mysql8 一键安装脚本
2年前 194
相关推荐
Elasticsearch 集群状态监控-李佳程的个人主页
Elasticsearch 集群状态监控
Elasticsearch 集群状态监控
2年前 48
Elasticsearch 集群工作原理-李佳程的个人主页
Elasticsearch 集群工作原理
Elasticsearch 集群工作原理
2年前 94
Beats 收集数据-李佳程的个人主页
Beats 收集数据
Beats 收集数据
2年前 54
Logstash 过滤-李佳程的个人主页
Logstash 过滤
Logstash 过滤
2年前 98
搜索
开启精彩搜索
使用 pg_ctl 启动关闭以及管理 PostgreSQL 数据库-李佳程的个人主页
633人已阅读
使用 pg_ctl 启动关闭以及管理 PostgreSQL 数据库
TOP1
Ceph-FS 文件存储-李佳程的个人主页
Ceph-FS 文件存储
2年前336人已阅读
TOP2
部署OpenVPN-李佳程的个人主页
部署OpenVPN
3年前293人已阅读
TOP3
OpenVPN 管理-李佳程的个人主页
OpenVPN 管理
3年前244人已阅读
TOP4
PostgreSQL 关闭数据库的三种方法-李佳程的个人主页
PostgreSQL 关闭数据库的三种方法
10个月前218人已阅读
TOP5
mysql8 一键安装脚本-李佳程的个人主页
mysql8 一键安装脚本
2年前194人已阅读
TOP6
Linux 简介-李佳程的个人主页
Linux 简介
11个月前188人已阅读
TOP7
ceph 集群维护-李佳程的个人主页
ceph 集群维护
2年前185人已阅读
TOP8
Nginx命令和信号-李佳程的个人主页
Nginx命令和信号
3年前183人已阅读
TOP9
安装KVM-李佳程的个人主页
安装KVM
3年前180人已阅读
TOP10
Zabbix 主动与被动模式-李佳程的个人主页
Zabbix 主动与被动模式
2年前174人已阅读
TOP11
KVM 存储管理-李佳程的个人主页
KVM 存储管理
3年前173人已阅读
TOP12
iptables（下）-李佳程的个人主页
iptables（下）
3年前169人已阅读
TOP13
Docker 介绍-李佳程的个人主页
Docker 介绍
2年前169人已阅读
TOP14
部署反向代理-李佳程的个人主页
部署反向代理
3年前169人已阅读
TOP15
部署 ceph-李佳程的个人主页
部署 ceph
2年前166人已阅读
TOP16
部署Percona XtraDB Cluster(PXC 5.7）-李佳程的个人主页
部署Percona XtraDB Cluster(PXC 5.7）
3年前164人已阅读
TOP17
kafka-李佳程的个人主页
kafka
3年前164人已阅读
TOP18
ceph 块设备 RBD-李佳程的个人主页
ceph 块设备 RBD
2年前160人已阅读
TOP19
磁盘常见工具-李佳程的个人主页
磁盘常见工具
3年前160人已阅读
TOP20
其他文章

Nginx 反向代理负载均衡

部署MHA实现Mysql高可用

Elasticsearch 部署

PostgreSQL 配置文件介绍

Elasticsearch 插件

Kubernetes 组件

实现 Master/Master 的 Keepalived 双主架构

Keepalived 实现 IPVS 的高可用性

Git 安装

实现 Java 应用源码编译并部署

ceph 存储池操作

Ansible使用01

zabbix 监控 Java 应用 Tomcat 服务

Redis 哨兵(Sentinel)

实现单主的 LVS-DR 模式

zabbix 用户 Users 和组 Groups 管理

网络测试诊断工具

半同步复制

zabbix 实现故障自愈功能

GitLab 的数据备份和恢复