概述

官方文檔：

輸入(ru)插件使 Logstash 能夠讀(du)取(qu)特定的事(shi)件源。

LogStash 提供了 50 + 種輸入插件，常見類(lei)型(xing)包括：

• 文件類：file, s3
• 網絡類：beats, tcp, udp, http, kafka
• 數據庫類：jdbc, mongodb，redis
• 系統類：syslog, eventlog, windows_eventlog
• 特殊類：stdin, generator

所有輸入(ru)插件都支持以下核心參數(shu)：

• add_field：值類型是哈希，默認值是{}，其主要作用是向事件添加字段
• codec：值類型是編解碼器，默認值是plain，
• enable_metric：值類型為布爾值，默認值是true，是否開啟metric指標
• id：值類型為字符串，此設置沒有默認值。為插件配置添加一個唯一的ID。
• tags：為事件添加標簽，值類型是數組
• type：值類型為字符串，type向此輸入處理的所有事件添加一個字段。可以在 Kibana 中搜索它

常用輸入插件-Stdin

Stdin是從標準輸入讀取事件。默認情況下，每個事件都被視為一行。如果需要合并多行，則需要使用多行編解碼器。
stdin插件主要用于測(ce)試和開發環(huan)(huan)境，不適合(he)生產(chan)環(huan)(huan)境的高(gao)吞(tun)吐(tu)量場(chang)景。

常用字段解釋

• add_field：值類型是哈希，默認值是{}，其主要作用是向事件添加字段
• codec：值類型是編解碼器，默認值是line，
• enable_metric：值類型為布爾值，默認值是true，是否開啟metric指標
• id：值類型為字符串，此設置沒有默認值。為插件配置添加一個唯一的ID。
• tags：為事件添加標簽，值類型是數組
• type：值類型為字符串，type向此輸入處理的所有事件添加一個字段。可以在 Kibana 中搜索它

實戰

[root@ELK01 ~/logstash/config]# cat stdin.conf
# 輸入
input {
    stdin {
        type => "stdin"
        id => "my_stdin"
        tags => ["stdin","my_stdin"]
        add_field => {
            "name" => "張三"
            "age" => 18
        }
        codec => line
    }
}

# 輸出，以debug方式輸出
output {
    stdout {
        codec => rubydebug
    }
}

啟動LogStash驗證

[root@ELK01 ~/logstash/config]# logstash -f ./stdin.conf
# 打印的日志信息省略
1 #輸入1，然后回車
{
          "age" => "18",                      #add_field添加的字段
          "host" => "ELK01",
          "name" => "張三",
          "type" => "stdin"                  # 指定的type類型
      "@version" => "1",
    "@timestamp" => 2025-06-15T05:34:25.042Z,
          "tags" => [                         #添加的tag
        [0] "stdin",
        [1] "my_stdin"
    ],
       "message" => "1",                     #輸入的字段

}

常用輸入插件-file

參考文檔：

在 Logstash 中，file輸入插件(jian)是最常用的輸入插件(jian)之一，主要(yao)用于從文(wen)(wen)(wen)件(jian)系統讀取數據(ju)，常見(jian)的使用場景(jing)包括收集日志文(wen)(wen)(wen)件(jian)、監(jian)控文(wen)(wen)(wen)件(jian)變更，file插件(jian)能夠讀取所有(you)的文(wen)(wen)(wen)件(jian)

常用字段解釋

• path（必需）
  • 功能：用于指定要監控的文件路徑，可以使用絕對路徑或相對路徑。
  • 注意事項：
    • 若路徑指向的是目錄而非文件，該插件會遞歸讀取目錄下的所有文件。
    • 要保證 Logstash 進程擁有讀取這些文件的權限。

path => ["/var/log/*.log", "/data/app/*.out"]  # 支持使用通配符匹配多個文件

• start_position
  • 功能：確定首次讀取文件時的起始位置。
  • 可選值：
    • beginning：從文件開頭開始讀取，適用于歷史數據遷移的場景。
    • end（默認值）：從文件末尾開始讀取，適合實時日志收集的情況。

start_position => "beginning"

• sincedb_path
  • 功能：指定 sincedb 文件的存儲位置，該文件用于記錄已讀取文件的偏移量。
  • 默認值：$HOME/.sincedb_<hash>

sincedb_path => "/data/logstash/sincedb"  # 生產環境建議使用專用目錄

