Logstash介紹及Input插件介紹

一、Logstash簡介

  Logstash是一個開源數據收集引擎，具有實時管道功能。Logstash可以動態地將來自不同數據源的數據統一起來，並將數據標准化到你所選擇的目的地。

 

Logstash管道有兩個必需的元素，輸入和輸出，以及一個可選元素過濾器。輸入插件從數據源那里消費數據，過濾器插件根據你的期望修改數據，輸出插件將數據寫入目的地。

 

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

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

Logstash 能夠動態地轉換和解析數據，不受格式或復雜度的影響：

利用 Grok 從非結構化數據中派生出結構
從 IP 地址破譯出地理坐標
將 PII(個人驗證信息) 數據匿名化，完全排除敏感字段
整體處理不受數據源、格式或架構的影響

輸出：選擇你的存儲，導出你的數據
盡管 Elasticsearch 是我們的首選輸出方向，能夠為我們的搜索和分析帶來無限可能，但它並非唯一選擇。
Logstash 提供眾多輸出選擇，您可以將數據發送到您要指定的地方，並且能夠靈活地解鎖眾多下游用例。

logstash是做數據采集的，類似於flume。
Logstash架構：

Batcher負責批量的從queue中取數據;
Queue分類：
In Memory ： 無法處理進程Crash、機器宕機等情況，會導致數據丟失
Persistent Queue In Disk：可處理進程Crash等情況，保證數據不丟失，保證數據至少消費一次，充當緩沖區，可以替代kafka等消息隊列的作用。

官網：https://www.elastic.co/cn/products/logstash

二、Input插件

Plugin	Description	Github repository
azure_event_hubs	Receives events from Azure Event Hubs	azure_event_hubs
beats	Receives events from the Elastic Beats framework	logstash-input-beats
cloudwatch	Pulls events from the Amazon Web Services CloudWatch API	logstash-input-cloudwatch
couchdb_changes	Streams events from CouchDB’s _changes URI	logstash-input-couchdb_changes
dead_letter_queue	read events from Logstash’s dead letter queue	logstash-input-dead_letter_queue
elasticsearch	Reads query results from an Elasticsearch cluster	logstash-input-elasticsearch
exec	Captures the output of a shell command as an event	logstash-input-exec
file	Streams events from files	logstash-input-file
ganglia	Reads Ganglia packets over UDP	logstash-input-ganglia
gelf	Reads GELF-format messages from Graylog2 as events	logstash-input-gelf
generator	Generates random log events for test purposes	logstash-input-generator
github	Reads events from a GitHub webhook	logstash-input-github
google_cloud_storage	Extract events from files in a Google Cloud Storage bucket	logstash-input-google_cloud_storage
google_pubsub	Consume events from a Google Cloud PubSub service	logstash-input-google_pubsub
graphite	Reads metrics from the graphite tool	logstash-input-graphite
heartbeat	Generates heartbeat events for testing	logstash-input-heartbeat
http	Receives events over HTTP or HTTPS	logstash-input-http
http_poller	Decodes the output of an HTTP API into events	logstash-input-http_poller
imap	Reads mail from an IMAP server	logstash-input-imap
irc	Reads events from an IRC server	logstash-input-irc
java_generator	Generates synthetic log events	core plugin
java_stdin	Reads events from standard input	core plugin
jdbc	Creates events from JDBC data	logstash-input-jdbc
jms	Reads events from a Jms Broker	logstash-input-jms
jmx	Retrieves metrics from remote Java applications over JMX	logstash-input-jmx
kafka	Reads events from a Kafka topic	logstash-input-kafka
kinesis	Receives events through an AWS Kinesis stream	logstash-input-kinesis
log4j	Reads events over a TCP socket from a Log4j SocketAppender object	logstash-input-log4j
lumberjack	Receives events using the Lumberjack protocl	logstash-input-lumberjack
meetup	Captures the output of command line tools as an event	logstash-input-meetup
pipe	Streams events from a long-running command pipe	logstash-input-pipe
puppet_facter	Receives facts from a Puppet server	logstash-input-puppet_facter
rabbitmq	Pulls events from a RabbitMQ exchange	logstash-input-rabbitmq
redis	Reads events from a Redis instance	logstash-input-redis
relp	Receives RELP events over a TCP socket	logstash-input-relp
rss	Captures the output of command line tools as an event	logstash-input-rss
s3	Streams events from files in a S3 bucket	logstash-input-s3
salesforce	Creates events based on a Salesforce SOQL query	logstash-input-salesforce
snmp	Polls network devices using Simple Network Management Protocol (SNMP)	logstash-input-snmp
snmptrap	Creates events based on SNMP trap messages	logstash-input-snmptrap
sqlite	Creates events based on rows in an SQLite database	logstash-input-sqlite
sqs	Pulls events from an Amazon Web Services Simple Queue Service queue	logstash-input-sqs
stdin	Reads events from standard input	logstash-input-stdin
stomp	Creates events received with the STOMP protocol	logstash-input-stomp
syslog	Reads syslog messages as events	logstash-input-syslog
tcp	Reads events from a TCP socket	logstash-input-tcp
twitter	Reads events from the Twitter Streaming API	logstash-input-twitter
udp	Reads events over UDP	logstash-input-udp
unix	Reads events over a UNIX socket	logstash-input-unix
varnishlog	Reads from the varnish cache shared memory log	logstash-input-varnishlog
websocket	Reads events from a websocket	logstash-input-websocket
wmi	Creates events based on the results of a WMI query	logstash-input-wmi
xmpp	Receives events over the XMPP/Jabber protocol	logstash-input-xmpp

舉例幾個常用的input來源：

注：以下為谷歌翻譯，可能會出現部分偏差。

Beats輸入插件

說明

該輸入插件使Logstash可以從Elastic Beats框架接收事件 。

以下示例顯示如何配置Logstash以在端口5044上偵聽傳入的Beats連接並索引到Elasticsearch。

input {
  beats {
    port => 5044
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][beat]}-%{[@metadata][version]}" 
  }
}

 

使用此處顯示的Logstash配置索引到Elasticsearch中的事件將類似於Beats直接索引到Elasticsearch中的事件。

如果未使用ILM，則將其設置index為， %{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}以便Logstash每天根據@timestampBeats事件的值創建一個索引。 

如果要傳送跨越多行的事件，則需要使用Filebeat中可用的配置選項來處理多行事件，然后再將事件數據發送到Logstash。您不能使用 多行編解碼器插件來處理多行事件。這樣做將導致無法啟動Logstash。

Versioned Beats Indices

為了最大程度地減少未來架構更改對Elasticsearch中現有索引和映射的影響，請配置Elasticsearch輸出以寫入版本索引。您為index設置指定的模式控制索引名稱：

index = > “％{[@@ metadata] [beat]}-％{[@@ metadata] [version]}-％{+ YYYY.MM.dd}”

%{[@metadata][beat]}

將索引名稱的第一部分設置為 beat 元數據字段的值，例如 filebeat 。

%{[@metadata][version]}

將名稱的第二部分設置為Beat版本，例如 7.4.2 。

%{+YYYY.MM.dd}

根據Logstash  @timestamp 字段將名稱的第三部分設置為日期。

此配置產生每日索引名稱，例如 filebeat-7.4.2-2019-11-27。

Beats的配置選項

該插件支持以下配置選項以及稍后介紹的“ 通用選項”。

