目前我们使用了 monit 作为监控服务应用,轻量级,易配置。由于是单机版,就产生了一个问题:多个监控节点同时产生内容相同的告警时,伴随而来的就是好几封一模一样的告警邮件。

最近通过学习 ELK 里的 logstash 组件,发现它有 throttleemail 插件,可以做到聚合告警。所以,马上验证一下!

logstash 的 pipline 配置样例如下:

input {
  beats {
    port => "5044"
  }
}

filter {
  if "monit.log" in [source] {
    grok {
      match => { "message" => "\[CST %{SYSLOGTIMESTAMP:[monit][timestamp]}\]%{SPACE}%{LOGLEVEL:[monit][level]}%{SPACE}\:%{SPACE}%{GREEDYDATA:[monit][log]}" }
      remove_field => "message"
    }
    throttle {
      before_count => -1 # 最小值
      after_count => 5 # 最大值
      period => 60 # 统计周期,单位:秒。
      max_age => 300 # 有效周期, 这个时间内不再触发同一个条件。单位:秒。
      key => "%{[monit][log]}" # 要统计的字段
      add_tag => "throttled"
    }
    date {
      match => [ "[monit][timestamp]", "MMM dd HH:mm:ss" ]
      remove_field => "[monit][timestamp]"
    }
  }
}

output {
  if "throttled" not in [tags] and [monit][level] == "error" {
    email {
      from => "admin@localhost"
      to => "haibin.li@somewhere.else"
      subject => "Monit@%{[beat][name]} 告警 - 事件级别:%{[monit][level]}"
      via => "sendmail"
      body => "事件内容: %{[monit][log]}\\n来自节点:%{[beat][name]}"
    }
  }
  elasticsearch {
    hosts => ["http://master:9200"]
    index => "monit-%{+YYYY.MM.dd}"
  }
}

应该比较容易看明白。关键是要理解 throttle 的定义。应用 grok 重构了日志后,统计 monit.log 这个字段,内容相同的,60 秒内出现次数在 5 以内的,logstash 给这条信息打上 throttled 标记,意思就是通过的。

但是,如果在 60 秒,有 7 个节点都产生了同一个告警内容,并且 monit.level 是 error 的,logstash 就会触发一封邮件。

在这个基础上,细化判断条件后,就可以再根据不同的事件,定义不同的邮件内容。比如某个事件应该发送邮件给具体某个邮件组,邮件内容里要包含的具体责任人和联系方式。