我正在使用 ELK(logstash、ES、Kibana)堆栈进行日志分析,并使用 Riemann 进行警报。我有日志,其中用户是由 logstash 解析的字段之一,我将事件从 riemann 输出插件发送到 riemann。
Logstash 解析日志,用户是字段之一。例如:日志解析
Timestamp user command-name
2014-06-07... root sh ./scripts/abc.sh
2014-06-08... sid sh ./scripts/xyz.sh
2014-06-08... abc sh ./scripts/xyz.sh
2014-06-09... root sh ./scripts/xyz.sh
日志存储:
riemann {
riemann_event => {
"service" => "logins"
"unique_user" => "%{user}"
}
}
因此用户值将是:root、sid、abc、root、sid、def 等。
所以我按用户拆分流,即每个唯一用户一个流。现在,我想在唯一用户数超过 3 时发出警报。我写了以下内容,但没有达到我的目的。
黎曼:
(streams
(where (service "logins")
(by :unique_user
(moving-time-window 3600
(smap (fn [events]
(let
[users (count events)]
(if (> users 3)
(email "abc@gmail.com")
))))))))
我是 Riemann 和 Clojure 的新手。任何帮助表示赞赏。