設置	輸入類型	需要
add_hostname	布爾值	沒有
cipher_suites	數組	沒有
client_inactivity_timeout	數	沒有
host	串	沒有
include_codec_tag	布爾值	沒有
port	數	是
ssl	布爾值	沒有
ssl_certificate	有效的文件系統路徑	沒有
ssl_certificate_authorities	數組	沒有
ssl_handshake_timeout	數	沒有
ssl_key	有效的文件系統路徑	沒有
ssl_key_passphrase	密碼	沒有
ssl_verify_mode	字符串，其中之一["none", "peer", "force_peer"]	沒有
ssl_peer_metadata	布爾值	沒有
tls_max_version	數	沒有
tls_min_version	數	沒有

 

cipher_suites

• 值類型為數組
• 默認值為 java.lang.String[TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384, TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384, TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384, TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256, TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256]@459cfcca

要使用的密碼套件列表，按優先級列出。

client_inactivity_timeout

• 值類型是數字
• 默認值為 60

X秒鍾不活動后，關閉空閑客戶端。

host

• 值類型為字符串
• 默認值為 "0.0.0.0"

要監聽的IP地址。

include_codec_tag

• 值類型為布爾值
• 默認值為 true

port

• 這是必需的設置。
• 值類型是數字
• 此設置沒有默認值。

要監聽的端口。

ssl

• 值類型為布爾值
• 默認值為 false

默認情況下，事件以純文本發送。您可以通過設置ssl為true並配置ssl_certificate和ssl_key選項來啟用加密。

ssl_certificate

• 值類型是路徑
• 此設置沒有默認值。

要使用的SSL證書。

ssl_certificate_authorities

• 值類型為數組
• 默認值為 []

根據這些權限驗證客戶證書。您可以定義多個文件或路徑。所有證書將被讀取並添加到信任存儲中。您需要配置ssl_verify_mode 到peer或force_peer啟用驗證。

ssl_handshake_timeout

• 值類型是數字
• 默認值為 10000

ssl握手不完整超時的時間（以毫秒為單位）

ssl_key

• 值類型是路徑
• 此設置沒有默認值。

要使用的SSL密鑰。注意：該密鑰必須為PKCS8格式，您可以使用OpenSSL對其 進行轉換以獲取更多信息。

ssl_key_passphrase

• 值類型為密碼
• 此設置沒有默認值。

要使用的SSL密鑰密碼。

ssl_verify_mode

• 值可以是任何的：none，peer，force_peer
• 默認值為 "none"

默認情況下，服務器不執行任何客戶端驗證。

peer將使服務器要求客戶端提供證書。如果客戶端提供證書，則將對其進行驗證。

force_peer將使服務器要求客戶端提供證書。如果客戶端不提供證書，則連接將關閉。

此選項需要與ssl_certificate_authoritiesCA 一起使用，並已定義了CA列表。

ssl_peer_metadata

• 值類型為布爾值
• 默認值為 false

允許在事件的元數據中存儲客戶端證書信息。

僅當ssl_verify_mode設置為peer或時，此選項才有效force_peer。

tls_max_version

• 值類型是數字
• 默認值為 1.2

加密連接允許的最大TLS版本。該值必須是以下之一：TLS 1.0的1.0，TLS 1.1的1.1，TLS 1.2的1.2

tls_min_version

• 值類型是數字
• 默認值為 1

加密連接允許的最低TLS版本。該值必須是以下之一：TLS 1.0的1.0，TLS 1.1的1.1，TLS 1.2的1.2

常用選項

所有輸入插件均支持以下配置選項：

設置	輸入類型	需要
add_field	雜湊	沒有
codec	編解碼器	沒有
enable_metric	布爾值	沒有
id	串	沒有
tags	數組	沒有
type	串	沒有

add_field

• 值類型為哈希
• 默認值為 {}

向事件添加字段

codec

• 值類型為編解碼器
• 默認值為 "plain"

用於輸入數據的編解碼器。輸入編解碼器是一種在數據輸入之前解碼數據的便捷方法，而無需在Logstash管道中使用單獨的過濾器。

enable_metric

• 值類型為布爾值
• 默認值為 true

默認情況下，為此特定插件實例禁用或啟用度量標准日志記錄，我們會記錄所有可以度量的數據，但是您可以禁用特定插件的度量標准收集。

id

• 值類型為字符串
• 此設置沒有默認值。

ID向插件配置添加唯一。如果未指定ID，Logstash將生成一個。強烈建議在您的配置中設置此ID。當您有兩個或多個相同類型的插件時（例如，如果您有2個拍子輸入），這特別有用。在這種情況下，添加命名ID將有助於在使用監視API時監視Logstash。

input { beats { id => "my_plugin_id" } }

tags

• 值類型為數組
• 此設置沒有默認值。

將任意數量的任意標簽添加到您的事件中。

這可以幫助以后進行處理。

type

• 值類型為字符串
• 此設置沒有默認值。

type向此輸入處理的所有事件添加一個字段。

 

新發現並從頭開始讀取。因此，復制的文件路徑不應位於要監視的文件名模式中（該path選項）。將檢測到截斷，並將“最后讀取”位置更新為零。

File輸入配置選項

該插件支持以下配置選項以及稍后介紹的“ 通用選項”。

 

持續時間設置可以以文本形式指定，例如“ 250 ms”，該字符串將轉換為十進制秒。有很多受支持的自然和縮寫持續時間，有關詳細信息，請參見string_duration。

設置	輸入類型	需要
close_older	數字或string_duration	沒有
delimiter	串	沒有
discover_interval	數	沒有
exclude	數組	沒有
file_chunk_count	數	沒有
file_chunk_size	數	沒有
file_completed_action	字符串，其中之一["delete", "log", "log_and_delete"]	沒有
file_completed_log_path	串	沒有
file_sort_by	字符串，其中之一["last_modified", "path"]	沒有
file_sort_direction	字符串，其中之一["asc", "desc"]	沒有
ignore_older	數字或string_duration	沒有
max_open_files	數	沒有
mode	字符串，其中之一["tail", "read"]	沒有
path	數組	是
sincedb_clean_after	數字或string_duration	沒有
sincedb_path	串	沒有
sincedb_write_interval	數字或string_duration	沒有
start_position	字符串，其中之一["beginning", "end"]	沒有
stat_interval	數字或string_duration	沒有

ose_older

• 值類型是數字或string_duration
• 默認值為 "1 hour"

文件輸入將關閉最后一次讀取指定持續時間（如果指定了數字，則為秒）的所有文件。根據文件是尾部還是讀取，這具有不同的含義。如果拖尾，並且傳入數據有很大的時間間隔，則可以關閉文件（允許打開其他文件），但是當檢測到新數據時，將排隊等待重新打開文件。如果正在讀取，則從讀取最后一個字節起，在closed_older秒后將關閉文件。如果將插件升級到4.1.0+，則此設置將保留，以實現向后兼容，並且讀取不會拖尾並且不切換為使用讀取模式。

delimiter

• 值類型為字符串
• 默認值為 "\n"

設置新行分隔符，默認為“ \ n”。請注意，在讀取壓縮文件時，不會使用此設置，而是使用標准Windows或Unix行結尾。

discover_interval

• 值類型是數字
• 默認值為 15

我們多長時間擴展一次path選項中的文件名模式，以發現要觀看的新文件。該值是的倍數stat_interval，例如，如果stat_interval是“ 500 ms”，則可以每15 X 500毫秒-7.5秒發現新文件文件。實際上，這是最好的情況，因為需要考慮讀取新內容所花費的時間。

exclude

• 值類型為數組
• 此設置沒有默認值。

