怎么將nginx日志導(dǎo)入elasticsearch

log_format?main????'$remote_addr?$http_x_forwarded_for?[$time_local]?$server_name?$request?'? ????????????'$status?$body_bytes_sent?$http_referer?'? ????????????'"$http_user_agent"?' ????????????'"$connection"?' ????????????'"$http_cookie"?' ????????????'$request_time?' ????????????'$upstream_response_time';

tar?-zxvf?filebeat-6.2.4-linux-x86_64.tar.gz?-c?/usr/local cd?/usr/local;ln?-s?filebeat-6.2.4-linux-x86_64?filebeat cd?/usr/local/filebeat

./filebeat?modules?enable?nginx

./filebeat?modules?list

vim?/usr/local/filebeat/blog_module_logstash.yml filebeat.modules: -?module:?nginx ?access: ??enabled:?true ??var.paths:?["/home/weblog/blog.cnfol.com_access.log"] ?#error: ?#?enabled:?true ?#?var.paths:?["/home/weblogerr/blog.cnfol.com_error.log"]   output.logstash: ?hosts:?["192.168.15.91:5044"]

./filebeat?-c?blog_module_logstash.yml?-e

tar?-zxvf?logstash-6.2.4.tar.gz?/usr/local cd?/usr/local;ln?-s?logstash-6.2.4?logstash 創(chuàng)建一個(gè)nginx日志的pipline文件 cd?/usr/local/logstash

vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns

forword?(?:%{ipv4}[,]?[?]?)+|%{word}

