I am currently using a node pipeline that looks like this:
{ "my-pipeline": { "description": "pipeline for my filebeat", "processors": [ { "json": { "field": "message", "add_to_root": true, "on_failure": [ { "rename": { "field": "message", "target_field": "originalMessage", "ignore_missing": true } }, { "set": { "field": "indexName", "value": "pipeline-errors" } }, { "set": { "field": "indexType", "value": "pipeline-error" } }, { "rename": { "field": "@timestamp", "target_field": "errorTimestamp", "ignore_missing": true } } ] } }, { "remove": { "field": "@timestamp", "ignore_failure": true } }, { "remove": { "field": "message", "ignore_failure": true } }, { "script": { "inline": "ctx._index = ctx.indexName; ctx._type=ctx.indexType; if (ctx.docVersion != null) {ctx._version = ctx.docVersion; ctx._version_type='external'}" } }, { "remove": { "field": "indexName", "ignore_failure": true } }, { "remove": { "field": "indexType", "ignore_failure": true } } ] } }
This pipeline is used simply to delete the log sent by the file bit. In the script processor, I look for the fields "indexName" and "indexType" and assign them to "_index" and "_type" respectively. Since I need to consider the version, the "version" field is included in the log (but this is not necessary, as some logs do not contain a version).
Using triggers of this pipeline:
org.elasticsearch.index.mapper.MapperParsingException: Cannot generate dynamic mappings of type [_version] for [_version] at org.elasticsearch.index.mapper.DocumentParser.createBuilderFromFieldType(DocumentParser.java:656) ~[elasticsearch-5.5.0.jar:5.5.0] at org.elasticsearch.index.mapper.DocumentParser.parseDynamicValue(DocumentParser.java:805) ~[elasticsearch-5.5.0.jar:5.5.0]
What I have tried so far ( updated 09-16 ):
- Replaced the field name with something like "docVersion", so that it doesn’t collide if its a keyword. This does not work either
- Tried to use ctx._source.version, this will raise a ScriptException [runtime error]; in the end, note that the _index and _type values come from ctx.indexName and ctx.indexType respectively
- I tried adding 'version_type = external' to the script, but I still get a MapperParsingException as above;
- Tried to use 'version_type = external_gte', but I also had the MapperParsingException property.
How to specify / use external versioning in elasticsearch documents when using node pipelines? if this is not possible compared to the script processor for pipelines, what are the options for using the external version when working with filebeat-to-elasticsearch so that the outdated version of the document is rejected?
Update 10-24-2017 It seems that this is a function that does not exist with the current version of elasticsearch (5.6 in my case). According to the check in the code , the IndexRequest in the pipeline runtime service does not contain a link to the document version or version type, thus the internal version. Perhaps this could be added as a function in future versions of elasticsearch.