排除項（與文件名匹配，不是完整路徑）。文件名模式在這里也是有效的。例如，如果您有

 路徑=> “ / var / log / *”

在尾模式下，您可能要排除壓縮文件：

 排除=> “ * .gz”

file_chunk_count

• 值類型是數字
• 默認值為 4611686018427387903

與結合使用時file_chunk_size，此選項設置在移至下一個活動文件之前從每個文件讀取多少塊（帶或條紋）。例如，file_chunk_count32和file_chunk_size32KB的a 將處理每個活動文件中的下一個1MB。由於默認值非常大，因此在移至下一個活動文件之前，該文件已有效讀取到EOF。

file_chunk_size

• 值類型是數字
• 默認值為32768（32KB）

從磁盤上以塊或塊的形式讀取文件內容，並從塊中提取行。請參閱file_chunk_count以了解為什么和何時從默認設置更改此設置。

file_completed_action

• 值可以是任何的：delete，log，log_and_delete
• 默認值為delete。

在read模式下，完成文件后應執行什么動作。如果指定了刪除，則文件將被刪除。如果指定了日志，則文件的完整路徑將記錄到file_completed_log_path設置中指定的文件中 。如果log_and_delete指定，則以上兩個動作都會發生。

file_completed_log_path

• 值類型為字符串
• 此設置沒有默認值。

完全讀取的文件路徑應附加到哪個文件。只有當指定文件這條道路file_completed_action是日志或log_and_delete。重要說明：此文件僅附加到-可能會變得很大。您負責文件輪換。

file_sort_by

• 值可以是任何的：last_modified，path
• 默認值為last_modified。

應該使用“監視”文件的哪個屬性對其進行排序。文件可以按修改日期或全路徑字母排序。以前，發現並因此“監視”的文件的處理順序取決於操作系統。

file_sort_direction

• 值可以是任何的：asc，desc
• 默認值為asc。

排序“監視”的文件時，在升序和降序之間進行選擇。如果最舊的數據優先，那么last_modified+ 的默認值asc很好。如果首先獲取最新數據更為重要，則選擇last_modified+ desc。如果對文件的完整路徑使用特殊的命名約定，則path+ 可能 asc會幫助控制文件處理的順序。

ignore_older

• 值類型是數字或string_duration
• 此設置沒有默認值。

當文件輸入發現在指定持續時間（如果指定了數字，則為秒）之前最后修改的文件時，將忽略該文件。發現后，如果修改了忽略的文件，則不再忽略它，並且讀取任何新數據。默認情況下，此選項是禁用的。請注意，該單位以秒為單位。

max_open_files

• 值類型是數字
• 此設置沒有默認值。

此輸入一次一次消耗的file_handles的最大數量是多少。如果您需要處理的文件數量超過此數目，請使用close_older關閉一些文件。不應將其設置為OS可以執行的最大操作，因為其他LS插件和OS進程需要文件句柄。內部默認設置為4095。

mode

• 值可以是tail或read。
• 默認值為tail。

您希望文件輸入以哪種模式操作。尾部幾個文件或讀取許多內容完整的文件。讀取模式現在支持gzip文件處理。如果指定了“讀取”，則將忽略以下其他設置：

1. start_position （始終從頭開始讀取文件）
2. close_older（到達EOF時文件自動關閉）

如果指定為“ read”，則注意以下設置：

1. ignore_older （舊文件未處理）
2. file_completed_action （處理文件時應采取什么措施）
3. file_completed_log_path （完整的文件路徑應記錄到哪個文件）

path

• 這是必需的設置。
• 值類型為數組
• 此設置沒有默認值。

用作輸入的文件的路徑。您可以在此處使用文件名模式，例如/var/log/*.log。如果您使用的模式/var/log/**/*.log，/var/log則將對所有*.log文件進行遞歸搜索。路徑必須是絕對路徑，不能是相對路徑。

您也可以配置多個路徑。請參閱Logstash配置頁面上的示例。

sincedb_clean_after

• 值類型是數字或string_duration
• 此設置的默認值為“ 2周”。
• 如果指定了一個數字，則將其解釋為天，可以是十進制，例如0.5是12小時。

現在，sincedb記錄具有與其關聯的最后一個活動時間戳記。如果在過去N天內未在跟蹤文件中檢測到任何更改，則它的sincedb跟蹤記錄將過期，並且不會保留。此選項有助於防止索引節點回收問題。Filebeat有一個有關inode回收的常見問題解答。

sincedb_path

• 值類型為字符串
• 此設置沒有默認值。

sincedb數據庫文件的路徑（保留跟蹤的日志文件的當前位置的路徑），該路徑將被寫入磁盤。默認情況下會將sincedb文件寫入到<path.data>/plugins/inputs/file NOTE：它必須是文件路徑，而不是目錄路徑

sincedb_write_interval

• 值類型是數字或string_duration
• 默認值為 "15 seconds"

自監視的日志文件的當前位置起寫入數據庫的頻率（以秒為單位）。

start_position

• 值可以是任何的：beginning，end
• 默認值為 "end"

選擇Logstash最初從哪個位置開始讀取文件：在開頭還是結尾。默認行為將文件視為實時流，因此從結尾開始。如果您要導入的舊數據，請將其設置為Beginning。

此選項僅修改文件為新文件且之前未曾查看過的“首次聯系”情況，即Logstash讀取的sincedb文件中未記錄當前位置的文件。如果以前已經看過文件，則此選項無效，並且將使用sincedb文件中記錄的位置。

stat_interval

• 值類型是數字或string_duration
• 默認值為 "1 second"

我們統計文件的頻率（以秒為單位），以查看它們是否已被修改。增加此間隔將減少我們進行的系統調用的數量，但會增加檢測新日志行的時間。

 

發現新文件並檢查它們是否已增長/收縮。該循環將休眠stat_interval數秒，然后再次循環。但是，如果文件已增長，則將讀取新內容並排隊。跨所有增長的文件進行讀取和排隊可能會花費一些時間，尤其是在管道擁塞的情況下。因此，整個循環時間是stat_interval和文件讀取時間的組合 。

常用選項

所有輸入插件均支持以下配置選項：

設置	輸入類型	需要
add_field	雜湊	沒有
codec	編解碼器	沒有
enable_metric	布爾值	沒有
id	串	沒有
tags	數組	沒有
type	串	沒有

add_field

• 值類型為哈希
• 默認值為 {}

向事件添加字段

codec

• 值類型為編解碼器
• 默認值為 "plain"

用於輸入數據的編解碼器。輸入編解碼器是一種在數據輸入之前解碼數據的便捷方法，而無需在Logstash管道中使用單獨的過濾器。

enable_metric

• 值類型為布爾值
• 默認值為 true

默認情況下，為此特定插件實例禁用或啟用度量標准日志記錄，我們會記錄所有可以度量的數據，但是您可以禁用特定插件的度量標准收集。

id

• 值類型為字符串
• 此設置沒有默認值。

ID向插件配置添加唯一。如果未指定ID，Logstash將生成一個。強烈建議在您的配置中設置此ID。當您有兩個或多個相同類型的插件時（例如，如果您有2個文件輸入），這特別有用。在這種情況下，添加命名ID將有助於在使用監視API時監視Logstash。

輸入{  文件{  id => “ my_plugin_id” } }

tags

• 值類型為數組
• 此設置沒有默認值。

將任意數量的任意標簽添加到您的事件中。

這可以幫助以后進行處理。

type

• 值類型為字符串
• 此設置沒有默認值。

type向此輸入處理的所有事件添加一個字段。

 JDBC配置選項