• ignore_older
  • 功能：設置一個時間閾值，超過該閾值的文件將被忽略。
  • 單位：秒、分鐘、小時、天等，例如3600（3600 秒）、2h（2 小時）。

ignore_older => "1d"  # 忽略超過1天未修改的文件

• exclude
  • 功能：用于排除不需要監控的文件。

exclude => ["*.gz", "*.bak"]  # 排除壓縮文件和備份文件

• stat_interval
  • 功能：設置檢查文件更新的時間間隔。
  • 默認值：1（秒）

stat_interval => 5  # 每5秒檢查一次文件更新

• discover_interval
  • 功能：設置掃描目錄以發現新文件的時間間隔，在監控目錄時會用到。
  • 默認值：15（秒）

discover_interval => 30  # 每30秒掃描一次目錄

• file_completed_action與file_completed_log_path
  • 用于處理已完成讀取的文件

file_completed_action => "log"  # 可選值: log, delete, archive
file_completed_log_path => "/var/log/logstash/completed_files.log"

• file_sort_by與file_sort_direction
  控制文件讀取順序

file_sort_by => "last_modified"  # 可選值: path, last_modified
file_sort_direction => "asc"     # 可選值: asc(升序), desc（降序）

• delimiter
  • 指定行分隔符（適用于非標準換行符的文件）

delimiter => "\r\n"  # Windows格式換行符

• max_open_files
  • 限制最大打開文件數（防止系統資源耗盡）

max_open_files => 1024  # 默認值為4095

• sincedb_write_interval
  • 設置 sincedb 寫入間隔（提高性能）

sincedb_write_interval => 15  # 每15秒寫入一次sincedb

• sincedb_clean_after
  • 清理過期的 sincedb 記錄

sincedb_clean_after => "7d"  # 清理7天未訪問的文件記錄，默認值為14天

• mode
  選擇監控模式（默認是tail，即追加模式）

mode => "read"  # 一次性讀取文件，讀完即關閉，默認值為tail

實戰

定義配置文件

[root@ELK01 ~/logstash/config]# cat file.conf
input {
    file {
        id => "my_app_log"
        codec => "plain"
        tags => ["file","myfile"]
        add_field => {
            "name" => "zhangsan"
            "age" => 18
        }
        type => "file"
        # 讀取日志的文件路徑
        path => "/tmp/apps*.log"
        # 讀取的開始位置
        # 該參數剩下的前提條件時"sincedb*"文件中沒有對新文件的記錄.
        start_position => beginning
        # sincedb文件
        sincedb_path => "/data/logstash/sincedb"
        ignore_older => "3d"
        # 讀取頻率，1秒
        stat_interval => 1
        max_open_files => 1024
        sincedb_write_interval => 10
        mode => tail
    }
}
output {
    stdout {
        codec => rubydebug
    }
}

啟動logstash

[root@ELK01 ~/logstash/config]# logstash -f ./file.conf
[2025-06-15T14:22:35,596][INFO ][logstash.javapipeline    ][main] Pipeline Java execution initialization time {"seconds"=>0.58}
[2025-06-15T14:22:35,657][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
[2025-06-15T14:22:35,717][INFO ][filewatch.observingtail  ][main][my_app_log] START, creating Discoverer, Watch with file and sincedb collections
[2025-06-15T14:22:35,734][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}

寫入內容至/tmp/apps01.log

[root@ELK01 /tmp]# echo hello logstash file >> apps.log

查看控制臺

{
      "@version" => "1",
          "host" => "ELK01",
          "path" => "/tmp/apps.log",
       "message" => "hello logstash file", # 我們寫入文件的內容
          "type" => "file",
          "name" => "zhangsan",
          "tags" => [
        [0] "file",
        [1] "myfile"
    ],
    "@timestamp" => 2025-06-15T06:24:17.191Z,
           "age" => "18"
}

查看sincedb文件(jian)

[root@ELK01 /tmp]# cat /data/logstash/sincedb
2490408 0 64768 44 1749968657.193113 /tmp/apps.log

# 2490408表示文件的inodb
# 0 表示磁盤分區
# 64768表示文件當前偏移量，單位字節
# 44表示記錄文件元數據（如權限、所有者）最后修改的時間戳（秒級）。
# /tmp/apps.log表示讀取的日志文件

持續更新其它插件