Hello. I’m attempting to leverage these Cloudflare Elastic integration config files and I’m running into an issue that is preventing the logs from being imported. Here is a detailed account of my issue:
I’m attempting to follow the instructions here:
I’m not using the Elastic Cloud, but my own installation. I’ve got both ElasticSearch and Kibana set up and talking to each other, the Cloudflare ingest pipelines imported, the index template imported, the AWS lambda function deployed, and logs flowing into S3. When the Lamdba function is trying to send the logs into Elastic is where I’m encountering the issue.
I do see a ton of these errors in my “cluster.log” file on my Elastic EC2 server every time it seems to be trying to index logs:
[2020-05-20T12:06:31,803][INFO ][o.e.a.b.TransportShardBulkAction] [logs-node-1] [Cloudflare-2020-05-18][0] mapping update rejected by primary
java.lang.IllegalArgumentException: mapper [source.geo.location] of different type, current_type [geo_point], merged_type [ObjectMapper]
Here is what I’m seeing in the AWS Cloudwatch logs from the Lamdba function:
So, this seems to be having an issue with the “geo-point” data type.
Looking in the Cloudflare-index-template.json file, I do see this:
And I can see this mapping in the “weekly” ingest pipeline:
Doing a quick bit of research, “geoip” seems to be available in Logstash:
I did not install Logstash since I didn’t think it was needed for this implementation. Looks like “geoip” might be used to derive all of the other properties (timezone, city, etc) all from the “ClientIP” field in the logs. However, I do see that this is also available as a default Elastic ingest processor:
When I run a quick API call against Elastic to look for available plugins, I do see geoip referenced:
GET “###.##.###.##:9243/_nodes/plugins” | python -m json.tool | grep geo
{
"classname": "org.elasticsearch.ingest.geoip.IngestGeoIpPlugin",
"description": "Ingest processor that uses looksup geo data based on ip adresses using the Maxmind geo database",
"elasticsearch_version": "7.7.0",
"extended_plugins": [],
"has_native_controller": false,
"java_version": "1.8",
"name": "ingest-geoip",
"version": "7.7.0"
}
So, it does seem that I have this installed, as far as I can tell. I did a real quick test to make sure geoip is working properly. I ran a couple of API commands to create a small pipeline with just the “geoip”:
curl --user <user>:<password> -X PUT "###.##.###.##:9243/_ingest/pipeline/testgeoip" -H "Content-Type: application/json" -d '{"description" : "Add geoip info","processors" : [{"geoip" : {"field" : "ip"}}]}'
I then created a small index using that pipeline with just a random IP:
curl --user <user>:<password> -X PUT "###.##.###.##:9243/my_index/_doc/my_id?pipeline=testgeoip" -H "Content-Type: application/json" -d '{"ip":"8.8.8.8"}'
I then fetched the contents of the index:
curl --user <user>:<password> -X GET "###.##.###.##:9243/my_index/_doc/my_id" | python -m json.tool
{
"_id": "my_id",
"_index": "my_index",
"_primary_term": 1,
"_seq_no": 0,
"_source": {
"geoip": {
"continent_name": "North America",
"country_iso_code": "US",
"location": {
"lat": 37.751,
"lon": -97.822
}
},
"ip": "8.8.8.8"
},
"_type": "_doc",
"_version": 1,
"found": true
}
So, it seems that geoip is working. However, the error message is targeting the “location” field specifically. Here, it looks to be an object (lat and lon values). I have also performed a more accurate test:
Create pipeline (pulled from the Cloudflare file):
PUT /_ingest/pipeline/jmggeoip
{
"description": "Jason Log Pipeline",
"processors": [
{
"geoip": {
"field": "ClientIP",
"target_field": "source.geo",
"properties": [
"ip",
"country_name",
"continent_name",
"region_iso_code",
"region_name",
"city_name",
"timezone",
"location"
]
}
}
]
}
Create index template mapping (pulled from the Cloudflare file):
PUT /_template/jmgtemplate
{
"index_patterns": [
"jmgindex-*"
],
"mappings": {
"properties": {
"source.geo": {
"properties": {
"ip": {
"type": "ip"
},
"postal_code": {
"type": "keyword"
},
"location": {
"type": "geo_point"
},
"dma_code": {
"type": "long"
},
"country_code3": {
"type": "keyword"
},
"latitude": {
"type": "float"
},
"longitude": {
"type": "float"
},
"region_name": {
"type": "keyword"
},
"city_name": {
"type": "keyword"
},
"timezone": {
"type": "keyword"
},
"country_code2": {
"type": "keyword"
},
"continent_code": {
"type": "keyword"
},
"country_name": {
"type": "keyword"
},
"region_code": {
"type": "keyword"
},
"continent_name": {
"type": "keyword"
},
"region_iso_code": {
"type": "keyword"
}
}
}
}
},
"settings": {
"index": {
"number_of_shards": "1",
"number_of_replicas": "1",
"mapping.ignore_malformed": true
}
}
}
Create index (index pattern matching above and pipline created above):
PUT /jmgindex-test/_doc/my_id?pipeline=jmggeoip
{“ClientIP”:“8.8.8.8”}
Fetch the index:
GET /jmgindex-test/_doc/my_id
This call returns the following information:
{
"_index" : "jmgindex-test",
"_type" : "_doc",
"_id" : "my_id",
"_version" : 1,
"_seq_no" : 0,
"_primary_term" : 1,
"found" : true,
"_source" : {
"source" : {
"geo" : {
"continent_name" : "North America",
"timezone" : "America/Chicago",
"ip" : "8.8.8.8",
"country_name" : "United States",
"location" : {
"lon" : -97.822,
"lat" : 37.751
}
}
},
"ClientIP" : "8.8.8.8"
}
}
So, as you can see, we are still getting latitude and longitude back. Now, let’s look at the field mapping
Now, we are properly mapping to “geo_point”. However, while this example seems to be working, the ingest process I set up for Cloudflare is not working. So, there must be something about the setup process that is missing. Just to reiterate, here is the error I’m getting when the Lamdba function tries to insert a log into ElasticSearch:
[2020-05-20T12:06:31,803][INFO ][o.e.a.b.TransportShardBulkAction] [logs-node-1] [Cloudflare-2020-05-18][0] mapping update rejected by primary
java.lang.IllegalArgumentException: mapper [source.geo.location] of different type, current_type [geo_point], merged_type [ObjectMapper]
This is why I’m hitting a wall. Everything “seems” to be setup properly from the Elastic side and I think the above proves the geo_point mapping and geoip functionality is working fine. I’m leaning towards something with the Lambda function at this point. Has anyone else run into this issue or have any input? I would appreciate some assistance in narrowing down this issue. Thanks in advance!