概述

官方文檔：

Logstash 是一(yi)款開源(yuan)數據(ju)(ju)(ju)收集(ji)引擎(qing)，具備(bei)實時流水(shui)線(xian)功能(neng)。Logstash 可以(yi)動態地整合來自(zi)不(bu)同數據(ju)(ju)(ju)源(yuan)的(de)(de)數據(ju)(ju)(ju)，并將其規范化到您(nin)選擇的(de)(de)目標位置。它能(neng)夠清理(li)并簡化您(nin)的(de)(de)所有數據(ju)(ju)(ju)，以(yi)適應各種高級下游分析和可視(shi)化用例(li)。

雖(sui)然(ran) Logstash 最(zui)初推動(dong)了日(ri)志收集領(ling)域的創新，但它的功能(neng)遠不(bu)止于(yu)此。任何(he)類型的事件(jian)都可以通(tong)過(guo)豐(feng)富的輸入、過(guo)濾和(he)輸出插件(jian)進行(xing)豐(feng)富和(he)轉換，而許多(duo)原(yuan)生(sheng)編解碼器則進一步簡化了數據采集流程。Logstash 通(tong)過(guo)利(li)用更大規(gui)模和(he)更多(duo)樣化的數據，加速(su)您的洞察。

Logstash 的主要(yao)特點包括：

• 多(duo)輸入源：Logstash 支持(chi)多(duo)種類(lei)型的輸入數據，包括(kuo)日(ri)志文件、系統消息隊列、數據庫等。

• 數(shu)據處理(li)：Logstash 可以(yi)對數(shu)據進行各種轉換和處理(li)，如過濾、解(jie)析、格式化等。

• 多(duo)輸出目標：Logstash 可以將數據發(fa)送到各(ge)種(zhong)目標，如(ru) Elasticsearch、Kafka、郵件通知等(deng)。

• 插(cha)件機制(zhi)：Logstash 提供了豐富的(de)插(cha)件，可(ke)以方便(bian)地擴展其功能(neng)。

• 與 Elasticsearch 和(he) Kibana 的集成(cheng)：Logstash 是 Elastic Stack（前稱 ELK Stack）的一部(bu)分(fen)，與 Elasticsearch 和(he) Kibana 有很好的集成(cheng)，可以方便(bian)地(di)進行數據搜索、存儲和(he)可視化。

LogStash的工作原理

參考文檔：

Logstash 的工作原理可以分為三(san)個主要(yao)步驟：輸(shu)(shu)入（Input）、過濾（Filter）和輸(shu)(shu)出（Output）。

• 輸入（Input）：Logstash 支持多種類型的輸入數據，包括日志文件、系統消息隊列、數據庫等。在配置文件中，你可以指定一個或多個輸入源。
• 過濾（Filter）：輸入數據被收集后，Logstash 可以對數據進行各種轉換和處理。例如，你可以使用 grok 插件來解析非結構化的日志數據，將其轉換為結構化的數據。你也可以使用 mutate 插件來修改數據，如添加新的字段、刪除字段、更改字段的值等。
• 輸出（Output）：處理后的數據可以被發送到一個或多個目標。Logstash 支持多種類型的輸出目標，包括 Elasticsearch、Kafka、郵件通知等。

這(zhe)三(san)個步驟是在 Logstash 的事件處理管道中(zhong)順序執行的。每個事件（例如(ru)，一(yi)行日志數據(ju)）都會經過輸入、過濾和(he)輸出這(zhe)三(san)個步驟。在過濾階段，如(ru)果一(yi)個事件被過濾器(qi)丟棄(qi)，那么它將(jiang)不會被發送到輸出目標。

LogStash的工作模型

參考文檔：

• 每個(ge)(ge) Input 啟動一個(ge)(ge)線程：Logstash 會(hui)為每個(ge)(ge)輸入插件(jian)啟動一個(ge)(ge)線程，這些線程并行運(yun)行，從各自的數據(ju)源獲取數據(ju)。

• 數據寫入(ru)隊列(lie)(lie)：輸入(ru)插件獲(huo)取的(de)數據會(hui)被寫入(ru)一(yi)個(ge)隊列(lie)(lie)。默認情況下，這(zhe)是一(yi)個(ge)存儲在(zai)內存中(zhong)(zhong)的(de)有界隊列(lie)(lie)，如果(guo) Logstash 意外(wai)停止，隊列(lie)(lie)中(zhong)(zhong)的(de)數據會(hui)丟失(shi)。為了防止數據丟失(shi)，Logstash 提(ti)供了兩個(ge)特性：

  • Persistent Queues：這個特性會將隊列存儲在磁盤上，即使 Logstash 意外停止，隊列中的數據也不會丟失。
  • Dead Letter Queues：這個特性會保存無法處理的事件。需要注意的是，這個特性只支持 Elasticsearch 作為輸出源。

