Logstash语法常用案例解析（一）
摘要
简述logstash的常用插件，以及简单的使用案例
一：基础运行
建议使用supervisor来管理ELK中的各个组件，方便同一管理
安装 https://www.aolens.cn/?p=809 有讲解
提供一个常用的配置：
[program:logstash] command=/opt/logstash/bin/logstash -f /opt/logstash/conf/index.conf numprocs=1 ;开几个进程 dirrectory=/opt/logstash user=root ;用户 stdout_logfile=/opt/logstash/logs/logstash.log stdout_logfile_maxbytes=1MB ;每个日志大小 stdout_logfile_backups=10 ;保留10个日志文件 stderr_logfile=/opt/logstash/logs/logstash_err.log stderr_logfile_maxbytes=1MB stderr_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
}} 将源数据写入redis
output{
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：调用命令执行
