Logstash 快速指南

---

---

Logstash - 簡介

Logstash 是一個基於過濾器/管道模式的工具，用於收集、處理和生成日誌或事件。它有助於集中和即時分析來自不同來源的日誌和事件。

Logstash 使用 JRuby 程式語言編寫，執行在 JVM 上，因此您可以在不同的平臺上執行 Logstash。它從幾乎所有型別的來源收集不同型別的資料，例如日誌、資料包、事件、事務、時間戳資料等。資料來源可以是社交資料、電子商務、新聞文章、CRM、遊戲資料、網路趨勢、金融資料、物聯網、移動裝置等。

Logstash 通用功能

Logstash 的通用功能如下：

• Logstash 可以從不同的來源收集資料併發送到多個目標。

• Logstash 可以處理所有型別的日誌資料，例如 Apache 日誌、Windows 事件日誌、網路協議上的資料、標準輸入的資料等等。

• Logstash 還可以處理 http 請求和響應資料。

• Logstash 提供各種過濾器，幫助使用者透過解析和轉換資料來找到更多有意義的資訊。

• Logstash 也可以用於處理物聯網中的感測器資料。

• Logstash 是開源的，並根據 Apache 許可證 2.0 版提供。

Logstash 關鍵概念

Logstash 的關鍵概念如下：

事件物件

它是 Logstash 中的主要物件，它封裝了 Logstash 管道中的資料流。Logstash 使用此物件來儲存輸入資料並在過濾器階段新增建立的額外欄位。

Logstash 為開發人員提供了事件 API 來操作事件。在本教程中，此事件以各種名稱引用，例如日誌資料事件、日誌事件、日誌資料、輸入日誌資料、輸出日誌資料等。

管道

它包含 Logstash 中從輸入到輸出的資料流階段。輸入資料進入管道，並以事件的形式進行處理。然後以使用者或最終系統期望的格式傳送到輸出目標。

輸入

這是 Logstash 管道的第一個階段，用於獲取 Logstash 中的資料以進行進一步處理。Logstash 提供各種外掛來從不同的平臺獲取資料。一些最常用的外掛是 - File、Syslog、Redis 和 Beats。

過濾器

這是 Logstash 的中間階段，事件的實際處理發生在此階段。開發人員可以使用 Logstash 預定義的正則表示式模式來建立序列，以區分事件中的欄位和可接受的輸入事件的條件。

Logstash 提供各種外掛來幫助開發人員將事件解析和轉換為所需結構。一些最常用的過濾器外掛是 - Grok、Mutate、Drop、Clone 和 Geoip。

輸出

這是 Logstash 管道的最後一個階段，輸出事件可以格式化為目標系統所需的結構。最後，它使用外掛將完全處理後的輸出事件傳送到目標。一些最常用的外掛是 - Elasticsearch、File、Graphite、Statsd 等。

Logstash 優點

以下幾點說明了 Logstash 的各種優點。

• Logstash 提供正則表示式模式序列來識別和解析任何輸入事件中的各種欄位。

• Logstash 支援各種 Web 伺服器和資料來源來提取日誌資料。

• Logstash 提供多個外掛來將日誌資料解析和轉換為任何使用者期望的格式。

• Logstash 是集中的，這使得它易於處理和收集來自不同伺服器的資料。

• Logstash 支援許多資料庫、網路協議和其他服務作為日誌事件的目標源。

• Logstash 使用 HTTP 協議，使使用者能夠升級 Elasticsearch 版本而無需同步升級 Logstash。

Logstash 缺點

以下幾點說明了 Logstash 的各種缺點。

• Logstash 使用 http，這會對日誌資料的處理產生負面影響。

• 使用 Logstash 有時可能有點複雜，因為它需要對輸入日誌資料有很好的理解和分析。

• 過濾器外掛不是通用的，因此使用者可能需要找到正確的模式序列以避免解析錯誤。

在下一章中，我們將瞭解 ELK 堆疊是什麼以及它如何幫助 Logstash。

Logstash - ELK 堆疊

ELK 代表 **Elasticsearch、Logstash** 和 **Kibana**。在 ELK 堆疊中，Logstash 從不同的輸入源提取日誌資料或其他事件。它處理事件，然後將其儲存在 Elasticsearch 中。Kibana 是一個 Web 介面，它訪問 Elasticsearch 中的日誌資料並將其視覺化。

Logstash 和 Elasticsearch

Logstash 提供輸入和輸出 Elasticsearch 外掛，用於讀取和寫入日誌事件到 Elasticsearch。Elasticsearch 作為輸出目標也由 Elasticsearch 公司推薦，因為它與 Kibana 相容。Logstash 透過 http 協議將資料傳送到 Elasticsearch。

Elasticsearch 提供批次上傳功能，這有助於將資料從不同的來源或 Logstash 例項上傳到集中的 Elasticsearch 引擎。ELK 比其他 DevOps 解決方案具有以下優點：

• ELK 堆疊更容易管理，並且可以擴充套件以處理 PB 級別的事件。

• ELK 堆疊架構非常靈活，並且它提供了與 Hadoop 的整合。Hadoop 主要用於存檔目的。Logstash 可以使用 Flume 直接連線到 Hadoop，Elasticsearch 提供了一個名為 **es-hadoop** 的聯結器來連線 Hadoop。

• ELK 的總擁有成本遠低於其替代方案。

Logstash 和 Kibana

Kibana 不直接與 Logstash 互動，而是透過資料來源（在 ELK 堆疊中是 Elasticsearch）。Logstash 從每個來源收集資料，Elasticsearch 以非常快的速度分析它，然後 Kibana 提供對該資料的可操作見解。

Kibana 是一個基於 Web 的視覺化工具，它幫助開發人員和其他人員分析 Logstash 在 Elasticsearch 引擎中收集的大量事件的變化。這種視覺化使得預測或檢視輸入源中錯誤或其他重要事件趨勢的變化變得容易。

Logstash - 安裝

要在系統上安裝 Logstash，我們應該按照以下步驟操作：

**步驟 1** - 檢查計算機中安裝的 Java 版本；它應該是 Java 8，因為它與 Java 9 不相容。您可以透過以下方式檢查：

在 Windows 作業系統 (OS) 中（使用命令提示符）：

> java -version 

在 UNIX OS 中（使用終端）：

$ echo $JAVA_HOME

**步驟 2** - 從以下地址下載 Logstash：

https://www.elastic.co/downloads/logstash.

• 對於 Windows OS，下載 ZIP 檔案。

• 對於 UNIX OS，下載 TAR 檔案。

• 對於 Debian OS，下載 DEB 檔案。

• 對於 Red Hat 和其他 Linux 發行版，下載 RPN 檔案。

• APT 和 Yum 實用程式也可以用於在許多 Linux 發行版中安裝 Logstash。

**步驟 3** - Logstash 的安裝過程非常簡單。讓我們看看如何在不同的平臺上安裝 Logstash。

