logstash

logstash

使用logstash input jdbc插件,从mysql导出数据到elasticsearch,发现乱码

LogstashViento 回复了问题 • 2 人关注 • 2 个回复 • 120 次浏览 • 4 天前 • 来自相关话题

mongodb停机后重启,logstash就一直连接不上

回复

Logstashexubo 发起了问题 • 1 人关注 • 0 个回复 • 120 次浏览 • 2016-12-30 13:16 • 来自相关话题

求助logstash对request的再次切分

回复

Logstashforsaken627 回复了问题 • 1 人关注 • 1 个回复 • 238 次浏览 • 2016-12-19 18:15 • 来自相关话题

logstash 写入ElasticSearch占用大量带宽

Logstashy676360 回复了问题 • 3 人关注 • 1 个回复 • 436 次浏览 • 2016-12-12 11:36 • 来自相关话题

logstash5.0使用kafka input插件出现异常

Logstashwangohyes 回复了问题 • 2 人关注 • 2 个回复 • 500 次浏览 • 2016-12-09 00:12 • 来自相关话题

filebeat+kafka+ELK 5.0

Logstashwangohyes 回复了问题 • 4 人关注 • 5 个回复 • 648 次浏览 • 2016-12-09 00:11 • 来自相关话题

怎么从多条子事件日志中分析出完整事件信息

回复

Logstashabird 发起了问题 • 1 人关注 • 0 个回复 • 192 次浏览 • 2016-11-24 18:02 • 来自相关话题

Logstash收集多台设备的beats日志信息,如何添加多台源设备的源ip字段并输出到Elasticsearch?

LogstashJea 回复了问题 • 2 人关注 • 1 个回复 • 261 次浏览 • 2016-11-24 16:50 • 来自相关话题

logstash5.0. install translate plugin 时报错

Logstashleighton_buaa 回复了问题 • 2 人关注 • 1 个回复 • 226 次浏览 • 2016-11-24 11:43 • 来自相关话题

Kafka+Logstash的问题

回复

Logstashdennishood 回复了问题 • 1 人关注 • 1 个回复 • 368 次浏览 • 2016-11-08 10:52 • 来自相关话题

条新动态, 点击查看
kennywu76

kennywu76 回答了问题 • 2014-12-02 18:12 • 2 个回复 不感兴趣

logstash获取时间的问题

赞同来自:

产生这个错误的原因是系统的locale设置不是english,因此logstash用到的joda无法理解Dec这样的月份格式。

