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

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

前几天，我们已经一步步搞定了一个业务日志从 mapping 设计到异常统计追踪上的用法。作为一个工程师，自评 100 分 —— But，领导找上门来说：你这个结构怎么搞的嘛，在 Kibana 上完全没法搜索！让客服和分析师怎么办？

因为 Kibana 上的输入框，默认使用 querystring 语法。这个里面压根没有对 nested object 的相关语法设计。

不过经过仔细查阅，发现原来 Kibana4 的搜索输入框，其实除了 querystring 以外，还支持 JSON 字符串的方式直接定义 query！其具体处理方式就是：把你输入的字符串判断一下是否是 JSON，如果是 JSON，直接替换进{"query": 这里}；如果不是，才生成一个 querystring query 放进 {"query":{"query_string":""}}

那我们来尝试一下把第三天写的那个 nested query 贴进搜索框里。内容是：

{

  "nested" : {

    "path" : "video_time_duration",

    "query" : {

      "match" : {

        "video_time_duration.type" : "1"

      }

    }

  }

}

意外发生了！Kibana4 竟然在页面上弹出一个错误提示，而且搜索栏的放大镜图标也变成不可以点击的灰色样式，敲回车同样没有反应：

当然我很确定我的数据是没问题的。这时候 Kibana4 的另一个特性救了我：它默认会把所有可修改的状态都 rison 序列化了放在 URL 里！于是我尝试直接在浏览器地址栏里输入下面这段 URL：

http://kibana:5601/#/discover?_g=()&_a=(columns:!(_source),index:%5Blogstash-mweibo-%5DYYYY.MM.DD,interval:auto,query:(nested:(path:video_time_duration,query:(term:(video_time_duration.type:1)))),sort:!('@timestamp',desc))

地址栏回车之后，页面刷新，看到搜索结果更新(如上图)！虽然搜索栏依然有报错，但实际上 nested query 生效了，我们在下面 search 里看到的都是成功过滤出来的『有过卡顿的视频播放记录』日志。

感谢 Kibana 如此开放的设计原则！

ps: 目前 nested aggregation 还没法像这样简单的绕过，不过已经有相关 pull request 在 review 中，或许 Kibana4.3/4.4 的时候就会合并了。有兴趣的同学，也可以跟我一样先睹为快哟：https://github.com/elastic/kibana/pull/5411

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

昨天我们通过 nested aggregation 计算出来，视频卡顿次数最多的是北京。不过这个结论似乎也没有什么奇怪的，北京的网民本身就多嘛。

Elasticsearch 还有一个有趣的聚合方式，叫 significant_terms。这时候就可以派上用场了！

我们把昨天的 query JSON 中，最后一段 sub agg 改成这样：

    "city_terms" : {

        "significant_terms" : {

            "field" : "geoip.city",

            "size" : "4"

        }

    }

重新运行请求，得到的响应结果是这样的：

"city_terms" : {

  "doc_count" : 2521720,

  "buckets" : [ {

    "key" : "武汉",

    "doc_count" : 85980,

    "score" : 0.1441705001066121,

    "bg_count" : 15347191

    }, {

    "key" : "北京",

    "doc_count" : 142761,

    "score" : 0.11808069152203737,

    "bg_count" : 43176384

    }, {

    "key" : "广州",

    "doc_count" : 104677,

    "score" : 0.10716870365361204,

    "bg_count" : 27274482

    }, {

    "key" : "郑州",

    "doc_count" : 59234,

    "score" : 0.09915501610550795,

    "bg_count" : 10587590

  } ]

}

大家一定发现了：第一名居然变成了武汉！

而且每个结果后面，还多出来了 score 和 bg_count 两个数据。这个 bg_count 是怎么回事呢？

这就是 significant_terms 的作用了。这个 agg 的大概计算步骤是这样：

1. 计算一个 term 在整个索引中的比例，作为背景计数(background)，这里是 15347191 / 2353406423；
2. 计算一个 term 在 parent agg 中的比例，作为前景计数(foreground)，这里是 85980 / 2521720；
3. 用 fgpercent 除以 bgpercent，得到这个 term 在 parent agg 的条件下比例凸显的可能性。

由于两个作分母的总数其实大家都是相等的，其实比较的就是各 term 的 doc_count / bg_count 了。

当然，实际的 score 不只是这么简单，还有其他综合因素。毕竟也不能给出来本身就没啥关注度的数据嘛。

我们还可以来验证一下『武汉』的 bg_count 是不是这个意思：

curl -XPOST 'http://10.19.0.67:9200/logstash-mweibo-2015.12.02/_count?pretty' -d '{

  "query" : {

    "match" : {

      "geoip.city" : "武汉"

    }

  }

}'

结果如下：

{

  "count" : 15347191,

  "_shards" : {

    "total" : 100,

    "successful" : 100,

    "failed" : 0

  }

}

数值完全对上了。没错，bg_count 就是『武汉』在整个索引里的总数。

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

话接上回，我们只是解决了写数据的问题，这种格式不太符合常规的数据怎么读，也需要我们相应的做出点改变。

今天以一个实际的例子来讲。我曾经处理过一份数据，记录的是视频播放的卡顿情况。其中有一个数组，每次卡顿就新增一个对象元素。所以设计的 mapping 如下：

         "video_time_duration" : {

           "type": "nested",

           "properties" : {

             "duration" : {

               "type" : "long",

               "doc_values" : true

             },

             "type" : {

               "type" : "long",

               "doc_values" : true

             }

           }

         },

其中 type 只有 0 或 1 两个可能，0 表示播放正常，1 表示卡顿。所以下面我们发一个请求，要求是计算这样的结果：