計划

可以將來自此插件的輸入安排為根據特定計划定期運行。此調度語法由rufus-scheduler支持。該語法類似於cron，具有一些特定於Rufus的擴展（例如時區支持）。

例子：

* 5 * 1-3 *	從一月到三月的每天凌晨5點每分鍾執行一次。
0 * * * *	將在每天每小時的第0分鍾執行。
0 6 * * * America/Chicago	每天早上6:00（UTC / GMT -5）執行。

可以在此處找到描述此語法的更多文檔。

狀態

插件將以sql_last_value元數據文件的形式保存配置文件中的參數last_run_metadata_path。執行查詢后，該文件將更新為的當前值sql_last_value。下次管道啟動時，將通過從文件中讀取來更新此值。如果 clean_run設置為true，則將忽略此值並將其sql_last_value設置為1970年1月1日；如果use_column_value為true，則將其設置為0 ，就好像從未執行過任何查詢一樣。

處理大型結果集

許多JDBC驅動程序使用該fetch_size參數來限制一次從游標到客戶端緩存中一次預取多少結果，然后再從結果集中檢索更多結果。這是使用jdbc_fetch_size配置選項在此插件中配置的。默認情況下，此插件未設置提取大小，因此將使用特定驅動程序的默認大小。

用法：

這是設置插件以從MySQL數據庫獲取數據的示例。首先，我們在當前路徑中放置適當的JDBC驅動程序庫（可以將其放置在文件系統上的任何位置）。在此示例中，我們使用用戶mysql連接到mydb數據庫，並希望在songs 表中輸入與特定藝術家匹配的所有行。以下示例演示了為此可能的Logstash配置。此示例中的選項將指示插件每分鍾在每分鍾執行一次此輸入語句。schedule

input {
  jdbc {
    jdbc_driver_library => "mysql-connector-java-5.1.36-bin.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "mysql"
    parameters => { "favorite_artist" => "Beethoven" }
    schedule => "* * * * *"
    statement => "SELECT * from songs where artist = :favorite_artist"
  }
}

配置SQL語句

此輸入需要sql語句。這可以通過字符串形式的語句選項傳入，也可以從文件（statement_filepath）中讀取。當配置中提供的SQL語句很大或很麻煩時，通常使用File選項。file選項僅支持一個SQL語句。該插件僅接受以下選項之一。它無法從文件以及statement配置參數中讀取語句。

配置多個SQL語句

當需要從不同的數據庫表或視圖中查詢和提取數據時，配置多個SQL語句很有用。可以為每個語句定義單獨的Logstash配置文件，或者在一個配置文件中定義多個語句。在單個Logstash配置文件中使用多個語句時，必須將每個語句定義為單獨的jdbc輸入（包括jdbc驅動程序，連接字符串和其他必需的參數）。

請注意，如果任何語句使用sql_last_value參數（例如，僅提取自上次運行以來更改的數據），則每個輸入都應定義自己的 last_run_metadata_path參數。否則，將導致不良行為，因為所有輸入都會將其狀態存儲到同一（默認）元數據文件中，從而有效覆蓋彼此的sql_last_value。

預定義的參數

一些參數是內置的，可以在查詢中使用。這是清單：

sql_last_value

用於計算要查詢的行的值。在運行任何查詢之前，將其設置為1970年1月1日（星期四），如果設置use_column_value為true，tracking_column則設置為 0 。在隨后的查詢運行之后，將對其進行相應的更新。

例：

input {
  jdbc {
    statement => "SELECT id, mycolumn1, mycolumn2 FROM my_table WHERE id > :sql_last_value"
    use_column_value => true
    tracking_column => "id"
    # ... other configuration bits
  }
} 

預處理語句

使用服務器端准備好的語句可以加快執行速度，因為服務器可以優化查詢計划和執行。

並非所有的JDBC可訪問技術都將支持准備好的語句。

隨着Prepared Statement支持的引入，帶來了不同的代碼執行路徑和一些新設置。現有的大多數設置仍然有用，但是有一些新的設置可供預讀語句讀取。使用布爾值設置use_prepared_statements啟用此執行模式。使用該prepared_statement_name設置為“ Prepared Statement”指定名稱，這將在本地和遠程標識“ Prepared Statement”，並且它在配置和數據庫中應該是唯一的。使用prepared_statement_bind_values數組設置來指定綁定值，:sql_last_value為前面提到的預定義參數使用精確的字符串（必要時多次）。該statement（或statement_path）設置仍保存SQL語句，但使用必須使用綁定變量?字符，按在prepared_statement_bind_values數組中找到的確切順序作為占位符。

目前不支持圍繞預備語句構建計數查詢，並且由於jdbc分頁在后台使用計數查詢，因此此時預備語句也不支持jdbc分頁。因此，使用准備好的語句時jdbc_paging_enabled，jdbc_page_size設置將被忽略。

例：

input {
  jdbc {
    statement => "SELECT * FROM mgd.seq_sequence WHERE _sequence_key > ? AND _sequence_key < ? + ? ORDER BY _sequence_key ASC"
    prepared_statement_bind_values => [":sql_last_value", ":sql_last_value", 4]
    prepared_statement_name => "foobar"
    use_prepared_statements => true
    use_column_value => true
    tracking_column_type => "numeric"
    tracking_column => "_sequence_key"
    last_run_metadata_path => "/elastic/tmp/testing/confs/test-jdbc-int-sql_last_value.yml"
    # ... other configuration bits
  }
}

Jdbc輸入配置選項編輯

該插件支持以下配置選項以及稍后介紹的“ 通用選項”。

設置	輸入類型	需要
clean_run	布爾值	沒有
columns_charset	Hash	沒有
connection_retry_attempts	數	沒有
connection_retry_attempts_wait_time	數	沒有
jdbc_connection_string	串	是
jdbc_default_timezone	串	沒有
jdbc_driver_class	串	是
jdbc_driver_library	串	沒有
jdbc_fetch_size	數	沒有
jdbc_page_size	數	沒有
jdbc_paging_enabled	布爾值	沒有
jdbc_password	密碼	沒有
jdbc_password_filepath	有效的文件系統路徑	沒有
jdbc_pool_timeout	數	沒有
jdbc_user	串	是
jdbc_validate_connection	布爾值	沒有
jdbc_validation_timeout	數	沒有
last_run_metadata_path	串	沒有
lowercase_column_names	布爾值	沒有
parameters	雜湊	沒有
plugin_timezone	字符串，其中之一["local", "utc"]	沒有
prepared_statement_bind_values	數組	沒有
prepared_statement_name	串	沒有
record_last_run	布爾值	沒有
schedule	串	沒有
sequel_opts	雜湊	沒有
sql_log_level	字符串，其中之一["fatal", "error", "warn", "info", "debug"]	沒有
statement	串	沒有
statement_filepath	有效的文件系統路徑	沒有
tracking_column	串	沒有
tracking_column_type	字符串，其中之一["numeric", "timestamp"]	沒有
use_column_value	布爾值	沒有
use_prepared_statements	布爾值	沒有

clean_run

• 值類型為布爾值
• 默認值為 false

是否應保留先前的運行狀態

columns_charset

• 值類型為哈希
• 默認值為 {}

特定列的字符編碼。此選項將覆蓋:charset指定列的選項。

例：

input { jdbc { ... columns_charset => { "column0" => "ISO-8859-1" } ... } }

connection_retry_attempts

• 值類型是數字
• 默認值為 1

嘗試連接數據庫的最大次數

connection_retry_attempts_wait_time

