How to remove dynamic mongodb oid's in logstash?

Question

Mongodb has 2 types of $oid reference -

Type 1 -

//MongoDB
city_id : "5fe3206428bf745876649fd3"

//Kafka Message
city_id : {
    "$oid": "5fe3206428bf745876649fd3"
}

Type 2 -

//MongoDB
city_ids : ["5fe3206428bf745876649fd3","5fe3206428bf745876649fd3","5fe3206428bf745876649fd3"]

//Kafka Message
city_ids : [
  {
    "$oid": "5fe3206428bf745876649fd3"
  },
  {
    "$oid": "5fe3206428bf745876649fd3"
  },
  {
    "$oid": "5fe3206428bf745876649fd3"
  }
]

How can I handle these two types in logstash, such that I get the exact data structure for elasticsearch as it is saved in MongoDB.

input {
  kafka {
        bootstrap_servers => "localhost:9092"
        decorate_events => true
        topics => ["users","organisations","cities"]
  }
}
filter { 
    json {
        source => "message"
        target => "json_payload"
    }

    json {
        source => "[json_payload][payload]"
        target => "payload"
    }
    
    mutate {
        rename => { "[payload]" => "document"}
        remove_field => ["message","json_payload","payload"]
        add_field => {
          "[es_index]" => "%{[@metadata][kafka][topic]}" 
          "[mongo_id]" => "%{[document][_id][$oid]}"
        }
    }
}
output {
    elasticsearch {
        hosts => ["localhost:9200"]
        index => "%{es_index}"
        document_id => "%{mongo_id}"
    }
    stdout {
      codec =>
        rubydebug {
            metadata => true
        }
    }
}

This is a follow through from a previous question.

user_11792977 · Answer

The following will do it dynamically for every field that contains $oid entries. It makes a lot of assumptions about the structure -- if it contains $oid entries then that is all that is retained.

    ruby {
        code => '
            event.to_hash.each { |k, v|
                if v.is_a? Hash
                    if v["$oid"]
                        event.set(k, v["$oid"])
                    end
                end
                if v.is_a? Array
                    if v[0]["$oid"]
                        a = []
                        v.each { |x| a << x["$oid"] }
                        event.set(k, a)
                    end
                end
            }
        '
    }

How to remove dynamic mongodb oid's in logstash?

1 Answers