logstash输入File、redis、filebeats、kafka

Logstash输入之File

File输入插件简介

input部分的file插件，file插件是从指定的文件中读取数据，虽然只是读取数据，但是却有两种操作方法：Tail模式和Read模式
Tail模式：在此模式下，指定的文件被视为一直在写入内容，例如我们的apache日志为access.log，我们的file插件去不间断的读取日志内容就像一直在使用tail -f命令一样，一直在读取文件，默认为Tail模式。
Read模式：在此模式下，指定的文件将被视为一个完整的文件，当logstash启动那一刻起，被指定文件的数据内容大小就被限制了，即使文件还是在写入，logstash也只会读取到刚启动的位置。

File输入插件配置选项

设置	输入类型	描述	默认值
close_older	number or string_duration	可选	1 hour
path	array	必须	无
exclude	array	可选	无
file_chunk_size	number	可选	32768=32kb
file_completed_action	值可以是:delete or log or log_and_delete	可选	delete
file_completed_log_path	string	可选	无
mode	值可以是:tail or read	可选	tail
start_position	值可以是:beginning or end	可选	end
stat_interval	值可以是:number or string_duration	可选	1 second

close_older

• 值类型是number或string_duration
• 默认值为“1 hour”
• 描述：
  文件输入关闭上次读取指定持续时间之前的任何文件（如果指定了数字、则为秒），这取决于是否对文件进行跟踪或读取，如果跟踪,并且输入数据存在较大的时间间隔，则可以关闭该文件（允许打开其它文件），但当检测到新数据时，将排队等待重新打开，如果正在读取，文件将在closed_older秒之后关闭，从读取最后一个字节开始，如果你将插件升级到4.1.0+，正在读取而没有跟踪，并且没有切换到使用读取模式，则保留此设置已实现向后兼容。

path

• 这是在file插件中必须要设置的选项
• 值类型为数组
• 没有默认值
• 作用：此设置是作用file输入插件的文件读取路径

举个栗子：

path => "var/log/access.log"
#如果多个文件路径可以写为数组格式
path => [ "var/log/access.log","/etc/httpd/logs/access.log" ]
#也可以进行模糊匹配
path => [ "var/log/*.log","/etc/httpd/**/*.log","/nginx/logs/*"]

exclude

