Unable to add new field with values of nested fields

Elastic Stack Logstash
1

I am attempting to create a geo-point data type so that I can utilize the map features in kibana. My index will not show as an option when I attempt to add a layer and I believe its because I do not have a geo-point datatype in my index. So I am attempting to create a new field that combines the longitude and latitude values from two other fields. See configs below:
Logstash:

filter {
if [log_type] == "suricata" {
json {
source => "message"
}
date {
match => [ "timestamp", "ISO8601" ]
}
if ![geoip] and [src_ip] !~ /^(10\.|192\.168\.)/ {
geoip {
add_tag => [ "GeoIP" ]
source => "src_ip"
}
}
if [geoip.latitude][geoip.longitude] {
mutate {
add_field => ["geoipgeopoint","%{[geoip.latitude][geoip.longitude]}"]
}
}
}
}

The fields geoip.latitude and geoip.longitude are added to events once the second if condition is matched succesfully. I've tried adding the add_field to the geoip filter but the filter isn't utilized for all events.
I am not seeing any error message in the logs but here is a snippet of the log file that shows a few lines before and after the new field I am attempting to create:

[2020-05-19T15:42:57,071][DEBUG][logstash.plugins.registry] On demand adding plugin to the registry {:name=>"geoip", :type=>"filter", :class=>LogStash::Filters::GeoIP}
[2020-05-19T15:42:57,082][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@add_tag = ["GeoIP"]
[2020-05-19T15:42:57,084][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@source = "src_ip"
[2020-05-19T15:42:57,086][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@id = "2307da31962e7901d09e6ecd33099645fbf9daaea2db320e76e18e09543da8e0"
[2020-05-19T15:42:57,087][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@enable_metric = true
[2020-05-19T15:42:57,088][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@remove_tag = []
[2020-05-19T15:42:57,090][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@add_field = {}
[2020-05-19T15:42:57,091][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@remove_field = []
[2020-05-19T15:42:57,092][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@periodic_flush = false
[2020-05-19T15:42:57,094][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@default_database_type = "City"
[2020-05-19T15:42:57,095][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@target = "geoip"
[2020-05-19T15:42:57,096][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@cache_size = 1000
[2020-05-19T15:42:57,098][DEBUG][logstash.filters.geoip   ] config LogStash::Filters::GeoIP/@tag_on_failure = ["_geoip_lookup_failure"]
[2020-05-19T15:42:57,109][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@add_field = {"geoipgeopoint"=>"%{[geoip.latitude][geoip.longitude]}"}
[2020-05-19T15:42:57,110][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@id = "46cb237597e7ad7c2133dc354e37e014806ce1baaba913e5ffddc82532c99288"
[2020-05-19T15:42:57,112][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@enable_metric = true
[2020-05-19T15:42:57,114][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@add_tag = []
[2020-05-19T15:42:57,115][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@remove_tag = []
[2020-05-19T15:42:57,117][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@remove_field = []
[2020-05-19T15:42:57,118][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@periodic_flush = false
[2020-05-19T15:42:57,120][DEBUG][logstash.filters.mutate  ] config LogStash::Filters::Mutate/@tag_on_failure = "_mutate_error"
[2020-05-19T15:43:04,505][DEBUG][org.logstash.config.ir.CompiledPipeline][main] Compiled filter
P[filter-mutate{"add_field"=>["geoipgeopoint", "%{[geoip.latitude][geoip.longitude]}"]}|[file]/etc/logstash/conf.d/04_suricata.conf:19:4:
mutate {
add_field => ["geoipgeopoint","%{[geoip.latitude][geoip.longitude]}"]
}
```] into org.logstash.config.ir.compiler.ComputeStepSyntaxElement@bc28d924

Template for index:

{
"suricata" : {
"order" : 0,
"index_patterns" : [
"my-suricata-*-*-*"
],
"settings" : {
"index" : {
"number_of_shards" : "3",
"number_of_replicas" : "0"
}
},
"mappings" : {
"properties" : {
"geoipgeopoint" : {
"type" : "geo_point"
}
}
},
"aliases" : { }
}
}

Data is still being parsed but I am not seeing new field in Kibana.

2

I tried the following config and can now see the new field but it doesn't show the value of the two fields it just shows those fields names:

Capture
Capture1531×148 28 KB 

filter {
        if [log_type] == "suricata" {
                json {
                        source => "message"
                }
                date {
                        match => [ "timestamp", "ISO8601" ]
                }
                if ![geoip] and [src_ip] !~ /^(10\.|192\.168\.)/ {
                        geoip {
                                add_tag => [ "GeoIP" ]
                                source => "src_ip"
                                add_field => {
                                        "geoip.coordinates" => "%{[geoip.latitude][geoip.longitude]}"
                                }
                        }
                }
#               if [geoip.latitude][geoip.longitude] {
#                       mutate {
#                               add_field => ["geoipgeopoint","%{[geoip.longitude]}"]
#                       }
#               }
        }
}

Here is the mapping for this index(only showing the portion that relates to geoip):

{
  "suricata" : {
    "order" : 0,
    "index_patterns" : [
      "my-suricata-*-*-*"
    ],
    "settings" : {
      "index" : {
        "number_of_shards" : "3",
        "number_of_replicas" : "0"
      }
    },
     "mappings": {
       "properties":{
         "geoip" : {
           "properties" : {
             "timezone" : {
               "type" : "text",
               "fields" : {
                 "keyword" : {
                   "ignore_above" : 256,
                   "type" : "keyword"
                 }
              }
            },
            "ip" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "latitude" : {
              "type" : "float"
            },
            "coordinates" : {
              "type" : "geo_point"
            },
            "continent_code" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "city_name" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "country_code2" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "country_name" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "dma_code" : {
              "type" : "long"
            },
            "country_code3" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "location" : {
              "properties" : {
                "lon" : {
                  "type" : "float"
                },
                "lat" : {
                  "type" : "float"
                }
              }
            },
            "region_name" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "postal_code" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "Location" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            },
            "longitude" : {
              "type" : "float"
            },
            "region_code" : {
              "type" : "text",
              "fields" : {
                "keyword" : {
                  "ignore_above" : 256,
                  "type" : "keyword"
                }
              }
            }
          }
        },
3

In logstash, if [geoip] is an object that contains a latitude field then you refer to it as

[geoip][latitude]

not as [geoip.latitude], which would be a field with a period in its name.

4

Hi Badger,
Thanks for helping me out with this....I tried making the changes to align with the suggestions you mentioned but I am still not getting any values populated to the new field. My updated configs below:

Capture1
Capture11316×304 50.1 KB

Logstash filter:

filter {
        if [log_type] == "suricata" {
                json {
                        source => "message"
                }
                date {
                        match => [ "timestamp", "ISO8601" ]
                }
                if ![geoip] and [src_ip] !~ /^(10\.|192\.168\.)/ {
                        geoip {
                                add_tag => [ "GeoIP" ]
                                source => "src_ip"
                                add_field => {
                                        "geoip.coordinates" => "%{[latitude][longitude]}"
                                }
                        }
                }
#               if [geoip.latitude][geoip.longitude] {
#                       mutate {
#                               add_field => ["geoipgeopoint","%{[geoip.longitude]}"]
#                       }
#               }
        }
}

I also tried
"geoip.coordinates" => "%{[geoip][latitude][geoip][longitude]}"
that's the reason the screenshot shows two separate outcomes for the geoip.coordinates field.

5

geoip filter will store the lat in [geoip][latitude] and store the lon in [geoip][longitude] respectively.

6

Hi @ptamba,

Thank you for your reply...I understand what you are saying but I want to combine those two field(longitude and latitude) values into one new field so that can utilize the geo-point data type. What would be the correct syntax for that?

@Badger

7

in elasticsearch , a geo_point is expressed as [lot,lan] in arrays, and “lat,lon” in string based on this docs

so in string, your coordinates will be

add_field => { 
   “[geoip][coordinates]” => “%{[geoip][latitude]},%{[geoip][longitude]}” 
}

actually your mapping here

already contains the coordinates and should already been populated automatically by logstash.

8

@ptamba - thank you for your help, I implemented your syntax and its now working as intended.
Here is my latest config just in case anybody else is having the same hiccup:

filter {
        if [log_type] == "suricata" {
                json {
                        source => "message"
                }
                date {
                        match => [ "timestamp", "ISO8601" ]
                }
                if ![geoip] and [src_ip] !~ /^(10\.|192\.168\.)/ {
                        geoip {
                                add_tag => [ "GeoIP" ]
                                source => "src_ip"
                                add_field => {
                                        "[geoip][coordinates]" => "%{[geoip][latitude]},%{[geoip][longitude]}"
                                }
                        }
                }
        }
}
9

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.