在logstash内部, input到filter, 以及filter到output, 消息都是通过一个队列来中转.

在我写hangout的第一个版本,也是这么做的,用ArrayBlockingQueue来中转消息, 上游几个线程把消息放在queue中, 下游再几个线程把queue中的消息消费走.

但是, 用下来之后, 发现在queue上面消耗的资源是相当的大,strace查看,非常大量的lock相关的系统调用, 现在的版本已经把queue去掉了. 想必Logstash也会有大量资源用在这一块.

zeromq中的Parallel Pipeline正好适合这个场景,而且文档中说是lock free的, 拿来和queue对比一下看.

在我自己的电脑上测试,2.6 GHz Intel Core i5.  一个主线程生成10,000,000个随机数, 分发给四个线程消费.

用Queue来实现, 需要约37秒, CPU使用率在150%. 用zeromq的ipc来传递消息, 只需要22秒, 期间CPU使用率在250%. 总的CPU使用时间都60秒左右.

不知道java中还有没有更合适的Queue可以用在这个场景中.至少zeromq和ArrayBlockingQueue相比, zeromq可以更快的处理消息, 但代价就是更高的CPU使用率.

代@childe 发文。

除了应用在日志系统外, 越来越多的业务数据也接入ES, 利用它天生强大的搜索性能和分布式可扩展, 可以为业务的精确快速灵活的搜索提供极大便利, 我觉得这是未来一个很好的方向.
但是, 对它ES各种各样的搜索方式, 你了解了吗?
我们来看几个”奇怪”的搜索.
## 奇怪的打分
### 奇怪的打分1
我们有个数据结构是

{

 “first_name”:”string”,

 “last_name”:”string”

}

插入了几条数据, 有诸葛亮 诸葛明 诸葛暗 诸葛黑, 还有个人名字很奇怪, 叫司马诸葛.
然后我们要搜索诸葛瑾, 虽然索引里面没有一个人叫这个名字, 但搜索出来诸葛亮也不错, 他们名字这么像, 说不定是亲兄弟, 可以顺藤摸瓜, 找到我们需要的信息呢.

{

    "query": {

        "multi_match": {

            "query":       "诸葛瑜",

            "type":        "most_fields",

            "fields":      [ “*_name” ]

        }

    }

}

但实际上呢, 司马诸葛这个人居然稳居搜索榜首位, 他是搞竞价排名了吧? 你知道其中的打分原理吗?
### 奇怪的打分2
我们有两条数据:

PUT /my_index/my_type/1

{

    "title": "Quick brown rabbits",

    "body":  "Brown rabbits are commonly seen."

}

PUT /my_index/my_type/2

{

    "title": "Keeping pets healthy",

    "body":  "My quick brown fox eats rabbits on a regular basis."

}

要搜索

{

    "query": {

        "bool": {

            "should": [

                { "match": { "title": "Brown fox" }},

                { "match": { "body":  "Brown fox" }}

            ]

        }

    }

}

第二条文档里面明确含有”brown fox”这个词组, 但是它的搜索得分比较低, 你知道为啥吗?
## and用在哪

{

    "query": {

        "multi_match": {

            "query":       "peter smith",

            "type":        "most_fields",

            "operator":    "and",

            "fields":      [ "first_name", "last_name" ]

        }

    }

}

你知道这个and代表什么吗?
是说
A: 姓和名里面都要含有"peter smith”,
还是说
B: 姓或者名里面要包含peter以及smith ?
还有, 怎么才能获得另外一个效果呢?
# 列表中的元素
我们有一条数据如下(按汉语分词)

{

 “时代”:”三国”,

 “姓名”: [“大司马”，“诸葛亮”]

}

我以词组的方式搜索:

{

    "query": {

        "match_phrase": {

            "姓名": "司马诸葛"

        }

    }

}

