elasticsearch - Geoip Logstash过滤器

Question

我有一个配置文件是这样的:

input {
  file {
      path => "/home/kibana/Documents/external_noise.log"
      type => "external_noise"
      start_position => "beginning"
      sincedb_path => "/dev/null"
  }
}
filter {

grok {
match => { 'message' => '%{CISCOTIMESTAMP:timestamp} %{WORD:action}%{SPACE}%{DATA:logsource} %{DATA:interface} %{GREEDYDATA:kvpairs}' }

     }


kv   {
source => "kvpairs"
field_split => ";"
value_split => ":"
remove_field => "kvpairs"
}
mutate {
    remove_field => [ "message" ]

}
geoip {

source => "src"
target => "geoip"
database => "/etc/logstash/GeoLiteCity.dat"
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}

 mutate {
convert => [ " [geoip][coordinates]", "float"]
}


date {
    match => [ "timestamp" , "MMM dd HH:mm:ss" ]
    target => "@timestamp"
    }

if "_grokparsefailure" in [tags] {
    drop {}
    }

}


output {

stdout {
    codec => rubydebug
        }
elasticsearch {
    action => "index"
    host => "localhost"
    index => "external-%{+dd.MM.YYYY}"
    workers => 1

}
}

我的示例日志文件是这些:

Jan 1 22:54:17 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 70.77.116.190; dst: %DSTIP%; proto: tcp; product: VPN-1 & FireWall-1; service: 445; s_port: 2612;
Jan 1 22:54:22 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 61.164.41.144; dst: %DSTIP%; proto: udp; product: VPN-1 & FireWall-1; service: 5060; s_port: 5069;
Jan 1 22:54:23 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 69.55.245.136; dst: %DSTIP%; proto: tcp; product: VPN-1 & FireWall-1; service: 445; s_port: 2970;
Jan 1 22:54:41 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 95.104.65.30; dst: %DSTIP%; proto: tcp; product: VPN-1 & FireWall-1; service: 445; s_port: 2565;
Jan 1 22:54:43 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 222.186.24.11; dst: %DSTIP%; proto: tcp; product: VPN-1 & FireWall-1; service: 2967; s_port: 6000;
Jan 1 22:54:54 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 74.204.108.202; dst: %DSTIP%; proto: udp; product: VPN-1 & FireWall-1; service: 137; s_port: 53038;
Jan 1 22:55:10 drop   %LOGSOURCE% >eth1 rule: 7; rule_uid: {C1336766-9489-4049-9817-50584D83A245}; src: 71.111.186.26; dst: %DSTIP%; proto: tcp; product: VPN-1 & FireWall-1; service: 445; s_port: 38548;

我尝试在Kibana上可视化 geoip，但显示未找到结果。我的 geoip配置有问题吗？我下载了数据库，对此我没有任何问题。但是，似乎 geoip无法读取我的存储IP地址的 src吗？另外，我扩展了字段表。我没有看到一些新的 geoip字段，其中包含有关被映射到实际地理位置的 src IP地址的信息。

需要一些帮助男孩

Answer 1

您唯一的问题是kv过滤器由于空格而无法正确拆分字段。

现在，当logstash解析您的日志时，您将收到如下事件:

{
      "@version" => "1",
    "@timestamp" => "2015-01-01T22:15:13.000Z",
          "host" => "iMac-de-Consulthys.local",
          "path" => "/home/kibana/Documents/external_noise.log",
          "type" => "external_noise",
     "timestamp" => "Jan 1 23:15:13",
        "action" => "drop",
     "logsource" => "%LOGSOURCE%",
     "interface" => ">eth1",
          " rule" => " 7",
      " rule_uid" => " {C1336766-9489-4049-9817-50584D83A245}",
           " src" => " 218.8.245.123",
           " dst" => " %DSTIP%",
         " proto" => " tcp",
       " product" => " VPN-1&FireWall-1",
       " service" => " 2967",
        " s_port" => " 6000",
}

您会注意到 kv过滤器提取的所有字段的开头都有一个空格。这意味着 geoip过滤器找不到 src字段。

因此，您要做的就是修改 kv过滤器以修剪键和值，如下所示:

kv   {
    source => "kvpairs"
    field_split => ";"
    value_split => ":"
    trim => "\s"                 <--- add this line
    trimkey => "\s"              <--- add this line
    remove_field => "kvpairs"
}

然后，您将获得带有正确创建的 geoip字段的精彩事件，如下所示:

{
      "@version" => "1",
    "@timestamp" => "2015-01-01T22:15:13.000Z",
          "host" => "iMac-de-Consulthys.local",
          "path" => "/home/kibana/Documents/external_noise.log",
          "type" => "external_noise",
     "timestamp" => "Jan 1 23:15:13",
        "action" => "drop",
     "logsource" => "%LOGSOURCE%",
     "interface" => ">eth1",
          "rule" => "7",
      "rule_uid" => "{C1336766-9489-4049-9817-50584D83A245}",
           "src" => "218.8.245.123",
           "dst" => "%DSTIP%",
         "proto" => "tcp",
       "product" => "VPN-1&FireWall-1",
       "service" => "2967",
        "s_port" => "6000",
         "geoip" => {
                      "ip" => "218.8.245.123",
           "country_code2" => "CN",
           "country_code3" => "CHN",
            "country_name" => "China",
          "continent_code" => "AS",
             "region_name" => "08",
               "city_name" => "Harbin",
                "latitude" => 45.75,
               "longitude" => 126.64999999999998,
                "timezone" => "Asia/Harbin",
        "real_region_name" => "Heilongjiang",
                "location" => [
            [0] 126.64999999999998,
            [1] 45.75
        ],
             "coordinates" => [
            [0] 126.64999999999998,
            [1] 45.75
        ]
    }
}