I am trying to load the logs into the elasticsearch from Kafka below is my config file.
input {
kafka { bootstrap_servers => "localhost:9092"
topics => ["elk104"] }}
filter{
mutate {
add_field => { "tmp_message" => "%{message}" }
split => ["message", "|"]
add_field => { "cef_version" => "%{[message][0]}" }
add_field => { "cef_device_vendor" => "%{[message][1]}" }
add_field => { "cef_device_product" => "%{[message][2]}" }
add_field => { "cef_device_version" => "%{[message][3]}" }
add_field => { "cef_sig_id" => "%{[message][4]}" }
add_field => { "cef_sig_name" => "%{[message][5]}" }
add_field => { "cef_sig_severity" => "%{[message][6]}" }
} kv {
field_split => ","
trim_key => "<>\[\],"
trim_value => "<>\[\],"
include_keys => []
}mutate {
rename => [ "src", "cef_traffic_src_ip"]
replace => { "message" => "%{tmp_message}" }
remove_field => [ "tmp_message" ]} geoip {
source => "cef_traffic_src_ip"}mutate {
rename => {
"longitude" => "[location][lon]"
"latitude" => "[location][lat]"}
convert => {"[location][lat]" => "float"}
convert => {"[location][lon]" => "float"}}}output {elasticsearch {
hosts => ["localhost:9200"]
index => "pioneer01-logstash"
template_name=>"elk_template"
workers => 1
}}
and for the template, I am using below configuration,
PUT _template/elk_template
{
"index_patterns": ["pioneer-*"],
"settings": {
"index": {
"refresh_interval": "5s"
}
},
"mappings": {
"dynamic_templates": [
{
"message_field": {
"path_match": "message",
"match_mapping_type": "string",
"mapping": {
"type": "text",
"norms": false
}
}
},
{
"string_fields": {
"match": "*",
"match_mapping_type": "string",
"mapping": {
"type": "text",
"norms": false,
"fields": {
"keyword": {
"type": "keyword",
"ignore_above": 256
}
}
}
}
}
],
"properties": {
"@timestamp": {
"type": "date"
},
"@version": {
"type": "keyword"
},
"location": {
"type": "geo_point"
},
"tags": {
"type": "keyword"
},
"latitude" : {
"type" : "half_float"
},
"longitude" : {
"type" : "half_float"
},
"ip": {
"type": "ip"
}
}
},
"aliases": {}
}
Still, when I look at the output in elasticsearch I can get the GeoIP results,
"geoip" : {
"city_name" : "Bucharest",
"country_code3" : "RO",
"region_code" : "B",
"timezone" : "Europe/Bucharest",
"ip" : "93.114.45.13",
"latitude" : 44.4354,
"continent_code" : "EU",
"country_code2" : "RO",
"region_name" : "Bucuresti",
"postal_code" : "052822",
"country_name" : "Romania",
"longitude" : 26.1033,
"location" : {
"lat" : 44.4354,
"lon" : 26.1033
}
}
But, when I try to create the dashboard in Kibana, I don't get the mapping of the points on the map there. can somebody help me with this?