Logstash                                                                                                                                           

-n 指定logstash例項名稱，如果沒有指定，預設是本地主機名
-f 從指定的檔案或資料夾中載入logstash的配置；如果是一個資料夾，它會載入裡面所有的檔案，或者可以加上萬用字元只加載特定格式的檔案
-w 允許filter和output的pipeline執行緒數量，預設是CPU核數
-b 每個 Logstash pipeline 執行緒，在執行具體的 filter 和 output 函式之前，最多能累積的日誌條數，預設是 125 條。越大效能越好，同樣也會消耗越多的 JVM 記憶體
-u 每個 Logstash pipeline 執行緒，在打包批量日誌的時候，最多等待幾毫秒。預設是 5 ms
-l 指定日誌輸出位置
-r 監控配置配置檔案，如果有變化則自動過載logstash
-e 使用給定的命令列字串作為配置，直接執行bin/log/logstash -e 配置預設是"input { stdin { type => stdin } }" "output { stdout { codec => rubydebug } }"
-t 檢查配置檔案是否有語法錯誤
-V 列印logstash版本
--log.level 指定日誌等級，包括trace、debug、info、warn、error、fatal，預設info
--http.host 指定web API繫結的主機，預設127.0.0.1
--http.port 指定web API的http埠，預設9600至9700

#############################後臺啟動logstash#############################
1.安裝supervisor

yum -y install supervisor --enablerepo=epel

2.在/etc/supervisord.conf配置檔案最後新增:

[program:logstash]
directory=/usr/local/logstash
command=/usr/local/logstash/bin/logstash -f /usr/local/logstash/etc/conf.d/ -r -l /var/log/logstash

3.啟動

service supervisord start

4.單獨控制一個子程序

supervisorctl stop logstash

#############################編解碼配置外掛#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/codec-plugins.html
#logstash不只是一個input|filter|output的資料流，而是一個input|decode|filter|encode|output的資料流。codec就是用來decode、encode事件的。
1.JSON編解碼
#直接輸入預定義好的JSON資料，可以省略filter/grok配置，降低logstash過濾的CPU負載消耗

input {
    file {
        path => "/opt/data/logs/bjape01-ngx-a1-172.16.3.2/nginx_access.log"
        codec => "json"
    }
}

2.multiline多行事件編碼

codec => multiline {
    pattern => "^\["
    negate => true
    what => "previous"
}

#pattern 要匹配的正則表示式，字串型別
#negate  正則表示式是否生效，布林型別，預設為flase
#what 未匹配的內容是向前合併還是向後後合併,previous,next兩個值選擇
#示例配置可以用於Log4j方式的應用程式日誌

3.line面向行的文字資料

4.plain空的純文字編解碼
#使用已經有定義框架輸入或輸出，例如logstash輸出轉存到elasticsearch

output {
        if [type] == "user_audit" {
            elasticsearch {
                            hosts => ["172.16.1.25","172.16.1.26","172.16.1.27"] 
                            index => 'user_audit-%{+YYYY-MM-dd}'
                            codec=>plain{charset=>"UTF-8"}
                          }
                    }
        }

#############################輸入外掛#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/input-plugins.html
1.標準輸入
示例：

input {
    stdin {
        add_field => {"key" => "value"}
        codec => "plain"
        tags => ["add"]
        type => "std"
    }
}

#add_field 向事件新增一個欄位，hash型別
#codec 設定編碼方式，預設是line
#tags 新增標記
#type 新增型別

2.檔案輸入

input {
    file {
        path => [
                "/opt/data/logs/idca-web1-172.16.3.2/apache_access.log",
                "/opt/data/logs/idca-web1-172.16.3.2/apache_error.log"
                ]
        stat_interval => 1
        discover_interval => 1
        type => "apache_log"
    }
}

#path 處理的檔案的路徑, 可以定義多個路徑
#stat_interval 每隔多久檢查一次被監聽檔案狀態（是否更新），預設是1秒
#discover_interval 每隔多久去檢查一次被監聽的path下是否有新檔案，預設是15秒
#start_position 從什麼位置開始讀取檔案資料，"beginning"從頭開始讀取，讀到最後一行不會終止，繼續tailf；"end"從結束位置tailf。預設值是"end"
    #坑：start_position僅在該檔案從未被監聽過的時候起作用，如果sincedb檔案中已經有這個檔案的inode記錄了，那麼logstash依然會衝記錄過的pos開始讀取資料。所以重複測試的時候每回需要刪除sincedb檔案，該檔案一般在安裝目錄下的data/plugins/inputs/file/中
    #優化：在file中新增sincedb_path => "/dev/null"，可以直接將sincedb寫到黑洞中