#input?{ #?stdin?{} #} #?從filebeat接受數(shù)據(jù) input?{ ?beats?{ ?port?=&gt;?5044 ?host?=&gt;?"0.0.0.0" ?} }  filter?{ ?#?添加一個(gè)調(diào)試的開(kāi)關(guān) ?mutate{add_field?=&gt;?{"[@metadata][debug]"=&gt;true}} ?grok?{ ?#?過(guò)濾nginx日志 ?#match?=&gt;?{?"message"?=&gt;?"%{nginxaccess_test2}"?} ?#match?=&gt;?{?"message"?=&gt;?'%{iporhost:clientip}?#?(?<http_x_forwarded_for>[^#]*)?#?[%{httpdate:[@metadata][webtime]}]?#?%{notspace:hostname}?#?%{word:verb}?%{uripathparam:request}?http/%{number:httpversion}?#?%{number:response}?#?(?:%{number:bytes}|-)?#?(?:"(?:%{notspace:referrer}|-)"|%{notspace:referrer}|-)?#?(?:"(?<http_user_agent>[^#]*)")?#?(?:"(?:%{number:connection}|-)"|%{number:connection}|-)?#?(?:"(?<cookies>[^#]*)")?#?%{number:request_time:float}?#?(?:%{number:upstream_response_time:float}|-)'?} ?#match?=&gt;?{?"message"?=&gt;?'(?:%{iporhost:clientip}|-)?(?:%{two_ip:http_x_forwarded_for}|%{ipv4:http_x_forwarded_for}|-)?[%{httpdate:[@metadata][webtime]}]?(?:%{hostname:hostname}|-)?%{word:method}?%{uripathparam:request}?http/%{number:httpversion}?%{number:response}?(?:%{number:bytes}|-)?(?:"(?:%{notspace:referrer}|-)"|%{notspace:referrer}|-)?%{qs:agent}?(?:"(?:%{number:connection}|-)"|%{number:connection}|-)?(?:"(?<cookies>[^#]*)")?%{number:request_time:float}?(?:%{number:upstream_response_time:float}|-)'?} ????match?=&gt;?{?"message"?=&gt;?'(?:%{iporhost:clientip}|-)?%{forword:http_x_forwarded_for}?[%{httpdate:[@metadata][webtime]}]?(?:%{hostname:hostname}|-)?%{word:method}?%{uripathparam:request}?http/%{number:httpversion}?%{number:response}?(?:%{number:bytes}|-)?(?:"(?:%{notspace:referrer}|-)"|%{notspace:referrer}|-)?%{qs:agent}?(?:"(?:%{number:connection}|-)"|%{number:connection}|-)?%{qs:cookie}?%{number:request_time:float}?(?:%{number:upstream_response_time:float}|-)'?} ?} ?#?將默認(rèn)的@timestamp(beats收集日志的時(shí)間)的值賦值給新字段@read_tiimestamp ?ruby?{? ?#code?=&gt;?"event.set('@read_timestamp',event.get('@timestamp'))" ?#將時(shí)區(qū)改為東8區(qū) ?code?=&gt;?"event.set('@read_timestamp',event.get('@timestamp').time.localtime?+?8*60*60)" ?} ?#?將nginx的日志記錄時(shí)間格式化 ?#?格式化時(shí)間?20/may/2015:21:05:56?+0000 ?date?{ ?locale?=&gt;?"en" ?match?=&gt;?["[@metadata][webtime]","dd/mmm/yyyy:hh:mm:ss?z"] ?} ?#?將bytes字段由字符串轉(zhuǎn)換為數(shù)字 ?mutate?{ ?convert?=&gt;?{"bytes"?=&gt;?"integer"} ?} ?#?將cookie字段解析成一個(gè)json ?#mutate?{ ?#?gsub?=&gt;?["cookies",';',','] ?#}? ?#?如果有使用到cdn加速http_x_forwarded_for會(huì)有多個(gè)ip，第一個(gè)ip是用戶(hù)真實(shí)ip ?if[http_x_forwarded_for]?=~?",?"{ ?????ruby?{ ?????????code?=&gt;?'event.set("http_x_forwarded_for",?event.get("http_x_forwarded_for").split(",")[0])' ????????} ????} ?#?解析ip，獲得ip的地理位置 ?geoip?{ ?source?=&gt;?"http_x_forwarded_for" ?#?#?只獲取ip的經(jīng)緯度、國(guó)家、城市、時(shí)區(qū) ?fields?=&gt;?["location","country_name","city_name","region_name"]? ?} ?#?將agent字段解析,獲得瀏覽器、系統(tǒng)版本等具體信息 ?useragent?{ ?source?=&gt;?"agent" ?target?=&gt;?"useragent" ?} ?#指定要?jiǎng)h除的數(shù)據(jù) ?#mutate{remove_field=&gt;["message"]} ?#?根據(jù)日志名設(shè)置索引名的前綴 ?ruby?{ ?code?=&gt;?'event.set("@[metadata][index_pre]",event.get("source").split("/")[-1])' ?}? ?#?將@timestamp?格式化為2019.04.23 ?ruby?{ ?code?=&gt;?'event.set("@[metadata][index_day]",event.get("@timestamp").time.localtime.strftime("%y.%m.%d"))' ?} ?#?設(shè)置輸出的默認(rèn)索引名 ?mutate?{ ?add_field?=&gt;?{ ??#"[@metadata][index]"?=&gt;?"%{@[metadata][index_pre]}_%{+yyyy.mm.dd}" ??"[@metadata][index]"?=&gt;?"%{@[metadata][index_pre]}_%{@[metadata][index_day]}" ?} ?} ?#?將cookies字段解析成json #?mutate?{ #?gsub?=&gt;?[ #??"cookies",?";",?",", #??"cookies",?"=",?":" #?] #?#split?=&gt;?{"cookies"?=&gt;?","} #?} #?json_encode?{ #?source?=&gt;?"cookies" #?target?=&gt;?"cookies_json" #?} #?mutate?{ #?gsub?=&gt;?[ #??"cookies_json",?',',?'","', #??"cookies_json",?':',?'":"' #?] #?} #?json?{ #?source?=&gt;?"cookies_json" #?target?=&gt;?"cookies2" #?} ?#?如果grok解析存在錯(cuò)誤，將錯(cuò)誤獨(dú)立寫(xiě)入一個(gè)索引 ?if?"_grokparsefailure"?in?[tags]?{ ?#if?"_dateparsefailure"?in?[tags]?{ ?mutate?{ ??replace?=&gt;?{ ??#"[@metadata][index]"?=&gt;?"%{@[metadata][index_pre]}_failure_%{+yyyy.mm.dd}" ??"[@metadata][index]"?=&gt;?"%{@[metadata][index_pre]}_failure_%{@[metadata][index_day]}" ??} ?} ?#?如果不存在錯(cuò)誤就刪除message ?}else{ ?mutate{remove_field=&gt;["message"]} ?} }  output?{ ?if?[@metadata][debug]{ ?#?輸出到rubydebuyg并輸出metadata ?stdout{codec?=&gt;?rubydebug{metadata?=&gt;?true}} ?}else{ ?#?將輸出內(nèi)容轉(zhuǎn)換成?"." ?stdout{codec?=&gt;?dots}? ?#?將輸出到指定的es ?elasticsearch?{ ??hosts?=&gt;?["192.168.15.160:9200"] ??index?=&gt;?"%{[@metadata][index]}" ??document_type?=&gt;?"doc" ?}? ?} }</cookies></cookies></http_user_agent></http_x_forwarded_for>

nohup?bin/logstash?-f?test_pipline2.conf?&