logstashの文法を簡単に記録します
5764 ワード
本文はいくつかの文法の簡単な列で示す1.json 非json+調整時間(東八区)+ログ時間生成+フィールド属性変更+切り取りフィールド長 注釈をつけるのがおっくうで、長すぎる.
input {
kafka{
bootstrap_servers => ["10.0.0.11:9092,10.0.0.12:9092,10.0.0.13:9092"]
group_id => "es"
topics => ["dhc"]
codec => json
}
}
filter {
if "dhc" in [tags]{
json {
source => "message"
}
if "." not in [value]{
mutate {
add_field => { "value_int" => "%{[value]}" }
}
}
mutate {
convert => { "value_int" => "integer" }
}
}
}
output {
# stdout {}
if "dhc" in [tags] {
elasticsearch {
hosts => ["10.0.0.11:9200","10.0.0.12:9200","10.0.0.13:9200"]
manage_template => false
index => "dhc-%{+yyyy.MM.dd}"
user => "elastic"
password => "111111"
}
}
}
input {
kafka{
bootstrap_servers => ["10.100.15.23:9092,10.100.15.24:9092,10.100.15.25:9092"]
group_id => "test"
auto_offset_reset => "earliest"
consumer_threads => "5"
decorate_events => "false"
topics => ["kafka"]
type => "bbs_log"
codec => json
}
}
filter {
if "km" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:times}\s*\S+\(.*))\>\s\S+\:\(.*))\,(?(.*))\>\s\S+%{GREEDYDATA:message_value}" }
}
date {
match => ["times", "ISO8601"]
remove_field => "times"
}
mutate{
convert => { "level" => "integer" }
}
}
if "shbkuc" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:times}\s*\S+\(.*))\>\s\S+\:\(.?))\,(?(.*))\>\s\S+%{GREEDYDATA:message_value}" }
}
date {
match => ["times", "ISO8601"]
remove_field => "times"
}
}
if "time" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:times}\s%{USER:info}\s(?(.*))\s(?(.*))\s\[(?(.{0,240}))\]" }
}
date {
match => ["times", "ISO8601"]
remove_field => "times"
}
if "(null)" not in [excute_time]{
mutate {
add_field => { "cuttime" => "%{[excute_time]}" }
}
}
mutate { convert => { "cuttime" => "integer" } }
}
if "test" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:DATE_time}\s*%{USER:server_name}\S+\s*%{INT:level}\,%{BASE16NUM:SNO}\,(?(.*))\;\s\S+%{GREEDYDATA:message_value}"
}
}
}
if "caserver" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:times}\s*%{USER:server_name}\S+\s*%{INT:level}\,(?(.*))\,(?(.*))\;\s\S+%{GREEDYDATA:message_value}" }
}
date {
match => ["times", "ISO8601"]
remove_field => "times"
}
mutate{
convert => { "level" => "integer" }
}
if "(null)" not in [excute_time]{
mutate {
# add_field => { "twotime" => "%{excute_time}" }
split => [ "excute_time","."]
add_field => { "cuttime" => "%{[excute_time][0]}" }
add_field => { "cuttime_last" => "%{[excute_time][1]}" }
}
# mutate {
# convert => { "cuttime" => "integer" }
# }
}
mutate { convert => { "cuttime" => "integer" } }
}
if "exuc" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:times}\s%{USER:server_name}\s*%{INT:level}\s(?(.*))\s(?(.*))\s\[%{GREEDYDATA:message_value}\]" }
}
date {
match => ["times", "ISO8601"]
remove_field => "times"
}
mutate{
convert => { "level" => "integer" }
}
if "(null)" not in [excute_time]{
mutate {
# add_field => { "twotime" => "%{excute_time}" }
split => [ "excute_time","."]
add_field => { "cuttime" => "%{[excute_time][0]}" }
add_field => { "cuttime_last" => "%{[excute_time][1]}" }
}
# mutate {
# convert => { "cuttime" => "integer" }
# }
}
mutate { convert => { "cuttime" => "integer" } }
}
if "json" in [tags]{
grok {
match => { "message" => "%{TIMESTAMP_ISO8601:times}\s*\S+\(.*))\>\s\S+\:\(.*))\,(?(.*))\>\s\S+%{GREEDYDATA:message_value}" }
}
date {
match => ["times", "ISO8601"]
# locale => "en"
# timezone => "+00:00"
remove_field => "times"
}
}
}
output {
# stdout {}
if "caserver" in [tags] {
elasticsearch {
hosts => ["10.100.15.23:9200","10.100.15.24:9200","10.100.15.25:9200"]
manage_template => false
index => "caserver-%{+yyyy.MM}"
user => "elastic"
password => "elk_123"
}
}
}