• 值類型是數字
• 默認值為 0.5

兩次嘗試之間休眠的秒數

jdbc_connection_string

• 這是必需的設置。
• 值類型為字符串
• 此設置沒有默認值。

JDBC連接字符串

jdbc_default_timezone

• 值類型為字符串
• 此設置沒有默認值。

時區轉換。Logstash（和Elasticsearch）期望時間戳以UTC術語表示。如果您的數據庫記錄了相對於另一個時區的時間戳，則將記錄該數據庫的時區，然后將此設置設置為數據庫使用的時區。但是，由於SQL不允許在時間戳字段中提供時區數據，因此我們無法逐條記錄地進行計算。此插件將以ISO8601格式的相對UTC時間自動將您的SQL時間戳字段轉換為Logstash時間戳。

使用此設置將手動分配指定的時區偏移，而不是使用本地計算機的時區設置。例如，您必須使用標准時區，例如America / Denver。

plugin_timezone

• 值可以是任何的：utc，local
• 默認值為 "utc"

如果您希望此插件將時間戳偏移到UTC以外的時區，則可以將此設置設置為local，插件將使用OS時區進行偏移調整。

注意：當指定plugin_timezone和/或時jdbc_default_timezone，偏移量調整在兩個地方進行，如果sql_last_value是時間戳，並且在語句中用作參數，則偏移量調整將從插件時區到數據時區，並且在處理記錄時，時間戳從數據庫時區偏移到插件時區。如果您的數據庫時區為UTC，則無需設置這些設置中的任何一個。

jdbc_driver_class

• 這是必需的設置。
• 值類型為字符串
• 此設置沒有默認值。

如果要使用Oracle JDBC ，則按https://github.com/logstash-plugins/logstash-input-jdbc/issues/43加載的JDBC驅動程序類，例如“ org.apache.derby.jdbc.ClientDriver” NB。驅動程序（ojdbc6.jar）正確jdbc_driver_class是"Java::oracle.jdbc.driver.OracleDriver"

jdbc_driver_library

• 值類型為字符串
• 此設置沒有默認值。

嘗試將JDBC邏輯抽象為mixin，以便在其他插件（輸入/輸出）中潛在地重用。當某人包含此模塊時，將調用此方法。將這些方法添加到給定的基礎中。第三方驅動程序庫的JDBC驅動程序庫路徑。如果需要多個庫，可以通過逗號分隔它們。

如果未提供，則插件將在Logstash Java類路徑中查找驅動程序類。

jdbc_fetch_size

• 值類型是數字
• 此設置沒有默認值。

JDBC提取大小。如果未提供，將使用各自的驅動程序默認值

jdbc_page_size

• 值類型是數字
• 默認值為 100000

JDBC頁面大小

jdbc_paging_enabled

• 值類型為布爾值
• 默認值為 false

JDBC啟用分頁

這將導致sql語句分解為多個查詢。每個查詢將使用限制和偏移量來集體檢索完整的結果集。限制大小通過設置jdbc_page_size。

請注意，不能保證查詢之間的順序。

jdbc_password

• 值類型為密碼
• 此設置沒有默認值。

JDBC密碼

jdbc_password_filepath

• 值類型是路徑
• 此設置沒有默認值。

JDBC密碼文件名

jdbc_pool_timeout

• 值類型是數字
• 默認值為 5

連接池配置。引發PoolTimeoutError之前等待獲取連接的秒數（默認為5）

jdbc_user

• 這是必需的設置。
• 值類型為字符串
• 此設置沒有默認值。

JDBC用戶

jdbc_validate_connection

• 值類型為布爾值
• 默認值為 false

連接池配置。使用前驗證連接。

jdbc_validation_timeout

• 值類型是數字
• 默認值為 3600

連接池配置。驗證連接的頻率（以秒為單位）

last_run_metadata_path

• 值類型為字符串
• 默認值為 "$HOME/.logstash_jdbc_last_run"

上次運行時間的文件路徑

lowercase_column_names

• 值類型為布爾值
• 默認值為 true

是否強制使用標識符字段的小寫

parameters

• 值類型為哈希
• 默認值為 {}

查詢參數的散列，例如 { "target_id" => "321" }

prepared_statement_bind_values

• 值類型為數組
• 默認值為 []

准備好的語句的綁定值數組。:sql_last_value是保留的預定義字符串

prepared_statement_name

• 值類型為字符串
• 默認值為 ""

准備好的語句的名稱。它在您的配置和數據庫中必須是唯一的

record_last_run

• 值類型為布爾值
• 默認值為 true

是否保存狀態 last_run_metadata_path

schedule

• 值類型為字符串
• 此設置沒有默認值。

定期運行語句的時間表，例如Cron格式：“ * * * * *”（每分鍾，每分鍾執行一次查詢）

默認情況下沒有時間表。如果沒有給出時間表，則該語句僅運行一次。

sequel_opts

• 值類型為哈希
• 默認值為 {}

常規/特定於供應商的續集配置選項。

可選連接池配置的示例max_connections-連接池的最大連接數

可以在此文檔頁面中找到特定於供應商的選項的示例：https : //github.com/jeremyevans/sequel/blob/master/doc/opening_databases.rdoc

sql_log_level

• 值可以是任何的：fatal，error，warn，info，debug
• 默認值為 "info"

記錄SQL查詢的日志級別，可接受的值是常見的致命，錯誤，警告，信息和調試值。默認值為info。

statement

• 值類型為字符串
• 此設置沒有默認值。

如果未定義，則即使未使用編解碼器，Logstash也會進行投訴。執行語句

要使用參數，請使用命名參數語法。例如：

“選擇*從MYTABLE WHERE id =：target_id”

在這里，“：target_id”是一個命名參數。您可以使用該parameters設置配置命名參數。

statement_filepath

• 值類型是路徑
• 此設置沒有默認值。

包含要執行的語句的文件的路徑

tracking_column

• 值類型為字符串
• 此設置沒有默認值。

值將被跟蹤的列use_column_value設置為true

tracking_column_type

• 值可以是任何的：numeric，timestamp
• 默認值為 "numeric"

跟蹤列的類型。目前僅“數字”和“時間戳”

use_column_value

• 值類型為布爾值
• 默認值為 false

設置為時true，將定義的 tracking_column值用作:sql_last_value。設置為時false，:sql_last_value反映上一次執行查詢的時間。

use_prepared_statements

• 值類型為布爾值
• 默認值為 false

設置為時true，啟用prepare語句用法

常用選項

所有輸入插件均支持以下配置選項：

設置	輸入類型	需要
add_field	雜湊	沒有
codec	編解碼器	沒有
enable_metric	布爾值	沒有
id	串	沒有
tags	數組	沒有
type	串	沒有

add_field

• 值類型為哈希
• 默認值為 {}

向事件添加字段

codec

• 值類型為編解碼器
• 默認值為 "plain"

用於輸入數據的編解碼器。輸入編解碼器是一種在數據輸入之前解碼數據的便捷方法，而無需在Logstash管道中使用單獨的過濾器。

enable_metric

• 值類型為布爾值
• 默認值為 true

默認情況下，為此特定插件實例禁用或啟用度量標准日志記錄，我們會記錄所有可以度量的數據，但是您可以禁用特定插件的度量標准收集。

id

• 值類型為字符串
• 此設置沒有默認值。

ID向插件配置添加唯一。如果未指定ID，Logstash將生成一個。強烈建議在您的配置中設置此ID。當您有兩個或多個相同類型的插件時，例如在您有2個jdbc輸入時，這特別有用。在這種情況下，添加命名ID將有助於在使用監視API時監視Logstash。