解决办法为在date filter里加一个locale=>"en"的设置:
filter {
d... 显示全部 »
产生这个错误的原因是系统的locale设置不是english,因此logstash用到的joda无法理解Dec这样的月份格式。

解决办法为在date filter里加一个locale=>"en"的设置:
filter {
date {
locale => "en"
..............
}
}
can

can 回答了问题 • 2014-12-23 19:42 • 3 个回复 不感兴趣

请教个关于logstash output 到es protocol的问题

赞同来自:

感谢@wood大神的指点,我这里总结下哈!
理论上来说,transport没有http的header带来的额外处理开销,所以速度应该要比http快。
node protocol和transport protocol的区别,主要就是node的话,logstash... 显示全部 »
感谢@wood大神的指点,我这里总结下哈!
理论上来说,transport没有http的header带来的额外处理开销,所以速度应该要比http快。
node protocol和transport protocol的区别,主要就是node的话,logstash会成为一个ES node加入集群,理论上性能比transport protocol更好,因为数据少一层转发。但是集群规模大了后,node protocol有一定副作用。 因为每个node上都有集群的状态信息,集群大了后,任何集群状态信息的改变都会在所有node同步,node数量越多,同步的代价越高
关于ES nodes过多可能导致的集群水平扩展问题可以参照这里:
http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/finite-scale.html
ES的transport protocol支持多个host,只是logstash的插件不支持,官网java client api有demo,依葫芦画瓢,改下logstash的ES插件就可以了:
http://www.elasticsearch.org/guide/en/elasticsearch/client/java-api/current/client.html
下面是http与transport比较的讨论帖,有兴趣的也可以看看,多了解了解:
http://elasticsearch-users.115913.n3.nabble.com/TransportClient-vs-HTTP-TransportClient-advantages-td4023014.html
差不多就这些了,希望对后面的同学有帮助!

使用logstash input jdbc插件,从mysql导出数据到elasticsearch,发现乱码

回复

LogstashViento 回复了问题 • 2 人关注 • 2 个回复 • 120 次浏览 • 4 天前 • 来自相关话题

mongodb停机后重启,logstash就一直连接不上

回复

Logstashexubo 发起了问题 • 1 人关注 • 0 个回复 • 120 次浏览 • 2016-12-30 13:16 • 来自相关话题

求助logstash对request的再次切分

回复

Logstashforsaken627 回复了问题 • 1 人关注 • 1 个回复 • 238 次浏览 • 2016-12-19 18:15 • 来自相关话题

logstash 写入ElasticSearch占用大量带宽

回复

Logstashy676360 回复了问题 • 3 人关注 • 1 个回复 • 436 次浏览 • 2016-12-12 11:36 • 来自相关话题

logstash5.0使用kafka input插件出现异常

回复

Logstashwangohyes 回复了问题 • 2 人关注 • 2 个回复 • 500 次浏览 • 2016-12-09 00:12 • 来自相关话题

filebeat+kafka+ELK 5.0

回复

Logstashwangohyes 回复了问题 • 4 人关注 • 5 个回复 • 648 次浏览 • 2016-12-09 00:11 • 来自相关话题

怎么从多条子事件日志中分析出完整事件信息

回复

Logstashabird 发起了问题 • 1 人关注 • 0 个回复 • 192 次浏览 • 2016-11-24 18:02 • 来自相关话题

Logstash收集多台设备的beats日志信息,如何添加多台源设备的源ip字段并输出到Elasticsearch?

回复

LogstashJea 回复了问题 • 2 人关注 • 1 个回复 • 261 次浏览 • 2016-11-24 16:50 • 来自相关话题

logstash5.0. install translate plugin 时报错

回复

Logstashleighton_buaa 回复了问题 • 2 人关注 • 1 个回复 • 226 次浏览 • 2016-11-24 11:43 • 来自相关话题

Kafka+Logstash的问题

回复

Logstashdennishood 回复了问题 • 1 人关注 • 1 个回复 • 368 次浏览 • 2016-11-08 10:52 • 来自相关话题

如何反向设置es mapping template

Elasticsearchrunc 发表了文章 • 1 个评论 • 263 次浏览 • 2016-08-31 14:26 • 来自相关话题

比如,由logstash打到es中的数据,除了其中一个字段比如message,其余字段都想设置为not analyzed,这种情况如何设置?貌似目前es只支持设置那些具体的字段为not analyzed,而不能反过来设置啊?
比如,由logstash打到es中的数据,除了其中一个字段比如message,其余字段都想设置为not analyzed,这种情况如何设置?貌似目前es只支持设置那些具体的字段为not analyzed,而不能反过来设置啊?

Day14: percolator接口在logstash中的运用

Advent三斗室 发表了文章 • 0 个评论 • 639 次浏览 • 2015-12-16 23:10 • 来自相关话题

我们都知道 Elasticsearch 除了普通的 search 接口以外,还有另一个 Percolator 接口,天生用来做实时过滤告警的。但是由于接口比较复杂,在目前的 ELK 体系中不是很容易运用。

而单纯从 Logstash 来做实时过滤报警,规则又不是很灵活。toplog.io 公司开发了一个 logstash-output-percolator插件,在有一定既定条件的情况下,成功运用上了 Percolator 方案。

这个插件的设计逻辑是:
通过 logstash-filter-checksum 自主生成 ES 文档的 _id;使用上一步生成的 _id 同时发送 logstash-output-elasticsearch 和 logstash-output-percolatorPercolator 接口一旦过滤成功,将 _id 发送给 Redis 服务器其他系统从 Redis 服务器中获取 _id 即可从 ES 里拿到实际数据
Percolator 接口的用法简单说是这样:

创建接口:curl -XPUT 'localhost:9200/patterns/.percolator/my-pattern-id' -d '{"query" : {"match" : {"message" : "ERROR"} } }'过滤测试:curl -XGET 'localhost:9200/my-index/my-type/_percolate' -d '{"doc" : {"message" : "ERROR: Service Apache failed to connect to MySQL"} }'要点就是把文档放在 doc 属性里发送到 _percolate 里。

对应的 Logstash 配置如下:filter {
checksum {
algorithm => "md5"
keys => ["message"]
}
}
output {
elasticsearch {
host => "localhost"
cluster => "my-cluster"
document_id => "%{logstash_checksum}"
index => "my-index"
}
percolator {
host => "es-balancer"
redis_host => ["localhost"]
document_id => "%{logstash_checksum}"
pattern_index => "patterns"
}
}连接上对应的 Redis,就可以看到报警信息了:$ redis-cli
127.0.0.1:6379> lrange percolator 0 1
1) "{\"matches\":[\"2\"],\"document_id\":\"a5d5c5f69b26ac0597370c9b1e7a8111\"}"想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
我们都知道 Elasticsearch 除了普通的 search 接口以外,还有另一个 Percolator 接口,天生用来做实时过滤告警的。但是由于接口比较复杂,在目前的 ELK 体系中不是很容易运用。

而单纯从 Logstash 来做实时过滤报警,规则又不是很灵活。toplog.io 公司开发了一个 logstash-output-percolator插件,在有一定既定条件的情况下,成功运用上了 Percolator 方案。

这个插件的设计逻辑是:
  1. 通过 logstash-filter-checksum 自主生成 ES 文档的 _id;
  2. 使用上一步生成的 _id 同时发送 logstash-output-elasticsearch 和 logstash-output-percolator
  3. Percolator 接口一旦过滤成功,将 _id 发送给 Redis 服务器
  4. 其他系统从 Redis 服务器中获取 _id 即可从 ES 里拿到实际数据

Percolator 接口的用法简单说是这样:

