Centos7 搭建logstash

下载并安装公共签名密钥：

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

创建一个名为 /etc/yum.repos.d/logstash.repo 的文件，并添加以下内容：

[logstash-7.x]
name=Elastic repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

安装logstash:

sudo yum install logstash

启动logstash

systemctl start logstash

 指定配置文件启动

/usr/share/logstash/bin/logstash -t -f /etc/logstash/conf.d/nginx.conf -t  #测试配置文件
Configuration OK
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/nginx_access.conf  #启动logstash/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/nginx_access.conf & #后台启动logstash

input plugin  让logstash可以读取特定的事件源。

 官网：https://www.elastic.co/guide/en/logstash/current/input-plugins.html

 事件源可以是从stdin屏幕输入读取，可以从file指定的文件，也可以从es，filebeat，kafka，redis等读取

stdin 标准输入

file   从文件读取数据

file{path => ['/var/log/nginx/access.log']  #要输入的文件路径type => 'nginx_access_log'start_position => "beginning"
}
# path  可以用/var/log/*.log,/var/log/**/*.log，如果是/var/log则是/var/log/*.log
# type 通用选项. 用于激活过滤器
# start_position 选择logstash开始读取文件的位置，begining或者end。

还有一些常用的例如：discover_interval，exclude，sincedb_path,sincedb_write_interval等可以参考官网

syslog  通过网络将系统日志消息读取为事件

syslog{port =>"514" type => "syslog"
}
# port 指定监听端口(同时建立TCP/UDP的514端口的监听)#从syslogs读取需要实现配置rsyslog：
# cat /etc/rsyslog.conf   加入一行
*.* @172.17.128.200:514　  #指定日志输入到这个端口，然后logstash监听这个端口，如果有新日志输入则读取
# service rsyslog restart   #重启日志服务

beats   从Elastic beats接收事件

beats {port => 5044   #要监听的端口
}
# 还有host等选项# 从beat读取需要先配置beat端，从beat输出到logstash。
# vim /etc/filebeat/filebeat.yml 
..........
output.logstash:
hosts: ["localhost:5044"]

kafka  将 kafka topic 中的数据读取为事件

kafka{bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"topics => ["access_log"]group_id => "logstash-file"codec => "json"
}
kafka{bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"topics => ["weixin_log","user_log"]  codec => "json"
}
# bootstrap_servers 用于建立群集初始连接的Kafka实例的URL列表。
# topics  要订阅的主题列表，kafka topics
# group_id 消费者所属组的标识符，默认为logstash。kafka中一个主题的消息将通过相同的方式分发到Logstash的group_id
# codec 通用选项，用于输入数据的编解码器。

filter plugin 过滤器插件，对事件执行中间处理

grok   解析文本并构造 。把非结构化日志数据通过正则解析成结构化和可查询化  

grok {match => {"message"=>"^%{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}$"}}
匹配nginx日志
# 203.202.254.16 - - [22/Jun/2018:16:12:54 +0800] "GET / HTTP/1.1" 200 3700 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/601.7.7 (KHTML, like Gecko) Version/9.1.2 Safari/601.7.7"
#220.181.18.96 - - [13/Jun/2015:21:14:28 +0000] "GET /blog/geekery/xvfb-firefox.html HTTP/1.1" 200 10975 "-" "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)"

 注意这里grok 可以有多个match匹配规则，如果前面的匹配失败可以使用后面的继续匹配。例如

 grok {match => ["message", "%{IP:clientip} - %{USER:user} \[%{HTTPDATE:raw_datetime}\] \"(?:%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion})\" (?:\"%{DATA:body}\" )?(?:\"%{DATA:cookie}\" )?%{NUMBER:response} (?:%{NUMBER:bytes:int}|-) \"%{DATA:referrer}\" \"%{DATA:agent}\" (?:(%{IP:proxy},? ?)*|-|unknown) (?:%{DATA:upstream_addr} |)%{NUMBER:request_time:float} (?:%{NUMBER:upstream_time:float}|-)"]match => ["message", "%{IP:clientip} - %{USER:user} \[%{HTTPDATE:raw_datetime}\] \"(?:%{WORD:verb} %{URI:request} HTTP/%{NUMBER:httpversion})\" (?:\"%{DATA:body}\" )?(?:\"%{DATA:cookie}\" )?%{NUMBER:response} (?:%{NUMBER:bytes:int}|-) \"%{DATA:referrer}\" \"%{DATA:agent}\" (?:(%{IP:proxy},? ?)*|-|unknown) (?:%{DATA:upstream_addr} |)%{NUMBER:request_time:float} (?:%{NUMBER:upstream_time:float}|-)"]       }

grok 语法：%{SYNTAX:SEMANTIC}   即 %{正则:自定义字段名}

                     官方提供了很多正则的grok pattern可以直接使用  :https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns  

                     grok debug工具： http://grokdebug.herokuapp.com

　　　　　　正则表达式调试工具： https://www.debuggex.com/

　　　　需要用到较多的正则知识，参考文档有：https://www.jb51.net/tools/zhengze.html

         自定义模式：   (?<字段名>the pattern)

        例如： 匹配 2018/06/27 14:00:54  

                (?<datetime>\d\d\d\d\/\d\d\/\d\d \d\d:\d\d:\d\d)

          得到结果：  "datetime": "2018/06/27 14:00:54"

date   日期解析  解析字段中的日期，然后转存到@timestam

[2018-07-04 17:43:35,503]
grok{match => {"message"=>"%{DATA:raw_datetime}"}
}
date{match => ["raw_datetime","YYYY-MM-dd HH:mm:ss,SSS"]remove_field =>["raw_datetime"]
}#将raw_datetime存到@timestamp 然后删除raw_datetime#24/Jul/2018:18:15:05 +0800
date {match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z]
}