logstsh架构

搜集—》过滤—》处理

Grok:匹配需要收集的字段信息

Date：处理日期类型

Geoip:添加地理位置信息

Useragent:提取请求用户信息

2、logstash安装

[root@localhost logstash]# tar -zxvf logstash-6.3.1.tar.gz
  mkdir config
[root@localhost logstash]# vi test.conf
input {
  stdin { }
}
output {
stdout {codec=>”rubydebug”}
}
[root@localhost logstash-6.3.1]# ./bin/logstash -f config/test.conf

3、logstsh操作

[root@localhost logstash-6.3.1]# vi test.conf
input {
  stdin {codec=>line}
}
output {
stdout {codec=>json}
}
[root@localhost logstash-6.3.1]# echo “foo
bar”./bin/logstash -f config/test.conf

logstsh input插件

stdin

输入插件：可以管道输入，也可以从终端交互输入

通用配置：

codec：类型为codec

type：类型为string自定义该事件类型，可用于后续判断

tags：类型为array，自定义事件的tag，可用于后续判断

add_field：类型为hash，为该事件添加字段

input{
stdin{
codec => “plain”
tags => [“test”]
type => “std”
add_field => {“key”=>”value”}
}
}
output{
stdout{
codec => “rubydebug”
}
}
[root@localhost logstash-6.3.1]# echo “test”|./bin/logstash -f config/test.conf
 
{
      “@version” => “1”,
           “key” => “value”,
       “message” => “test”,
          “type” => “std”,
          “tags” => [
        [0] “test”
    ],
          “host” => “localhost”,
    “@timestamp” => 2019-03-24T12:20:16.334Z
}

file

从文件读取数据，如常见的日志文件

配置：

path => [“/var/log/**/*.log”,”/var/log/message”] 文件位置

exclude => “*.gz” 不读取哪些文件

sincedb_path => “/var/log/message” 记录sincedb文件路径

start_position => “beginning” 或者”end” 是否从头读取文件

stat_interval => 1000 单位秒，定时检查文件是否有更新，默认1S

input {
 file {
  path => [“/home/elk/logstsh/config/nginx_logs”]
  start_position => “beginning”
  type => “web”
 }
}
output {
 stdout {
  codec => “rubydebug”
 }
}
 
{
          “path” => “/home/elk/logstsh/config/nginx_logs”,
       “message” => “79.136.114.202 – – [04/Jun/2015:07:06:35 +0000] \”GET /downloads/product_1 HTTP/1.1\” 404 334 \”-\” \”Debian APT-HTTP/1.3 (0.8.16~exp12ubuntu10.22)\””,
    “@timestamp” => 2019-03-24T12:47:20.900Z,
          “host” => “localhost”,
          “type” => “web”,
      “@version” => “1”
}
 

Elasticsearch

logstsh filter

Filter是logstsh功能强大的原因，它可以对数据进行丰富的处理，比如解析数据、删除字段、类型转换等

date：日期解析

grok：正则匹配解析

dissect：分割符解析

mutate：对字段作处理，比如重命名、删除、替换等

json：按照json解析字段内容到指定字段中

geoip：增加地理位置数据

ruby：利用ruby代码来动态修改logstsh Event

./logstash -f ../../config/test.conf

{“logdate”:”Jan 01 2018 12:02:08”}

Grok

正则匹配

%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] “%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}” %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent}

93.180.71.3 – – [17/May/2015:08:05:32 +0000] “GET /downloads/product_1 HTTP/1.1” 304 0 “-” “Debian APT-HTTP/1.3 (0.8.16~exp12ubuntu10.21)”

93.180.71.3 – – [17/May/2015:08:05:23 +0000] “GET /downloads/product_1 HTTP/1.1” 304 0 “-” “Debian APT-HTTP/1.3 (0.8.16~exp12ubuntu10.21)”

Logstsh ouput

stdout
file :
file {
path => “/var/log/web.log”
codec => line {format => “%{message}”}
}
elasticsearch :
elasticsearch {
                hosts => [“http://192.168.14.10:9200”]
                index => “logstash-%{type}-%{+YYYY.MM.dd}”
        }