创建接口:
curl -XPUT 'localhost:9200/patterns/.percolator/my-pattern-id' -d '{"query" : {"match" : {"message" : "ERROR"} } }'
过滤测试:
curl -XGET 'localhost:9200/my-index/my-type/_percolate' -d '{"doc" : {"message" : "ERROR: Service Apache failed to connect to MySQL"} }'
要点就是把文档放在 doc 属性里发送到 _percolate 里。

对应的 Logstash 配置如下:
filter {
checksum {
algorithm => "md5"
keys => ["message"]
}
}
output {
elasticsearch {
host => "localhost"
cluster => "my-cluster"
document_id => "%{logstash_checksum}"
index => "my-index"
}
percolator {
host => "es-balancer"
redis_host => ["localhost"]
document_id => "%{logstash_checksum}"
pattern_index => "patterns"
}
}
连接上对应的 Redis,就可以看到报警信息了:
$ redis-cli
127.0.0.1:6379> lrange percolator 0 1
1) "{\"matches\":[\"2\"],\"document_id\":\"a5d5c5f69b26ac0597370c9b1e7a8111\"}"
想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day13: ipip.net介绍

Advent三斗室 发表了文章 • 0 个评论 • 733 次浏览 • 2015-12-16 23:05 • 来自相关话题

Geo 定位在 ELK 应用中是非常重要和有用的一个环节。不幸的是:GeoIP 本身在国内的准确度实在堪忧。高春辉近年成立了一个项目,专注收集细化 IP 地址在国内的数据:http://www.ipip.net。数据分为免费版和收费版两种。项目提供了不少客户端,有趣的是,有社区贡献了一个 Logstash 插件:https://github.com/bittopaz/logstash-filter-ipip。

用法很简单:filter {
ipip {
source => "clientip"
target => "ipip"
}
}生成的 JSON 数据结构类似下面这样:{
"clientip" : "",
"ipip" : {
"country" : "",
"city" : "",
"carrier" : "",
"province" : ""
}
}不过这个插件只实现了收费版的数据库基础格式。免费版的支持,收费版高级的经纬度、基站位置等,都没有随着更新。事实上,我们可以通过 ipip 官方的 Java 库,实现一个更灵活的 logstash-filter-ipip_java 插件出来,下期见。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
Geo 定位在 ELK 应用中是非常重要和有用的一个环节。不幸的是:GeoIP 本身在国内的准确度实在堪忧。高春辉近年成立了一个项目,专注收集细化 IP 地址在国内的数据:http://www.ipip.net。数据分为免费版和收费版两种。项目提供了不少客户端,有趣的是,有社区贡献了一个 Logstash 插件:https://github.com/bittopaz/logstash-filter-ipip

用法很简单:
filter {
ipip {
source => "clientip"
target => "ipip"
}
}
生成的 JSON 数据结构类似下面这样:
{
"clientip" : "",
"ipip" : {
"country" : "",
"city" : "",
"carrier" : "",
"province" : ""
}
}
不过这个插件只实现了收费版的数据库基础格式。免费版的支持,收费版高级的经纬度、基站位置等,都没有随着更新。事实上,我们可以通过 ipip 官方的 Java 库,实现一个更灵活的 logstash-filter-ipip_java 插件出来,下期见。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day10: 如何处理数组形式的JSON日志

Advent三斗室 发表了文章 • 0 个评论 • 1046 次浏览 • 2015-12-16 22:57 • 来自相关话题

ELK 收集业务日志的来源,除了应用服务器以外,还有很大一部分来自客户端。考虑到客户端网络流量的因素,一般实现上都不会要求实时上报数据,而是攒一批,等到手机连上 WIFI 网络了,再统一发送出来。所以,这类客户端日志一般都有几个特点:
预先已经记录成 JSON 了;日志主体内容是一个巨大无比的数组,数据元素才是实际的单次日志记录;一次 POST 会有几 MB 到几十 MB 大小。

在处理这类数据的时候,第一关是别让数据超长直接给丢弃了(说的就是你啊,Rsyslog);第二关就是拆分 JSON 数组,把几十 MB 数据扔 ES 字段里,显然是不利于搜索和统计需求的。今天我们就来说说怎么拆分 JSON 数组。

假设收到的是这么一段日志:
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}首先我们知道可以在读取的时候把 JSON 数据解析成 LogStash::Event 对象:
input {
tcp {
codec => json
}
}但是怎么把解析出来的 logs 字段拆分成多个 event 呢?这里我们可以用一个已有插件:logstash-filter-split。
filter {
split {
field => "logs"
}
date {
match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
remove_fields => ["logs", "timestamp"]
}
}这样,就可以得到两个 event 了:
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","@timestamp":"2015-12-10T09:55:00Z","reason":"****"}
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10T09:56:12Z","tracert":"****"}看起来可能跟这个插件的文档描述不太一样。文档上写的是通过 terminator 字符,切割 field 字符串成多个 event。但实际上,field 设置是会自动判断的,如果 field 内容是字符串,就切割字符串成为数组再循环;如果内容已经是数组了,直接循环:
original_value = event[@field]