輸入{  jdbc {  id => “ my_plugin_id” } }

tags

• 值類型為數組
• 此設置沒有默認值。

將任意數量的任意標簽添加到您的事件中。

這可以幫助以后進行處理。

type

• 值類型為字符串
• 此設置沒有默認值。

type向此輸入處理的所有事件添加一個字段。

Kafka輸入配置選項

其中一些選項映射到Kafka選項。有關更多詳細信息，請參見https://kafka.apache.org/documentation。

設置	輸入類型	需要
auto_commit_interval_ms	串	沒有
auto_offset_reset	串	沒有
bootstrap_servers	串	沒有
check_crcs	串	沒有
client_id	串	沒有
connections_max_idle_ms	串	沒有
consumer_threads	數	沒有
decorate_events	布爾值	沒有
enable_auto_commit	串	沒有
exclude_internal_topics	串	沒有
fetch_max_bytes	串	沒有
fetch_max_wait_ms	串	沒有
fetch_min_bytes	串	沒有
group_id	串	沒有
heartbeat_interval_ms	串	沒有
jaas_path	有效的文件系統路徑	沒有
kerberos_config	有效的文件系統路徑	沒有
key_deserializer_class	串	沒有
max_partition_fetch_bytes	串	沒有
max_poll_interval_ms	串	沒有
max_poll_records	串	沒有
metadata_max_age_ms	串	沒有
partition_assignment_strategy	串	沒有
poll_timeout_ms	數	沒有
receive_buffer_bytes	串	沒有
reconnect_backoff_ms	串	沒有
request_timeout_ms	串	沒有
retry_backoff_ms	串	沒有
sasl_jaas_config	串	沒有
sasl_kerberos_service_name	串	沒有
sasl_mechanism	串	沒有
security_protocol	字符串，其中之一["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"]	沒有
send_buffer_bytes	串	沒有
session_timeout_ms	串	沒有
ssl_endpoint_identification_algorithm	串	沒有
ssl_key_password	密碼	沒有
ssl_keystore_location	有效的文件系統路徑	沒有
ssl_keystore_password	密碼	沒有
ssl_keystore_type	串	沒有
ssl_truststore_location	有效的文件系統路徑	沒有
ssl_truststore_password	密碼	沒有
ssl_truststore_type	串	沒有
topics	數組	沒有
topics_pattern	串	沒有
value_deserializer_class	串	沒有

 

auto_commit_interval_ms

• 值類型為字符串
• 默認值為 "5000"

消費者抵消被提交給Kafka的頻率（以毫秒為單位）。

auto_offset_reset

• 值類型為字符串
• 此設置沒有默認值。

Kafka中沒有初始偏移量或偏移量超出范圍時該怎么辦：

• 最早：將偏移量自動重置為最早的偏移量
• 最新：自動將偏移量重置為最新偏移量
• none：如果未找到消費者組的先前偏移量，則向消費者拋出異常
• 其他：向消費者拋出異常。

bootstrap_servers

• 值類型為字符串
• 默認值為 "localhost:9092"

Kafka實例的URL列表，用於建立與集群的初始連接。host1:port1,host2:port2這些網址的形式應為這些url 。這些網址僅用於初始連接，以發現完整的集群成員身份（可能會動態更改），因此該列表不需要包含完整的服務器集（不過，您可能需要多個服務器，以防服務器宕機）。

check_crcs

• 值類型為字符串
• 此設置沒有默認值。

自動檢查消耗的記錄的CRC32。這樣可以確保不會發生在線或磁盤損壞消息的情況。此檢查會增加一些開銷，因此在尋求極端性能的情況下可能會禁用該檢查。

client_id

• 值類型為字符串
• 默認值為 "logstash"

發出請求時傳遞給服務器的ID字符串。其目的是通過允許包含邏輯應用程序名稱來跟蹤IP /端口以外的請求源。

connections_max_idle_ms

• 值類型為字符串
• 此設置沒有默認值。

在此配置指定的毫秒數后關閉空閑連接。

consumer_threads

• 值類型是數字
• 默認值為 1

理想情況下，您應該擁有與分區數一樣多的線程，以實現完美的平衡-線程多於分區意味着某些線程將處於空閑狀態

decorate_events

• 值類型為布爾值
• 默認值為 false

向事件添加Kafka元數據（如主題，消息大小）的選項。這將添加一個名為kafkalogstash事件的字段，其中包含以下屬性：

• topic：此消息與之相關的主題
• consumer_group：此事件中曾經閱讀的消費者組
• partition：此消息與之關聯的分區
• offset：與該消息關聯的分區的偏移量
• key：包含消息密鑰的ByteBuffer

enable_auto_commit

• 值類型為字符串
• 默認值為 "true"

如果為true，則定期將使用方已經返回的消息的偏移量提交給Kafka。當處理失敗時，將使用此已提交的偏移量作為開始消耗的位置。

exclude_internal_topics

• 值類型為字符串
• 此設置沒有默認值。

內部主題的記錄（例如偏移量）是否應向消費者公開。如果設置為true，則從內部主題接收記錄的唯一方法是訂閱該主題。

fetch_max_bytes

• 值類型為字符串
• 此設置沒有默認值。

服務器為獲取請求應返回的最大數據量。這不是絕對最大值，如果提取的第一個非空分區中的第一條消息大於此值，則仍將返回該消息以確保使用者可以取得進展。

fetch_max_wait_ms

• 值類型為字符串
• 此設置沒有默認值。

如果沒有足夠的數據立即滿足，服務器在響應提取請求之前將阻塞的最長時間fetch_min_bytes。此時間應小於或等於在poll_timeout_ms

fetch_min_bytes

• 值類型為字符串
• 此設置沒有默認值。

服務器為獲取請求應返回的最小數據量。如果沒有足夠的數據，則請求將等待該數據積累，然后再回答請求。

group_id

• 值類型為字符串
• 默認值為 "logstash"

該消費者所屬的組的標識符。使用者組是一個邏輯訂戶，恰好由多個處理器組成。主題中的消息將以相同的方式分發到所有Logstash實例group_id

heartbeat_interval_ms

• 值類型為字符串
• 此設置沒有默認值。

消費者協調員之間心跳的預期時間。心跳用於確保消費者的會話保持活動狀態，並在新消費者加入或離開小組時促進重新平衡。該值必須設置為小於 session.timeout.ms，但通常應設置為不大於該值的1/3。可以將其調整得更低，以控制正常重新平衡的預期時間。

jaas_path

• 值類型是路徑
• 此設置沒有默認值。

Java身份驗證和授權服務（JAAS）API為Kafka提供用戶身份驗證和授權服務。此設置提供JAAS文件的路徑。Kafka客戶端的樣本JAAS文件：

KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useTicketCache=true renewTicket=true serviceName="kafka"; };

請注意，在配置文件中指定jaas_path和kerberos_config將把它們添加到全局JVM系統屬性中。這意味着，如果您有多個Kafka輸入，則所有這些都將共享 jaas_path和kerberos_config。如果不希望這樣做，則必須在不同的JVM實例上運行Logstash的單獨實例。

kerberos_config

• 值類型是路徑
• 此設置沒有默認值。

kerberos配置文件的可選路徑。這是krb5.conf樣式，如https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html中所述

key_deserializer_class

• 值類型為字符串
• 默認值為 "org.apache.kafka.common.serialization.StringDeserializer"

用於反序列化記錄鍵的Java類