**注意** - 不要在安裝資料夾中放置任何空格或冒號。

• **Windows OS** - 解壓 zip 包，Logstash 就安裝好了。

• **UNIX OS** - 將 tar 檔案解壓到任何位置，Logstash 就安裝好了。

$tar –xvf logstash-5.0.2.tar.gz

對於 Linux OS 使用 APT 實用程式：

• 下載並安裝公共簽名金鑰：

$ wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

• 儲存儲存庫定義：

$ echo "deb https://artifacts.elastic.co/packages/5.x/apt stable main" | sudo
   tee -a /etc/apt/sources.list.d/elastic-5.x.list

• 執行更新：

$ sudo apt-get update

• 現在您可以使用以下命令進行安裝：

$ sudo apt-get install logstash

**對於 Debian Linux OS 使用 YUM 實用程式** -

• 下載並安裝公共簽名金鑰：

$ rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

• 在“/etc/yum.repos.d/”目錄中帶有 .repo 字尾的檔案中新增以下文字。例如，**logstash.repo**

[logstash-5.x]
name = Elastic repository for 5.x packages
baseurl = https://artifacts.elastic.co/packages/5.x/yum
gpgcheck = 1
gpgkey = https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled = 1
autorefresh = 1
type = rpm-md

• 您現在可以使用以下命令安裝 Logstash：

$ sudo yum install logstash

**步驟 4** - 轉到 Logstash 主目錄。在 bin 資料夾中，在 Windows 的情況下執行 **elasticsearch.bat** 檔案，或者您可以使用命令提示符和終端執行相同的操作。在 UNIX 中，執行 Logstash 檔案。

我們需要指定輸入源、輸出源和可選過濾器。為了驗證安裝，您可以使用標準輸入流 (stdin) 作為輸入源和標準輸出流 (stdout) 作為輸出源來執行它，使用基本配置。您也可以使用 **–e** 選項在命令列中指定配置。

在 Windows 中：

> cd logstash-5.0.1/bin
> Logstash -e 'input { stdin { } } output { stdout {} }'

在 Linux 中：

$ cd logstash-5.0.1/bin
$ ./logstash -e 'input { stdin { } } output { stdout {} }'

**注意** - 在 Windows 的情況下，您可能會收到一個錯誤，指出未設定 JAVA_HOME。為此，請在環境變數中將其設定為“C:\Program Files\Java\jre1.8.0_111”或您安裝 java 的位置。

**步驟 5** - Logstash Web 介面的預設埠為 9600 到 9700，在 **logstash-5.0.1\config\logstash.yml** 中定義為 **http.port**，它將選擇給定範圍內的第一個可用埠。

我們可以透過瀏覽 **https://:9600** 來檢查 Logstash 伺服器是否已啟動並正在執行，或者如果埠不同，則請檢查命令提示符或終端。我們可以看到分配的埠為“成功啟動 Logstash API 端點 {:port ⇒ 9600}”。它將返回一個 JSON 物件，其中包含以下關於已安裝 Logstash 的資訊：

{
   "host":"manu-PC", 
   "version":"5.0.1",
   "http_address":"127.0.0.1:9600",
   "build_date":"2016-11-11T22:28:04+00:00",
   "build_sha":"2d8d6263dd09417793f2a0c6d5ee702063b5fada",
   "build_snapshot":false
}

Logstash - 內部架構

在本章中，我們將討論 Logstash 的內部架構和不同元件。

Logstash 服務架構

Logstash 處理來自不同伺服器和資料來源的日誌，它充當發件人。發件人用於收集日誌，它們安裝在每個輸入源中。像 **Redis、Kafka** 或 **RabbitMQ** 這樣的代理是緩衝區，用於儲存資料供索引器使用，可能存在多個代理作為故障轉移例項。

像 **Lucene** 這樣的索引器用於索引日誌以獲得更好的搜尋效能，然後將輸出儲存在 Elasticsearch 或其他輸出目標中。輸出儲存中的資料可用於 Kibana 和其他視覺化軟體。

Logstash 內部架構

Logstash 管道包含三個元件：**輸入、過濾器** 和 **輸出**。輸入部分負責指定和訪問輸入資料來源，例如 **Apache Tomcat 伺服器** 的日誌資料夾。

解釋 Logstash 管道的示例

Logstash 配置檔案包含有關 Logstash 三個元件的詳細資訊。在這種情況下，我們建立一個名為 **Logstash.conf** 的檔案。

以下配置捕獲來自輸入日誌“inlog.log”的資料，並將其寫入輸出日誌“outlog.log”，而無需任何過濾器。

Logstash.conf

Logstash 配置檔案僅使用輸入外掛從 **inlog.log** 檔案複製資料，並使用輸出外掛將日誌資料重新整理到 **outlog.log** 檔案。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/inlog.log"
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/outlog.log"
   }
}

執行 Logstash

Logstash 使用 **–f** 選項來指定配置檔案。

C:\logstash\bin> logstash –f logstash.conf

inlog.log

以下程式碼塊顯示輸入日誌資料。

Hello tutorialspoint.com

outlog.log

Logstash 的輸出包含 message 欄位中的輸入資料。Logstash還會向輸出新增其他欄位，例如時間戳、輸入源路徑、版本、主機和標籤。

{
   "path":"C:/tpwork/logstash/bin/log/inlog1.log",
   "@timestamp":"2016-12-13T02:28:38.763Z",
   "@version":"1", "host":"Dell-PC",
   "message":" Hello tutorialspoint.com", "tags":[]
}

如您所見，Logstash 的輸出包含比輸入日誌提供的資料更多內容。輸出包含源路徑、時間戳、版本、主機名和標籤，這些用於表示額外的訊息，例如錯誤。

我們可以使用過濾器來處理資料，使其符合我們的需求。在接下來的示例中，我們使用過濾器來獲取資料，該過濾器將輸出限制為僅包含類似 GET 或 POST 的動詞，後跟一個**唯一資源識別符號**的資料。

Logstash.conf

在此 Logstash 配置中，我們添加了一個名為**grok**的過濾器來過濾輸入資料。與模式序列輸入日誌匹配的輸入日誌事件，只有在出現錯誤的情況下才會到達輸出目標。Logstash 在不匹配 grok 過濾器模式序列的輸出事件中新增名為“_grokparsefailure”的標籤。