#exclude 不想被監聽的檔案可以排除出去
#sincedb_path 配置sincedb檔案位置
#sincedb_write_interval 每隔多久寫一次sincedb檔案，預設15秒

3.syslog輸入

input {
    syslog {
        port => "514"
    }
}

4.collectd輸入

#############################過濾器配置#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/filter-plugins.html
1.grok正則捕獲
#用於匹配對應格式的message，然後過濾出來

filter {
  grok {
    break_on_match => false
    patterns_dir => ["/usr/local/logstash/etc/conf.d/patterns"]
    match => { "message" => "\[%{TIMESTAMP_ISO8601:time}\] \[%{GLASSFISHVERSION:gfversion}\] %{OTHER}"}
  }
   grok {
    break_on_match => false
    patterns_dir => ["/usr/local/logstash/etc/conf.d/patterns"]
    match => {"path" => "/%{USER}/%{USER}/%{USER}/%{USER:host}/%{OTHER}" }
    overwrite => [ "host" ]
  }
}

#break_on_match 表示匹配即停止，預設為true，如果希望grok嘗試所有模式，設定為false
#patterns_dir 將grok表示式統一寫到檔案中，該選項指定檔案目錄。grok表示式可以寫到配置檔案中，但是日誌格式可能有多種，每種都寫一行自己的表示式就可能會出問題，所以建議將所有的grok表示式統一寫到一個地方。

#grok表示式格式：

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
#第一行用普通的正則表示式來定義一個grok表示式；第二行用以定義好的grok表示式來定義另一個grok表示式。logstash內建了多種grok表示式，檢視https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns，同樣可以自定義grok表示式。

%{NUMBER:num}
%{NUMBER:num:int}
#grok表示式的列印複製格式，第一行中NUMBER表示匹配的模式，num是匹配內容的標識，代表匹配的內容；第二行中int表示轉換匹配內容的格式，預設是字串格式。

#overwrite用來重寫message欄位，上面例子第二個grok中match的是一個路徑，但是我們只想保留路徑中的一個host欄位，所以overwrite => [ "host" ]就可以了
 
2.date時間處理
#日誌產生到logstash處理一般都會有一段時間差，導致logstash實際處理時間都比日誌產生時間晚。logstash-filter-date外掛可以解析日誌中的時間，變成LogStash:Timestamp物件，然後轉存到@timestamp欄位裡，作為事件的時間戳；如果未使用date外掛或沒有解析到時間，logstash將使用日誌輸入的時間作為事件的時間戳。

date {
      match => [ "time", "MMM dd yyyy HH:mm:ss",
            "MMM  d yyyy, HH:mm:ss", "ISO8601" ]
    }

#配置選項包括：locale、match、tag_on_failure、target、timezone。常用的是match，值的型別同樣是列表，預設是[]
#例項中的time就是grok中匹配欄位的標識，由我們自定義。
#date外掛支援五種時間格式，常用的是ISO8601和Joda-Time庫

ISO8601：類似"2018-10-19T14:25:25.905+0800"，這是我們的北京時間，如果是UTC時間後面跟的是一個大寫的Z（時區偏移），nginx、apache、java應用都可以設定為這種格式。
UNIX：UNIX時間戳格式，記錄從1970年起至今的總秒數。
UNIX_MS：從1970年至今的總毫秒數。
TAI64N：tai64n格式，很少用。
Joda-Time庫：y年，yyyy例如2018，yy例如18；M月，MMMM例如January，MMM例如Jan，MM例如01，M例如1；d日，dd例如01，d例如1；H時，HH例如01，H例如1；m分，mm例如01，m例如1；s秒，ss例如01，s例如1

#############################輸出外掛#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/output-plugins.html
1.輸出到Elasticsearch

output {
    elasticsearch {
        hosts => ["172.16.1.25","172.16.1.26","172.16.1.27"]
        index => "%{type}-%{+YYYY-MM-dd}"
        codec=>plain{charset=>"UTF-8"}
    }
}

#hosts Elasticsearch地址，列表形式，可以寫多個。
#index 索引名，寫入Elasticsearch索引的名稱，可以使用變數。注意：%{+YYYY-MM-dd}}在語法解析時，看到已+開頭的，就會自動認為後面是時間格式，所以之前處理過程中不要給自定義欄位取加號開頭的名字；索引名中不要有大寫字母，否則Elasticsearch會報InvalidindexNameException