• 值类型是数组
• 没有默认值
• 描述：当我们使用path配置项指定了一个/nginx/logs/*下的所有日志都被读取的时候，假如此目录下有已被压缩的文件，例如是 *.tar.gz or *.gz结尾的压缩文件，我们可以使用exclude来排除它

举个栗子：

path => "/var/log/*"
exclude => "*.gz"

file_chunk_size

• 值类型默认为数字
• 默认值为32768(32kb)
• 描述：文件内存以块的形式从磁盘中读取，并从块中提取行，块的默认大小为32768，

举个栗子：
我们把默认的块大小32768改为了65536

file_chunk_size => "65536"

file_completed_action

• 值类型可以是：delete or log or log_and_delete
• 默认值为delete
• 描述：在read读取模式下，当文件读取完成时应该执行什么操作？

1）delete：如果指定了delete，那么文件读取完成后，将被删除。
2）log：如果指定了log，文件读取完成后，文件的完整路径会记录到 file_completed_log_path 指定的文件路径中
3）log_and_delete：如果指定了 log_and_delete，那么文件读取完成后将会被删除，并且将文件的完整路径记录到 file_completed_log_path 指定的文件路径中。

file_completed_log_path

• 值类型是string字符串
• 没有默认值
• 描述：指定路径，只有设置file_completed_action的值为log或者log_and_delete时，此值才生效

举个栗子：

 file_completed_action => "log_and_delete"file_completed_log_path => "/usr/local/logstash-7.0.0/file_completed/"

mode

• 值类型为 tail or read
• 默认值为tail
• 描述：此模式即是我们开篇将的两种读取模式，一种为tail跟踪，一种为read读取，
• tail模式下不可能读取*.gz的压缩文件，read模式下支持*.gz文件的读取
  如果指定了read模式，则以下几个配置项将失效

1) start_position (始终从头开始读取文件)
2) close_older (达到EOF时文件会自动关闭)

如果指定read模式，则注意以下配置：

ignore_older (不能处理旧文件)
file_completed_action (处理文件后应执行什么操作)
file_completed_log_path (应该将完成的文件路径记录到哪个文件)

举个栗子：

    mode => "read"file_chunk_size => "65536"file_completed_action => "log_and_delete"file_completed_log_path => "/usr/local/logstash-7.0.0/file_completed/"

start_position

• 值可以是beginning or end
• 默认值end
• 描述：选择logstash最初开始读取文件的位置，开头或结尾，默认为end，代表对一个文件使用tail -f命令，beginning为从文件行首开始读取内容。

举个栗子：

start_position => "beginning"

stat_interval

• 值类型是number or string_duration
• 默认值为1 second
• 描述：统计文件的频率的间隔，以秒为单位，看看他们是否被修改，增加这个间隔会减少系统调用的次数，但是会增加检测新日志行的时间。

通用配置选项

设置	输入类型	描述
add_field	hash	可选
codec	codec	可选
enable_metric	boolean	可选
id	string	可选
tags	array	可选
type	string	可选

add_field
哈希是一组键值对的集合，以 key => value格式置顶，多个键值条目由空格分隔

• 值类型是哈希值
• 默认为{ }
• 作用：向配置中添加字段

举个栗子：

add_field => { "Key1" => "value1""Key2" => "value2"...
}
或者也可以写成这样
add_field => { "Key1" => "value1" "Key2" => "value2" }

codec
用于输入数据的编解码器。输入编解码器是一种在数据进入输入之前解码数据的便捷方法，无需在Logstash管道中使用单独的过滤器。

• 值类型是编解码器
• 默认是为"line"
• 作用：编解码器可用于输入和输出，输入编解码器提供了一种方法的方法，可以在数据输入之前对其进行解码，输出编解码器可在数据离开输出之前对其进行编码，也就是对数据进行了一次拆封在封装，封装格式为codec所指定的格式，如果在logstash中设置了编解码器，那么就不需要在logstash管道中再单独使用过滤功能了
  举个栗子：
  我们在输出管道配置文件中设置编解码器的格式为json，默认为line，当数据输出的时候，就是已经经过了编解码器拆封再封装的json格式数据。

codec => "json"

enable_metric
默认情况下，我们会记录我们能记录的所有度量，但是你可以禁用特定插件的度量集合

• 值类型是布尔，布尔型的值只有两个结果，true和false
• 默认值为true
  举个例子：

enable_metric => false

id

• 值类型为字符串
• 此类型没有默认值
  向插件配置添加唯一ID，如果没有指定ID，Logstash将生产一个ID，强烈建议在配置中设置此ID，当你有两个或者多个相同类型插件时，这尤其有用，例如，如果你有两个stdin输入，在本例中添加一个命名ID将有助于在使用监视API时监视Logstash
  大概意思为：在我们stdin输入中添加一个标识符ID，以防多个stdin混淆
  举个栗子：
  我们ID即是为name字段，添加了一个标示而已

 name =>“Hello world”

tags

• 值类型为数组
• 没有默认值
• 作用：在配置中添加标记，有助于以后的处理
  举个栗子：

users => [ {id => 1, name => bob}, {id => 2, name => jane} ]

type
此类型用的较多，主要是为了根据类型值来进行判断输出指向或者过滤内容

• 值类型为字符串
• 没有默认值
• 作用：type指定一个字符串，同样可用作标示，一般会根据判断type值来进行输出或过滤

File输入插件示例配置

下面配置请结合上面讲的选项含义结合参考

input {file {path => [ "/etc/httpd/logs/*","/etc/httpd/logs/*" ]type => "apache"exclude => "*.gz"start_position => "beginning"add_field => { "application" => "httpd" }tags => "web"}file {path => "/var/log/messages"type => "syslog"mode => "read"file_chunk_size => "65536"file_completed_action => "log_and_delete"file_completed_log_path => "/usr/local/logstash-7.0.0/file_completed/"}
}
filter {
}
output {stdout {codec => "rubydebug"}
}

其它的file插件命令选项还有很多很多，但file插件我们常用到的命令选项就是path，更多file插件命令选项请看：https://www.elastic.co/guide/en/logstash/7.0/plugins-inputs-file.html

Logstash输入之Redis

Redis插件简介

通过redis输入插件将从redis实例中读取事件，它同时支持redis通道和列表以及在logstash中使用list命令。

Redis输入插件配置选项

设置	输入类型	描述
batch_count	number	可选
data_type	string	必选
db	number	可选
host	string	可选
key	string	必选
password	password	可选
port	number	可选
threads	number	可选
timeout	number	可选

batch_count

• 值类型为number数字
• 默认为125
• 描述：使用EVAL从redis返回的事件数量

data_type

• 必须设置项
• 值类型可以是:list,channel,pattern_channel其中之一
• 没有默认值
• 描述：指定列表或通道，如果redis数据类型为list，那么将BLPOP键。如果redis数据类型是channel，那么将订阅密钥，如果redis数据类型是pattern_channel，那么同样是订阅密钥。

db

• 值类型为number数字
• 默认为0
• 描述：指定redis的数据号，redis默认有16个数据库，编号为0-15，指定从哪个数据库中读取数据

host

• 值类型为number数字
• 默认值为"127.0.0.1"
• 描述：指定redis的IP地址或者主机名

key

• 必须设置
• 值类型为string字符串
• 没有默认值
• 描述：指定reids列表或通道名称

password

• 值类型为password密码
• 没有默认值
• 描述：如果reids设置了密码认证，则需要启用password选项来指定redis连接密码，redis默认不存在身份验证

port

• 值类型为number数字
• 默认值为6379
• 描述：指定redis的端口

threads

• 值类型为number数字
• 默认为1
• 描述：指定redis输入插件的线程数量，因为reids是单线程的中间件，所以默认为1即可

timeout

• 值类型为number数字
• 默认为5秒
• 描述：logstash初始和redis连接超时时间（以秒为单位）

通用配置选项

设置	输入类型	描述
add_field	hash	可选
codec	codec	可选
enable_metric	boolean	可选
id	string	可选
tags	array	可选
type	string	可选

add_field
哈希是一组键值对的集合，以 key => value格式置顶，多个键值条目由空格分隔

• 值类型是哈希值
• 默认为{ }
• 作用：向配置中添加字段

举个栗子：

add_field => { "Key1" => "value1""Key2" => "value2"...
}
或者也可以写成这样
add_field => { "Key1" => "value1" "Key2" => "value2" }

codec
用于输入数据的编解码器。输入编解码器是一种在数据进入输入之前解码数据的便捷方法，无需在Logstash管道中使用单独的过滤器。

• 值类型是编解码器
• 默认是为"line"
• 作用：编解码器可用于输入和输出，输入编解码器提供了一种方法的方法，可以在数据输入之前对其进行解码，输出编解码器可在数据离开输出之前对其进行编码，也就是对数据进行了一次拆封在封装，封装格式为codec所指定的格式，如果在logstash中设置了编解码器，那么就不需要在logstash管道中再单独使用过滤功能了
  举个栗子：
  我们在输出管道配置文件中设置编解码器的格式为json，默认为line，当数据输出的时候，就是已经经过了编解码器拆封再封装的json格式数据。

codec => "json"

enable_metric
默认情况下，我们会记录我们能记录的所有度量，但是你可以禁用特定插件的度量集合

• 值类型是布尔，布尔型的值只有两个结果，true和false
• 默认值为true
  举个例子：

enable_metric => false

id

• 值类型为字符串
• 此类型没有默认值
  向插件配置添加唯一ID，如果没有指定ID，Logstash将生产一个ID，强烈建议在配置中设置此ID，当你有两个或者多个相同类型插件时，这尤其有用，例如，如果你有两个stdin输入，在本例中添加一个命名ID将有助于在使用监视API时监视Logstash
  大概意思为：在我们stdin输入中添加一个标识符ID，以防多个stdin混淆
  举个栗子：
  我们ID即是为name字段，添加了一个标示而已

 name =>“Hello world”

tags

• 值类型为数组
• 没有默认值
• 作用：在配置中添加标记，有助于以后的处理
  举个栗子：

users => [ {id => 1, name => bob}, {id => 2, name => jane} ]

type
此类型用的较多，主要是为了根据类型值来进行判断输出指向或者过滤内容

• 值类型为字符串
• 没有默认值
• 作用：type指定一个字符串，同样可用作标示，一般会根据判断type值来进行输出或过滤

Redis输入插件示例配置

下面配置请结合上面讲的选项含义结合参考

input {redis {batch_count => "300"data_type => "list"db => "3"key => "list"host => "10.150.30.246"port => "6003"timeout => "30"}
}
output {stdout {codec => rubydebug}
}

Logstash输入之Filebeats

Filebeats输入插件简介

通过filebeats插件可以使logstash接收到filebeats所传输来的数据

Filebeats输入插件配置选项

设置	输入类型	描述	默认值
port	number	必选	无
add_hostname	boolean	可选	false
cipher_suites	array	可选
client_inactivity_timeout	number	可选	60
host	string	可选	0.0.0.0
include_codec_tag	boolean	可选	true
ssl	boolean	可选	false
ssl_certificate	file path	可选	无
ssl_certificate_authorities	array	可选	[]
ssl_handshake_timeout	number	可选	10000ms
ssl_key	file path	可选	无
ssl_key_passphrase	password	可选	无
ssl_verify_mode	string	可选	none
ssl_peer_metadata	boolean	可选	false
tls_max_version	number	可选	1.2
tls_min_version	number	可选	1

port

• 必须设置
• 值类型是数字
• 描述：在logstash的input部分中filebeats插件指定filebeats的端口

add_hostname

• 值类型是布尔型
• 默认值为false
• 描述：标志，以确定是否使用hostname字段中beat提供的值降host字段添加到事件中。

cipher_stites

• 值类型是数组
• 默认值是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
• 描述：在闲置60秒后关闭和filebeats端连接

host

• 值类型是字符串
• 默认值是0.0.0.0
• 描述：指定filebeats端的IP地址

include_codec_tag

• 值类型是布尔型
• 默认值为true

SSL

• 值类型为布尔
• 默认为false
• 描述：默认filebeats会以明文文本的形式进行发送数据，当我们将此值设置为true并配置了ssl_certificate和ssl_key选项来启用加密

ssl_certificate

• 值类型为路径
• 没有默认值
• 描述：指定要使用的SSL证书路径

ssl_key

• 值类型为路径
• 没有默认值
• 描述：指定要使用的SSL密钥路径

ssl_certificate_authorities

• 值类型为数组
• 默认值为[]
• 描述：针对这些权限验证客户端证书；你可以定义多个文件或路径；所有证书都将被读取并且添加到信任区；需要将ssl_verify_mode配置为peer或者force_peer来启用验证

ssl_handshake_timeout

• 值类型是数字
• 默认值是10000ms
• 描述：未完成ssl握手超时的时间，单位：毫秒

ssl_key_passphrase

• 默认值是密码
• 没有默认值
• 描述：要使用的SSL密钥密码

ssl_verify_mode

• 值类型可以是:none、peer、force_peer中的任意一个
• 默认值为none
• 描述：默认情况下为none，服务器不做任何客户端验证。
  peer：peer模式中服务器将请求客户端提供证书，如果客户端提供证书，将对其进行验证。
  force_peer：将使用服务器请求客户端提供证书，如果客户端不提供证书，服务器将主动关闭连接。
  注意：此选项需要与ssl_certificate_authority和定义的ca列表一起使用。

ssl_peer_metadata

• 值类型为布尔
• 默认为false
• 描述：启用在事件的元数据中存储客户端证书信息；只有当ssl_verify_mode设置为peer或者force_peer时，此选项才有效。

tls_max_version

• 值类型为数字
• 默认值为1.2
• 描述：允许加密连接的TLS最大版本，该值必须是以下值之一1.0 for TLS 1.0, 1.1 for TLS 1.1, 1.2 for TLS 1.2

tls_min_version

• 值类型为数字
• 默认值为1
• 描述：允许加密连接的最小TLS版本。该值必须是以下值之一:1.0 for TLS 1.0, 1.1 for TLS 1.1, 1.2 for TLS 1.2

通用配置选项

设置	输入类型	描述
add_field	hash	可选
codec	codec	可选
enable_metric	boolean	可选
id	string	可选
tags	array	可选
type	string	可选

add_field
哈希是一组键值对的集合，以 key => value格式置顶，多个键值条目由空格分隔

• 值类型是哈希值
• 默认为{ }
• 作用：向配置中添加字段

举个栗子：

add_field => { "Key1" => "value1""Key2" => "value2"...
}
或者也可以写成这样
add_field => { "Key1" => "value1" "Key2" => "value2" }

codec
用于输入数据的编解码器。输入编解码器是一种在数据进入输入之前解码数据的便捷方法，无需在Logstash管道中使用单独的过滤器。

• 值类型是编解码器
• 默认是为"line"
• 作用：编解码器可用于输入和输出，输入编解码器提供了一种方法的方法，可以在数据输入之前对其进行解码，输出编解码器可在数据离开输出之前对其进行编码，也就是对数据进行了一次拆封在封装，封装格式为codec所指定的格式，如果在logstash中设置了编解码器，那么就不需要在logstash管道中再单独使用过滤功能了
  举个栗子：
  我们在输出管道配置文件中设置编解码器的格式为json，默认为line，当数据输出的时候，就是已经经过了编解码器拆封再封装的json格式数据。

codec => "json"

enable_metric
默认情况下，我们会记录我们能记录的所有度量，但是你可以禁用特定插件的度量集合

• 值类型是布尔，布尔型的值只有两个结果，true和false
• 默认值为true
  举个例子：

enable_metric => false

id

• 值类型为字符串
• 此类型没有默认值
  向插件配置添加唯一ID，如果没有指定ID，Logstash将生产一个ID，强烈建议在配置中设置此ID，当你有两个或者多个相同类型插件时，这尤其有用，例如，如果你有两个stdin输入，在本例中添加一个命名ID将有助于在使用监视API时监视Logstash
  大概意思为：在我们stdin输入中添加一个标识符ID，以防多个stdin混淆
  举个栗子：
  我们ID即是为name字段，添加了一个标示而已

 name =>“Hello world”

tags

• 值类型为数组
• 没有默认值
• 作用：在配置中添加标记，有助于以后的处理
  举个栗子：

users => [ {id => 1, name => bob}, {id => 2, name => jane} ]

type
此类型用的较多，主要是为了根据类型值来进行判断输出指向或者过滤内容

• 值类型为字符串
• 没有默认值
• 作用：type指定一个字符串，同样可用作标示，一般会根据判断type值来进行输出或过滤

Filebeats输入插件示例配置

下面配置请结合上面讲的选项含义结合参考

input {beats {port => 5044host => "10.150.30.242"client_inactivity_timeout => 120type => filebeats}
}
output {stdout {codec => "rubydebug"}

Logstash7.0输入之Kafka

Kafka输入插件简介

在logstash使用kafka插件将从kafka中读取数据，此插件需要使用kafka client 2.1.0客户端版本。有关代理兼容性，请参阅kafka官方兼容性参考资料。
Kafka插件支持通过以下方式连接到kakfa：

• SSL (需要kafka插件版本在3.0.0及以上版本)
• Kerberos SASL (需要Kafka插件版本5.1.0及以上版本)
  不过不用担心，现在logstash对kafka输入插件提供的版本为v9.0.0
  默认情况下，安全认证是禁用的，但可以根据需要打开安全认证

Logstash Kafka使用者使用Kafka处理组管理并使用默认的偏移管理策略。
默认情况下，Logstash Kafka实例组成一个逻辑组Consumer Group来订阅Kafka，每个Logstash Kafka使用者可以运行多个线程来增加读取吞吐量。或者，您可以使用相同的group_id运行多个Logstash实例，将负载分散到物理机器上。kafka中的消息将分发给所有具有相同group_id的Logstash实例。

有关更多信息，请参阅：Kafka更多信息
Kafka消费者配置：Kafka消息者配置

Kafka输入插件配置选项

设置	输入类型	描述
auto_commit_interval_ms	string	可选
auto_offset_reset	string	可选
bootstrap_servers	string	可选
check_crcs	string	可选
client_id	string	可选
connections_max_idle_ms	string	可选
consumer_threads	number	可选
decorate_events	boolean	可选
enable_auto_commit	string	可选
exclude_internal_topics	string	可选
fetch_max_bytes	string	可选
fetch_max_wait_ms	string	可选
fetch_min_bytes	string	可选
group_id	string	可选
heartbeat_interval_ms	string	可选
jaas_path	path	可选
kerberos_config	path	可选
key_deserializer_class	string	可选
max_partition_fetch_bytes	string	可选
max_poll_interval_ms	string	可选
max_poll_records	string	可选
metadata_max_age_ms	string	可选
partition_assignment_strategy	string	可选
poll_timeout_ms	number	可选
receive_buffer_bytes	string	可选
reconnect_backoff_ms	string	可选
request_timeout_ms	string	可选
retry_backoff_ms	string	可选
sasl_kerberos_service_name	string	可选
sasl_mechanism	string	可选
security_protocol	string	可选
send_buffer_bytes	string	可选
session_timeout_ms	string	可选
ssl_endpoint_identification_algorithm	string	可选
ssl_key_password	password	可选
ssl_keystore_location	path	可选
ssl_keystore_password	password	可选
ssl_keystore_type	string	可选
ssl_truststore_location	path	可选
ssl_truststore_password	password	可选
ssl_truststore_type	string	可选
topics	array	可选
topics_pattern	string	可选
value_deserializer_class	string	可选

auto_commit_interval_ms

• 值类型是string字符串
• 默认值是5000
• 描述：Kafka消费者将偏移量提交给Kafka的频率，单位为毫秒5000ms=5s

auto_offset_reset

• 值类型是string字符串
• 没有默认值
• 描述：Kafka中没有初始偏移量或者偏移量超出范围怎么办？以下四种值
  1）earliest：Kafka自动将偏移量重置为最老的偏移量
  2）latest：Kafka自动将偏移量重置为最新的偏移量
  3）none：Kafka如果没有找到消费者组原来的偏移量，则向消费者抛出异常
  4）anything else：Kafka向消费者抛出异常

bootstrap_servers

• 值类型为string字符串
• 默认值是localhost:9092
• 描述：Logstash连接Kafka实例的列表，如果为Kafka集群，可以写为 host1:port1,host2:port2，逗号分隔

check_crcs

• 值类型是string字符串
• 没有默认值
• 描述：自动检查所使用记录的CRC32。这确保不会发生消息的联机或磁盘损坏。这种检查增加了一些开销，所以在寻求极端性能的情况下可能会禁用它。

client_id

• 值类型为string字符串
• 默认值为logstash
• 描述：发出请求时传递给服务器的ID字符串，这样做的目的是允许通过包含逻辑应用程序名称来跟踪超出ip/port的请求源

connections_max_idle_ms

• 值类型为string字符串
• 没有默认值
• 描述：在此配置指定的毫秒后关闭Logstash与Kafka的空闲连接

consumer_threads

• 值类型是number数字
• 默认值为1
• 描述：Kafka消费者线程数，建议与Kafka Topic中的分区数量相同，过多的线程会处于空闲状态，浪费资源

decorate_events

• 值类型为boolean布尔
• 默认值为false
• 描述：该选项可将Kafka元数据(如Topic，消息大小)添加到事件中，这将向logstash事件添加一个名为kafka的字段，该字段包含以下属性:
  1）Topic：此消息与关联的主题
  2）consumer_group：消费者组用于在此事件中读取
  3）partition：此消息关联的分区
  4）offset：与此消息关联分区的偏移量
  5）key：包含消息键的字节缓冲区

enable_auto_commit

• 值类型为string字符串
• 默认值为true
• 描述：如果为真，则定期会把Logstash消费过Kafka的消息偏移量提交给Kafka，当流程失败时，将使用这个提交的偏移量作为开始消费的位置

exelude_internal_topics

• 值类型为string字符串
• 没有默认值
• 描述：是否将内部主题(如偏移量)的记录暴露给消费者,如果设置为true，则从内部主题接收记录的唯一方法是订阅它

fetch_max_bytes

• 值类型是string字符串
• 没有默认值
• 描述：服务器应该为获取请求返回的最大数据量。这不是一个绝对最大值，如果fetch的第一个非空分区中的第一个消息大于这个值，那么仍然会返回该消息，以确保使用者能够取得进展。

fetch_max_wait_ms

• 值类型是string字符串
• 没有默认值
• 描述：如果没有足够的数据满足fetch_min_bytes，服务器将在响应fetch请求之前阻塞的最长时间。这应该小于或等于poll_timeout_ms中使用的超时

fetch_min_bytes

• 值类型为string字符串
• 没有默认值
  服务器应该为获取请求返回的最小数据量。如果没有足够的数据可用，则请求将等待大量数据累积后才响应请求。

group_id

• 值类型是string字符串
• 默认为logstash
• 描述：此Logstash消费者所属的消费者组名称，消费者组是由多个消费者实例的单个订阅服务器，Topic中的消息将分发给所有具有相同group_id的logstash实例

heartbeat_interval_ms

• 值类型为string字符串
• 没有默认值
• 描述：消费者协调器的预期心跳间隔时间。心跳用于确保用户会话保持活跃，并在新用户加入或离开组时促进再平衡。该值必须设置为session.timeout以下。，但通常应设置不高于该值的1/3。它可以调整甚至更低，以控制正常再平衡的预期时间。

jaas_path

• 值类型为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

• 值类型为path路径
• 没有默认值
• 描述：kerberos配置文件的可选路径。这是krb5.conf样式，详见https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html

key_deserializer_class

• 值类型是string字符串
• 默认值是”org.apache.kafka.common.serialization.StringDeserializer”
• 描述：用于反序列化记录密钥的JAVA类

max_poll_interval_ms

• 值类型是string字符串
• 没有默认值
• 描述：使用消费者组时，poll()调用之间的最大延迟，这就为消费者在获取更多记录之前空闲的时间设置了上限，如果在此超时过期之前未调用poll(),则认为使用者失败，消费者组将重新平衡，以便将分区重新分配给另一个成员。配置request_timeout_ms的值必须始终大于max_poll_interval_ms

max_poll_records

• 值类型为string字符串
• 没有默认值
• 描述：在一次对poll()的调用中返回的最大记录数

poll_timeout_ms

• 值类型为number数字
• 默认值为100
• 描述：Kafka消费者将等待从主题接收新消息

receive_buffer_bytes

• 默认值为string数字
• 没有默认值
• 描述：读取数据时使用TCP接收缓冲区(SO_RCVBUF)的大小

reconnect_backoff_ms

• 值类型为string字符串
• 没有默认值
• 描述：尝试重新连接到给定主机之前等待的时间。这避免了在紧密循环中重复连接到主机。此回退适用于消费者发送给代理的所有请求。

request_timeout_ms

• 值类型为string字符串
• 没有默认值
• 描述：Kafka消费者等待请求响应的最大时间，如果超过此时间没有接收到响应，消费者将在必要时重新发送请求，直到重新发送请求耗尽时请求失败

sasl_kerberos_service_name

• 值类型为string字符串
• 没有默认值
• 描述：Kafka代理运行的Kerberos主体名称。这可以在Kafka的JAAS配置或Kafka的配置中定义

sasl_mechanism

• 值类型是string字符串
• 默认值是”GSSAPI”
• 描述：用于客户端连接的SASL机制。这可以是安全提供者可用的任何机制。GSSAPI是默认机制。

security_protocol

• 值类型可以是：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL
• 默认值为PLAINTEXT
• 描述：Kafka消费者要使用的安全协议，可以上以上几种协议

send_buffer_bytes

• 值类型是字符串
• 没有默认值
• 描述：发送数据时使用TCP发送缓冲区(SO_SNDBUF)的大小

session_timeout_ms

• 值类型是字符串
• 没有默认值
• 描述：超时之后，如果poll_timeout_ms未调用，则将消费者标记为死，并为消费者组触发重新平衡操作group_id

topics

• 值类型是数组array
• 默认值是[“logstash”]
• 描述：要订阅的主题列表，默认为[“logstash”]

topic_pattern

• 值类型是字符串string
• 没有默认值
• 描述：要订阅的主题正则表达式模式。使用此配置时将忽略主题配置。

Kafka输入插件示例配置

cat /opt/application/logstash-7.0.0/conf/logstash-kafka.conf 
input {kafka {auto_offset_reset => "earliest"bootstrap_servers => "10.150.55.94:9092,10.150.55.95:9092,10.150.50.223:9092"consumer_threads => "3"group_id => "logstash_group"heartbeat_interval_ms => ""request_timeout_ms => "30000"topics => "nginx_log"}
}
filter {json {source => "message"}
}
output {stdout {codec => rubydebug}
}

检查配置

send_buffer_bytes

• 值类型是字符串
• 没有默认值
• 描述：发送数据时使用TCP发送缓冲区(SO_SNDBUF)的大小

session_timeout_ms

• 值类型是字符串
• 没有默认值
• 描述：超时之后，如果poll_timeout_ms未调用，则将消费者标记为死，并为消费者组触发重新平衡操作group_id

topics

• 值类型是数组array
• 默认值是[“logstash”]
• 描述：要订阅的主题列表，默认为[“logstash”]

topic_pattern

• 值类型是字符串string
• 没有默认值
• 描述：要订阅的主题正则表达式模式。使用此配置时将忽略主题配置。

Kafka输入插件示例配置

cat /opt/application/logstash-7.0.0/conf/logstash-kafka.conf 
input {kafka {auto_offset_reset => "earliest"bootstrap_servers => "10.150.55.94:9092,10.150.55.95:9092,10.150.50.223:9092"consumer_threads => "3"group_id => "logstash_group"heartbeat_interval_ms => ""request_timeout_ms => "30000"topics => "nginx_log"}
}
filter {json {source => "message"}
}
output {stdout {codec => rubydebug}
}

检查配置