Logstash基礎操作-Filter

Grok配置案例：

##啟動文件配置：
#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{}
}
filter?{
grok?{
match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\
%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
???}
}
output?{
??stdout{
????codec?=>?"rubydebug"
??}
}
##輸出文件內容
172.16.213.132?[07/Feb/2018:16:24:19?+0800]?"GET?/?HTTP/1.1"?403?5039
##顯示內容
{
??????"@version"?=>?"1",
????"@timestamp"?=>?2019-11-10T06:02:42.865Z,
??????????"host"?=>?"localhost.localdomain",
???????"message"?=>?"172.16.213.132?[07/Feb/2018:16:24:19?+0800]?\"GET?/?HTTP/1.1\"?403?5039",
?????"timestamp"?=>?"07/Feb/2018:16:24:19?+0800",
?????????"bytes"?=>?"5039",
??????"response"?=>?"403",
??????"clientip"?=>?"172.16.213.132",
??????"referrer"?=>?"\"GET?/?HTTP/1.1\""
}

Grok 過濾重復字段

##?配置文件
#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{
?}
}
filter?{
??grok?{
??match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\?
??%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
??remove_field?=>?["message"]
???}
}
output?{
??stdout{
??codec?=>?"rubydebug"
??}
}

Grok搭配Date時間插件配置

#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{
??}
}
filter?{
grok?{
?match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\?
?%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
?remove_field?=>?["message"]
???}
date?{
??match?=>?["timestamp",?"dd/MMMM/yyyy:HH:mm:ss?Z"]
??}
}
output?{
??stdout{
??codec?=>?"rubydebug"
??}
}

Date 過濾重復得字段配置

#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{
??}
}
filter?{
?grok?{
???match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\?
???%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
???remove_field?=>?["message"]
???}
date?{
??match?=>?["timestamp",?"dd/MMMM/yyyy:HH:mm:ss?Z"]
??
??}
mutate?{
???remove_field?=>?[?"timestamp"?]??
??}
}
output?{
?stdout{
??codec?=>?"rubydebug"
??}
}

綜合練習配置參數

#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{
??}
}
filter?{
??grok?{
???match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\?
???%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
???remove_field?=>?["message"]
??}
?date?{
??match?=>?["timestamp",?"dd/MMMM/yyyy:HH:mm:ss?Z"]?
??}
?mutate{
????rename?=>?{"response"?=>?"response_new"}
????gsub?=>?["referrer",?"\"",?""]
????remove_field?=>?[?"timestamp"?]
????split?=>?["clientip",?"."]
??}
}
output?{
?stdout{
??codec?=>?"rubydebug"
??}
}

Geoip 地理位置插件操作方式

#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{
??}
}
filter?{
????grok?{
?????match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\?
?????%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
?????remove_field?=>?["message"]
???}
???date?{
????match?=>?["timestamp",?"dd/MMMM/yyyy:HH:mm:ss?Z"]?
??}
???mutate{
??????remove_field?=>?[?"timestamp"?]
??}
??geoip?{
????source?=>?"clientip"
????database?=>?"/usr/local/include/GeoLite2-ASN_20191105/GeoLite2-ASN.mmdb"
???}
}
output?{
??stdout{
????codec?=>?"rubydebug"
??}?
}

Geoip輸出指定屬性值

#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{
??}
}
filter?{
????grok?{
?????match?=>?["message","%{IP:clientip}\?\[%{HTTPDATE:timestamp}\]\?
?????%{QS:referrer}\?%{NUMBER:response}\?%{NUMBER:bytes}"]
?????remove_field?=>?["message"]
???}
???date?{
????match?=>?["timestamp",?"dd/MMMM/yyyy:HH:mm:ss?Z"]
??}
???mutate{
??????remove_field?=>?[?"timestamp"?]
??}
geoip?{
source?=>?"clientip"
#database?=>?"/usr/local/include/GeoLite2-Country_20191015/GeoLite2-Country.mmdb"
database?=>?"/usr/local/include/GeoLite2-City_20191105/GeoLite2-City.mmdb"
fields?=>?["city_name",?"region_name",?"country_name",?"ip",?"latitude",?"longitude",?"timezone"]
???}
}
output?{
??stdout{
????codec?=>?"rubydebug"
??}
}
模擬數據：
36.7.152.182?[07/Feb/2018:16:24:19?+0800]?"GET?/?HTTP/1.1"?403?5039

綜合實戰

#?Sample?Logstash?configuration?for?creating?a?simple
#?Beats?->?Logstash?->?Elasticsearch?pipeline.
input?{
??stdin{}
}
filter{
grok{
??match?=>?{"message"?=>?"%{TIMESTAMP_ISO8601:localtime}\|\~\|%{IP:clientip}
??\|\~\|%{GREEDYDATA:http_user_agent}\|\~\|%{GREEDYDATA:url}
??\|\~\|%{GREEDYDATA:mediaid}\|\~\|%{GREEDYDATA:osid}"}
??remove_field?=>?[?"message"?]
???}
date?{
????match?=>?["localtime",?"yyyy-MM-dd'T'HH:mm:ssZZ"]
????target?=>?"@timestamp"
???}
mutate?{
??????remove_field?=>?["localtime"]
???}
geoip?{
?source?=>?"clientip"
?#database?=>?"/usr/local/include/GeoLite2-Country_20191015/GeoLite2-Country.mmdb"
?database?=>?"/usr/local/include/GeoLite2-City_20191105/GeoLite2-City.mmdb"
?fields?=>?["city_name",?"region_name",?"country_name",?"ip",?"latitude",?"longitude",?"timezone"]
??}
}
output?{
???stdout?{
???codec?=>?"rubydebug"
???}
}
示例：2018-02-09T10:57:42+08:00|~|123.87.240.97|~|Mozilla/5.0
(iPhone;CPU?iPhone?OS?11_2_2?like?Mac?OS?X)
AppleWebKit/604.4.7?Version/11.0?Mobile/15C202?Safari/604.1
|~|http://m.sina.cn/cm/ads_ck_wap.html
|~|12434785489009|~|DF45566587855P