logstash配置文件
- 喜欢使用ruby语法,作为过滤器,定制化比较程度高,但是语法会比较繁琐
- 下面是处理 Java日志logstash 配置文件
input {
beats {
port => 5056
}
}
filter {
if[logSource]="java-0"{
#第一步,分割日志,增加字段,ruby语法
ruby {
init => "@kname = ['logLvel','timeSort','times','logFile','threadMsg','tmeRequestInfo']"
code => "
new_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').delete('[').split(']'))])
new_event.remove('@timestamp')
event.append(new_event)
"
}
#首先删除不要的字段,能提高性能
mutate{
#删除不必要的字段
remove_field => "tags"
remove_field => "beat"
remove_field => "@version"
remove_field => "message"
}
#字段转换
mutate{
remove_field => ["type","tags","input_type","fields"]
}
}
#[DEBUG][103955134][2018/07/26 00:12:04170][BaseJdbcLogger.java][pool-1-thread-73][ooo Using Connection [com.mysql.jdbc.JDBC4Connection@9cf9250]]
if[logSource]="java-dctorder"{
#第一步,分割日志,增加字段,ruby语法
ruby {
init => "@kname = ['logLvel','timeSort','times','logFile','threadMsg','tmeRequestInfo']"
code => "
new_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').delete('[').split(']'))])
new_event.remove('@timestamp')
event.append(new_event)
"
}
#首先删除不要的字段,能提高性能
mutate{
#删除不必要的字段
remove_field => "tags"
remove_field => "beat"
remove_field => "@version"
remove_field => "message"
}
#字段转换
mutate{
remove_field => ["type","tags","input_type","fields"]
}
}
}
output {
## 通过判断发送到elastic
if [logSource]== "java-dctorder" {
## 将错误日志放到redis中，以便后续处理
if[logLvel]== "ERROR"{
redis {
host => ["host:prot"]
id => "my_plugin_id_0003"
key => "key"
password => "password"
data_type => "list"
db => "0"
}
}
elasticsearch {
#输出到elastic 的用户名和密码没有可以不填
user => 'elastic'
password => 'password'
hosts => "127.0.0.1:9200"
index => "java-dctorder-%{+YYYY.MM.dd}"
}
}
if [logSource]=="java-0"{
#输出到elastic 的用户名和密码没有可以不填
elasticsearch {
user => 'elastic'
password => 'password'
hosts => "127.0.0.1:9200"
index => "java-dboss-%{+YYYY.MM.dd}"
}
}
}
##抓取php日志的配置文件
input {
beats {
port => 5044
}
}
filter {
#第一步,分割日志,增加字段,ruby语法
ruby {
init => "@kname = ['logLvel','api','times','clientip','userId','methodMsg','serviceMsg','ThreadMsg','infoMsg']"
code => "
new_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').delete('[').split(']'))])
new_event.remove('@timestamp')
event.append(new_event)
"
}
#字段转换
mutate{
remove_field => ["type","tags","input_type","fields"]
}
}
output {
elasticsearch {
#输出到elastic 的用户名和密码没有可以不填
user => 'elastic'
password => 'Om?BiI1Aliaw$VW+4&hr'
hosts => "127.0.0.1:9200"
index => "logstash-phpik-%{+YYYY.MM.dd}"
}
if [logLvel] == "ERROR"{
redis {
host => ["host:port"]
id => "my_plugin_id_0001"
key => "LOGSTASH_ERROR_LOG_LIST"
password => "lLBiOoOk6lb9"
data_type => "list"
db => "255"
}
}
if [logLvel] == "WARN"{
redis {
host => ["host:port"]
id => "my_plugin_id_0002"
key => "LOGSTASH_WARN_LOG_LIST"
password => "password"
data_type => "list"
db => "255"
}
}
}
- 后台启动 nohup ./logstash -f ../config/conf/logstash-java.conf &
