Logstash 是一個開源的資料收集引擎，它具有備實時資料傳輸能力。它可以統一過濾來自不同源的資料，並按照開發者的制定的規範輸出到目的地。

一、原理

 Logstash 通過管道進行運作，管道有兩個必需的元素，輸入和輸出，還有一個可選的元素，過濾器。

輸入外掛從資料來源獲取資料，過濾器外掛根據使用者指定的資料格式修改資料，輸出外掛則將資料寫入到目的地。如下圖：

輸入：採集各種樣式、大小和來源的資料

資料往往以各種各樣的形式，或分散或集中地存在於很多系統中。Logstash 支援各種輸入選擇 ，可以在同一時間從眾多常用來源捕捉事件。能夠以連續的流式傳輸方式，輕鬆地從您的日誌、指標、Web 應用、資料儲存以及各種 AWS 服務採集資料。

過濾器：實時解析和轉換資料

資料從源傳輸到儲存庫的過程中，Logstash 過濾器能夠解析各個事件，識別已命名的欄位以構建結構，並將它們轉換成通用格式，以便更輕鬆、更快速地分析和實現商業價值。

Logstash 能夠動態地轉換和解析資料，不受格式或複雜度的影響：

• 利用 Grok 從非結構化資料中派生出結構
• 從 IP 地址破譯出地理座標
• 將 PII 資料匿名化，完全排除敏感欄位
• 整體處理不受資料來源、格式或架構的影響

輸出：選擇你的儲存，匯出你的資料

儘管 Elasticsearch 是我們的首選輸出方向，能夠為我們的搜尋和分析帶來無限可能，但它並非唯一選擇。

Logstash 提供眾多輸出選擇，您可以將資料傳送到您要指定的地方，並且能夠靈活地解鎖眾多下游用例。

二、安裝 

2.1、下載

Logstash 依賴 JDK1.8 ，因此在安裝之前請確保機器已經安裝和配置好 JDK1.8。

Logstash 下載：https://artifacts.elastic.co/downloads/logstash/logstash-6.5.1.rpm

2.2、安裝

yum install logstash-6.5.1.rpm

檢視啟動檔案來了解下檔案目錄

vim /etc/systemd/system/logstash.service

常用的目錄：

/etc/default/logstash
/etc/sysconfig/logstash
/usr/share/logstash/bin/logstash
/etc/logstash

2.3、設定執行許可權

• 方法1，修改

logstash.service

vim /etc/systemd/system/logstash.service 
User=root #直接以root賬戶允許     
Group=root

• 方法2

chown -R logstash:logstash /var/lib/logstash     
chown -R logstash:logstash /var/log/logstash 

2.4 、開機啟動 

systemctl start logstash.service      
systemctl enable logstash.service      
systemctl status logstash.service 

2.5、我們先來一個簡單的案例：

cd /usr/share/logstash
bin/logstash -e 'input { stdin { } } output { stdout {} }'

可能會有點慢

這時候輸入個Hello World

在生產環境中，Logstash 的管道要複雜很多，可能需要配置多個輸入、過濾器和輸出外掛。

因此，需要一個配置檔案管理輸入、過濾器和輸出相關的配置。配置檔案內容格式如下：

#　輸入
input {
  ...
}

# 過濾器
filter {
  ...
}

# 輸出
output {
  ...
}

根據自己的需求在對應的位置配置 輸入外掛、過濾器外掛、輸出外掛 和 編碼解碼外掛 即可。

三、外掛用法

 Logstash 每讀取一次資料的行為叫做事件。

在 Logstach 目錄中建立一個配置檔案，名為 logstash.conf（名字任意）。

cd /usr/share/logstash

3.1 輸入外掛

輸入外掛允許一個特定的事件源可以讀取到 Logstash 管道中，配置在 input {} 中，且可以設定多個。

修改配置檔案：

input {
    # 從檔案讀取日誌資訊
    file {
        path => "/var/log/elasticsearch/"
        type => "elasticsearch"
        start_position => "beginning"
    }
}