Logstash 提供了許多內建正則表示式模式，用於解析流行的伺服器日誌，例如 Apache。此處使用的模式需要一個類似 get、post 等的動詞，後跟一個統一資源識別符號。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/inlog2.log"
   }
}
filter {
   grok {
      match => {"message" => "%{WORD:verb} %{URIPATHPARAM:uri}"}
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/outlog2.log"
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

C:\logstash\bin> logstash –f  Logstash.conf

inlog2.log

我們的輸入檔案包含兩個由預設分隔符（即換行符）分隔的事件。第一個事件與 GROk 中指定的模式匹配，第二個事件不匹配。

GET /tutorialspoint/Logstash
Input 1234

outlog2.log

我們可以看到，第二個輸出事件包含“_grokparsefailure”標籤，因為它與 grok 過濾器模式不匹配。使用者還可以使用輸出外掛中的**“if”**條件來刪除輸出中這些不匹配的事件。

{
   "path":"C:/tpwork/logstash/bin/log/inlog2.log",
   "@timestamp":"2016-12-13T02:47:10.352Z","@version":"1","host":"Dell-PC","verb":"GET",
   "message":"GET /tutorialspoint/logstash", "uri":"/tutorialspoint/logstash", "tags":[]
}
{
   "path":"C:/tpwork/logstash/bin/log/inlog2.log",
   "@timestamp":"2016-12-13T02:48:12.418Z", "@version":"1", "host":"Dell-PC",
   "message":"t 1234\r", "tags":["_grokparsefailure"]
}

Logstash - 收集日誌

使用 shipper 收集來自不同伺服器或資料來源的日誌。shipper 是安裝在伺服器上的 Logstash 例項，它訪問伺服器日誌並將日誌傳送到特定的輸出位置。

它主要將輸出傳送到 Elasticsearch 進行儲存。Logstash 從以下來源獲取輸入：

• STDIN
• Syslog
• 檔案
• TCP/UDP
• Microsoft Windows 事件日誌
• Websocket
• Zeromq
• 自定義擴充套件

使用 Apache Tomcat 7 伺服器收集日誌

在此示例中，我們使用檔案輸入外掛收集安裝在 Windows 上的 Apache Tomcat 7 伺服器的日誌，並將它們傳送到另一個日誌。

logstash.conf

在這裡，Logstash 被配置為訪問本地安裝的 Apache Tomcat 7 的訪問日誌。檔案外掛的路徑設定中使用正則表示式模式來從日誌檔案中獲取資料。這包含其名稱中的“access”，並且它添加了 apache 型別，這有助於區分集中式目標源中的 apache 事件和其他事件。最後，輸出事件將顯示在 output.log 中。

input {
   file {
      path => "C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/*access*"
      type => "apache"
   }
} 
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

C:\logstash\bin> logstash –f  Logstash.conf

Apache Tomcat 日誌

訪問 Apache Tomcat 伺服器及其 Web 應用程式（**https://:8080**）以生成日誌。Logstash 即時讀取日誌中的更新資料，並將其儲存在配置檔案中指定的 output.log 中。

Apache Tomcat 根據日期生成新的訪問日誌檔案，並將訪問事件記錄到該檔案中。在我們的例子中，它是在 Apache Tomcat 的 **logs** 目錄中的 localhost_access_log.2016-12-24.txt。

0:0:0:0:0:0:0:1 - - [
   25/Dec/2016:18:37:00 +0800] "GET / HTTP/1.1" 200 11418
0:0:0:0:0:0:0:1 - munish [
   25/Dec/2016:18:37:02 +0800] "GET /manager/html HTTP/1.1" 200 17472
0:0:0:0:0:0:0:1 - - [
   25/Dec/2016:18:37:08 +0800] "GET /docs/ HTTP/1.1" 200 19373
0:0:0:0:0:0:0:1 - - [
   25/Dec/2016:18:37:10 +0800] "GET /docs/introduction.html HTTP/1.1" 200 15399

output.log

您可以在輸出事件中看到，添加了一個 type 欄位，並且事件存在於 message 欄位中。

{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt",
   "@timestamp":"2016-12-25T10:37:00.363Z","@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - - [25/Dec/2016:18:37:00 +0800] \"GET /
   HTTP/1.1\" 200 11418\r","type":"apache","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt","@timestamp":"2016-12-25T10:37:10.407Z",
   "@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - munish [25/Dec/2016:18:37:02 +0800] \"GET /
   manager/html HTTP/1.1\" 200 17472\r","type":"apache","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt","@timestamp":"2016-12-25T10:37:10.407Z",
   "@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - - [25/Dec/2016:18:37:08 +0800] \"GET /docs/
   HTTP/1.1\" 200 19373\r","type":"apache","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   localhost_access_log.2016-12-25.txt","@timestamp":"2016-12-25T10:37:20.436Z",
   "@version":"1","host":"Dell-PC",
   "message":"0:0:0:0:0:0:0:1 - - [25/Dec/2016:18:37:10 +0800] \"GET /docs/
   introduction.html HTTP/1.1\" 200 15399\r","type":"apache","tags":[]
}

使用 STDIN 外掛收集日誌

在本節中，我們將討論另一個使用**STDIN 外掛**收集日誌的示例。

logstash.conf

這是一個非常簡單的示例，其中 Logstash 讀取使用者在標準輸入中輸入的事件。在我們的例子中，它是命令提示符，它將事件儲存在 output.log 檔案中。

input {
   stdin{}
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

C:\logstash\bin> logstash –f  Logstash.conf

在命令提示符中寫入以下文字：

使用者輸入了以下兩行。Logstash 透過分隔符設定分隔事件，其預設值為“\n”。使用者可以透過更改檔案外掛中分隔符的值來更改它。

Tutorialspoint.com welcomes you
Simply easy learning

output.log

以下程式碼塊顯示輸出日誌資料。

{
   "@timestamp":"2016-12-25T11:41:16.518Z","@version":"1","host":"Dell-PC",
   "message":"tutrialspoint.com welcomes you\r","tags":[]
}
{
   "@timestamp":"2016-12-25T11:41:53.396Z","@version":"1","host":"Dell-PC",
   "message":"simply easy learning\r","tags":[]
}

Logstash - 支援的輸入

Logstash 支援來自不同來源的大量日誌。它與下面解釋的著名來源一起工作。

從指標收集日誌

系統事件和其他時間活動記錄在指標中。Logstash 可以訪問系統指標中的日誌並使用過濾器對其進行處理。這有助於以自定義方式向用戶顯示事件的即時饋送。指標根據指標過濾器的**flush_interval 設定**重新整理，預設設定為 5 秒。

我們透過收集和分析透過 Logstash 執行的事件並在命令提示符上顯示即時饋送來跟蹤 Logstash 生成的測試指標。

logstash.conf

此配置包含一個由 Logstash 提供的生成器外掛，用於測試指標，並將 type 設定設定為“generated”以進行解析。在過濾階段，我們僅使用“if”語句處理具有 generated 型別的行。然後，metrics 外掛計算 meter 設定中指定的欄位。metrics 外掛每隔 5 秒（在**flush_interval**中指定）重新整理一次計數。

最後，使用**codec 外掛**進行格式化，將過濾器事件輸出到標準輸出（如命令提示符）。Codec 外掛使用[events][rate_1m]值在 1 分鐘滑動視窗中輸出每秒事件。

input {
   generator {
     	type => "generated"
   }
}
filter {
   if [type] == "generated" {
      metrics {
         meter => "events"
         add_tag => "metric"
      }
   }
}
output {
   # only emit events with the 'metric' tag
   if "metric" in [tags] {
      stdout {
         codec => line { format => "rate: %{[events][rate_1m]}"
      }
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logsaths –f logstash.conf

stdout（命令提示符）

rate: 1308.4
rate: 1308.4
rate: 1368.654529135342
rate: 1416.4796003951449
rate: 1464.974293984808
rate: 1523.3119444107458
rate: 1564.1602979542715
rate: 1610.6496496890895
rate: 1645.2184750334154
rate: 1688.7768007612485
rate: 1714.652283095914
rate: 1752.5150680019278
rate: 1785.9432934744932
rate: 1806.912181962126
rate: 1836.0070454626025
rate: 1849.5669494173826
rate: 1871.3814756851832
rate: 1883.3443123790712
rate: 1906.4879113216743
rate: 1925.9420717997118
rate: 1934.166137658981
rate: 1954.3176526556897
rate: 1957.0107444542625

從 Web 伺服器收集日誌

Web 伺服器會生成大量關於使用者訪問和錯誤的日誌。Logstash 幫助使用輸入外掛從不同的伺服器提取日誌，並將它們儲存在集中式位置。

我們正在從本地 Apache Tomcat 伺服器的**stderr 日誌**提取資料，並將其儲存在 output.log 中。

logstash.conf

此 Logstash 配置檔案指示 Logstash 讀取 apache 錯誤日誌並新增名為“apache-error”的標籤。我們可以簡單地使用檔案輸出外掛將其傳送到 output.log。

input {
   file {
      path => "C:/Program Files/Apache Software Foundation/Tomcat 7.0 /logs/*stderr*"
      type => "apache-error"  
   }
} 
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>Logstash –f Logstash.conf

輸入日誌示例

這是 Apache Tomcat 中發生伺服器事件時生成的示例**stderr 日誌**。

C:\Program Files\Apache Software Foundation\Tomcat 7.0\logs\ tomcat7-stderr.2016-12-25.log

Dec 25, 2016 7:05:14 PM org.apache.coyote.AbstractProtocol start
INFO: Starting ProtocolHandler ["http-bio-9999"]
Dec 25, 2016 7:05:14 PM org.apache.coyote.AbstractProtocol start
INFO: Starting ProtocolHandler ["ajp-bio-8009"]
Dec 25, 2016 7:05:14 PM org.apache.catalina.startup.Catalina start
INFO: Server startup in 823 ms

output.log

{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"Dec 25, 2016 7:05:14 PM org.apache.coyote.AbstractProtocol start\r",
   "type":"apache-error","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"INFO: Starting ProtocolHandler [
      \"ajp-bio-8009\"]\r","type":"apache-error","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"Dec 25, 2016 7:05:14 PM org.apache.catalina.startup.Catalina start\r",
   "type":"apache-error","tags":[]
}
{
   "path":"C:/Program Files/Apache Software Foundation/Tomcat 7.0/logs/
   tomcat7-stderr.2016-12-25.log","@timestamp":"2016-12-25T11:05:27.045Z",
   "@version":"1","host":"Dell-PC",
   "message":"INFO: Server startup in 823 ms\r","type":"apache-error","tags":[]
}

從資料來源收集日誌

首先，讓我們瞭解如何配置 MySQL 進行日誌記錄。在 MySQL 資料庫伺服器的 [mysqld] 下的**my.ini 檔案**中新增以下幾行。

在 Windows 中，它位於 MySQL 的安裝目錄內，位於：

C:\wamp\bin\mysql\mysql5.7.11

在 UNIX 中，您可以在 – /etc/mysql/my.cnf 中找到它

general_log_file   = "C:/wamp/logs/queries.log"
general_log = 1

logstash.conf

在此配置檔案中，檔案外掛用於讀取 MySQL 日誌並將其寫入 ouput.log。

input {
   file {
      path => "C:/wamp/logs/queries.log"
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

queries.log

這是由在 MySQL 資料庫中執行的查詢生成的日誌。

2016-12-25T13:05:36.854619Z   2 Query		select * from test1_users
2016-12-25T13:05:51.822475Z    2 Query	select count(*) from users
2016-12-25T13:05:59.998942Z    2 Query         select count(*) from test1_users

output.log

{
   "path":"C:/wamp/logs/queries.log","@timestamp":"2016-12-25T13:05:37.905Z",
   "@version":"1","host":"Dell-PC",
   "message":"2016-12-25T13:05:36.854619Z    2 Query\tselect * from test1_users",
   "tags":[]
}
{
   "path":"C:/wamp/logs/queries.log","@timestamp":"2016-12-25T13:05:51.938Z",
   "@version":"1","host":"Dell-PC",
   "message":"2016-12-25T13:05:51.822475Z    2 Query\tselect count(*) from users",
   "tags":[]
}
{
   "path":"C:/wamp/logs/queries.log","@timestamp":"2016-12-25T13:06:00.950Z",
   "@version":"1","host":"Dell-PC",
   "message":"2016-12-25T13:05:59.998942Z    2 Query\tselect count(*) from test1_users",
   "tags":[]
}

Logstash - 解析日誌

Logstash 使用輸入外掛接收日誌，然後使用過濾器外掛來解析和轉換資料。日誌的解析和轉換根據輸出目標中存在的系統執行。Logstash 解析日誌資料，僅轉發所需的欄位。之後，這些欄位將轉換為目標系統相容且易於理解的形式。

如何解析日誌？

日誌的解析是透過使用**GROK**(圖形化知識表示)模式來執行的，您可以在 Github 上找到它們：

https://github.com/elastic/logstash/tree/v1.4.2/patterns.

Logstash 將日誌資料與指定的 GROK 模式或模式序列匹配，以解析日誌，例如“%{COMBINEDAPACHELOG}”，這通常用於 apache 日誌。

解析後的資料結構更清晰，更易於搜尋和執行查詢。Logstash 在輸入日誌中搜索指定的 GROK 模式，並提取日誌中的匹配行。您可以使用 GROK 偵錯程式來測試您的 GROK 模式。

GROK 模式的語法是 %{SYNTAX:SEMANTIC}。Logstash GROK 過濾器採用以下形式：

%{PATTERN:FieldName}

這裡，PATTERN 代表 GROK 模式，fieldname 是欄位的名稱，它表示輸出中解析的資料。

例如，使用線上 GROK 偵錯程式 https://grokdebugger.com/

輸入

日誌中出現錯誤行的示例：

[Wed Dec 07 21:54:54.048805 2016] [:error] [pid 1234:tid 3456829102]
   [client 192.168.1.1:25007] JSP Notice:  Undefined index: abc in
   /home/manu/tpworks/tutorialspoint.com/index.jsp on line 11

GROK 模式序列

此 GROK 模式序列與包含時間戳、日誌級別、程序 ID、事務 ID 和錯誤訊息的日誌事件匹配。

\[(%{DAY:day} %{MONTH:month} %{MONTHDAY} %{TIME} %{YEAR})\] \[.*:%{LOGLEVEL:loglevel}\]
   \[pid %{NUMBER:pid}:tid %{NUMBER:tid}\] \[client %{IP:clientip}:.*\]
   %{GREEDYDATA:errormsg}

輸出

輸出採用 JSON 格式。

{
   "day": [
      "Wed"
   ],
   "month": [
      "Dec"
   ],
   "loglevel": [
      "error"
   ],
   "pid": [
      "1234"
   ],
   "tid": [
      "3456829102"
   ],
   "clientip": [
      "192.168.1.1"
   ],
   "errormsg": [
      "JSP Notice:  Undefined index: abc in
      /home/manu/tpworks/tutorialspoint.com/index.jsp on line 11"
   ]
}

Logstash - 過濾器

Logstash 在輸入和輸出之間的管道中間使用過濾器。Logstash 的過濾器測量操作和建立事件，例如**Apache-Access**。許多過濾器外掛用於管理 Logstash 中的事件。這裡，在**Logstash 聚合過濾器**的示例中，我們正在過濾資料庫中每個 SQL 事務的持續時間並計算總時間。

安裝聚合過濾器外掛

使用 Logstash-plugin 實用程式安裝聚合過濾器外掛。對於 Windows，Logstash-plugin 是 Logstash **bin 資料夾**中的批處理檔案。

>logstash-plugin install logstash-filter-aggregate

logstash.conf

在此配置中，您可以看到三個“if”語句，用於**初始化、遞增**和**生成**事務的總持續時間，即**sql_duration**。聚合外掛用於新增存在於輸入日誌的每個事件中的 sql_duration。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - 
            %{NOTSPACE:taskid} - %{NOTSPACE:logger} - 
            %{WORD:label}( - %{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"    
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logstash –f logstash.conf 

input.log

以下程式碼塊顯示輸入日誌資料。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END - end

output.log

如配置檔案中指定的，最後一個“if”語句中記錄器為 – TRANSACTION_END，它列印總事務時間或 sql_duration。這在 output.log 中以黃色突出顯示。

{
   "path":"C:/tpwork/logstash/bin/log/input.log","@timestamp": "2016-12-22T19:04:37.214Z",
   "loglevel":"INFO","logger":"TRANSACTION_START","@version": "1","host":"wcnlab-PC",
   "message":"8566 - TRANSACTION_START - start\r","tags":[]
}
{
   "duration":320,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.366Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 320\r","taskid":"48566","tags":[]
}
{
   "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.373Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 200\r","taskid":"48566","tags":[]
}
{
   "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-22T19:04:38.380Z","loglevel":"INFO","logger":"TRANSACTION_END",
   "@version":"1","host":"wcnlab-PC","label":"end",
   "message":" INFO - 48566 - TRANSACTION_END - end\r","taskid":"48566","tags":[]
}

Logstash - 轉換日誌

Logstash 提供各種外掛來轉換解析的日誌。這些外掛可以**新增、刪除**和**更新**日誌中的欄位，以便更好地理解和在輸出系統中進行查詢。

我們使用**Mutate 外掛**在輸入日誌的每一行中新增一個欄位名稱 user。

安裝 Mutate 過濾器外掛

要安裝 mutate 過濾器外掛；我們可以使用以下命令。

>Logstash-plugin install Logstash-filter-mutate

logstash.conf

在此配置檔案中，Mutate 外掛在 Aggregate 外掛之後新增，以新增新欄位。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [ "message", "%{LOGLEVEL:loglevel} -
         %{NOTSPACE:taskid} - %{NOTSPACE:logger} -
         %{WORD:label}( - %{INT:duration:int})?" ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ; 
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
   mutate {
      add_field => {"user" => "tutorialspoint.com"}
   }
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output.log"
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logstash –f logstash.conf

input.log

以下程式碼塊顯示輸入日誌資料。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END - end

output.log

您可以看到輸出事件中有一個名為“user”的新欄位。

{
   "path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.383Z",
   "@version":"1",
   "host":"wcnlab-PC",
   "message":"NFO - 48566 - TRANSACTION_START - start\r",
   "user":"tutorialspoint.com","tags":["_grokparsefailure"]
}
{
   "duration":320,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.383Z","loglevel":"INFO","logger":"SQL",
   "@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 320\r",
   "user":"tutorialspoint.com","taskid":"48566","tags":[]
}
{
   "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.399Z","loglevel":"INFO",
   "logger":"SQL","@version":"1","host":"wcnlab-PC","label":"transaction1",
   "message":" INFO - 48566 - SQL - transaction1 - 200\r",
   "user":"tutorialspoint.com","taskid":"48566","tags":[]
}
{
   "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
   "@timestamp":"2016-12-25T19:55:37.399Z","loglevel":"INFO",
   "logger":"TRANSACTION_END","@version":"1","host":"wcnlab-PC","label":"end",
   "message":" INFO - 48566 - TRANSACTION_END - end\r",
   "user":"tutorialspoint.com","taskid":"48566","tags":[]
}

Logstash - 輸出階段

輸出是 Logstash 管道中的最後一個階段，它將輸入日誌中的過濾器資料傳送到指定的目的地。Logstash 提供多個輸出外掛，可以將過濾後的日誌事件儲存到各種不同的儲存和搜尋引擎中。

儲存日誌

Logstash 可以將過濾後的日誌儲存在**檔案、Elasticsearch 引擎、stdout、AWS CloudWatch**等中。Logstash 還可以使用**TCP、UDP、Websocket**等網路協議將日誌事件傳輸到遠端儲存系統。

在 ELK 堆疊中，使用者使用 Elasticsearch 引擎來儲存日誌事件。在下面的示例中，我們將為本地 Elasticsearch 引擎生成日誌事件。

安裝 Elasticsearch 輸出外掛

我們可以使用以下命令安裝 Elasticsearch 輸出外掛。

>logstash-plugin install Logstash-output-elasticsearch

logstash.conf

此配置檔案包含一個 Elasticsearch 外掛，該外掛將輸出事件儲存在本地安裝的 Elasticsearch 中。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [ "message", "%{LOGLEVEL:loglevel} -
      %{NOTSPACE:taskid} - %{NOTSPACE:logger} -  
      %{WORD:label}( - %{INT:duration:int})?" ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
   mutate {
      add_field => {"user" => "tutorialspoint.com"}
   }
}
output {
   elasticsearch {
      hosts => ["127.0.0.1:9200"]
   }
}

Input.log

以下程式碼塊顯示輸入日誌資料。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END - end

啟動本地主機上的 Elasticsearch

要在本地主機上啟動 Elasticsearch，您應該使用以下命令。

C:\elasticsearch\bin> elasticsearch

Elasticsearch 準備就緒後，您可以透過在瀏覽器中鍵入以下 URL 來檢查它。

https://:9200/

響應

以下程式碼塊顯示本地主機上 Elasticsearch 的響應。

{
   "name" : "Doctor Dorcas",
   "cluster_name" : "elasticsearch",
   "version" : {
      "number" : "2.1.1",
      "build_hash" : "40e2c53a6b6c2972b3d13846e450e66f4375bd71",
      "build_timestamp" : "2015-12-15T13:05:55Z",
      "build_snapshot" : false,
      "lucene_version" : "5.3.1"
   },
   "tagline" : "You Know, for Search"
}

**注意** - 有關 Elasticsearch 的更多資訊，您可以點選以下連結。

https://tutorialspoint.tw/elasticsearch/index.html

現在，使用上述 Logstash.conf 執行 Logstash

>Logstash –f Logstash.conf

在將上述文字貼上到輸出日誌中之後，該文字將由 Logstash 儲存在 Elasticsearch 中。您可以透過在瀏覽器中鍵入以下 URL 來檢查儲存的資料。

https://:9200/logstash-2017.01.01/_search?pretty

響應

這是儲存在索引 Logstash-2017.01.01 中的 JSON 格式資料。

{
   "took" : 20,
   "timed_out" : false,
   "_shards" : {
      "total" : 5,
      "successful" : 5,
      "failed" : 0
   },
   "hits" : {
      "total" : 10,
      "max_score" : 1.0,
      "hits" : [ {
         "_index" : "logstash-2017.01.01",
         "_type" : "logs",
         "_id" : "AVlZ9vF8hshdrGm02KOs",
         "_score" : 1.0,
         "_source":{
            "duration":200,"path":"C:/tpwork/logstash/bin/log/input.log", 
            "@timestamp":"2017-01-01T12:17:49.140Z","loglevel":"INFO",
            "logger":"SQL","@version":"1","host":"wcnlab-PC",
            "label":"transaction1",
            "message":" INFO - 48566 - SQL - transaction1 - 200\r",
            "user":"tutorialspoint.com","taskid":"48566","tags":[]
         }
      },
      {
         "_index" : "logstash-2017.01.01",
         "_type" : "logs",
         "_id" : "AVlZ9vF8hshdrGm02KOt",
         "_score" : 1.0,
         "_source":{
            "sql_duration":520,"path":"C:/tpwork/logstash/bin/log/input.log",
            "@timestamp":"2017-01-01T12:17:49.145Z","loglevel":"INFO",
            "logger":"TRANSACTION_END","@version":"1","host":"wcnlab-PC",
            "label":"end",
            "message":" INFO - 48566 - TRANSACTION_END - end\r",
            "user":"tutorialspoint.com","taskid":"48566","tags":[]
         }
      }
   }
}

Logstash - 支援的輸出

Logstash 提供多個外掛來支援各種資料儲存或搜尋引擎。日誌的輸出事件可以傳送到輸出檔案、標準輸出或 Elasticsearch 等搜尋引擎。Logstash 中有三種受支援的輸出型別，它們是：

• 標準輸出
• 檔案輸出
• 空輸出

現在讓我們詳細討論每個輸出。

標準輸出 (stdout)

它用於將過濾後的日誌事件作為資料流生成到命令列介面。這是一個將資料庫事務的總持續時間生成到 stdout 的示例。

logstash.conf

此配置檔案包含一個標準輸出外掛，用於將總 sql_duration 寫入標準輸出。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid}
            - %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   if [logger] == "TRANSACTION_END" {
      stdout {
         codec => line{format => "%{sql_duration}"}
      }
   }
}

注意 − 如果尚未安裝，請安裝聚合過濾器。

>logstash-plugin install Logstash-filter-aggregate

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logstash –f logsatsh.conf

Input.log

以下程式碼塊顯示輸入日誌資料。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END – end

stdout（在 Windows 中為命令提示符，在 UNIX 中為終端）

這是總 sql_duration 320 + 200 = 520。

520

檔案輸出

Logstash 還可以將過濾器日誌事件儲存到輸出檔案中。我們將使用上述示例，將輸出儲存到檔案中，而不是 STDOUT。

logstash.conf

此 Logstash 配置檔案指示 Logstash 將總 sql_duration 儲存到輸出日誌檔案中。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input1.log"
   }
} 
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid} -
            %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" 
      ]
   }
   if [logger] == "TRANSACTION_START" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] = 0"
         map_action => "create"
      }
   }
   if [logger] == "SQL" {
      aggregate {
         task_id => "%{taskid}"
         code => "map['sql_duration'] ||= 0 ;
            map['sql_duration'] += event.get('duration')"
      }
   }
   if [logger] == "TRANSACTION_END" {
      aggregate {
         task_id => "%{taskid}"
         code => "event.set('sql_duration', map['sql_duration'])"
         end_of_task => true
         timeout => 120
      }
   }
}
output {
   if [logger] == "TRANSACTION_END" {
      file {
         path => "C:/tpwork/logstash/bin/log/output.log"
         codec => line{format => "%{sql_duration}"}
      }
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logstash –f logsatsh.conf

input.log

以下程式碼塊顯示輸入日誌資料。

INFO - 48566 - TRANSACTION_START - start
INFO - 48566 - SQL - transaction1 - 320
INFO - 48566 - SQL - transaction1 - 200
INFO - 48566 - TRANSACTION_END – end

output.log

以下程式碼塊顯示輸出日誌資料。

520

空輸出

這是一個特殊的輸出外掛，用於分析輸入和過濾器外掛的效能。

Logstash - 外掛

Logstash 為其管道的三個階段（輸入、過濾器和輸出）提供各種外掛。這些外掛幫助使用者從各種來源捕獲日誌，例如 Web 伺服器、資料庫、網路協議等。

捕獲後，Logstash 可以根據使用者的需求將資料解析並轉換為有意義的資訊。最後，Logstash 可以將這些有意義的資訊傳送或儲存到各種目標來源，例如 Elasticsearch、AWS Cloudwatch 等。

輸入外掛

Logstash 中的輸入外掛幫助使用者從各種來源提取和接收日誌。使用輸入外掛的語法如下：

Input {
   Plugin name {
      Setting 1……
      Setting 2……..
   }
}

您可以使用以下命令下載輸入外掛：

>Logstash-plugin install Logstash-input-<plugin name>

Logstash-plugin 實用程式位於 Logstash 安裝目錄的bin 資料夾中。下表列出了 Logstash 提供的輸入外掛。

外掛設定

所有外掛都有其特定的設定，有助於指定外掛中的重要欄位，例如埠、路徑等。我們將討論一些輸入外掛的設定。

檔案

此輸入外掛用於直接從輸入源中存在的日誌或文字檔案提取事件。它的工作方式類似於 UNIX 中的 tail 命令，儲存最後讀取的遊標，只讀取輸入檔案中新追加的資料，但這可以透過使用 star_position 設定來更改。以下是此輸入外掛的設定。

Elasticsearch

此特定外掛用於讀取 Elasticsearch 叢集中的搜尋查詢結果。以下是此外掛中使用的設定：

eventlog

此輸入外掛從 Windows 伺服器的 win32 API 讀取資料。以下是此外掛的設定：

Twitter

此輸入外掛用於從其流式 API 收集 Twitter 的 feed。下表描述了此外掛的設定。

TCP

TCP 用於透過 TCP 套接字獲取事件；它可以從使用者連線或伺服器讀取，這取決於 mode 設定。下表描述了此外掛的設定：

Logstash – 輸出外掛

Logstash 支援各種輸出源和不同的技術，例如資料庫、檔案、電子郵件、標準輸出等。

使用輸出外掛的語法如下：

output {
   Plugin name {
      Setting 1……
      Setting 2……..
   }
}

您可以使用以下命令下載輸出外掛：

>logstash-plugin install logstash-output-<plugin name>

Logstash-plugin 實用程式位於 Logstash 安裝目錄的 bin 資料夾中。下表描述了 Logstash 提供的輸出外掛。

所有外掛都有其特定的設定，這有助於在外掛中指定重要的欄位，例如埠、路徑等。我們將討論一些輸出外掛的設定。

Elasticsearch

Elasticsearch 輸出外掛使 Logstash 能夠將輸出儲存在 Elasticsearch 引擎的特定叢集中。這是使用者常用的選擇之一，因為它包含在 ELK Stack 包中，因此為 DevOps 提供端到端解決方案。下表描述了此輸出外掛的設定。

Email

電子郵件輸出外掛用於在 Logstash 生成輸出時通知使用者。下表描述了此外掛的設定。

Http

此設定用於透過 http 將輸出事件傳送到目標。此外掛具有以下設定：

stdout

stdout 輸出外掛用於將輸出事件寫入命令列介面的標準輸出。在 Windows 中是命令提示符，在 UNIX 中是終端。此外掛具有以下設定：

statsd

它是一個網路守護程式，用於透過 UDP 將矩陣資料傳送到目標後端服務。在 Windows 中是命令提示符，在 UNIX 中是終端。此外掛具有以下設定：

過濾器外掛

Logstash 支援各種過濾器外掛，用於將輸入日誌解析和轉換為更結構化且易於查詢的格式。

使用過濾器外掛的語法如下：

filter {
   Plugin name {
      Setting 1……
      Setting 2……..
   }
}

您可以使用以下命令下載過濾器外掛：

>logstash-plugin install logstash-filter-<plugin name>

Logstash-plugin 實用程式位於 Logstash 安裝目錄的 bin 資料夾中。下表描述了 Logstash 提供的輸出外掛。

Codec 外掛

Codec 外掛可以是輸入或輸出外掛的一部分。這些外掛用於更改或格式化日誌資料的表示方式。Logstash 提供多個 Codec 外掛，如下所示：

構建您自己的外掛

您還可以建立適合您需求的 Logstash 自定製外掛。Logstash-plugin 實用程式用於建立自定製外掛。在這裡，我們將建立一個過濾器外掛，它將在事件中新增一條自定製訊息。

生成基本結構

使用者可以使用 logstash-plugin 實用程式的 generate 選項生成必要的文 件，也可以在 GitHub 上找到。

>logstash-plugin generate --type filter --name myfilter --path c:/tpwork/logstash/lib

這裡，**type** 選項用於指定外掛是輸入、輸出還是過濾器。在此示例中，我們正在建立名為 **myfilter** 的過濾器外掛。path 選項用於指定要建立外掛目錄的路徑。執行上述命令後，您將看到建立了一個目錄結構。

開發外掛

您可以在外掛目錄的 **\lib\logstash\filters** 資料夾中找到外掛的程式碼檔案。副檔名為 **.rb**。

在我們的例子中，程式碼檔案位於以下路徑：

C:\tpwork\logstash\lib\logstash-filter-myfilter\lib\logstash\filters\myfilter.rb

我們將訊息更改為：default ⇒ "Hi, You are learning this on tutorialspoint.com" 並儲存檔案。

安裝外掛

要安裝此外掛，需要修改 Logstash 的 Gemfile。您可以在 Logstash 的安裝目錄中找到此檔案。在我們的例子中，它位於 **C:\tpwork\logstash**。使用任何文字編輯器編輯此檔案，並在其中新增以下文字。

gem "logstash-filter-myfilter",:path => "C:/tpwork/logstash/lib/logstash-filter-myfilter"

在上述命令中，我們指定了外掛的名稱以及可以在哪裡找到它以進行安裝。然後，執行 Logstash-plugin 實用程式來安裝此外掛。

>logstash-plugin install --no-verify

測試

在這裡，我們在之前的示例之一中添加了 **myfilter**：

logstash.conf

此 Logstash 配置檔案在 grok 過濾器外掛之後，在 filter 部分包含 myfilter。

input {
   file {
      path => "C:/tpwork/logstash/bin/log/input1.log"
   } 
}
filter {
   grok {
      match => [
         "message", "%{LOGLEVEL:loglevel} - %{NOTSPACE:taskid} -
            %{NOTSPACE:logger} - %{WORD:label}( - %{INT:duration:int})?" ]
   }
   myfilter{}
}
output {
   file {
      path => "C:/tpwork/logstash/bin/log/output1.log"
      codec => rubydebug
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logstash –f logsatsh.conf

input.log

以下程式碼塊顯示輸入日誌資料。

INFO - 48566 - TRANSACTION_START - start

output.log

以下程式碼塊顯示輸出日誌資料。

{
   "path" => "C:/tpwork/logstash/bin/log/input.log",
   "@timestamp" => 2017-01-07T06:25:25.484Z,
   "loglevel" => "INFO",
   "logger" => "TRANSACTION_END",
   "@version" => "1",
   "host" => "Dell-PC",
   "label" => "end",
   "message" => "Hi, You are learning this on tutorialspoint.com",
   "taskid" => "48566",
   "tags" => []
}

在 Logstash 上釋出

開發人員還可以透過將其上傳到 github 並遵循 Elasticsearch 公司定義的標準化步驟，將其自定製外掛釋出到 Logstash。

有關釋出的更多資訊，請參閱以下 URL：

https://www.elastic.co/guide/en/logstash/current/contributing-to-logstash.html

Logstash - 監控 API

Logstash 提供 API 來監控其效能。這些監控 API 提取有關 Logstash 的執行時指標。

節點資訊 API

此 API 用於獲取有關 Logstash 節點的資訊。它以 JSON 格式返回 OS、Logstash 管道和 JVM 的資訊。

您可以使用以下 URL 向 Logstash 傳送 **get** 請求來提取資訊：

GET https://:9600/_node?pretty

響應

以下是節點資訊 API 的響應。

{
   "host" : "Dell-PC",
   "version" : "5.0.1",
   "http_address" : "127.0.0.1:9600",
   
   "pipeline" : {
      "workers" : 4,
      "batch_size" : 125,
      "batch_delay" : 5,
      "config_reload_automatic" : false,
      "config_reload_interval" : 3
   },
   "os" : {
      "name" : "Windows 7",
      "arch" : "x86",
      "version" : "6.1",
      "available_processors" : 4
   },
   "jvm" : {
      "pid" : 312,
      "version" : "1.8.0_111",
      "vm_name" : "Java HotSpot(TM) Client VM",
      "vm_version" : "1.8.0_111",
      "vm_vendor" : "Oracle Corporation",
      "start_time_in_millis" : 1483770315412,
      
      "mem" : {
         "heap_init_in_bytes" : 16777216,
         "heap_max_in_bytes" : 1046937600,
         "non_heap_init_in_bytes" : 163840,
         "non_heap_max_in_bytes" : 0
      },
      "gc_collectors" : [ "ParNew", "ConcurrentMarkSweep" ]
   }
}

您還可以透過在 URL 中新增其名稱來獲取管道、作業系統和 JVM 的特定資訊。

GET https://:9600/_node/os?pretty
GET https://:9600/_node/pipeline?pretty
GET https://:9600/_node/jvm?pretty

外掛資訊 API

此 API 用於獲取有關 Logstash 中已安裝外掛的資訊。您可以透過向下面提到的 URL 傳送 get 請求來檢索此資訊：

GET https://:9600/_node/plugins?pretty

響應

以下是外掛資訊 API 的響應。

{
   "host" : "Dell-PC",
   "version" : "5.0.1",
   "http_address" : "127.0.0.1:9600",
   "total" : 95,
   "plugins" : [ {
      "name" : "logstash-codec-collectd",
      "version" : "3.0.2"
   },
   {
      "name" : "logstash-codec-dots",
      "version" : "3.0.2"
   },
   {
      "name" : "logstash-codec-edn",
      "version" : "3.0.2"
   },
   {
      "name" : "logstash-codec-edn_lines",
      "version" : "3.0.2"
   },
   ............
}

節點統計資訊 API

此 API 用於以 JSON 物件提取 Logstash 的統計資訊（記憶體、程序、JVM、管道）。您可以透過向下面提到的 URL 傳送 get 請求來檢索此資訊：

GET https://:9600/_node/stats/?pretty
GET https://:9600/_node/stats/process?pretty
GET https://:9600/_node/stats/jvm?pretty
GET https://:9600/_node/stats/pipeline?pretty

熱執行緒 API

此 API 檢索有關 Logstash 中熱執行緒的資訊。熱執行緒是 Java 執行緒，其 CPU 使用率很高，並且執行時間長於正常執行時間。您可以透過向下面提到的 URL 傳送 get 請求來檢索此資訊：

GET https://:9600/_node/hot_threads?pretty

使用者可以使用以下 URL 以更易於閱讀的形式獲取響應。

GET https://:9600/_node/hot_threads?human = true

Logstash - 安全性和監控

在本章中，我們將討論 Logstash 的安全性和監控方面。

監控

Logstash 是一個非常好的工具，用於監控生產環境中的伺服器和服務。生產環境中的應用程式會生成不同型別的日誌資料，例如訪問日誌、錯誤日誌等。Logstash 可以使用過濾器外掛計算或分析錯誤、訪問或其他事件的數量。此分析和計數可用於監控不同的伺服器及其服務。

Logstash 提供了諸如 **HTTP Poller** 之類的外掛來監控網站狀態監控。在這裡，我們正在監控一個名為 **mysite** 的網站，該網站託管在本地 Apache Tomcat 伺服器上。

logstash.conf

在此配置檔案中，http_poller 外掛用於在 interval 設定中指定的時間間隔後命中外掛中指定的站點。最後，它將站點的狀態寫入標準輸出。

input {
   http_poller {
      urls => {
         site => "https://:8080/mysite"
      }
      request_timeout => 20
      interval => 30
      metadata_target => "http_poller_metadata"
   }
}
output {
   if [http_poller_metadata][code] == 200 {
      stdout {
         codec => line{format => "%{http_poller_metadata[response_message]}"}
      }
   }
   if [http_poller_metadata][code] != 200 {
      stdout {
         codec => line{format => "down"}
      }
   }
}

執行 Logstash

我們可以使用以下命令執行 Logstash。

>logstash –f logstash.conf

stdout

如果站點正常執行，則輸出將為：

Ok

如果我們使用 Tomcat 的 **Manager App** 停止站點，輸出將更改為：

down

安全

Logstash 提供了大量功能，用於與外部系統進行安全通訊並支援身份驗證機制。所有 Logstash 外掛都支援透過 HTTP 連線進行身份驗證和加密。

使用 HTTP 協議的安全

在 Logstash 提供的各種外掛（例如 Elasticsearch 外掛）中，有使用者和密碼等設定用於身份驗證。

elasticsearch {
   user => <username>
   password => <password>
}

另一種身份驗證是用於 Elasticsearch 的 **PKI（公鑰基礎設施）**。開發人員需要在 Elasticsearch 輸出外掛中定義兩個設定才能啟用 PKI 身份驗證。

elasticsearch {
   keystore => <string_value>
   keystore_password => <password>
}

在 HTTPS 協議中，開發人員可以使用授權機構的證書進行 SSL/TLS。

elasticsearch {
   ssl => true
   cacert => <path to .pem file>
}

使用傳輸協議的安全

要將傳輸協議與 Elasticsearch 一起使用，使用者需要將協議設定設定為傳輸。這避免了 JSON 物件的反序列化，並提高了效率。

基本身份驗證與在 Elasticsearch 輸出協議中的 http 協議中執行的身份驗證相同。

elasticsearch {
   protocol => “transport”
   user => <username>
   password => <password>
}

PKI 身份驗證還需要將 SSL 設定為 true，並在 Elasticsearch 輸出協議中使用其他設定：

elasticsearch {
   protocol => “transport”
   ssl => true
   keystore => <string_value>
   keystore_password => <password>
}

最後，SSL 安全性比其他通訊安全方法需要更多設定。

elasticsearch {
   ssl => true
   ssl => true
   keystore => <string_value>
   keystore_password => <password>
   truststore => 
   truststore_password => <password>
}

Logstash 的其他安全優勢

Logstash 可以幫助輸入系統源防止拒絕服務攻擊等攻擊。監控日誌並分析這些日誌中的不同事件可以幫助系統管理員檢查傳入連線和錯誤的變化。這些分析可以幫助檢視伺服器上是否正在發生或即將發生攻擊。

Elasticsearch 公司的其他產品，例如 **x-pack** 和 **filebeat**，提供了一些與 Logstash 安全通訊的功能。