出现了播放卡顿的用户，单次卡顿时长在10到200ms的，最常见于哪些城市？

下面是我们最终的查询请求 JSON：

{

  "size" : 0,

  "query" : {

    "nested" : {

      "path" : "video_time_duration",

      "query" : {

        "match" : {

          "video_time_duration.type" : "1"

        }

      }

    }

  },

  "aggs" : {

    "video" : {

      "nested" : {

        "path" : "video_time_duration"

      },

      "aggs" : {

        "filter_type" : {

          "filter" : {

            "term" : {

              "video_time_duration.type" : "1"

            }

          },

          "aggs" : {

            "duration_ranges" : {

              "range" : {

                "field" : "video_time_duration.duration",

                "ranges" : [

                  { "from" : 10, "to" : 200 }

                ]

              },

              "aggs" : {

                "city" : {

                  "reverse_nested": {},

                  "aggs" : {

                    "city_terms" : {

                      "terms" : {

                        "field" : "geoip.city"

                      }

                    }

                  }

                }

              }

            }

          }

        }

      }

    }

  }

}

很明显的可以看到对 nested object 里存的数据，不管是做 query 还是 agg，都需要显式的加上"nested": {"path" : "video_time_duration" 的声明。这样，才能保证我们取到的 duration 数值是对应 type 为卡顿的，而不是流畅播放的。

大家可能注意到，我同时在 query 和 aggFilter 中重复了一场 term 过滤。其中这次 nested query 是不必要的，除了作为语法展示以外，也有一个减少 hits 数的作用。但是和一般的请求不同的是，这里不可以去掉 nested agg 里的 term filter，因为 nested query 只是拿到『有过卡顿』的数据 id。不加 filter，聚合 duration 的时候，会把卡过但也流畅过的那部分都计算在内。

另一个要点：当我们过滤好 nested 数据的时候，要取顶层其他字段的内容，在 sub agg 里是无法直接获取的，需要额外使用一次 reverse_nested 来跳出这个 nested path，才可以恢复正常的 agg 路径。

最终得到的响应如下：

{

  "took" : 4672,

  "timed_out" : false,

  "_shards" : {

    "total" : 100,

    "successful" : 100,

    "failed" : 0

  },

  "hits" : {

    "total" : 9560309,

    "max_score" : 0.0,

    "hits" : [ ]

  },

  "aggregations" : {

    "video" : {

      "doc_count" : 33713503,

      "filter_type" : {

        "doc_count" : 25441559,

        "duration_ranges" : {

          "buckets" : [ {

            "key" : "10.0-200.0",

            "from" : 10.0,

            "from_as_string" : "10.0",

            "to" : 200.0,

            "to_as_string" : "200.0",

            "doc_count" : 2521720,

            "city" : {

              "doc_count" : 2521720,

              "city_terms" : {

                "doc_count_error_upper_bound" : 0,

                "sum_other_doc_count" : 2267886,

                "buckets" : [ {

                    "key" : "北京",

                    "doc_count" : 142761

                  }, {

                    "key" : "广州",

                    "doc_count" : 104677

                  }

                ]

              }

            }

          } ]

        }

      }

    }

  }

}

响应数据中，我们可以直接看这些 hits 和 doc_count 数据。他们表示：

1. 一共命中了『有过卡顿』的视频播放次数：9560309；
2. 其中记录下来的播放间隔 33713503 次；
3. 里面有 25441559 次是卡顿(减一下即 8271944 次是流畅咯)；
4. 里面卡顿时长在 10-200 ms 的是 2521720 次；
5. 这些卡顿出现最多的在北京，发生了 142761 次。

数据蛮有意思吧。ES 能告诉你的还不止这点。更有趣的，明天见。

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

Elasticsearch 中有些高级特性，可能不太常用，但是在恰当场景下，又非常有效果。今天，我们来说说 nested object。

我们都知道，Elasticsearch 宣传中是 schemaless 的。但实际使用中，并不是完全的随意。比如过多的 kv 切割，会导致 mapping 大小暴涨，对集群稳定性是个不小的挑战。

以 urlparams 为例，下面这段 urlparams 直接通过 logstash-filter-kv 切割得到的结果，需要在 mapping 中占用 4 个字段的定义。

"urlparams" : {

    "uid" : "1234567890",

    "action" : "payload",

    "t" : "1449053032000",

    "pageid" : "v6"

}

如果哪个开发一时想不开(我真的碰到过)，把 urlparams 写成 uid=123456789&action=payload&1449053032000=t&pageid=v6，那基本上整个 ES 集群就会被过于频繁的 mapping 更新搞挂了。

这时候，我们修改一下 mapping 定义：

{

  "accesslog" : {

    "properties" : {

      "urlparams" : {

        "type" : "nested",

        "properties" : {

            "key" : { "type" : "string", "index" : "not_analyzed", "doc_values" : true },

            "value" : { "type" : "string", "index" : "not_analyzed", "doc_values" : true }

        }

      }

    }

  } 

}

同时在 Logstash 的 filter 配置中添加一段：

if [urlargs] {

    ruby {

        init => "@kname = ['key','value']"

        code => "event['urlparams'] = event['urlargs'].split('&').collect {|i| Hash[@kname.zip(i.split('='))]}"

        remove_field => [ "urlargs","uri","request" ]

    }

}

生成的 JSON 数据变成这个样子：

"urlargs": [

    { "key": "uid", "value": "1234567890" },

    { "key": "action", "value": "payload" },

    { "key": "1449053032000", "value": "t" },

    { "key": "pageid", "value": "v6" }

]

这样，再错乱的 urlparams，也不会发生 mapping 变更，导致集群故障了！

想了解更全面的 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 哟。