下载并安装公共签名密钥：
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]
}
——allure-pytest的基础使用)