max_partition_fetch_bytes

• 值類型為字符串
• 此設置沒有默認值。

服務器將返回的每個分區的最大數據量。用於請求的最大總內存為#partitions * max.partition.fetch.bytes。此大小必須至少與服務器允許的最大消息大小一樣大，否則，生產者有可能發送大於消費者可以獲取的消息。如果發生這種情況，使用者可能會在嘗試在某個分區上獲取大消息時陷入困境。

max_poll_interval_ms

• 值類型為字符串
• 此設置沒有默認值。

使用使用者組管理時，調用poll（）之間的最大延遲。這為使用者在獲取更多記錄之前可以處於空閑狀態的時間設置了上限。如果在此超時到期前未調用poll（），則認為使用方失敗，該組將重新平衡以將分區重新分配給另一個成員。配置的值request_timeout_ms必須始終大於max_poll_interval_ms

max_poll_records

• 值類型為字符串
• 此設置沒有默認值。

一次調用poll（）返回的最大記錄數。

metadata_max_age_ms

• 值類型為字符串
• 此設置沒有默認值。

以毫秒為單位的時間段，在此之后，即使我們沒有看到任何分區領導更改也可以強制刷新元數據以主動發現任何新的代理或分區

partition_assignment_strategy

• 值類型為字符串
• 此設置沒有默認值。

客戶端用於在使用者實例之間分配分區所有權的分區分配策略的類名稱。映射到Kafka partition.assignment.strategy設置，默認設置為org.apache.kafka.clients.consumer.RangeAssignor。

poll_timeout_ms

• 值類型是數字
• 默認值為 100

時間卡夫卡消費者將等待接收來自主題的新消息

receive_buffer_bytes

• 值類型為字符串
• 此設置沒有默認值。

讀取數據時要使用的TCP接收緩沖區（SO_RCVBUF）的大小。

reconnect_backoff_ms

• 值類型為字符串
• 此設置沒有默認值。

嘗試重新連接到給定主機之前要等待的時間。這樣可以避免以緊密的循環重復連接到主機。此退回適用於消費者發送給代理的所有請求。

request_timeout_ms

• 值類型為字符串
• 此設置沒有默認值。

該配置控制客戶端等待請求響應的最長時間。如果超時之前仍未收到響應，則客戶端將在必要時重新發送請求，如果重試已用盡，則會使請求失敗。

retry_backoff_ms

• 值類型為字符串
• 此設置沒有默認值。

嘗試對給定主題分區重試失敗的獲取請求之前等待的時間。這避免了在緊密的循環中反復進行獲取和失敗。

sasl_jaas_config

• 值類型為字符串
• 此設置沒有默認值。

此插件實例本地的JAAS配置設置，與使用通過進行配置的配置文件（jaas_path通過JVM共享）相反。這允許每個插件實例都有其自己的配置。

如果同時設置sasl_jaas_config和jaas_path配置，則此處的設置優先。

