本节书摘来自华章出版社《ELK Stack权威指南（第2版）》一书中的第2章，第2.4节，作者 饶琛琳 更多章节内容可以访问云栖社区“华章计算机”公众号查看。
2.4 输出插件
2.4.1 输出到Elasticsearch
Logstash早期有三个不同的Elasticsearch插件。到1.4.0版本的时候，开发者彻底重写了LogStash::Outputs::Elasticsearch插件。从此，我们只需要用这一个插件，就能任意切换使用Elasticsearch集群支持的各种不同协议了。
1.配置示例
2.解释
批量发送 在过去的版本中，主要由本插件的'flush_size'和'idle_flush_time'两个参数共同控制Logstash向Elasticsearch发送批量数据的行为。以上面示例来说：Logstash会努力攒到20 000条数据一次性发送出去，但是如果10秒钟内也没攒够20 000条，Logstash还是会以当前攒到的数据量发一次。
默认情况下，'flush_size'是500条，'idle_flush_time'是1秒。这也是很多人改大了'flush_size'也没能提高写入ES性能的原因—Logstash还是1秒钟发送一次。
从5.0开始，这个行为有了另一个前提：'flush_size'的大小不能超过Logstash运行时的命令行参数设置的'batch_size'，否则将以'batch_size'为批量发送的大小。
索引名 写入的Elasticsearch索引的名称，这里可以使用变量。为了更贴合日志场景，Logstash提供了%{+YYYY.MM.dd}这种写法。在语法解析的时候，看到以+号开头的，就会自动认为后面是时间格式，尝试用时间格式来解析后续字符串。所以，之前处理过程中不要给自定义字段取个加号开头的名字……
此外，注意索引名中不能有大写字母，否则Elasticsearch在日志中会报InvalidIndex-NameException，但是Logstash不会报错，这个错误比较隐晦，也容易掉进这个坑中。
轮询 gstash 1.4.2在transport和HTTP协议的情况下是固定连接指定host发送数据。从1.5.0开始，host可以设置数组，它会从节点列表中选取不同的节点发送数据，达到Round-Robin负载均衡的效果。
3.不同版本的协议沿革
1.4.0版本之前，有logstash-output-elasticsearch、 logstash-output-elasticsearch_http、logstash- output-elasticsearch_river三个插件。
1.4.0到2.0版本之间，配合Elasticsearch废弃river方法，只剩下logstash-output-elasticsearch一个插件，同时实现了node、transport、http三种协议。
2.0版本开始，为了兼容性和调试方便，logstash-output-elasticsearch改为只支持HTTP协议。想继续使用node或者transport协议的用户，需要单独安装logstash-output-elasticsearch_ java插件。
一个小集群里，继续使用logstash-output-elasticsearch_java的node协议是最方便了。
Logstash以Elasticsearch的client节点身份（即不存数据不参加选举）运行。如果你运行下面这行命令，你就可以看到自己的Logstash进程名，对应的node.role值是c：
# curl 127.0.0.1:9200/_cat/nodes?v
host ip heap.percent ram.percent load node.role master name
local 192.168.0.102 7 c - logstash-local-1036-2012
local 192.168.0.2 7 d * Sunstreak
Logstash 1.5以后，也不再分发一个内嵌的Elasticsearch服务器。如果你想变更node协议下的这些配置，在$PWD/elasticsearch.yml文件里写自定义配置即可，Logstash会尝试自动加载这个文件。
对于拥有很多索引的大集群，你可以用transport协议。Logstash进程会转发所有数据到你指定的某台主机上。这种协议跟上面的node协议是不同的。node协议下的进程是可以接收到整个Elasticsearch集群状态信息的，当进程收到一个事件时，它就知道这个事件应该存在集群内哪个机器的分片里，所以它就会直接连接该机器发送这条数据。而transport协议下的进程不会保存这个信息，在集群状态更新（节点变化，索引变化都会发送全量更新）时，就不会对所有的Logstash进程也发送这种信息。更多Elasticsearch集群状态的细节，参阅http://www.elasticsearch.org/guide。
如果你已经有现成的Elasticsearch集群，但是版本跟Logstash自带的又不太一样，建议你使用http协议。Logstash会使用POST方式发送数据。
4.数据重复问题
经常有读者问，为什么 Logstash 在有多个 conf 文件的情况下，进入ES（Elasticsecrrch）的数据会重复，几个 conf 数据就会重复几次。其实问题原因在之前章节提到过，output 段顺序执行，没有对日志 type 进行判断的各插件配置都会全部执行一次。在 output 段对 type 进行判断的语法如下所示：
output {
if [type] == "nginxaccess" {
elasticsearch { }
}
}
5.模板
Elasticsearch支持给索引预定义设置和mapping（前提是你用的Elasticsearch版本支持这个API，不过估计应该都支持）。Logstash自带有一个优化好的模板，内容如下:
这其中的关键设置包括：
template for index-pattern：只有匹配logstash-*的索引才会应用这个模板。有时候我们会变更Logstash的默认索引名称，记住你也得通过PUT方法上传可以匹配你自定义索引名的模板。当然，我更建议的做法是，把你自定义的名字放在“logstash-”后面，变成index =>"logstash-custom-%{+yyyy.MM.dd}"这样。
refresh_interval for indexing：Elasticsearch是一个近实时搜索引擎。它实际上是每1秒钟刷新一次数据。对于日志分析应用，我们用不着这么实时，所以Logstash自带的模板修改成了5秒钟。你还可以根据需要继续放大这个刷新间隔以提高数据写入性能。
multi-field with keyword：Elasticsearch 会自动使用自己的默认分词器（空格、点、斜线等分割）来分析字段。分词器对于搜索和评分是非常重要的，但是大大降低了索引写入和聚合请求的性能。所以Logstash模板定义了一种叫“多字段”（multi-field）类型的字段。这种类型会自动添加一个以“.keyword”结尾的字段，并给这个字段设置为不启用分词器。简单说，你想获取URL字段的聚合结果的时候，不要直接用url，而是用url.keyword作为字段名。当你还对分词字段发起聚合和排序请求的时候，直接提示无法构建fielddata了！
在Logstash 5.0中，同时还保留携带了针对Elasticsearch 2.x的template文件，在那里，通过旧版本的mapping配置，达到和新版本相同的行为效果：对应统计字段明确设置"index":"not_analyzed","doc_values":true，以及对分词字段加上对fielddata 的{"format":"disabled"}。
half_float：Elasticsearch 5.0新引入了half_float类型。比标准的float类型占用更少的资源，提供更好的性能。在明确自己数值范围较小的时候可用。刚巧，经纬度就是一个数值范围很小的数据。
geo_point：Elasticsearch支持geo_point类型，geo distance聚合等等。比如说，你可以请求某个geo_point点方圆10千米内数据点的总数。在Kibana的tilemap类型面板里，就会用到这个类型的数据。
6.其他模板配置建议
order：如果你有自己单独定制template的想法，很好。这时候有几种选择：
在logstash-output-elasticsearch配置中开启manage_template => false选项，然后一切自己动手；
在logstash-output-elasticsearch配置中开启template =>"/path/to/your/tmpl.json"选项，让logstash来发送你自己写的template文件；
避免变更Logstash里的配置，而是另外发送一个template，利用Elasticsearch的templates order功能。
这个order功能，就是Elasticsearch在创建一个索引的时候，如果发现这个索引同时匹配上了多个template，那么就会先应用order数值小的template设置，然后再应用一遍order数值高的作为覆盖，最终达到一个merge的效果。比如，对上面这个模板已经很满意，只想修改一下refresh_interval，那么只需要新写一个：
{
"order" : 1,
"template" : "logstash-*",
"settings" : {
"index.refresh_interval" : "20s"
}
}
然后运行curl -XPUT http:// localhost:9200/_template/template_newid -d '@/path/to/your/tmpl.json' 即可。Logstash默认的模板，order是0，id是logstash，通过logstash-output-elasticsearch的配置选项template_name修改。你的新模板就不要跟这个名字冲突了。
_fields_name：日志场景最重要的瓶颈首先在于入库速度。所以Logstash默认模板中，通过对一些字段的“norms”：false和“include_in_all”：false设定来减少计算量，提高入库速度。其实还有一个内置的字段，在条件许可的情况下，是可以关闭的，就是_fields_name。
这个字段和_all有些类似，_all里面记录的是所有字段的值，而_fields_name里记录的是所有字段的名字。这个作用是可以大大加速诸如exists、missing类查询的速度。但是会导致30%左右的写入速度损耗。在日志场景中，也是可以关闭的，代码如下：
…
"_all" : {"enabled" : true, "norms" : false},
“_fields_name” : { “enabled” : false },
…
2.4.2 发送email
配置示例如下：
logstash-output-email插件支持SMTP协议和sendmail两种方式，通过via参数设置。SMTP方式有较多的options参数可配置。sendmail只能利用本机上的sendmail服务来完成。文档上描述了Mail库支持的sendmail配置参数，但实际代码中没有相关处理，不要被迷惑了。
2.4.3 调用系统命令执行
logstash-output-exec插件的运用也非常简单，如下所示，将Logstash切割成的内容作为参数传递给命令。这样，在每个事件到达该插件的时候，都会触发这个命令的执行。
output {
exec {
command =>"sendsms.pl \"%{message}\" -t %{user}"
}
}
需要注意的是。这种方式是每次都重新开始执行一次命令并退出。本身是比较慢速的处理方式（程序加载，网络建联等都有一定的时间消耗）。最好只用于少量的信息处理场景，比如不适用nagios的其他报警方式。示例就是通过短信发送消息。
2.4.4 保存成文件
通过日志收集系统将分散在数百台服务器上的数据集中存储在某中心服务器上，这是运维最原始的需求。早年的scribed，甚至直接就把输出的语法命名为<store>。Logstash当然也能做到这点。
和LogStash::Inputs::File不同，LogStash::Outputs::File里可以使用sprintf format格式来自动定义输出到带日期命名的路径。
配置示例如下：
output {
file {
path =>"/path/to/%{+yyyy}/%{+mm}/%{+dd}/%{+HH}/%{host}.log.gz"
message_format =>"%{message}"
gzip => true
}
}
使用output/file插件首先需要注意的就是message_format参数。插件默认是输出整个event的JSON形式数据的。这可能跟大多数情况下使用者的期望不符。大家可能只是希望按照日志的原始格式保存就好了。所以需要定义为%{message}，当然，前提是在之前的filter插件中，你没有使用remove_field或者update等参数删除或修改%{message}字段的内容。
另一个非常有用的参数是gzip。gzip格式是一个非常奇特而友好的格式。其格式包括有：
10字节的头，包含幻数、版本号以及时间戳。
可选的扩展头，如原文件名。
文件体，包括DEFLATE压缩的数据。
8字节的尾注，包括CRC-32校验和以及未压缩的原始数据长度。
这样gzip就可以一段一段的识别出来数据—反过来说，也就是可以一段一段压缩了，添加在后面！
这对于我们流式添加数据简直太棒了！
你或许见过网络流传的parallel命令行工具并发处理数据的神奇文档，但在自己用的时候总见不到效果。实际上就是因为：文档中处理的gzip文件，可以分开处理然后再合并的，而你的用法却不一定可以。
这里需要注意两点：
1）按照Logstash标准，其实应该可以把数据格式的定义改在codec插件中完成，但是 logstash-output-file插件内部实现中跳过了@codec.decode这步，所以codec设置无法生效！
2）按照Logstash标准，配置参数的值可以使用event sprintf格式。但是logstash-output-file插件对event.sprintf(@path)的结果，还附加了一步inside_file_root?校验（个人猜测是为了防止越权到其他路径），这个file_root是通过直接对path参数分割/符号得到的。如果在sprintf格式中带有/符号，那么被切分后的结果就无法正确解析了。所以，如下所示配置，虽然看起来是正确的，实际效果却不对，正确写法应该是本节之前的配置示例那样。
output {
file {
path => "/path/to/%{+yyyy/MM/dd}/%{host}.log.gz"
codec => line {
format => "%{message}"
}
}
}
2.4.5 报警发送到Nagios
Logstash中有两个output插件是与Nagios有关的。logstash-output-nagios插件发送数据给本机的nagios.cmd管道命令文件，logstash-output-nagios_nsca插件则是调用send_nsca命令以NSCA协议格式把数据发送给Nagios服务器（远端或者本地皆可）。
1. nagios.cmd
nagios.cmd是Nagios服务器的核心组件。Nagios事件处理和内外交互都是通过这个管道文件来完成的。
使用CMD方式，需要自己保证发送的Logstash事件符合Nagios事件的格式。即必须在filter阶段预先准备好nagios_host和nagios_service字段；此外，如果在filter阶段也准备好nagios_annotation和nagios_level字段，这里也会自动转换成nagios事件信息。
如果不打算在filter阶段提供nagios_level，那么也可以在该插件中通过参数配置。
所谓nagios_level，即我们通过nagiosplugin检查数据时的返回值。其取值范围和含义如下：
“0”，代表“OK”，服务正常。
“1”，代表“WARNNING”，服务警告，一般nagios plugin命令中使用-w参数设置该阈值。
“2”，代表“CRITICAL”，服务危急，一般nagios plugin命令中使用-c参数设置该阈值。
“3”，代表“UNKNOWN”，未知状态，一般会在timeout等情况下出现。
默认情况下，该插件会以“CRITICAL”等级发送报警给Nagios服务器。
nagios.cmd文件的具体位置，可以使用command_file参数设置。默认位置是“/var/lib/nagios3/rw/nagios.cmd”。
关于和nagios.cmd交互的具体协议说明，有兴趣的读者请阅读Using external commands in Nagios一文（http://archive09.linux.com/feature/153285），这是《Learning Nagios 3.0》书中内容节选。
2. NSCA
NSCA是一种标准的Nagios分布式扩展协议。分布在各机器上的send_nsca进程主动将监控数据推送给远端Nagios服务器的NSCA进程。
当Logstash跟Nagios服务器没有在同一个主机上运行的时候，就只能通过NSCA方式来发送报警了—当然也必须在Logstash服务器上安装send_nsca命令。
Nagios事件所需要的几个属性在上一段中已经有过描述。不过在使用这个插件的时候，不要求提前准备好，而是可以在该插件内部定义参数：
output {
nagios_nsca {
nagios_host =>"%{host}"
nagios_service =>"logstash_check_%{type}"
nagios_status =>"2"
message_format =>"%{@timestamp}: %{message}"
host =>"nagiosserver.domain.com"
}
}
这里请注意，host和nagios_host两个参数，分别是用来设置Nagios服务器的地址，和报警信息中有问题的服务器地址。
关于NSCA原理，架构和配置说明，还不了解的读者请阅读官方网站Using NSClient++ from nagios with NSCA一节（http://nsclient.org/nscp/wiki/doc/usage/nagios/nsca）。
3.其他类似插件
除了Nagios以外，Logstash同样可以发送信息给其他常见监控系统，方式和Nagios大同小异：
logstash-output-ganglia插件通过UDP协议，发送gmetric型数据给本机/远端的gmond或者gmetad。
logstash-output-zabbix插件调用本机的zabbix_sender命令发送。
2.4.6 statsd
statsd最早是2008年Flickr公司用Perl写的针对graphite、datadog等监控数据后端存储开发的前端网络应用，2011年Etsy公司用Nodejs重构。用于接收、写入、读取和聚合时间序列数据，包括即时值和累积值等。
graphite是用Python模仿RRDtools写的时间序列数据库套件，包括三个部分：
carbon：一个Twisted守护进程，监听处理数据。
whisper：存储时间序列的数据库。
webapp：一个用Django框架实现的网页应用。
statsd项目涉及多个模块同时部署，步骤比较多，这里单独给读者介绍一下部署方法。首先通过如下几步安装graphite：
1）安装cairo和pycairo库：
# yum -y install cairo pycairo
2）pip安装：
Graphite以树状结构存储监控数据，所以statsd也是如此。所以发送给statsd的数据的key也一定得是“first.second.tree.four”这样的形式。而在logstash-output-statsd插件中，就会以三个配置参数来拼接成这种形式：
namespace.sender.metric
其中namespace和sender都是直接设置的，而metric又分为好几个不同的参数可以分别设置。statsd支持的metric类型如下：
increment
示例语法：increment => ["nginx.status.%{status}"]。该配置即可在statsd中生成对应的nginx.status.200, nginx.status.206, nginx.status.304, nginx.status.404, nginx.status.502, nginx.status.503, nginx.status.504等一系列监控项。同时，在statsd内配置的一个时间周期内，各状态码的次数，会自动累加到各自监控项里。
decrement
语法同increment。不过是递减而不是递增。
count
示例语法：count => {"nginx.bytes" =>"%{bytes}"}。该配置可以在statsd中生成一个nginx.bytes监控项。而每条Nginx访问记录的响应字节数，累加成带宽。
gauge
语法同count。gauge和count（在rrdtool中叫counter）两种计数器的区别从早年的rrdtool时代就被反复强调，即gauge直接存储原始数值，不累加。
set
示例语法：set => {"online"=>"%{user_id}"}。是新版statsd才支持的功能，可以用来做去重计算，比如在线人数统计。
timing
示例语法：timing => ["nginx.requesttime"=>"%{request_time}"]。该配置可以在statsd中生成一个nginx.requesttime监控项，记录时间周期内，Nginx访问记录的响应时间的数值统计情况，包括平均值，最大值，最小值，标准差，百分比值等。
关于这些metric类型的详细说明，请阅读statsd文档：https://github.com/etsy/statsd/blob/master/docs/metric_types.md。
推荐阅读
Etsy发布nodejs版本statsd的博客：Measure Anything, Measure Everything(http://codeascraft.etsy.com/2011/02/15/measure-anything-measure-everything/)
Flickr发布statsd的博客：Counting & Timing(http://code.flickr.net/2008/10/27/counting-timing/)
Librato有关statsd协作的博客：Using StatsD with Librato(http://support.metrics.librato.com/knowledgebase/articles/77199-using-statsd-with-librato)
2.4.7 标准输出stdout
和之前logstash-input-stdin插件一样，logstash-output-stdout插件也是最基础和简单的输出插件。同样在这里简单介绍一下，作为输出插件的一个共性了解。
配置示例如下：
output {
stdout {
codec => rubydebug
workers => 2
}
}
输出插件统一具有一个参数是workers。Logstash为输出做了多线程的准备。
其次是codec设置。codec的作用在之前已经讲过。可能除了logstash-codec-multiline，其他codec插件本身并没有太多的设置项。所以一般省略掉后面的配置区段。换句话说。上面配置示例的完全写法应该是：
output {
stdout {
codec => rubydebug {
}
workers => 2
}
}
单就logstash-output-stdout插件来说，其最重要和常见的用途就是调试。所以在不太有效的时候，加上命令行参数-vv运行，查看更多详细调试信息。
2.4.8 TCP发送数据
虽然之前我们已经提到过不建议直接使用LogStash::Inputs::TCP和LogStash::Outputs::TCP 做转发工作，不过在实际交流中，发现确实有不少朋友觉得这种简单配置足够使用，因而不愿意多加一层消息队列的。所以，还是把Logstash如何直接发送TCP数据也稍微提点一下。
配置示例如下：
output {
tcp {
host =>"192.168.0.2"
port => 8888
codec => json_lines
}
}
在收集端采用TCP方式发送给远端的TCP端口。这里需要注意的是，默认的Codec选项是json。而远端的LogStash::Inputs::TCP的默认Codec选项却是plain！所以不指定各自的Codec，对接肯定是失败的。
另外，由于IO BUFFER的原因，即使是两端共同约定为json依然无法正常运行，接收端会认为一行数据没结束，一直等待直至自己OutOfMemory！
所以，正确的做法是，发送端指定Codec为json_lines，这样每条数据后面会加上一个回车，接收端指定Codec为json_lines或者json均可，这样才能正常处理。包括在收集端已经切割好的字段，也可以直接带入收集端使用了。
2.4.9 输出到HDFS
数据写入HDFS是很多日志收集系统的最终目的。不过Logstash偏巧不是其中之一。到目前为止，Logstash还没有官方支持的直接写入HDFS的插件。而在社区，则有两种不同的解决方案可供选择。下面分别介绍。
1.通过HTTP接口
插件源码地址见：https://github.com/dstore-dbap/logstash-webhdfs
该插件使用Hadoop的WebHDFS接口，其本质就是发送POST数据，可以说实现起来比较简单。未来logstash-plugins官方可能也会收这个插件。
配置示例如下：
2.通过Java接口
插件源码地址见：https://github.com/avishai-ish-shalom/logstash-hdfs
该插件使用Hadoop的HDFS接口，利用JRuby可以直接导入Java类的特性，直接使用了org.apache.hadoop.fs.FileSystem等类来实现。
配置示例如下：
output {
hdfs {
path =>"/path/to/output_file.log"
enable_append => true
}
}
因为需要导入各种Hadoop的jar包，所以这个运行比较麻烦。Logstash-1.4.2上的运行命令示例如下：
