按唯一 IP 计算的 Logstash 计数

Question

我正在尝试使用 Logstash 进行一些日志分析。

我需要从 Apache 访问日志中计算唯一 IP，然后我需要将它们与计数过滤器进行匹配，以确定是否会发送电子邮件。

类似这样的： 如果在 5 分钟内发现来自唯一 IP 的 10+ 次访问，我需要发送一封带有此 IP 的电子邮件。

什么是最好的解决方案？

Answer 1

这样做非常困难——要做到这一点，您需要为每个 IP 地址创建一个计量器。一旦每个 IP 地址都有一个仪表，您就需要查看它的 rate_5m 并确定它是否超过了您的阈值（注意 rate_5m 是过去 5 分钟内的每秒速率）。一旦您决定需要发送警报，您可能希望在其中包含 IP 地址（因此我们需要使用 ruby​​ 过滤器提取它）......总而言之，我不确定我会曾经在生产中使用过这样的东西，因为它可能会疯狂地消耗内存（因为每个 ip 地址的仪表）。

filter {
  metrics {
    meter =>  "%{ip}"
    add_tag =>  ["metric"]
  }
  ruby { code => '
    ip = nil
    if event["tags"].include? "metric"
        event.to_hash.each do |key,value|
            if key.end_with?(".rate_5m") and value > 0.2
                ip = key[0..-9]
            end
        end
    end
    if ip
      event["ip"] = ip
      event["tags"] = ["alert"]
    end
    '
  }
}

output {
  if "alert" in [tags] {
    email { ... }
  }
}

您可能会编写一个更智能的自定义过滤器，使用趋势算法之类的东西来查找计数趋势更高的 IP 地址。