ELK技术栈—Logstash—Input插件
原文作者：歸來朝歌
原文地址：logstash之Input插件
?
1、stdin標(biāo)準(zhǔn)輸入和stdout標(biāo)準(zhǔn)輸出
Logsrtash含有兩個(gè)非常重要的基礎(chǔ)插件，input與output；首先執(zhí)行命令：
bin/logstash -e 'input { stdin { } } output { stdout { codec ? => rubydebug } }' ?程序啟動(dòng)之后輸入：hello logstash
2、監(jiān)控日志文件變化
Logstash 使用一個(gè)名叫?FileWatch?的 Ruby?Gem 庫來監(jiān)聽文件變化。這個(gè)庫支持 glob 展開文件路徑，而且會(huì)記錄一個(gè)叫?.sincedb的數(shù)據(jù)庫文件來跟蹤被監(jiān)聽的日志文件的當(dāng)前讀取位置。所以，不要擔(dān)心 logstash 會(huì)漏過你的數(shù)據(jù)（數(shù)據(jù)量并不是特別大的前提下）
創(chuàng)建logstash_conf文件(名字自定義)：monitor_file.conf
input{file{path => "/home/angel/servers/logstash-5.5.2/monitor_log/ "type => "log"start_position => "beginning"} } output{stdout{codec=>rubydebug} }啟動(dòng)之后，logstash就會(huì)監(jiān)控 /home/angel/servers/logstash-5.5.2/monitor_log/logs.txt文件，如果發(fā)現(xiàn)文件中有內(nèi)容，則以標(biāo)準(zhǔn)輸出的形式輸出到控制臺(tái)；
檢測(cè)logstash 配置文件
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_file.conf -t啟動(dòng)logstash：
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_file.conf測(cè)試：
echo "hello logstash" >> /home/angel/servers/logstash-5.5.2/monitor_log/logs.txt ?配置參數(shù)說明：
- Path=>表示監(jiān)控的文件路徑
- Type=>給類型打標(biāo)記，用來區(qū)分不同的文件類型。
- Start_postion=>從哪里開始記錄文件，默認(rèn)是從結(jié)尾開始標(biāo)記，要是你從頭導(dǎo)入一個(gè)文件就把改成”beginning”.
- discover_interval=>多久去監(jiān)聽path下是否有文件，默認(rèn)是15s
- exclude=>排除什么文件
- close_older=>一個(gè)已經(jīng)監(jiān)聽中的文件，如果超過這個(gè)值的時(shí)間內(nèi)沒有更新內(nèi)容，就關(guān)閉監(jiān)聽它的文件句柄。默認(rèn)是3600秒，即一個(gè)小時(shí)。
- sincedb_path=>監(jiān)控庫存放位置(默認(rèn)的讀取文件信息記錄在哪個(gè)文件中)。默認(rèn)在：/data/plugins/inputs/file。
- sincedb_write_interval=> logstash 每隔多久寫一次 sincedb 文件，默認(rèn)是 15 秒。
- stat_interval=>logstash 每隔多久檢查一次被監(jiān)聽文件狀態(tài)（是否有更新），默認(rèn)是 1 秒。
3、tcp插件
TCP是一種網(wǎng)絡(luò)傳輸控制協(xié)議，很多公司的數(shù)據(jù)不一定是在本地的，而是在傳輸網(wǎng)絡(luò)的；這個(gè)時(shí)候使用TCP建立連接后，通信雙方就可以進(jìn)行數(shù)據(jù)傳輸了；Logstash提供了TCP插件，TCP插件可以監(jiān)控某個(gè)端口，當(dāng)數(shù)據(jù)打入logstash監(jiān)聽的端口隊(duì)列的時(shí)候，logstash就可以進(jìn)行數(shù)據(jù)的采集;
| codec=> | 可選項(xiàng) （編碼解碼） |
| data_timeout=> | 可選項(xiàng) （超時(shí)時(shí)間，秒為單位。如果設(shè)置-1,則永不超時(shí)，默認(rèn)是5） |
| host=> | 可選項(xiàng) （主機(jī)地址，字符串類型，如"localhost"或者"192.168.0.1?） |
| mode=> | 可選項(xiàng) （值是["server","client"]其中之一,默認(rèn)是server） |
| port=> | 必填項(xiàng) （遠(yuǎn)程監(jiān)聽的端口） |
| ssl_cacert=> | 可選項(xiàng)，ssl認(rèn)證相關(guān) |
| ssl_cert=> | ? |
| ssk_key=> | ? |
| ssl_enable=> | 是否開啟ssl認(rèn)證 |
| tags=> | 可選項(xiàng) 用于增加一些標(biāo)簽，這個(gè)標(biāo)簽可能在后續(xù)的處理中起到標(biāo)志的作用 |
| type=> | 可選項(xiàng) 標(biāo)記事件類型,通過type判斷 |
vim monitor_tcp.conf
input{tcp {port => 9876mode => "server" ?//值是["server","client"]其中之一,默認(rèn)是server ?ssl_enable => false} } output{stdout{} }啟動(dòng)并且監(jiān)聽到9876端口;
檢測(cè)logstash 配置文件
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_tcp.conf -t啟動(dòng)logstash：
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_tcp.conf寫一段java的socket代碼向9876發(fā)送數(shù)據(jù)：
public static void main(String[] args) throws Exception{// 向服務(wù)器端發(fā)送請(qǐng)求，服務(wù)器IP地址和服務(wù)器監(jiān)聽的端口號(hào)Socket client = new Socket("hadoop01", 9876); ?// 通過printWriter 來向服務(wù)器發(fā)送消息PrintWriter printWriter = new PrintWriter(client.getOutputStream());System.out.println("連接已建立...");for(int i=0;i<10;i++){// 發(fā)送消息printWriter.println("hello logstash , 這是第"+i+" 條消息"); ?printWriter.flush();} ? }?
4、syslog插件
syslog**機(jī)制負(fù)責(zé)記錄內(nèi)核和應(yīng)用程序產(chǎn)生的日志信息，管理員可以通過查看日志記錄，來掌握系統(tǒng)狀況**。默認(rèn)系統(tǒng)已經(jīng)安裝了rsyslog.直接啟動(dòng)即可。
vim /etc/rsyslog.conf，增加如下代碼，將日志信息發(fā)送至6789端口：
*.* @@hadoop01:6789重啟日志：
/etc/init.d/rsyslog restartvim monitor_syslog.conf
input{tcp{port=> 6789type=> syslog}udp{port=> 6789type=> syslog} } ? filter{if [type] == "syslog" {grok {match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }add_field => [ "received_at", "%{@timestamp}" ]add_field => [ "received_from", "%{host}" ]}date {match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]}} } ? output{stdout{codec=> rubydebug} }啟動(dòng)：
bin/logstash -f /home/angel/servers/logstash-5.5.2/logstash_conf/monitor_syslog.conf通過telnet hadoop01 6789傳輸數(shù)據(jù)，【注意，在logstash中的grok是正則表達(dá)式，用來解析當(dāng)前數(shù)據(jù)】
Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154] Jun 05 08:00:00 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' denied Jun 05 08:10:00 louis CRON[620]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log) Jun 05 08:05:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'. ?日志截圖
5、logstash的codec監(jiān)控nginx日志
Codec 來自 Coder/decoder兩個(gè)單詞的首字母縮寫；Logstash 不只是一個(gè)input | filter | output 的數(shù)據(jù)流，而是一個(gè)input | decode | filter | encode | output?的數(shù)據(jù)流，codec 就是用來decode、encode 事件的。簡(jiǎn)單說，就是在logstash讀入的時(shí)候，通過codec編碼解析日志為相應(yīng)格式，從logstash輸出的時(shí)候，通過codec解碼成相應(yīng)格式。如果當(dāng)前的日志是codec的，可以避免使用grok，從而提高開發(fā)效率和性能
編輯nginx文件，設(shè)置log格式：
log_format json '{"@timestamp":"$time_iso8601",''"@version":"1",''"host":"$server_addr",''"client":"$remote_addr",''"size":$body_bytes_sent,''"responsetime":$request_time,''"domain":"$host",''"url":"$uri",''"status":"$status"}'; access_log logs/access_codec.log json;截圖：
Nginx生成kibana瀏覽器的日志：
server {listen 80;server_name hadoop01;location / {proxy_pass http://hadoop01:5601;proxy_set_header Upgrade $http_upgrade;proxy_set_header Connection 'upgrade';proxy_set_header Host $host;proxy_cache_bypass $http_upgrade;}}?
?
vim nginx_codec.conf
啟動(dòng)nginx：
/usr/local/nginx/sbin啟動(dòng)logstash：
bin/logstash -f /home/angel/logstash-5.5.2/logstash_conf/nginx_codec.conf6、合并多行數(shù)據(jù)（Multiline）
在處理日志時(shí)，除了訪問日志外，還要處理運(yùn)行時(shí)日志，該日志大都用程序?qū)懙?#xff0c;比如 log4j。運(yùn)行時(shí)日志跟訪問日志最大的不同是，運(yùn)行時(shí)日志是多行，也就是說，連續(xù)的多行才能表達(dá)一個(gè)意思。multiline 插件來說，有三個(gè)設(shè)置比較重要：negate、pattern 和 what
| negate | ? | 類型是boolean | ? | 默認(rèn)false | 如果沒有匹配，那么否定正則表達(dá)式 |
| Pattern | ?? | 類型是string | ?? | 沒有默認(rèn)值 | 要匹配的正則表達(dá)式 |
| What | previous或者next | 沒有默認(rèn)值 | 如果正則表達(dá)式匹配了，那么該事件是屬于（屬于下一次觸發(fā)的事件）還是上一個(gè)（觸發(fā)的事件） |
【注意】使用mutiline之前，需要安裝插件：
bin/logstash-plugin install logstash-filter-multiline //如果報(bào)錯(cuò)，請(qǐng)使用： bin/logstash-plugin install --version 3.0.4 logstash-filter-multiline//如果還是報(bào)錯(cuò)，請(qǐng)下載logstash-filter-multiline-3.0.4.gem，然后安裝 bin/logstash-plugin install /home/angel/logstash-5.5.2/vendor/bundle/jruby/1.9/cache/logstash-filter-multiline-3.0.4.gem- 百度搜索：logstash-filter-multiline-3.0.4.gem
- 將gem文件放入這個(gè)理：/home/angel/logstash-5.5.2/vendor/bundle/jruby/1.9/cache
- bin/logstash-plugin install /home/angel/logstash-5.5.2/vendor/bundle/jruby/1.9/cache/logstash-filter-multiline-3.0.4.gem
vim multiline.conf
input {file{path=>"/home/angel/logstash-5.5.2/logs/mutil/mu_log.out"type=>"runtimelog"codec=> multiline {pattern => "^\["negate => truewhat => "previous"}start_position=>"beginning"sincedb_path=>"/home/angel/logstash-5.5.2/logs/mutil/sincedb-access"ignore_older=>0} } filter{multiline{pattern => "^\["negate => truewhat => "previous"}date {match => [ "timestamp", "yyyy-MM-dd-HH:mm:ss" ]locale => "cn"}grok {match=>["message","\[%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level}\] %{GREEDYDATA:msg}"]} } output{stdout{codec => rubydebug} }啟動(dòng)
bin/logstash -f /home/angel/logstash-5.5.2/logstash_conf/multiline.conf向/home/angel/logstash-5.5.2/logs/mutil/mu_log.out文件中輸入測(cè)試日志：
echo "[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over. [16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null order by product_category.ORDERS asc [16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over. [16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over. [16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is nullorder by product_category.ORDERS desc [16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null order by product_category.ORDERS asc [16-04-12 03:40:07 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over" >> /home/angel/logstash-5.5.2/logs/mutil/mu_log.out?
?
超強(qiáng)干貨來襲 云風(fēng)專訪：近40年碼齡，通宵達(dá)旦的技術(shù)人生總結(jié)
以上是生活随笔為你收集整理的ELK技术栈—Logstash—Input插件的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： ELK技术栈—Kibana
- 下一篇： 并查集—岛屿数量