if original_value.is_a?(Array)
splits = original_value
elsif original_value.is_a?(String)
splits = original_value.split(@terminator, -1)
else
raise LogStash::ConfigurationError, "Only String and Array types are splittable. field:#{@field} is of type = #{original_value.class}"
end

return if splits.length == 1

splits.each do |value|
next if value.empty?

event_split = event.clone
@logger.debug("Split event", :value => value, :field => @field)
event_split[(@target || @field)] = value
filter_matched(event_split)

yield event_split
end
event.cancel顺带提一句:这里 yield 在 Logstash 1.5.0 之前,实现有问题,生成的新事件,不会继续执行后续 filter,直接进入到 output 阶段。也就是说,如果你用 Logstash 1.4.2 来执行上面那段配置,生成的两个事件会是这样的:{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10 17:56:12","tracert":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
ELK 收集业务日志的来源,除了应用服务器以外,还有很大一部分来自客户端。考虑到客户端网络流量的因素,一般实现上都不会要求实时上报数据,而是攒一批,等到手机连上 WIFI 网络了,再统一发送出来。所以,这类客户端日志一般都有几个特点:
  1. 预先已经记录成 JSON 了;
  2. 日志主体内容是一个巨大无比的数组,数据元素才是实际的单次日志记录;
  3. 一次 POST 会有几 MB 到几十 MB 大小。


在处理这类数据的时候,第一关是别让数据超长直接给丢弃了(说的就是你啊,Rsyslog);第二关就是拆分 JSON 数组,把几十 MB 数据扔 ES 字段里,显然是不利于搜索和统计需求的。今天我们就来说说怎么拆分 JSON 数组。

假设收到的是这么一段日志:
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
首先我们知道可以在读取的时候把 JSON 数据解析成 LogStash::Event 对象:
input {
tcp {
codec => json
}
}
但是怎么把解析出来的 logs 字段拆分成多个 event 呢?这里我们可以用一个已有插件:logstash-filter-split。
filter {
split {
field => "logs"
}
date {
match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
remove_fields => ["logs", "timestamp"]
}
}
这样,就可以得到两个 event 了:
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","@timestamp":"2015-12-10T09:55:00Z","reason":"****"}
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10T09:56:12Z","tracert":"****"}
看起来可能跟这个插件的文档描述不太一样。文档上写的是通过 terminator 字符,切割 field 字符串成多个 event。但实际上,field 设置是会自动判断的,如果 field 内容是字符串,就切割字符串成为数组再循环;如果内容已经是数组了,直接循环:
    original_value = event[@field]

if original_value.is_a?(Array)
splits = original_value
elsif original_value.is_a?(String)
splits = original_value.split(@terminator, -1)
else
raise LogStash::ConfigurationError, "Only String and Array types are splittable. field:#{@field} is of type = #{original_value.class}"
end

return if splits.length == 1

splits.each do |value|
next if value.empty?

event_split = event.clone
@logger.debug("Split event", :value => value, :field => @field)
event_split[(@target || @field)] = value
filter_matched(event_split)

yield event_split
end
event.cancel
顺带提一句:这里 yield 在 Logstash 1.5.0 之前,实现有问题,生成的新事件,不会继续执行后续 filter,直接进入到 output 阶段。也就是说,如果你用 Logstash 1.4.2 来执行上面那段配置,生成的两个事件会是这样的:
{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10 17:56:12","tracert":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day9: Elasticsearch template的order

Advent三斗室 发表了文章 • 0 个评论 • 1149 次浏览 • 2015-12-10 01:13 • 来自相关话题

ELK Stack 在入门学习过程中,必然会碰到自己修改定制索引映射(mapping)乃至模板(template)的问题。
这时候,不少比较认真看 Logstash 文档的新用户会通过下面这段配置来制定自己的模板策略:output {
elasticsearch {
host => "127.0.0.1"
manage_template => true
template => "/path/to/mytemplate"
template_name => "myname"
}
}
然而随后就发现,自己辛辛苦苦修改出来的模板,通过 curl -XGET 'http://127.0.0.1:9200/_template/myname' 看也确实上传成功了,但实际新数据索引创建出来,就是没生效!

这个原因是:Logstash 默认会上传一个名叫 logstash 的模板到 ES 里。如果你在使用上面这个配置之前,曾经运行过 Logstash(一般来说都会),那么 ES 里就已经存在这么一个模板了。你可以curl -XGET 'http://127.0.0.1:9200/_template/logstash' 验证。

这个时候,ES 里就变成有两个模板,logstash 和 myname,都匹配 logstash-* 索引名,要求设置一定的映射规则了。

ES 会按照一定的规则来尝试自动 merge 多个都匹配上了的模板规则,最终运用到索引上:https://www.elastic.co/guide/e ... lates

其中要点就是:template 是可以设置 order 参数的!而不写这个参数,默认的 order 值就是 0。order 值越大,在 merge 规则的时候优先级越高。

所以,解决这个问题的办法很简单:在你自定义的 template 里,加一行,变成这样:
{
"template" : "logstash-*",
"order" : 1,
"settings" : { ... },
"mappings" : { ... }
}当然,其实如果只从 Logstash 配置角度出发,其实更简单的办法是:直接修改原来默认的 logstash 模板,然后模板名称也不要改,就好了:
output {
elasticsearch {
host => "127.0.0.1"
manage_template => true
template_overwrite => true
}
}想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
ELK Stack 在入门学习过程中,必然会碰到自己修改定制索引映射(mapping)乃至模板(template)的问题。
这时候,不少比较认真看 Logstash 文档的新用户会通过下面这段配置来制定自己的模板策略:
output {
elasticsearch {
host => "127.0.0.1"
manage_template => true
template => "/path/to/mytemplate"
template_name => "myname"
}
}

然而随后就发现,自己辛辛苦苦修改出来的模板,通过 curl -XGET 'http://127.0.0.1:9200/_template/myname' 看也确实上传成功了,但实际新数据索引创建出来,就是没生效!

这个原因是:Logstash 默认会上传一个名叫 logstash 的模板到 ES 里。如果你在使用上面这个配置之前,曾经运行过 Logstash(一般来说都会),那么 ES 里就已经存在这么一个模板了。你可以curl -XGET 'http://127.0.0.1:9200/_template/logstash' 验证。

这个时候,ES 里就变成有两个模板,logstash 和 myname,都匹配 logstash-* 索引名,要求设置一定的映射规则了。

ES 会按照一定的规则来尝试自动 merge 多个都匹配上了的模板规则,最终运用到索引上:https://www.elastic.co/guide/e ... lates

其中要点就是:template 是可以设置 order 参数的!而不写这个参数,默认的 order 值就是 0。order 值越大,在 merge 规则的时候优先级越高。

所以,解决这个问题的办法很简单:在你自定义的 template 里,加一行,变成这样:
{
"template" : "logstash-*",
"order" : 1,
"settings" : { ... },
"mappings" : { ... }
}
当然,其实如果只从 Logstash 配置角度出发,其实更简单的办法是:直接修改原来默认的 logstash 模板,然后模板名称也不要改,就好了:
output {
elasticsearch {
host => "127.0.0.1"
manage_template => true
template_overwrite => true
}
}
想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day7: hangout 替代 logstash-input-kafka

Advent三斗室 发表了文章 • 2 个评论 • 2082 次浏览 • 2015-12-08 00:54 • 来自相关话题

用 Logstash 接收 Kafka 里的业务日志再写入 Elasticsearch 已经成为一个常见的选择。但是大多数人随后就会碰到一个问题:logstash-input-kafka 的性能上不去!

这个问题,主要是由于 Logstash 用 JRuby 实现,所以数据从 Kafka 下来到最后流转进 Logstash 里,要经过四五次 Ruby 和 Java 之间的数据结构转换,大大浪费和消耗了 CPU 资源。作为优化,我们可以通过修改默认的 logstash-input-kafka 的 codec 配置为 line,把 Jrjackson 处理流程挪到 logstash-filter-json 里多线程处理,但是也只能提高一倍性能而已。

Logstash 开发组目前也在实现纯 Java 版的 logstash-core-event,但是最终能提高多少,也是未知数。

那么在 Logstash 性能提上去之前,围绕 Kafka 还有什么办法能高效又不失灵活的做到数据处理并写入 Elasticsearch 呢?今天给大家推荐一下携程网开源的 hangout。

hangout 采用 YAML 格式配置语法,跟 Elasticsearch 一样,省去了 Logstash 解析 DSL 的复杂度。下面一段配置是 repo 中自带的 example 示例:
inputs:
- Kafka:
codec: plain
encoding: UTF8 # defaut UTF8
topic:
app: 2
consumer_settings:
group.id: hangout
zookeeper.connect: 192.168.1.200:2181
auto.commit.interval.ms: "1000"
socket.receive.buffer.bytes: "1048576"
fetch.message.max.bytes: "1048576"
num.consumer.fetchers: "4"
- Kafka:
codec: json
topic:
web: 1
consumer_settings:
group.id: hangout
zookeeper.connect: 192.168.1.201:2181
auto.commit.interval.ms: "5000"

filters:
- Grok:
match:
- '^(?<logtime>\S+) (?<user>.+) (-|(?<level>\w+)) %{DATA:msg}$'
remove_fields: ['message']
- Add:
fields:
test: 'abcd'
if:
- '<#if message??>true</#if>'
- '<#if message?contains("liu")>true<#elseif message?contains("warn")>true</#if>'
- Date:
src: logtime
formats:
- 'ISO8601'
remove_fields: ['logtime']
- Lowercase:
fields: ['user']
- Add:
fields:
me: 'I am ${user}'
- Remove:
fields:
- logtime
- Trim:
fields:
- user
- Rename:
fields:
me: he
user: she
- Gsub:
fields:
she: ['c','CCC']
he: ['(^\w+)|(\w+$)','XXX']
- Translate:
source: user
target: nick
dictionary_path: /tmp/app.dic
- KV:
source: msg
target: kv
field_split: ' '
value_split: '='
trim: '\t\"'
trimkey: '\"'
include_keys: ["a","b","xyz","12"]
exclude_keys: ["b","c"] # b in excluded
tag_on_failure: "KVfail"
remove_fields: ['msg']
- Convert:
fields:
cs_bytes: integer
time_taken: float
- URLDecode:
fields: ["query1","query2"]

outputs:
- Stdout:
if:
- '<#if user=="childe">true</#if>'
- Elasticsearch:
cluster: hangoutcluster
hosts:
- 192.168.1.200
index: 'hangout-%{user}-%{+YYYY.MM.dd}'
index_type: logs # default logs
bulk_actions: 20000 #default 20000
bulk_size: 15 # default 15 MB
flush_interval: 10 # default 10 seconds
concurrent_requests: 0 # default 0, concurrent_requests设置成大于0的数, 意思着多线程处理, 以我应用的经验,还有是一定OOM风险的,强烈建议设置为0
- Kafka:
broker_list: 192.168.1.200:9092
topic: test2其 pipeline 设计和 Logstash 不同的是:整个 filter 和 output 流程,都在 Kafka 的 consumer 线程中完成。所以,并发线程数完全是有 Kafka 的 partitions 设置来控制的。

实际运行下来,hangout 比 Logstash 确实在处理能力,尤其是 CPU 资源消耗方面,性价比要高出很多。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
用 Logstash 接收 Kafka 里的业务日志再写入 Elasticsearch 已经成为一个常见的选择。但是大多数人随后就会碰到一个问题:logstash-input-kafka 的性能上不去!

这个问题,主要是由于 Logstash 用 JRuby 实现,所以数据从 Kafka 下来到最后流转进 Logstash 里,要经过四五次 Ruby 和 Java 之间的数据结构转换,大大浪费和消耗了 CPU 资源。作为优化,我们可以通过修改默认的 logstash-input-kafka 的 codec 配置为 line,把 Jrjackson 处理流程挪到 logstash-filter-json 里多线程处理,但是也只能提高一倍性能而已。

Logstash 开发组目前也在实现纯 Java 版的 logstash-core-event,但是最终能提高多少,也是未知数。

那么在 Logstash 性能提上去之前,围绕 Kafka 还有什么办法能高效又不失灵活的做到数据处理并写入 Elasticsearch 呢?今天给大家推荐一下携程网开源的 hangout

hangout 采用 YAML 格式配置语法,跟 Elasticsearch 一样,省去了 Logstash 解析 DSL 的复杂度。下面一段配置是 repo 中自带的 example 示例:
inputs:
- Kafka:
codec: plain
encoding: UTF8 # defaut UTF8
topic:
app: 2
consumer_settings:
group.id: hangout
zookeeper.connect: 192.168.1.200:2181
auto.commit.interval.ms: "1000"
socket.receive.buffer.bytes: "1048576"
fetch.message.max.bytes: "1048576"
num.consumer.fetchers: "4"
- Kafka:
codec: json
topic:
web: 1
consumer_settings:
group.id: hangout
zookeeper.connect: 192.168.1.201:2181
auto.commit.interval.ms: "5000"

filters:
- Grok:
match:
- '^(?<logtime>\S+) (?<user>.+) (-|(?<level>\w+)) %{DATA:msg}$'
remove_fields: ['message']
- Add:
fields:
test: 'abcd'
if:
- '<#if message??>true</#if>'
- '<#if message?contains("liu")>true<#elseif message?contains("warn")>true</#if>'
- Date:
src: logtime
formats:
- 'ISO8601'
remove_fields: ['logtime']
- Lowercase:
fields: ['user']
- Add:
fields:
me: 'I am ${user}'
- Remove:
fields:
- logtime
- Trim:
fields:
- user
- Rename:
fields:
me: he
user: she
- Gsub:
fields:
she: ['c','CCC']
he: ['(^\w+)|(\w+$)','XXX']
- Translate:
source: user
target: nick
dictionary_path: /tmp/app.dic
- KV:
source: msg
target: kv
field_split: ' '
value_split: '='
trim: '\t\"'
trimkey: '\"'
include_keys: ["a","b","xyz","12"]
exclude_keys: ["b","c"] # b in excluded
tag_on_failure: "KVfail"
remove_fields: ['msg']
- Convert:
fields:
cs_bytes: integer
time_taken: float
- URLDecode:
fields: ["query1","query2"]

outputs:
- Stdout:
if:
- '<#if user=="childe">true</#if>'
- Elasticsearch:
cluster: hangoutcluster
hosts:
- 192.168.1.200
index: 'hangout-%{user}-%{+YYYY.MM.dd}'
index_type: logs # default logs
bulk_actions: 20000 #default 20000
bulk_size: 15 # default 15 MB
flush_interval: 10 # default 10 seconds
concurrent_requests: 0 # default 0, concurrent_requests设置成大于0的数, 意思着多线程处理, 以我应用的经验,还有是一定OOM风险的,强烈建议设置为0
- Kafka:
broker_list: 192.168.1.200:9092
topic: test2
其 pipeline 设计和 Logstash 不同的是:整个 filter 和 output 流程,都在 Kafka 的 consumer 线程中完成。所以,并发线程数完全是有 Kafka 的 partitions 设置来控制的。

实际运行下来,hangout 比 Logstash 确实在处理能力,尤其是 CPU 资源消耗方面,性价比要高出很多。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day6:用logstash-input-http_poller模拟nginxbeat

Advent三斗室 发表了文章 • 0 个评论 • 965 次浏览 • 2015-12-07 00:18 • 来自相关话题

Elastic 公司最近推出了 beats 系列,在官方的 packet/top/file{beat} 之外,社区也自发制作了一些比如 docker/nginx/

不过很可惜的是:nginxbeat 只支持两个数据来源:标准的 ngx_http_stub_status_module 和商业版 Nginx Plus 的ngx_http_status_module

我们都知道,ngx_http_stub_status_module 输出的信息太少,除了进程级别的连接数,啥都没有。那么,在使用开源版本 Nginx 的我们,还有别的办法么?

在官网的第三方模块列表里,发现了一个韩国人写的 nginx-module-vts。这个扩展可以做到 vhost 级别的状态信息输出。(我知道国人还有很多类似的统计扩展,但是没上官网,不便普及,就忽略吧)

但是,不懂 Golang 的话,没法自己动手实现一个 nginx-vts-beat 啊。怎么办?

其实我们可以用 logstash-input-http_poller 实现类似的功能。

首先,我们要给自己的 Nginx 加上 vts 扩展。编译方式这里就不讲了,和所有其他第三方模块一样。配置方式详见README。我们这里假设是按照核心和非核心接口来统计 URL 的状态:
http {
vhost_traffic_status_zone;

map $uri $filter_uri {
default 'non-core';
/2/api/timeline core;
~^/2/api/unread core;
}

server {
vhost_traffic_status_filter_by_set_key $filter_uri;
location /status {
auth_basic "Restricted";
auth_basic_user_file pass_file;
vhost_traffic_status_display;
vhost_traffic_status_display_format json;
}
}
}然后我们需要下面一段 Logstash 配置来定期获取这个数据:

input {
http_poller {
urls => {
0 => {
method => get
url => "http://localhost:80/status/format/json"
headers => {
Accept => "application/json"
}
auth => {
user => "YouKnowIKnow"
password => "IKnowYouDonotKnow"
}
}
1 => {
method => get
url => "http://localhost:80/status/con ... up%3D*"
headers => {
Accept => "application/json"
}
auth => {
user => "YouKnowIKnow"
password => "IKnowYouDonotKnow"
}
}
}
request_timeout => 60
interval => 60
codec => "json"
}
}这样,就可以每 60 秒,获得一次 vts 数据,并重置计数了。

注意,urls 是一个 Hash,所以他的执行顺序是根据 Hash.map 来的,为了确保我们是先获取数据再重置,这里干脆用 0, 1 来作为 Hash 的 key,这样顺序就没问题了。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
Elastic 公司最近推出了 beats 系列,在官方的 packet/top/file{beat} 之外,社区也自发制作了一些比如 docker/nginx/

不过很可惜的是:nginxbeat 只支持两个数据来源:标准的 ngx_http_stub_status_module 和商业版 Nginx Plus 的ngx_http_status_module

我们都知道,ngx_http_stub_status_module 输出的信息太少,除了进程级别的连接数,啥都没有。那么,在使用开源版本 Nginx 的我们,还有别的办法么?

在官网的第三方模块列表里,发现了一个韩国人写的 nginx-module-vts。这个扩展可以做到 vhost 级别的状态信息输出。(我知道国人还有很多类似的统计扩展,但是没上官网,不便普及,就忽略吧)

但是,不懂 Golang 的话,没法自己动手实现一个 nginx-vts-beat 啊。怎么办?

其实我们可以用 logstash-input-http_poller 实现类似的功能。

首先,我们要给自己的 Nginx 加上 vts 扩展。编译方式这里就不讲了,和所有其他第三方模块一样。配置方式详见README。我们这里假设是按照核心和非核心接口来统计 URL 的状态:
http {
vhost_traffic_status_zone;

map $uri $filter_uri {
default 'non-core';
/2/api/timeline core;
~^/2/api/unread core;
}

server {
vhost_traffic_status_filter_by_set_key $filter_uri;
location /status {
auth_basic "Restricted";
auth_basic_user_file pass_file;
vhost_traffic_status_display;
vhost_traffic_status_display_format json;
}
}
}
然后我们需要下面一段 Logstash 配置来定期获取这个数据:

input {
http_poller {
urls => {
0 => {
method => get
url => "http://localhost:80/status/format/json"
headers => {
Accept => "application/json"
}
auth => {
user => "YouKnowIKnow"
password => "IKnowYouDonotKnow"
}
}
1 => {
method => get
url => "http://localhost:80/status/con ... up%3D*"
headers => {
Accept => "application/json"
}
auth => {
user => "YouKnowIKnow"
password => "IKnowYouDonotKnow"
}
}
}
request_timeout => 60
interval => 60
codec => "json"
}
}
这样,就可以每 60 秒,获得一次 vts 数据,并重置计数了。

注意,urls 是一个 Hash,所以他的执行顺序是根据 Hash.map 来的,为了确保我们是先获取数据再重置,这里干脆用 0, 1 来作为 Hash 的 key,这样顺序就没问题了。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day1: 怎样让Logstash每次都从头读文件?

Advent三斗室 发表了文章 • 0 个评论 • 1615 次浏览 • 2015-12-05 11:56 • 来自相关话题

Advent Calendar 是各大技术社区每年 12 月大多会举办的一个系列活动。原意是圣诞节前夕的小礼品,延伸为每天一篇技术小分享的意思。最常见的包括 Perl Advent、sysadmin advent、web advent、performance advent 等。个人从 2009 年开始每年都看,从2013 年开始偶尔会参加其他社区的 advent 写作。今年考虑自己在 ELK Stack 上专注较多,在历次技术大会和最终出版的《ELK Stack权威指南》之外,又有一些新的发现和收获,干脆尝试一把自己一个人的 advent,也算是对 ELK 小知识的一种查漏补缺。

今天是 12 月 1 日,第一天,开天辟地,让我们也从最简单而又容易被忽略的一个小技巧开始吧!

每个上手 ELK 的新用户,肯定都需要测试一下读取文件输出到终端这步。在 Logstash 中,也就是配置这样一段:input {
file {
path => ["/data/test.log"]
}
}
output {
stdout {
codec => rubydebug
}
}不过很多新人的测试随后就卡在第二步了:当你修改一下配置,准备添加一段 filter 配置再重复运行 logstash 命令时,发现终端一直停滞没有输出。

这是因为:Logstash 会记录自己读取文件内容的偏移量到一个隐藏文件里,默认情况下,下次启动,他会从这个偏移量继续往后读,避免重复读取数据。

这个隐藏文件,叫做 $HOME/.sincedb_****。过去很多文档,在解释了这个原理后,都会告诉大家解决办法:每次重新运行 logstash 命令之前,删除掉家目录下的 sincedb 隐藏文件。

但是这种办法很笨,不是么?

今天告诉大家一个更方便的办法,改用下面这段 Logstash 配置:
input {
file {
path => ["/data/test.log"]
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
output {
stdout {
codec => rubydebug
}
}要点就在这行 sincedb_path => "/dev/null" 了!该参数用来指定 sincedb 文件名,但是如果我们设置为 /dev/null这个 Linux 系统上特殊的空洞文件,那么 logstash 每次重启进程的时候,尝试读取 sincedb 内容,都只会读到空白内容,也就会理解成之前没有过运行记录,自然就从初始位置开始读取了!

好了,第一天就是这样。更多内容,敬请期待。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。 查看全部
Advent Calendar 是各大技术社区每年 12 月大多会举办的一个系列活动。原意是圣诞节前夕的小礼品,延伸为每天一篇技术小分享的意思。最常见的包括 Perl Advent、sysadmin advent、web advent、performance advent 等。个人从 2009 年开始每年都看,从2013 年开始偶尔会参加其他社区的 advent 写作。今年考虑自己在 ELK Stack 上专注较多,在历次技术大会和最终出版的《ELK Stack权威指南》之外,又有一些新的发现和收获,干脆尝试一把自己一个人的 advent,也算是对 ELK 小知识的一种查漏补缺。

今天是 12 月 1 日,第一天,开天辟地,让我们也从最简单而又容易被忽略的一个小技巧开始吧!

每个上手 ELK 的新用户,肯定都需要测试一下读取文件输出到终端这步。在 Logstash 中,也就是配置这样一段:
input {
file {
path => ["/data/test.log"]
}
}
output {
stdout {
codec => rubydebug
}
}
不过很多新人的测试随后就卡在第二步了:当你修改一下配置,准备添加一段 filter 配置再重复运行 logstash 命令时,发现终端一直停滞没有输出

这是因为:Logstash 会记录自己读取文件内容的偏移量到一个隐藏文件里,默认情况下,下次启动,他会从这个偏移量继续往后读,避免重复读取数据。

这个隐藏文件,叫做 $HOME/.sincedb_****。过去很多文档,在解释了这个原理后,都会告诉大家解决办法:每次重新运行 logstash 命令之前,删除掉家目录下的 sincedb 隐藏文件。

但是这种办法很笨,不是么?

今天告诉大家一个更方便的办法,改用下面这段 Logstash 配置:
input {
file {
path => ["/data/test.log"]
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
output {
stdout {
codec => rubydebug
}
}
要点就在这行 sincedb_path => "/dev/null" 了!该参数用来指定 sincedb 文件名,但是如果我们设置为 /dev/null这个 Linux 系统上特殊的空洞文件,那么 logstash 每次重启进程的时候,尝试读取 sincedb 内容,都只会读到空白内容,也就会理解成之前没有过运行记录,自然就从初始位置开始读取了!

好了,第一天就是这样。更多内容,敬请期待。

想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

elasticsearch logstash kibana beats 资料分享

资料分享abcdef 发表了文章 • 1 个评论 • 1352 次浏览 • 2015-12-02 14:40 • 来自相关话题

ELK系列文章推荐 http://www.ttlsa.com/log-system/elk/    写的还不错。
ELK系列文章推荐 http://www.ttlsa.com/log-system/elk/    写的还不错。
  Logstash是一个开源的日志收集管理工具。