示例（為Azure Event Hub設置）：

  input { kafka { sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='auser' password='apassword';" } }

sasl_kerberos_service_name

• 值類型為字符串
• 此設置沒有默認值。

Kafka代理運行時使用的Kerberos主體名稱。這可以在Kafka的JAAS配置中定義，也可以在Kafka的配置中定義。

sasl_mechanism

• 值類型為字符串
• 默認值為 "GSSAPI"

用於客戶端連接的SASL機制。這可以是安全提供程序可用的任何機制。GSSAPI是默認機制。

security_protocol

• 值可以是任何的：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL
• 默認值為 "PLAINTEXT"

要使用的安全協議，可以是PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL之一

send_buffer_bytes

• 值類型為字符串
• 此設置沒有默認值。

發送數據時要使用的TCP發送緩沖區（SO_SNDBUF）的大小

session_timeout_ms

• 值類型為字符串
• 此設置沒有默認值。

超時后，如果poll_timeout_ms未調用，則將使用者標記為已死，並為由標識的組觸發重新平衡操作group_id

ssl_endpoint_identification_algorithm

• 值類型為字符串
• 默認值為 "https"

端點識別算法，默認為"https"。設置為空字符串""以禁用端點驗證

ssl_key_password

• 值類型為密碼
• 此設置沒有默認值。

密鑰存儲文件中私鑰的密碼。

ssl_keystore_location

• 值類型是路徑
• 此設置沒有默認值。

如果需要客戶端身份驗證，則此設置存儲密鑰庫路徑。

ssl_keystore_password

• 值類型為密碼
• 此設置沒有默認值。

如果需要客戶端身份驗證，則此設置存儲密鑰庫密碼

ssl_keystore_type

• 值類型為字符串
• 此設置沒有默認值。

密鑰庫類型。

ssl_truststore_location

• 值類型是路徑
• 此設置沒有默認值。

用於驗證Kafka經紀人證書的JKS信任庫路徑。

ssl_truststore_password

• 值類型為密碼
• 此設置沒有默認值。

信任庫密碼

ssl_truststore_type

• 值類型為字符串
• 此設置沒有默認值。

信任庫類型。

topics

• 值類型為數組
• 默認值為 ["logstash"]

要訂閱的主題列表，默認為[“ logstash”]。

topics_pattern

• 值類型為字符串
• 此設置沒有默認值。

要訂閱的主題正則表達式模式。使用此配置時，主題配置將被忽略。

value_deserializer_class

• 值類型為字符串
• 默認值為 "org.apache.kafka.common.serialization.StringDeserializer"

用於反序列化記錄值的Java類

常用選項

所有輸入插件均支持以下配置選項：

設置	輸入類型	需要
add_field	雜湊	沒有
codec	編解碼器	沒有
enable_metric	布爾值	沒有
id	串	沒有
tags	數組	沒有
type	串	沒有

 

add_field

• 值類型為哈希
• 默認值為 {}

向事件添加字段

codec

• 值類型為編解碼器
• 默認值為 "plain"

用於輸入數據的編解碼器。輸入編解碼器是一種在數據輸入之前解碼數據的便捷方法，而無需在Logstash管道中使用單獨的過濾器。

enable_metric

• 值類型為布爾值
• 默認值為 true

默認情況下，為此特定插件實例禁用或啟用度量標准日志記錄，我們會記錄所有可以度量的數據，但是您可以禁用特定插件的度量標准收集。

id

• 值類型為字符串
• 此設置沒有默認值。

ID向插件配置添加唯一。如果未指定ID，Logstash將生成一個。強烈建議在您的配置中設置此ID。當您有兩個或多個相同類型的插件時，例如在您有2個kafka輸入時，這特別有用。在這種情況下，添加命名ID將有助於在使用監視API時監視Logstash。

input { kafka { id => "my_plugin_id" } }

tags

• 值類型為數組
• 此設置沒有默認值。

將任意數量的任意標簽添加到您的事件中。

這可以幫助以后進行處理。

type

• 值類型為字符串
• 此設置沒有默認值。

type向此輸入處理的所有事件添加一個字段。

 

均不支持批處理所使用的操作。

Redis輸入配置選項

該插件支持以下配置選項以及稍后介紹的“ 通用選項”。

設置	輸入類型	需要
batch_count	數	沒有
data_type	字符串，其中之一["list", "channel", "pattern_channel"]	是
db	數	沒有
host	串	沒有
path	串	沒有
key	串	是
password	密碼	沒有
port	數	沒有
ssl	布爾值	沒有
threads	數	沒有
timeout	數	沒有
command_map	hash	沒有

另請參閱通用選項，以獲取所有輸入插件支持的選項列表。

 

batch_count

• 值類型是數字
• 默認值為 125

使用EVAL從Redis返回的事件數。

data_type

• 這是必需的設置。
• 值可以是任何的：list，channel，pattern_channel
• 此設置沒有默認值。

指定列表或頻道。如果data_type為list，則我們將對密鑰進行BLPOP鎖定。如果data_type為channel，那么我們將訂閱該密鑰。如果data_type為pattern_channel，則我們將訂閱該密鑰。

db

• 值類型是數字
• 默認值為 0

Redis數據庫號。

host

• 值類型為字符串
• 默認值為 "127.0.0.1"

Redis服務器的主機名。

path

• 值類型為字符串
• 此設置沒有默認值。
• 如果兩者都指定，路徑將覆蓋主機配置。

Redis服務器的unix套接字路徑。

key

• 這是必需的設置。
• 值類型為字符串
• 此設置沒有默認值。

Redis列表或通道的名稱。

password

• 值類型為密碼
• 此設置沒有默認值。

用於驗證的密碼。默認情況下沒有身份驗證。

port

• 值類型是數字
• 默認值為 6379

要連接的端口。

ssl

• 值類型為布爾值
• 默認值為 false

啟用S​​SL支持。

threads

• 值類型是數字
• 默認值為 1

timeout

• 值類型是數字
• 默認值為 5

初始連接超時（以秒為單位）。

command_map

• 值類型為哈希
• 此設置沒有默認值。
• key是默認命令名稱，value是重命名的命令。

以“舊名稱”⇒“新名稱”的形式配置重命名的redis命令。Redis允許在其協議中重命名或禁用命令，請參見：https : //redis.io/topics/security

常用選項

所有輸入插件均支持以下配置選項：

設置	輸入類型	需要
add_field	hash	沒有
codec	編解碼器	沒有
enable_metric	布爾值	沒有
id	串	沒有
tags	數組	沒有
type	串	沒有

詳細信息 

add_field

• 值類型為哈希
• 默認值為 {}

向事件添加字段

codec

• 值類型為編解碼器
• 默認值為 "json"

用於輸入數據的編解碼器。輸入編解碼器是一種在數據輸入之前解碼數據的便捷方法，而無需在Logstash管道中使用單獨的過濾器。

enable_metric

• 值類型為布爾值
• 默認值為 true

默認情況下，為此特定插件實例禁用或啟用度量標准日志記錄，我們會記錄所有可以度量的數據，但是您可以禁用特定插件的度量標准收集。

id

• 值類型為字符串
• 此設置沒有默認值。

ID向插件配置添加唯一。如果未指定ID，Logstash將生成一個。強烈建議在您的配置中設置此ID。當您有兩個或兩個以上相同類型的插件時，例如在您有2個redis輸入時，此功能特別有用。在這種情況下，添加命名ID將有助於在使用監視API時監視Logstash。

input { redis { id => "my_plugin_id" } }

tags

• 值類型為數組
• 此設置沒有默認值。

將任意數量的任意標簽添加到您的事件中。

這可以幫助以后進行處理。

type

• 值類型為字符串
• 此設置沒有默認值。

type向此輸入處理的所有事件添加一個字段。

 

 

TCP輸入插件

Tcp輸入配置選項

該插件支持以下配置選項以及稍后介紹的“ 通用選項”。

設置	輸入類型	需要
host	串	沒有
mode	字符串，其中之一["server", "client"]	沒有
port	數	是
proxy_protocol	布爾值	沒有
ssl_cert	有效的文件系統路徑	沒有
ssl_certificate_authorities	數組	沒有
ssl_enable	布爾值	沒有
ssl_extra_chain_certs	數組	沒有
ssl_key	有效的文件系統路徑	沒有
ssl_key_passphrase	密碼	沒有
ssl_verify	布爾值	沒有
tcp_keep_alive	布爾值	沒有
dns_reverse_lookup_enabled	布爾值	沒有

 

host

• 值類型為字符串
• 默認值為 "0.0.0.0"

當mode為時server，要監聽的地址。當mode為時client，要連接的地址。

mode

• 值可以是任何的：server，client
• 默認值為 "server"

運行模式。server偵聽客戶端連接， client連接到服務器。

port

• 這是必需的設置。
• 值類型是數字
• 此設置沒有默認值。

當mode為時server，要監聽的端口。當mode為時client，要連接的端口。

proxy_protocol

• 值類型為布爾值
• 默認值為 false

代理協議支持，目前僅支持v1 http://www.haproxy.org/download/1.5/doc/proxy-protocol.txt

ssl_cert

• 值類型是路徑
• 此設置沒有默認值。

PEM格式的證書路徑。該證書將顯示給連接的客戶端。

ssl_certificate_authorities

• 值類型為數組
• 默認值為 []

根據這些權限驗證客戶證書或證書鏈。您可以定義多個文件或路徑。所有證書將被讀取並添加到信任存儲中。

ssl_enable

• 值類型為布爾值
• 默認值為 false

啟用S​​SL（必須設置其他ssl_選項才能生效）。

ssl_extra_chain_certs

• 值類型為數組
• 默認值為 []

額外的X509證書的路徑數組。這些證書與證書一起使用以構造提供給客戶端的證書鏈。

ssl_key

• 值類型是路徑
• 此設置沒有默認值。

對應於指定證書（PEM格式）的私鑰的路徑。

ssl_key_passphrase

• 值類型為密碼
• 默認值為 nil

私鑰的SSL密鑰密碼。

ssl_verify

• 值類型為布爾值
• 默認值為 true

根據CA驗證SSL連接另一端的身份。對於輸入，將字段設置sslsubject為客戶端證書的字段。

tcp_keep_alive

• 值類型為布爾值
• 默認值為 false

指示套接字使用TCP保持活動。使用操作系統默認設置保持活動設置。

dns_reverse_lookup_enabled

• 值類型為布爾值
• 默認值為 true

通過禁用此設置可以避免DNS反向查找。如果禁用，添加到事件的地址元數據將包含在TCP層指定的源地址，並且IP不會解析為主機名。

常用選項

所有輸入插件均支持以下配置選項：

設置	輸入類型	需要
add_field	雜湊	沒有
codec	編解碼器	沒有
enable_metric	布爾值	沒有
id	串	沒有
tags	數組	沒有
type	串	沒有

詳細信息編輯

add_field

• 值類型為哈希
• 默認值為 {}

向事件添加字段

codec

• 值類型為編解碼器
• 默認值為 "line"

用於輸入數據的編解碼器。輸入編解碼器是一種在數據輸入之前解碼數據的便捷方法，而無需在Logstash管道中使用單獨的過濾器。

enable_metric

• 值類型為布爾值
• 默認值為 true

默認情況下，為此特定插件實例禁用或啟用度量標准日志記錄，我們會記錄所有可以度量的數據，但是您可以禁用特定插件的度量標准收集。

id

• 值類型為字符串
• 此設置沒有默認值。

ID向插件配置添加唯一。如果未指定ID，Logstash將生成一個。強烈建議在您的配置中設置此ID。當您有兩個或多個相同類型的插件時（例如，如果您有2個tcp輸入），這特別有用。在這種情況下，添加命名ID將有助於在使用監視API時監視Logstash。

input { tcp { id => "my_plugin_id" } }

tags

• 值類型為數組
• 此設置沒有默認值。

將任意數量的任意標簽添加到您的事件中。

這可以幫助以后進行處理。

type

• 值類型為字符串
• 此設置沒有默認值。

type向此輸入處理的所有事件添加一個字段。