• 多個 Pipeline Worker 處理數據(ju)：Logstash 會啟動多個 Pipeline Worker，每(mei)個 Worker 會從隊(dui)列中取出一批數據(ju)，然后執(zhi)行過濾器和輸出插件。Worker 的數量(liang)和每(mei)次處理的數據(ju)量(liang)可以(yi)在(zai)配置(zhi)文件中設置(zhi)。

LogStash下載安裝

下載地址：
在此地(di)址下你(ni)可以下載LogStash不(bu)同版(ban)本、不(bu)同平臺的安(an)裝包

安裝JDK

LogStash是以Java語言編寫的，安裝JDK可以查看這篇文章：//www.xtjzw.net/huangSir-devops/p/18919758

LogStash與Java版本對應關系，可以(yi)閱讀這篇文章：

我這(zhe)里(li)安裝的(de)是(shi)LogStash7.17.26，JDK版本選擇的(de)是(shi)JDK11

# 下載
[root@master ~]# wget //mirrors.huaweicloud.com/openjdk/11.0.2/openjdk-11.0.2_linux-x64_bin.tar.gz
[root@master ~]# ll openjdk-11.0.2_linux-x64_bin.tar.gz
-rw-r--r-- 1 root root 187513052 Jan 18  2019 openjdk-11.0.2_linux-x64_bin.tar.gz
 
# 解壓
[root@master ~]# tar -xvf openjdk-11.0.2_linux-x64_bin.tar.gz
 
# 創建軟連接
[root@master ~]# ln -s /root/jdk-11.0.2 /usr/local/jdk11
[root@master ~]# ll /usr/local/jdk11
lrwxrwxrwx 1 root root 16 Jun 14 21:09 /usr/local/jdk11 -> /root/jdk-11.0.2/
 
# 配置環境變量
[root@master ~]# vim /etc/profile
# 根據實際安裝路徑修改
export JAVA_HOME=/usr/local/jdk11/
export PATH=$JAVA_HOME/bin:$PATH
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
 
# 加載環境變量
[root@master ~]# source /etc/profile
 
# 驗證
[root@master ~]# java -version
openjdk version "11.0.2" 2019-01-15
OpenJDK Runtime Environment 18.9 (build 11.0.2+9)
OpenJDK 64-Bit Server VM 18.9 (build 11.0.2+9, mixed mode)

下載部署

我這(zhe)里以7.17.26版本為(wei)準

# 下載
[root@ELK01 /data/software]# wget //artifacts.elastic.co/downloads/logstash/logstash-7.17.26-linux-x86_64.tar.gz

# 解壓
[root@ELK01 /data/software]# tar -xvf logstash-7.17.26-linux-x86_64.tar.gz

# 創建軟連接
[root@ELK01 /data/software]# ln -s /data/software/logstash-7.17.26 /usr/local/logstash
[root@ELK01 /data/software]# ll /usr/local/logstash
lrwxrwxrwx 1 root root 31 Jun 15 12:09 /usr/local/logstash -> /data/software/logstash-7.17.26/

# 配置環境變量
[root@ELK01 ~]# echo 'export PATH=$PATH:/usr/local/logstash/bin' >> /etc/profile
[root@ELK01 ~]# source /etc/profile

驗證LogStash

啟動LogStash

[root@ELK01 ~]# logstash -e "input { stdin { type => stdin } }  output { stdout { codec => rubydebug } }"
...# 啟動日志省略
1 # 輸入1
{
       "message" => "1",  # 輸出1
          "host" => "ELK01",
      "@version" => "1",
    "@timestamp" => 2025-06-15T04:15:58.847Z,
          "type" => "stdin"
}
2 # 輸入2
{
       "message" => "2", # 輸出2
          "host" => "ELK01",
      "@version" => "1",
    "@timestamp" => 2025-06-15T04:16:00.123Z,
          "type" => "stdin"
}
3 # 輸入3
{
       "message" => "3", # 輸出3
          "host" => "ELK01",
      "@version" => "1",
    "@timestamp" => 2025-06-15T04:16:01.564Z,
          "type" => "stdin"
}
4 # 輸入4
{
       "message" => "4", # 輸出4
          "host" => "ELK01",
      "@version" => "1",
    "@timestamp" => 2025-06-15T04:16:03.969Z,
          "type" => "stdin"
}