能搜索到吗?
上面这些其实都是[elasticsearch Definitive Guide](https://www.elastic.co/guide)里面的几个小例子, 欢迎大家继续去那里寻找答案和其他各种小技巧.
 

看到前一天, Medcl 介绍了Beat, 我想今天我就介绍一下算是同一个领域的, 我们的一个小产品吧, 同样也基于elastic旗下的logstash-forwarder. 我真的不是来打广告的, 就是第一次写, 没经验, 看着前一天的文章, 顺手就想到了.

在日志收集系统中, 从kafkf到ES这条路是没问题了, 但散布在各个服务器上采集日志的agent用logstash实在是太重了, 而且效率也低. 特别是我们有大量的windows服务器, 找一个合适的agent居然不是想象中的容易.

logstash-forwarder对于日志文件的探测和offset记录, deadtime等配置都非常适合我们, 但惟一不支持吐数据到kafak,对我们来说是一个遗憾. 我和oliver(https://github.com/oliveagle)做过一点改造之后, 让她支持了这个功能.

目前我们所有iis服务器已经部署了这个应用, 效率高, 占资源小, 可以数据压缩, 支持简单的格式切割, 实乃windows居家必备(我真不是来打广告的). golang客户端, 还能直接发送到kafka, 想想就很贴心~

贴上一段配置瞅瞅先, 启一个进程采集nginx和tomcat日志, 分别吐到kafka的2个topic中.

{

    "files": [

    {

        "paths": [

            "/var/log/nginx/*.log"

            ],

        "Fields":{

            "type":"nginx"

        },

        "DeadTime": "30m"

    },

    {

        "paths": [

            "/var/log/tomcat/*.log",

            "/var/log/tomcat/*/*.log"

            ],

        "Fields":{

            "type":"tomcat"

        },

        "DeadTime": "30m"

    }

    ],

    "kafka": {

        "broker_list": ["10.0.0.1:9092","10.0.0.2:9092"],

        "topic_id": "topic_name_change_it_{{.type}}",

        "compression_codec": "gzip"

    }

}

再简单介绍一下参数吧,

• DeadTime:30m 是说超过30分钟没有更新, 就不会再继续跟踪这个文件了(退出goroutine)
• “Fields”:{ “type”:”tomcat” } , 会在每条日志中增加配置的字段
• path目前就是用的golang官方库, 好像是还不支持递归多层目录查找, 反正我翻了一下文档, 没有找到.

grok还不支持, 但简单的分割是可以的

"files": [

        {

            "paths": [

                "d:\\target.txt"

                ],

            "FieldNames": ["datetime", "datetime", "s_ip", "cs_method", "cs_uri_stem", "cs_uri_query", "s_port", "time_taken"],

            "Delimiter": "\\s+",

            "QuoteChar": "\""

        }

]

以上配置就是说按空白符把日志切割来, 塞到对应的字段中去. 第一个第二个合在一起, 放在datetime字段中.

其实还是有不少要完善的地方, 比如说没有带上机器的Hostname, 以及日志的路径. 在很多时候, 这些信息还是很有用的, 我们也会继续完善.

现在放在了https://github.com/childe/logs ... kafka, 有需要的同学,可以去看下.

Advent接力传到我这里了，今天我给大家介绍一下Beats，刚好前几天也有好多人问我它是干嘛的，之前的上海我有分享过Beats的内容，PPT在这里：

https://pan.baidu.com/s/1eS157 ... -6-18 


事实上Beats是一系列产品的统称，属于ElasticStack里面收集数据的这一层：Data Shipper Layer，包括以下若干Beats：

1. PacketBeat，用来嗅探和分析网络流量，如HTTP、MySQL、Redis等
2. TopBeat，用来收集系统的监控信息，功能如其名，类似*nix下的top命令，只不过所有的信息都会发送给后端的集中存储：Elasticsearch，这样你就可以很方便的监控所有的服务器的运行情况了
3. FileBeat，用来收集数据源是文件的数据，比如常见的系统日志、应用日志、网站日志等等，FIleBeat思路来自Logstash-forwarder，Beats团队加入之后重构改写而成，解决的就是Logstash作为Agent采集时占用太多被收集系统资源的问题，Beats家族都是Golang编写，效率高，占用内存和CPU比较少，非常适合作为agent跑着服务器上
4. 。。。

所以Beats其实是一套框架，另外的一个子项目Libbeat，就是所有beats都共用的模块，封装了所有的公共的组件，如配置管理、公共基础类、协议的解析处理、与Elasticsearch的操作等等，你可以很方便基于它实现你自己的beats，这也是Beats的目标，希望将来会出现更多的Beats，做各种各样的事情。
 
另外PacketBeat比较特殊，它又是网络协议抓包和处理的一个框架，目前支持了常见的一些协议，要扩展未知的协议其实非常简单，PacketBeat作为一个框架，数据抓包和后续的存储已经帮你处理好了，你只需要实现你的协议的解码操作就行了，当然这块也是最难和最业务相关的。
 
关于PacketBeat我回头再单独写一篇文章来介绍怎样编写一个PacketBeat的协议扩展吧，PacketBeat扩展的其它协议最终还是需要和PacketBeat集成在一起，也就是最终你的代码是要和PacketBeat的代码在一个工程里面的，而其它的Beats使用Libbeat完全是单独的Beat，如Filebeat和TopBeat，完全是独立打包和独立运行，这个也是两大Beats的主要区别。
 
随便提一下，现在所有的这些Beats已经合并到一个项目里面来方便管理了，golang，you know：https://github.com/elastic/beats
 
现在社区已经提交了的Beats：
https://www.elastic.co/guide/e ... .html
 
明后天在Beijing的ArchSummit2015，我将在Elastic展台，欢迎过来骚扰，领取Elastic的各种贴纸，还有限量的印有Elastic的T恤，数量有限哦
 
今天的Advent就这些吧。
Advent接力活动，规则：http://elasticsearch.cn/article/20
 

我们都知道 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 哟。

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 哟。

很多从 MySQL 转过来的 Elasticsearch 用户总是很习惯的问一个问题：『怎么在 ES 里实现 join 操作？』过去，我们的回答一般都是：通过类似宽表的思路，将数据平铺在一个索引里。不过，最近另一家 Lucene 开发商给出了另一个方案，他们开发了一个 Elasticsearch 插件，实现了 filter 层面的 join，GitHub 项目地址见：https://github.com/sirensolutions/siren-join

不过需要提醒一下的是：filter 层面的意思，就是只相当于是 SQL 里的 exists 操作。所以目前对这个插件也不要抱有太大期望。今天我们来稍微演示一下。

安装和其他 ES 插件一样：

# bin/plugin -i solutions.siren/siren-join/1.0

注意 siren-join v1.0 只支持 ES 1.7 版本，2.0 版本支持据说正在开发中。

我们 bulk 上传这么一段数据：

{"index":{"_index":"index1","_type":"type","_id":"1"}}

{"id":1, "foreign_key":"13"}

{"index":{"_index":"index1","_type":"type","_id":"2"}}

{"id":2}

{"index":{"_index":"index1","_type":"type","_id":"3"}}

{"id":3, "foreign_key": "2"}

{"index":{"_index":"index1","_type":"type","_id":"4"}}

{"id":4, "foreign_key": "14"}

{"index":{"_index":"index1","_type":"type","_id":"5"}}

{"id":5, "foreign_key": "2"}

{"index":{"_index":"index2","_type":"type","_id":"1"}}

{"id":"1", "tag": "aaa"}

{"index":{"_index":"index2","_type":"type","_id":"2"}}

{"id":"2", "tag": "aaa"}

{"index":{"_index":"index2","_type":"type","_id":"3"}}

{"id":"3", "tag": "bbb"}

{"index":{"_index":"index2","_type":"type","_id":"4"}}

{"id":"4", "tag": "ccc"}

注意，siren-join 要求用来 join 的字段必须数据类型一致。所以，当我们要用 index2 的 id 和 index1 的foreign_key 做 join 的时候，这两个字段就要保持一致，这里为了演示，特意都改成字符串。那么我们发起一个请求如下：

# curl -s -XPOST 'http://localhost:9200/index1/_coordinate_search?pretty' -d '

{

    "query":{

        "filtered":{

            "query":{

                "match_all":{}

            },

            "filter":{

                "filterjoin":{

                    "foreign_key":{

                        "index":"index2",

                        "type":"type",

                        "path":"id",

                        "query":{

                            "terms":{

                                "tag":["aaa"]

                            }

                        }

                    }

                }

            }

        }

    },

    "aggs":{

        "avg":{

            "avg":{

                "field":"id"

            }

        }

    }

}'

意即：从 index2 中搜索 q=tag:aaa 的数据的 id，查找 index1 中对应 foreign_key 的文档的 id 数据平均值。响应结果如下：

{

    "coordinate_search" : {

        "actions" : [ {

            "relations" : {

                "from" : {

                    "indices" : [ ],

                    "types" : [ ],

                    "field" : "id"

                },

                "to" : {

                    "indices" : null,

                    "types" : null,

                    "field" : "foreign_key"

                }

            },

            "size" : 2,

            "size_in_bytes" : 20,

            "is_pruned" : false,

            "cache_hit" : true,

            "took" : 0

        } ]

    },

    "took" : 2,

    "timed_out" : false,

    "_shards" : {

        "total" : 5,

        "successful" : 5,

        "failed" : 0

    },

    "hits" : {

        "total" : 2,

        "max_score" : 1.0,

        "hits" : [ {

            "_index" : "index1",

            "_type" : "type",

            "_id" : "5",

            "_score" : 1.0,

            "_source":{"id":5, "foreign_key": "2"}

        }, {

            "_index" : "index1",

            "_type" : "type",

            "_id" : "3",

            "_score" : 1.0,

            "_source":{"id":3, "foreign_key": "2"}

        } ]

    },

    "aggregations" : {

        "avg" : {

            "value" : 4.0

        }

    }

}

响应告诉我们：从 index2 中搜索到 2 条参与 join 的文档，在 index1 中命中 2 条数据，最后求平均值为 4.0。

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