# filter {
#
# }

output {
    # 標準輸出
    stdout { codec => rubydebug }
}

其中，messages 為系統日誌。

 儲存檔案。鍵入：

bin/logstash -f logstash.conf

3.2 輸出外掛

輸出外掛將事件資料傳送到特定的目的地，配置在 output {} 中，且可以設定多個。

修改配置檔案：

input {
    # 從檔案讀取日誌資訊
    file {
        path => "/var/log/error.log"
        type => "error"
        start_position => "beginning"
    }
    
}

# filter {
#
# }

output {
    # 輸出到 elasticsearch
    elasticsearch {
        hosts => ["192.168.50.70:9200"]
        index => "error-%{+YYYY.MM.dd}"
    }
}

儲存檔案。鍵入：

bin/logstash -f logstash.conf

3.3 編碼解碼外掛

編碼解碼外掛本質是一種流過濾器，配合輸入外掛或輸出外掛使用。

從上圖中，我們發現一個問題：Java 異常日誌被拆分成單行事件記錄到 Elasticsearch 中，這不符合開發者或運維人員的檢視習慣。因此，我們需要對日誌資訊進行編碼將多行事件轉成單行事件記錄起來。

我們需要配置 Multiline codec 外掛，這個外掛可以將多行日誌資訊合併成一行，作為一個事件處理。

Logstash 預設沒有安裝該外掛，需要開發者自行安裝。鍵入：

bin/logstash-plugin install logstash-codec-multiline

修改配置檔案：

input {
    # 從檔案讀取日誌資訊
    file {
        path => "/var/log/error.log"
        type => "error"
        start_position => "beginning"
        # 使用 multiline 外掛
        codec => multiline {
            # 通過正則表示式匹配，具體配置根據自身實際情況而定
            pattern => "^\d"
            negate => true
            what => "previous"
        }
    }

}

# filter {
#
# }

output {
    # 輸出到 elasticsearch
    elasticsearch {
        hosts => ["192.168.2.41:9200"]
        index => "error-%{+YYYY.MM.dd}"
    }
}

儲存檔案。鍵入：

bin/logstash -f logstash.conf

4.4 過濾器外掛

過濾器外掛位於 Logstash 管道的中間位置，對事件執行過濾處理，配置在 filter {}，且可以配置多個。

本次測試使用 grok 外掛演示，grok 外掛用於過濾雜亂的內容，將其結構化，增加可讀性。

安裝：

bin/logstash-plugin install logstash-filter-grok

修改配置檔案：

input {
     stdin {}
}


filter {
     grok {
       match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER
:duration}" }
     }
}


output {
     stdout {
        codec => "rubydebug"
     }
}

儲存檔案。鍵入：

bin/logstash -f logstash.conf

啟動成功後，我們輸入：

55.3.244.1 GET /index.html 15824 0.043

控制檯返回：

[[email protected] logstash-5.6.3]# bin/logstash -f logstash.conf 
Sending Logstash's logs to /root/logstash-5.6.3/logs which is now configured via log4j2.properties
[2017-10-30T08:23:20,456][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.3/modules/fb_apache/configuration"}
[2017-10-30T08:23:20,459][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.3/modules/netflow/configuration"}
[2017-10-30T08:23:21,447][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>"main", "pipeline.workers"=>1, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>125}
The stdin plugin is now waiting for input:
[2017-10-30T08:23:21,516][INFO ][logstash.pipeline        ] Pipeline main started
[2017-10-30T08:23:21,573][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
55.3.244.1 GET /index.html 15824 0.043
{
      "duration" => "0.043",
       "request" => "/index.html",
    "@timestamp" => 2017-10-30T15:23:23.912Z,
        "method" => "GET",
         "bytes" => "15824",
      "@version" => "1",
          "host" => "localhost.localdomain",
        "client" => "55.3.244.1",
       "message" => "55.3.244.1 GET /index.html 15824 0.043"
}

輸入的內容被匹配到相應的名字中。