Logstash语法常用案例解析（一）
摘要
简述logstash的常用插件，以及简单的使用案例
一：基础运行
建议使用supervisor来管理ELK中的各个组件，方便同一管理
安装 https://www.aolens.cn/?p=809 有讲解
提供一个常用的配置：
[program:logstash]command=/opt/logstash/bin/logstash -f /opt/logstash/conf/index.confnumprocs=1 ;开几个进程dirrectory=/opt/logstashuser=root ;用户stdout_logfile=/opt/logstash/logs/logstash.logstdout_logfile_maxbytes=1MB ;每个日志大小stdout_logfile_backups=10 ;保留10个日志文件stderr_logfile=/opt/logstash/logs/logstash_err.logstderr_logfile_maxbytes=1MBstderr_logfile_backups=10
运行参数：
启动logstash服务（常用supervisor守护进程）
./bin/logstash -f /etc/logstash/conf.d/* -t #检查配置文件是否ok
./bin/logstash -f conf.d/nginx.conf -w 5 -l /var/log/logstash/
二：配置语法
1，区域：（section）
Logstash用｛｝来定义区域。可以在区域中定义多个插件区域，插件区域内可以定义键值对
eg:
input { # 输入数据 file { path=["/var/log/messages","/var/log/*.log"] type="system" start_position="beginning" } }filter{ # 数据过滤处理 if[type]=="system"{ grok{ match=["message",%{COMBINEDAPACHELOG}] } }}output{ # 数据处理输出 stdout{ codec=rubydebug }}2，数据类型：
string--普通字符串
name => "Hello world"
name => 'It\'s a beautiful day'
array--数组可以是单个或者多个字符串值。
path => [ "/var/log/messages", "/var/log/*.log" ]
path => "/data/mysql/mysql.log"
hash--键值对，注意多个键值对用空格分隔，而不是逗号。
match => {
"field1" => "value1"
"field2" => "value2"
... }
Codec--用来表示数据编码。用于input和output段。便于数据的处理。
codec => "json"
number--必须是有效的数值，浮点数或者整数。
port => 33
boolean--布尔值必须是TRUE或者false。
ssl_enable => true
bytes--指定字节单位。默认是byte。
my_bytes => "1113" # 1113 bytes
my_bytes => "10MiB" # 10485760 bytes
my_bytes => "100kib" # 102400 bytes Binary (Ki,Mi,Gi,Ti,Pi,Ei,Zi,Yi) 单位1024
my_bytes => "180 mb" # 180000000 bytes SI (k,M,G,T,P,E,Z,Y) 单位基于1000
password--一个单独的字符串。
my_password => "password"
path--代表有效的操作系统路径。
my_path => "/tmp/logstash"
3：字段引用
Logstash配置中要使用字段的值，只需要把字段的名字写在中括号[]里。只要是input输入的值，都可以引用
eg:
[geoip][location][-1]
4，条件判断
表达式支持的操作符
==(等于),!=(不等),<(小于),>(大于),<=(小等),>=(大等)
=~(匹配正则),!~(不匹配正则)
in(包含),not in(不包含)
and(与),or(或),nand(与非),xor(非或)
()(复合表达式),!()(取反复合表达式结果)
三：常用插件
1，插件管理
./bin/plugin -h
install
uninstall
update
list
eg:
bin/plugin install logstash-output-webhdfs
bin/plugin update logstash-input-tcp
2，常用插件 input,output,filter,codec
2.1 Input插件
stdin：标准输入,常用于测试，
input { stdin { type = "string" tags = ["add"] codec="plain" }}file：从文件系统中读取文件，类似linux下的tail -f。 最常用
input { file { path = ["/var/log/*.log","/var/log/message"] ＃ logstash只支持文件的绝对路径 type = "system" ＃ type记录文件类型，定义的变量为全局变量，其他插件都可以调用 start_position = "beginning" }}redis：从redis服务器读取，同时使用redis channel和redis list。
input{ redis{ data_type="list" key="logstash-nginx" host="192.168.1.250" port=6379 db=1 threads=5 }} 将源数据写入redisoutput{ redis{ host="192.168.1.250" port=6379 db=1 data_type="list" key="logstash-nginx" }}TCP/UDP： 输入
#nc127.0.0.18888 /var/log/nginx/access.json # 可以是json文件直接传值 #echo'{"name":"liuziping","age":"18"}' |nc127.0.0.18888 input { tcp { port = 8888 # 定义tcp监听端口 codec="json" # 规定传入的数据为json格式，k/v结构方便分析 mode = "server" }}
syslog：监听在514端口的系统日志信息，并解析成RFC3164格式。
input { syslog { port = "514" }}beats: 通过Filebeat发送事件。
2.2:Output 插件
stdout：标准输出
output { stdout { codec = rubydebug workers = 2 }}file ：保存成文件
output { file { path = "/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz" message_format = "%{message}" gzip = true }}elasticsearch：保存进elasticsearch ，也是最为重要的
output { elasticsearch { hosts => ["192.168.0.2:9200"] # 或者cluster => "ClusterName" index => "logstash-%{type}-%{+YYYY.MM.dd}" #索引名，统一格式，方便kibana导入，会讲统一类型的日志，全部导入 这里的type＝input中的type值 document_type => "nginx" workers => 1 #启动一个进程 flush_size => 20000 # 攒够20000 条数据一次性发给ES，默认500条 idle_flush_time => 10 # 如果10s内没攒够 20000 条也发一次给ES,默认1s template_overwrite => true }}redis：保存到redis中在input插件中已讲解
TCP：输出TCP
output { tcp { host = "192.168.0.2" port = 8888 codec = json_lines } }Email:发送邮件
exec：调用命令执行
