Parsing JSON from CouchDB to ElasticSearch via Logstash

Question

I am not able to parse JSON from CouchDB to Elasticsearch index in the desired way. My CouchDB data looks like this:

{
  "_id": "56161609157031561692637",
  "_rev": "4-4119e8df293a6354be4c9fd7e8b12e68",
  "deleteFlag": "N",
  "entryUser": "John",
  "parameter": "{"id":"14188","rcs_p":null,"rcs_e":null,"dep_p":null,"dep_e":null,"dep_place":null,"rcf_p":null,"rcf_e":null,"rcf_place":null,"dlv_p":"3810","dlv_e":"1569","seg_no":null,"trans_type":"incoming","trans_service":"delivery"}",
  "physicalId": "0",
  "recordDate": "2020-12-28T17:50:16+05:45",
  "tag": "CARGO",
  "uId": "56161609157031561692637",
  "~version": "CgMBKgA="
}

What I am trying to do is be able to search using the nested field of the parameter of the above JSON. When I put the data in ES index it is stored like this:

{
  "_index": "del3",
  "_type": "_doc",
  "_id": "XRCV9XYBx5PRwauO--qO",
  "_version": 1,
  "_score": 0,
  "_source": {
    "@version": "1",
    "doc_as_upsert": true,
    "doc": {
      "physicalId": "0",
      "recordDate": "2020-12-27T12:56:45+05:45",
      "tag": "CARGO",
      "~version": "CgMBGgA=",
      "uId": "48541609052212485430933",
      "_rev": "3-937bf92e6010afec13664b1d9d06844b",
      "deleteFlag": "N",
      "entryUser": "John",
      "parameter": "{"id":"4038","rcs_p":null,"rcs_e":null,"dep_p":null,"dep_e":null,"dep_place":null,"rcf_p":null,"rcf_e":null,"rcf_place":null,"dlv_p":"5070","dlv_e":"2015","seg_no":null,"trans_type":"incoming","trans_service":"delivery"}"
    },
    "@timestamp": "2021-01-12T07:53:33.978Z"
  },
  "fields": {
    "@timestamp": [
      "2021-01-12T07:53:33.978Z"
    ],
    "doc.recordDate": [
      "2020-12-27T07:11:45.000Z"
    ]
  }
}

I want to be able to access the fields inside the parameter (id, rcs_p, rcs_e, ..) in Elasticsearch.

Here is my logstash.conf file:

input {
    couchdb_changes {
        host => "<host_name>"
        port => 5984
        db => "mychannel_asset$management"
        keep_id => false
        keep_revision => true
        #initial_sequence => 0
        always_reconnect => true
        sequence_path => "/usr/share/logstash/config/seqfile"
    }
}

filter {
        json {
                source => "[parameter]"
                remove_field => ["[parameter]"]
        }
}

output {
    if([doc][tag] == "CARGO") {
        elasticsearch {
            hosts => ["http://elasticsearch:9200"]
            index => "del3"
            user => elastic
            password => changeme
        }
    }
}

How do I achieve my desired result? I also tried to do by creating a custom template by defining a nested type for parameter but no luck yet. Any help would be appreciated.

Answer 1

I think you did almost everything right. I'm not too sure about the actual structure, but one of these might work:

filter {
    json {
        source => "parameter"
        target => "parameter"
    }
}

filter {
    json {
        source => "[doc][parameter]"
        target => "[doc][parameter]"
    }
}

I don't know how CouchDB source input plugins works but it seems to be putting everything under doc object.