如何删除logstash中的动态mongodb oid？

Question

Mongodb有2种$oid引用-

类型 1 -

//MongoDB
city_id : "5fe3206428bf745876649fd3"

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

类型 2 -

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

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

我如何在logstash中处理这两种类型，以便我得到elasticsearch的确切数据结构，因为它保存在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
        }
    }
}

这是对之前question 的跟进。

Answer 1

以下将对包含 $oid 条目的每个字段动态执行此操作。它对结构做了很多假设——如果它包含 $oid 条目，那么这就是保留的全部内容。

    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
            }
        '
    }