9.6 KiB
7.Logstash输入插件(Input)以及编码插件(Codec)详解
1、Logstash输入插件(Input)以及编码插件(Codec)详解
在上面文章中我们介绍了logstash的基本使用，主要讲述了logstash的input（输入）和output（输出）功能，其实logstash的功能远远不止于此，logstash之所以功能强大和流行，还与其丰富的过滤器插件是分不开的，过滤器提供的并不单单是过滤的功能，还可以对进入过滤器的原始数据进行复杂的逻辑处理，甚至添加独特的事件到后续流程中。本节就重点讲述下logstash过滤器插件的使用方法。
1.1、Logstash基本语法组成
Logstash配置文件有如下三部分组成，其中input、output部分是必须配置，filter部分是可选配置，而filter就是过滤器插件，可以在这部分实现各种日志过滤功能。
input {
输入插件
}
filter {
过滤匹配插件
}
output {
输出插件
}
下面我们将依次进行介绍。
1.2、Logstash输入插件(Input)
Logstash的输入插件主要用来接收数据，Logstash支持多种数据源，常见的有读取文件、标准输入、读取syslog日志、读取网络数据等，这里分别介绍下每种接收数据源的配置方法。
1、 读取文件(File)
logstash使用一个名为filewatch的ruby gem库来监听文件变化,并通过一个叫.sincedb的数据库文件来记录被监听的日志文件的读取进度（时间戳），这个sincedb数据文件的默认路径在 <path.data>/plugins/inputs/file下面，文件名类似于.sincedb_452905a167cf4509fd08acb964fdb20c，而<path.data>表示logstash插件存储目录，默认是LOGSTASH_HOME/data。
看下面一个事件配置文件：
input {
file {
path => ["/var/log/messages"]
type => "system"
start_position => "beginning"
}
}
output {
stdout{
codec=>rubydebug
}
}
这个配置是监听并接收本机的/var/log/messages文件内容，start_position表示按时间戳记录的地方开始读取，如果没有时间戳则从头开始读取，有点类似cat命令，默认情况下，logstash会从文件的结束位置开始读取数据，也就是说logstash进程会以类似tail -f命令的形式逐行获取数据。type用来标记事件类型,通常会在输入区域通过type标记事件类型。
假定/var/log/messages中输入的内容为如下信息：
Feb 6 17:20:11 logstashserver yum[15667]: Installed: vsftpd-2.2.2-24.el6.x86_64
那么经过logstash后，输出内容为如下JSON格式：
{
"@version" => "1",
"host" => " logstashserver",
"path" => "/var/log/messages",
"@timestamp" => 2018-02-02T02:02:36.361Z,
"message" => "Feb 6 17:20:11 logstashserver yum[15667]: Installed: vsftpd-2.2.2-24.el6.x86_64",
"type" => "system"
}
从输出可以看出，除了增加了前四个字段外，message字段是真正的输出内容，将输入信息原样输出，最后还有一个type字段，这是在input中定义的一个事件类型，也被原样输出了，在后面的过滤插件中会用到这个type字段。
2、标准输入(Stdin)
stdin是从标准输入获取信息，关于stdin的使用，前面章节已经做过了一些简单的介绍，这里再看一个稍微复杂一点的例子，下面是一个关于stdin的事件配置文件：
input{
stdin{
add_field=>{"key"=>"iivey"}
tags=>["add1"]
type=>"test1"
}
}
output {
stdout{
codec=>rubydebug
}
}
如果输入"hello world",可以在终端看到如下输出信息：
{
"@timestamp" => 2018-02-05T10:46:38.211Z,
"@version" => "1",
"host" => "logstashserver",
"message" => "hello world",
"type" => "test1",
"key" => "iivey",
"tags" => [
[0] "add1"
]
}
type和tags是logstash的两个特殊字段, type一般会放在input中标记事件类型, tags主要用于在事件中增加标签，以便在后续的处理流程中使用，主要用于filter或output阶段。
3、读取 Syslog日志
syslog是linux下一个功能强大的系统日志收集系统，它的升级版本有rsyslog和syslog-ng，在主流的Linux发行版centos6.x/7.x中，默认是rsyslog，升级版本都涵盖sysLog的常用功能，不过在功能和性能上更为出色。rsyslog日志收集系统既可以记录信息在本地文件中，也可以通过网络发送到接收的rsyslog服务器。本节介绍下如何将rsyslog收集到的日志信息发送到logstash中，这里以centos7.5为例，需要做如下两个步骤的操作：
首先，在需要收集日志的服务器上找到rsyslog的配置文件/etc/rsyslog.conf,添加如下内容：
*.* @@172.16.213.120:5514
其中，172.16.213.120是logstash服务器的地址。5514是logstash启动的监听端口。
接着，重启rsyslog服务：
[root@kafkazk1 logstash]# systemctl restart rsyslog
然后，在logstash服务器上创建一个事件配置文件，内容如下：
input {
syslog {
port => "5514"
}
}
output {
stdout{
codec=>rubydebug
}
}
此时, kafkazk1主机上的日志信息都会到logstash中来, 假定输入的内容为如下信息：
Feb 5 17:42:26 kafkazk1 systemd: Stopped The nginx HTTP and reverse proxy server.
那么经过logstash后，输出内容为如下JSON格式：
{
"severity" => 6,
"@timestamp" => 2018-02-05T09:42:26.000Z,
"@version" => "1",
"host" => "172.16.213.51",
"program" => "systemd",
"message" => "Stopped The nginx HTTP and reverse proxy server.\n",
"priority" => 30,
"logsource" => "kafkazk1",
"facility" => 3,
"severity_label" => "Informational",
"timestamp" => "Feb 5 17:42:26",
"facility_label" => "system"
}
从JSON格式的输出可以看到增加了几个字段，并且输入的信息经过LogStash后被自动切割成多个字段了，这其实就是下面将要讲到的filter的功能。很多时候，我们为了能更详细的分析日志信息，需要将输入的日志分成多个字段，然后通过不同的字段维度进行多重分析。
在数据量较大的时候，LogStash读取syslog日志会存在很大性能瓶颈，因而如果遇到大数据量场景，建议使用"LogStash::Inputs::TCP"和"LogStash::Filters::Grok"配合实现同样的syslog功能！
4、读取TCP网络数据
下面的事件配置文件就是通过"LogStash::Inputs::TCP"和"LogStash::Filters::Grok"配合实现syslog功能的例子，这里使用了logstash的TCP/UDP插件读取网络数据：
input {
tcp {
port => "5514"
}
}
filter {
grok {
match => { "message" => "%{SYSLOGLINE}" }
}
}
output {
stdout{
codec=>rubydebug
}
}
其中，5514端口是logstash启动的tcp监听端口。注意这里用到了日志过滤"LogStash::Filters::Grok"功能，下面马上会介绍到。
"LogStash::Inputs::TCP"最常见的用法就是结合nc命令导入旧数据。在启动logstash进程后，在另一个终端运行如下命令即可导入旧数据：
[root@kafkazk1 app]# nc 172.16.213.120 5514 < /var/log/secure
通过这种方式，就把/var/log/secure的内容全部导入到了logstash中，当nc命令结束时，数据也就导入完成了。
1.3、Logstash编码插件(Codec)
在前面介绍的例子中，其实我们就已经用过编码插件codec了，也就是这个rubydebug，它就是一种codec，虽然它一般只会用在stdout插件中，作为配置测试或者调试的工具。
编码插件(Codec)可以在logstash输入或输出时处理不同类型的数据，同时，还可以更好更方便的与其它自定义格式的数据产品共存，比如fluent、netflow、collectd等通用数据格式的其它产品。因此，Logstash不只是一个input-->filter-->output的数据流，而是一个input-->decode-->filter-->encode-->output的数据流。
Codec支持的编码格式常见的有plain、json、json_lines等。下面依次介绍。
1、codec插件之plain
plain是一个空的解析器，它可以让用户自己指定格式，也就是说输入是什么格式，输出就是什么格式。下面是一个包含plain编码的事件配置文件：
input{
stdin {
}
}
output{
stdout {
codec => "plain"
}
}
在启动logstash进程后，输入什么格式的数据，都会原样输出，这里不再过多说明。
2、codec插件之json、json_lines
如果发送给logstash的数据内容为json格式,可以在input字段加入codec=>json来进行解析，这样就可以根据具体内容生成字段，方便分析和储存。如果想让logstash输出为json格式，可以在output字段加入codec=>json，下面是一个包含json编码的事件配置文件：
input {
stdin {
}
}
output {
stdout {
codec => json
}
}
同理，在启动logstash进程后，如果输入“hello world”，那么输出信息为：
{"@version":"1","host":"logstashserver","@timestamp":"2018-02-07T04:14:51.096Z","message":"hello world"}
这就是json格式的输出，可以看出，json每个字段是key:values格式，多个字段之间通过逗号分隔。有时候，如果json文件比较长，需要换行的话，那么就要用json_lines编码格式了。这个就不再介绍了。
@版权声明：51CTO独家出品，未经允许不能转载，否则追究法律责任
原网址: 访问
创建于: 2020-06-28 14:55:12
目录: default
标签: blog.51cto.com