[TOC] ## logstash模塊組成 Logstash由三個組件構造成，分別是input、filter以及output。三個組件的工作流為： | 組件 | 功能 | | --- | --- | | input | 收集數據 | | filter | 處理數據 | | output | 輸出數據 | filter過濾器插件非必須，但filter插件才能體現logtash的強大。 ~~~yml input{ 　　輸入插件 } filter{ 　　過濾器插件 } outer{ 　　輸出插件 } ~~~ add\_field、remove\_field、add\_tag、remove\_tag 是所有 Logstash 插件都有。相關使用方法看字段名就可以知道。 ### 常用啟動參數 |參數 |說明 |舉例| | --- | --- | --- | |-e |用命令行里的配置參數啟動實例 |./bin/logstash -e ‘input {stdin {}} output {stdout {}}’| |-f |指定啟動實例的配置文件 |./bin/logstash -f config/test.conf| |-t |測試配置文件的正確性 |./bin/logstash-f config/test.conf -t| |-l |指定日志文件名稱 |./bin/logstash-f config/test.conf -l logs/test.log| |-w |指定filter線程數量(5) |./bin/logstash-f config/test.conf -w 8| ### 數據類型 | 類型 | 示例 | | --- | --- | | bool | debug => true | | bytes | my\_bytes => "113" # 113 bytes | | string | host => "hostname" | | number | port => 214 | | array | match =>\[ "/var/log/messages", "/var/log/\*.log" \] | | hash | options => {key1 => "value1",key2 => "value2" } | ### 條件判斷 logshash支持if...else的條件判斷,支持下面的操作符： 等性：==, !=, <, >, <=, >= 正則：=~, !~ 包含：in, not in 布爾：and, or, nand, xor 取反：! ## input模塊 `input`[插件官方詳解:](https://www.elastic.co/guide/en/logstash/current/input-plugins.html) ### input是怎么樣接收日志的 logstash使用一個名為filewatch的ruby gem庫來監聽文件變化，讀取進度記錄到一個叫.sincedb的數據庫文件中。 這個文件的默認路徑在`plugins/inputs/file`下面。 ### 從file-stdin-beat-redis讀取示例 logstash的input模塊可以從各種輸入類型獲取數據,如文件(file),日志(log),數據庫(mysql,reids)等,各個輸入類型涉及的內部關鍵字不一樣,下面以常用的幾個輸入類型講解 **從文件`file`輸入** ~~~ input{ file{ path => ["/var/log/messages"] 　　} } #其他選項 path 必須的選項，文件路徑,可以多個,中括號包圍 exclude 排除不想被監聽的文件 close_older 已經監聽的文件多久未更新就關閉監聽(3600秒 1小時) ignore_older 檢查文件列表時，若有文件的最后修改時間超過(84600秒 一天)就忽略 discover_interval 多久去查一次被監聽的path下是否有新文件(15秒) sincedb_path 自定義 sincedb 文件到其他位置。 sincedb_write_interval 每隔多久寫一次 sincedb 文件(15秒) stat_interval 被監聽文件的檢查頻率(1秒) start_position 從什么位置開始讀取文件數據,值"end(默認)/beginning" ~~~ **標準輸入`stdin{}`** ```sh input{ stdin{ type => "stdin" } } ``` **filebeat輸入** ~~~ input { beats { hosts => "10.0.0.11" port => "5044" } } ~~~ **redis輸入** ~~~ input { redis { data_type => "list" key => "filebeat-1011" host => "10.0.0.11" port => 6379 password => 'abcd1234e' db => "2" threads => 5 codec => "json" } } ~~~ ### input通用參數 下列6個參數,是所有input插件都支持的參數 ```sh codec 輸入編解碼器,在數據輸入之前解碼數據(line) add_field 向輸入數據中添加一個K/V字段 enable_metric 禁用或啟用日志記錄 id 設置stdin的ID,不設置隨機生成,多個時stdin有用 tags 給數據數據添加標簽,方便后續處理 type 向此輸入的數據添加一個字段,主要用于過濾器 ``` `codec`參數,在input/output中都有,接下來專門講解 ## CODEC編碼插件 編碼插件（codec）可以在logstash輸入或輸出時處理不同類型的數據，同時，還可以更好更方便的與其他自定義格式的數據產品共存，比如：fluent、netflow、collectd等通用數據格式的其他產品。 因此，logstash不只是一個input-->filter-->output的數據流，而且是一個input-->decode-->filter-->encode-->output的數據流。 常用的codec插件有plain，json，multiline等 ### **plain插件：** plain是最簡單的編碼插件，你輸入什么信息，就返回什么信息 ~~~sh input { stdin { codec => plain {} } } ~~~ ### **json插件：** 有時候logstash采集的日志是JSON格式，那我們可以在input字段加入codec => json來進行解析，這樣就可以根據具體內容生成字段，方便分析和儲存。 如果想讓logstash輸出為json格式，可以在output字段加入codec=>json。 如果數據為json格式，可直接使用該插件，從而省掉filter/grok的配置，降低過濾器的cpu消耗 ~~~sh input { stdin { codec => json } } ~~~ ### **multiline插件：** 用于合并多行數據 java類程序,一條有用的數據可能輸出在很多行,分析日志時得把這些日志按一行處理,multiline插件用于解決此類問題。 例:tomcat的日志catalina.out有很多調試的日志，日志都以時間戳格式"20-Apr-2016 11:29:28.535"開頭，那么我們可以配置如下： ~~~sh input { file{ path => "xxx/tomcat/ogs/catalina.out" tags => ["api-core"] codec => multiline { pattern => "^\d{2}\:\d{2}\:\d{2}\.\d{3}" auto_flush_interval => 10 negate => true what => "previous" } stat_interval => "1" } } ~~~ pattern 為正則表達式匹配 negate true表示不匹配正則表達式。默認false what 如果匹配，事件屬于previous(上一個)或者next(下一個)事件 auto_flush_interval 多少秒沒有新數據，之前積累的多行數據轉換為一個事件 >以上的配置解釋：不匹配pattern時間戳格式開頭的行，都歸屬到上一個事件中，等待10秒沒有新數據產生，那么最后一個時間戳格式后的所有行數據就是最后一個事件。 ## output模塊 [output官方插件地址:](https://www.elastic.co/guide/en/logstash/6.8/output-plugins.html) Logstash的output模塊,相比于input模塊來說是一個輸出模塊,output模塊集成了大量的輸出插件,可以輸出到指定文件,也可輸出到指定的網絡端口,當然也可以輸出數據到ES. ### output通用參數 | 參數名 | 參數說明 | | --- | --- | | workers | 設置輸出線程數量 | | codec | 輸出編解碼器,對輸出的數據編碼(line) | | enable_metric | 禁用或啟用日志記錄| ### 輸出到標準輸出(stdout) 標準輸出默認使用`rubydebug`,還可以使用`json` ~~~bash #使用rubydebug output { stdout { codec => rubydebug } } #使用json output { stdout { codec => json } } ~~~ ### 輸出到redis ```sh output { redis{ codec => plain data_type => list db => 3 host => ["10.0.0.11:6379"] key => xxx password => xxx port => 6379 timeout => 5 } } ``` ### 輸出到Elasticsearch 到es的輸出參數有很多,但重要的就是hosts和index,其他參數可以參考官方文檔: [es參數](https://www.elastic.co/guide/en/logstash/6.8/plugins-outputs-elasticsearch.html) ```sh output { elasticsearch { hosts => ["10.0.0.12:9200"] index => logstash-%{+YYYY.MM.dd} } } ``` ### 增加條件的判斷的output 在logstash進行output時,可以根據type和tag值不同,做不同的條件判斷,輸出不同的數據,如: ```sh output{ if "api-01" in [tags] { elasticsearch { hosts => ["10.0.0.12:9200"] index => "pro-api-01-%{+YYYY.MM.dd}" } } if "api-02" in [tags] { elasticsearch{ hosts => ["10.0.0.12:9200"] index => "pro-api-02-%{+YYYY.MM.dd}" } } } ```