在实际的项目中需要对线上日志做实时分析跟统计，这一套方案可以用现有的ELK(ElasticSearch, Logstash, Kibana)方案既可以满足，关于这个方案的具体的步骤可以参考网上的解决方案。但如果只想统计某个错误码(http状态码，业务错误码)在指定时间内出现多少次然后就触发一个告警或者某个指令动作(邮件或者是调用已经写好的http接口，例如例如微信平台来通知告警信息等)，这种需求可以用logstash进行实现，并且这种方案比较轻量级，很容易实现。这里以在linux平台为主。
下载
首先下载 最新的logstash 压缩包，https://download.elastic.co/logstash/logstash/logstash-2.3.2.zip ，下载后用unzip命令进行解压只指定的目录，解压后会生成一个目录logstash-2.3.2 这个目录，里面包含下面这些文件：
基本配置
开始可以通过一个小的介绍了解logstash是怎么回事，链接里面有一些基本的demo，有兴趣可以参考里面的写一下。链接如下：
http://udn.yyuap.com/doc/logstash-best-practice-cn/get_start/hello_world.html
运行命令如下：
假设在与logstash 平级目录新建一个 配置文件，sample.conf ,这个配置文件就是定义 input filter output 这3块，其主要表示 input的来源是什么(文件或命令之类的),filter 代表对于输入的内容按照规则怎么来过滤，output 就是按照过滤的规则进行输出 。
最简单的写法如下：
# bin/logstash -e 'input{stdin{}}output{stdout{codec=>rubydebug}}'
如果是在sample.conf 中写入则如下：
input {
stdin{}
}
filter {
}
output {
stdout {}
}
# bin/logstash -f sample.conf
特殊配置
在实际的业务中可能往往要对日志做过滤，然后根据某些关键字段进行统计，通过统计结果来决定需要做什么。logstash 本身对于syslog,redis等日志在过滤的时候一般不需要自己写正则表达式去过滤，这些都是通用的，在filter中可以直接通过%{定义的常量}来获取日志内容，已经自动支持的过滤的日志包含下面这些：
这个列表不全，对于已经支持的这些可以直接到github 上进行搜索(https://github.com/logstash/logstash/tree/v1.4.2/patterns)，上面都有这些日志过滤的正则表达式,这些是属于通用配置。
往往在业务日志中，所打印的业务关键错误码以及格式并非与通用格式相匹配，那么此时就需要自己针对自己的业务日志格式，写出正则表达式来过滤自己想要的信息。
这里就以日志中的errno 来为例，所实现的目的就是 当这个错误码在30s内出现3次就把这个错误输出出来或告警。
在logstash中正则表达式叫做grok表达式，其规则跟perl与ruby的正则表达式一样。 也有在线的grok正则表达式校验工具：http://grokdebug.herokuapp.com/
这里就以下面的日志格式为例：log.txt
begin 10001 end
begin 10001 end
begin 10001 end
在sample.conf 中，grok 是写在filter 中的，这里既可以直接写正则表达式，也可以引用外部文件的正则表达。
方式1：
grok {
match => {
"message" => "\s+(?<code>\d+?)\s+"
}
}
方式2就是预先在 patterns/ 目录下创建一个文件，这个文件里面含有正则表达式，例如创建文件的名称为 test ，内容如下：
ERRORCODE \s+(?<code>\d+?)\s+
grok {
patterns_dir => "../../patterns/test"
match => ["message", "%{ERRORCODE:code:int}"]
}：
很显然这里我们习惯采用方式2，因为正则匹配相关的可以单独在一个文件，在sample.conf 就直接引用正则匹配中字段值，例如上面就直接用"ERRORCODE"， message 中%后面花括号中的写法规则如下：
%{PATTERN_NAME:capture_name:data_type}
以上面完子为例，完整的 sample.conf 内容如下：
input {
# 以日志文件作为来源
file {
# 文件路径
path => "../../log.txt"
}
}
filter {
# 正则解析
grok {
# 增加自定义的正则，这里我把自定义的正则文件，加在了这个目录下
patterns_dir => "../../patterns/test"
# 从message字段中解析出 code
match => ["message", "%{ERRORCODE:code:int}"]
# 删除原来的message字段
remove_field => [ "message" ]
# 增加WARN，作为标记
add_tag => "warn"
}
# 计数
metrics {
# 计数器数据保存的字段名 code的值就是上面解析出来的
meter => "events_%{code}"
# 增加"metric"，作为标记
add_tag => "metric"
# 3秒内的message数据才统计，避免延迟
# ignore_older_than =>
# 每隔300秒统计一次
flush_interval => ""
# 每301秒清空计数器
clear_interval => ""
}
# 如果event中标记为“metric”的
if "metric" in [tags] {
# 执行ruby代码
ruby {
# 如果count为10001的数量小于3条，就忽略此事件(即不发送任何消息)。
code => "event.cancel if event['events_10001']['count'] < 3"
}
}
}
output {
# 输出到console
stdout {
codec => rubydebug
}
}
上面配置结束后，由于开启 debug 模式，所以当每次往 log.txt 写入 "begin 10001 end" 在控制台下都会有输出，输入内容下：
{
"@version" => "",
"@timestamp" => "2016-11-14T13:14:14.374Z",
"path" => "../../log.txt",
"host" => "xxxx",
"code" => "",
"tags" => [
[] "warn"
]
}
当在30s内输入的次数大于3次的时候，输出的内容如下：
{
"@version" => "",
"@timestamp" => "2016-11-14T13:14:20.148Z",
"message" => "xxx",
"events_" => {
"WARN" => {
"count" => ,
"rate_1m" => 2.1864058457089883,
"rate_5m" => 3.814999397602868,
"rate_15m" => 4.1951645878441255
}
},
"tags" => [
[] "metric"
]
}
通过上面的结构可以看出我们在输出的时候还可以写表达式，例如只有当30s内大于3次才输出信息的话可以在 output 中加一个判断，逻辑如下：
output {
# 如果event中标记为“metric”，表示只发送计数的消息。
if "metric" in [tags] {
# 执行一个简单的命令
exec {
command => "echo \"hello: %{events_10001.count}\""
}
}
}
到这里已经结束，关于logstash中的 统计相关的规则可以参考：http://udn.yyuap.com/doc/logstash-best-practice-cn/filter/metrics.html
输出的形式也是多种的 可以对接ES,触发http接口，这里就以输出为例。