1. Input Plugin

1.1 從檔案輸入

從檔案讀取資料，如常見的日誌檔案。檔案讀取通常要解決幾個問題：

1.1.1 logstash-input-file配置

配置：

1.1.2 path

例如：

• /var/log/*.log：匹配/var/log目錄下以.log結尾的所有檔案
• /var/log/**/*.log：匹配/var/log所有子目錄下以.log結尾的檔案
• /var/log/{app1,app2,app3}/*.log：匹配/var/log目錄下app1，app2，app3子目錄中以.log結尾的檔案

1.1.3 sincedb

通過logstash-input-file外掛匯入了一些本地日誌檔案時，logstash會通過一個名為sincedb的獨立檔案中來跟蹤記錄每個檔案中的當前位置。這使得停止和重新啟動Logstash成為可能，並讓它在不丟失在停止Logstashwas時新增到檔案中的行數的情況下繼續執行。sincedb預設儲存在當前使用者的Home目錄下。

在除錯的時候，我們可能希望取消sincedb的記錄功能，使檔案每次都能從頭開始讀取。此時，我們可以這樣來做：

input {
    file {
        path => ["檔案路徑"]
        start_position => "beginning"
        sincedb_path => "/dev/null"
    }
}

1.2 從HTTP輸入

input {
  http {
    port => 埠號
  }
}

1.3 從TCP輸入

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 埠號
    codec => json_lines
  }
}

2. Codec Plugin

Codec Plugin作用於Input和Output Plugin，負責將資料在原始與Logstash Event之間轉換，常見的codec有：

• plain：讀取原始內容
• dots：將內容簡化為點進行輸出
• rubydebug：將Logstash Events按照ruby格式輸出，方便除錯
• line：處理帶有換行符的內容
• json：處理json格式的內容
• multiline：處理多行資料的內容

2.1 multiline

當一個Event的message由多行組成時，需要使用multiline，常見的情況是處理日誌資訊，如下所示：

Exception in thread "main" java.lang.NullPointerException
	at com.example.myproject.Book.getTitle(Book.java:16)
	at com.example.myproject.Author.getBookTitles(Author.java:25)
	at com.example.myproject.Bootstrap.main(Bootstrap.java:14)

主要設定引數如下：

例如，匹配日誌檔案：

input {
	file {
		path => ["檔案路徑"]
       start_position => "beginning"
		codec => multiline {
			pattern => "^\s" #匹配以空格開頭的行
			what => "previous"
		}
	}
}

3. Filter Plugin

Filter是Logstash功能強大的主要原因，它可以對Logstash Event進行豐富的處理，比如解析資料、刪除欄位、型別轉換等等，常見的有如下幾個：

3.1 date外掛

date外掛可以將日期字串解析為日期型別，然後替換@timestamp欄位或者指定其他欄位：

filter {
	date {
		match => ["logdate", "MM dd yyy HH:mm:ss"]
	}
}

通過以上配置，我們可以將

{
	"logdate": "Jan 01 2018 12:02:2018"
	"target": "@timestamp"
}

轉換為：

{
	"@version" => "1"
	"host" => "..."
	"logdate" => Jan01 2018 12:02:08
	"@timestamp" => "2018-01-01T04:02:08.000Z"
}

3.2 grok外掛

grok外掛用於解析日誌內容，例如我們要將下面這一條日誌解析成json陣列。

83.149.9.216 [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

那麼我們就可以這樣來配置grok：

%{IPORHOST:clientip} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{DATA:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} (?:-|%{NUMBER:bytes:int}) %{QS:referrer} %{QS:agent}

grok語法：

例子：%{SYNTAX:SEMANTIC}
SYNTAX為grok pattern的名稱，SEMANTIC為賦值欄位名稱。%{NUMBER:duration}可以匹配數值型別，但是grok匹配出的內容都是字串型別，可以通過在最後指定為int或者float來強轉型別：%{NUMBER:duration:int}

grok內建pattern：logstash-patterns-core

自定義grok pattern
我們通過pattern_definitions引數，以鍵值對的方式定義pattern名稱和內容。也可以通過pattern_dir引數，以檔案的形式讀取pattern。

filter {
	grok {
		match => {
			"message" => "%{SERVICE:service}"
		}
		pattern_definitions => {
			"SERVICE" => "[a-z0-9]{10,11}"
		}
	}
}

3.3 dissect外掛

基於分隔符原理解析資料，解決grok解析時消耗過多cpu資源的問題。

同樣的日誌資訊，dissect可以這樣來寫：

83.149.9.216 [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

dissect配置：

%{clientip} [%{timestamp}] "%{request}" %{response} %{bytes} "%{referrer}" "%{agent}"

但是，正因為dissect語法簡單，因此它能處理的場景比較有限。它只能處理格式相似，且有分隔符的字串。它的語法如下：

• %{}裡面是欄位
• 兩個%{}之間是分隔符。

例子1：
有以下日誌：

Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool

我想要把前面的日期和時間解析到同一個欄位中，那麼就可以這樣來做：

filter {
    dissect {
        mapping => {
        	"message" => "%{ts} %{+ts} %{+ts} %{src} %{prog}[%{pid}]: %{msg}"
        }
    }
}

例子2：
如果有以下日誌：

name=hushukang&age=28

我想要把它解析成：

{
	"name": "hushukang",
	"age": 28
}

那麼，可以這樣來寫dissect:

filter {
    dissect {
        mapping => {
            "message" => "%{?key1}=%{&key1}&%{?key2}=%{&key2}"
        }
        convert_datatype => {
            age => "int"
        }
    }
}

%{?}代表忽略匹配值，但是賦予欄位名，用於後續匹配用。%{&}代表將匹配值賦到指定欄位中。convert_datatype可以將指定字串轉為int或者float型別。

3.4 mutate外掛

mutate是使用最頻繁的外掛，可以對欄位進行各種操作，比如重新命名、刪除、替換、更新等，主要操作如下：

• convert型別轉換
• gsub字串替換
• split、join、merge字串切割、數組合併為字串、數組合併為陣列
• rename欄位重新命名
• update、replace欄位內容更新或替換。它們都可以更新欄位的內容，區別在於update只在欄位存在時生效，而replace在欄位不存在時會執行新增欄位的操作
• remove_field刪除欄位

mutate {
    convert => {"age" => "integer"}
    gsub => [
        "birthday", "-", "/",
        "image", "\/", "_"
    ]
    split => {"hobities" => ","}
    rename => {"hobities" => "hobityList"}
    replace => {"icon" => "%{image}"}
    remove_field => ["message"]
}

3.5 json外掛

將欄位內容為json格式的資料解析出來。
語法如下：

filter {
	json {
		source => "message"
		target => "msg_json"
	}
}

如果不指定target的話，那麼filter會把解析出來的json資料直接放到根級別。

3.6 geoip外掛

常用的外掛，根據ip地址提供對應的地域資訊，比如經緯度、城市名等，方便進行地理資料分析。

語法如下：

filter {
	geoip {
		source => "clientIp"
	}
}

4. Output Plugin

4.1 輸出到控制檯

多用於除錯

output {
	stdout {
		codec => rubydebug
	}
}

4.2 輸出到檔案

實現將分散在多地的檔案統一到一處的需求，比如將所有web機器的web日誌收集到1個檔案中，從而方便查閱資訊

output {
	file {
		path => "檔案路徑"
		codec => line {format => %{message}}
	}
}

4.3 輸出到elasticsearch

output {
  elasticsearch {
    hosts => ["http://192.168.3.12:9200"]
    index => "logstash-%{+YYYY.MM.dd}"
    document_type => "_doc"
    user => "使用者名稱"
    password => "密碼"
  }
}