yeah, it works, but new problem occurs.
I add metadata in the filter:
filter {
mutate {
add_field => {"@fields" => "%{fields}"}
rename => { "[operation]" => "[@metadata][operation]" }
rename => { "[id]" => "[@metadata][id]" }
}
json {
source => "@fields"
remove_field => ["@fields","@version","@timestamp","fields"]
}
}
and i use the id metadata to update a doc when the operation is set to update:
if [@metadata][operation] == "update" {
elasticsearch {
hosts => ["xx.xx.xx.xx:9200"]
index => "im"
action => "update"
doc_as_upsert => true
timeout => 300
document_id => "%{[@metadata][id]}"
#user => "elastic"
#password => "changeme"
}
}
the result is create a new doc other than update the exists one:
