Filter mutate replace a field for specific value

Elastic Stack Logstash
1

Hello,

I use logstash 7.17.

I want to create (or replace if existing) a field when I have a specific value. Here is a snippet of my code :

filter { 
    grok {
        match => {
            "[topic]" => "%{GREEDYDATA}_%{GREEDYDATA:topicCountry}-%{GREEDYDATA:topicCurrency}" 
        }
    }
    if ([relatedParty][role] == "parent") {
        mutate {
            replace => {
                "[test][parentId]" => "%{topicCountry}_%{topicCurrency}_%{[relatedParty][id]}"
            }
        }
    } else if ([relatedParty][role] == "child") {
        mutate {
            replace => {
                "[test][childId]" => "%{topicCountry}_%{topicCurrency}_%{[relatedParty][id]}"
            }
        }    
    }
[...]
}

And I have the following json in input

[{"_id": "1TEST6", "relatedParty": [{"referredType": "test", "role": "child", "id": "1234"}, {"referredType": "test", "role": "parent", "id": "5678"}], }]

But with in the output, I don't see the field [test][childId] or [test][parentId]. How should I create the fields ?

Any help would be appreciated, thank you in advance

2

[relatedParty] is an array. So [relatedParty][role] does not exist. If there at most two entries in the array then you could use

if ([relatedParty][0][role] == "parent") {
    mutate {
        replace => {
            "[test][parentId]" => "%{topicCountry}_%{topicCurrency}_%{[relatedParty][0][id]}"
        }
    }
} else if ([relatedParty][0][role] == "child") {
    mutate {
        replace => {
            "[test][childId]" => "%{topicCountry}_%{topicCurrency}_%{[relatedParty][0][id]}"
        }
    }
}
if ([relatedParty][1][role] == "parent") {
    mutate {
        replace => {
            "[test][parentId]" => "%{topicCountry}_%{topicCurrency}_%{[relatedParty][1][id]}"
        }
    }
} else if ([relatedParty][1][role] == "child") {
    mutate {
        replace => {
            "[test][childId]" => "%{topicCountry}_%{topicCurrency}_%{[relatedParty][1][id]}"
        }
    }
}

If there is a possibility of multiple children then you would need to use a ruby filter.

3

If I try to use ruby will something like this work ?

items_party = event.get("[relatedParty]")
            if items_party
                items_party.each do |item, value|
                if item["role"] == "child"
                    child_id = item["id"]
                    event.set("[test][childId]", "#{country}_#{topicCurrency}_#{child_id}")
                elsif item["role"] == "parent"
                    parent_id = item["id"]
                    event.set("[hierarchy][parentId]", "#{country}_#{topicCurrency}_#{parent_id}")
                end
            end
4

You would need to use event.get to access country / topicCurrency from the event but other than that it looks reasonable.

5

Thank to your response I was able to do the following code that worked for me :

filter {
  grok {
          match => {
              "[@metadata][kafka][topic]" => "%{GREEDYDATA:environment}_%{GREEDYDATA:topicCountry}-%{GREEDYDATA:topicCurrency}" 
          }
      }
      ruby {
          code => '
              topic_country = event.get("[topicCountry]")
              topic_currency = event.get("[topicCurrency]")
              items_party = event.get("[relatedParty]")
              if items_party
                  items_party.each do |item, value|
                      item_role = item["role"]
                      item_id = item["id"]
                      if item_role == "child"
                          event.set("[test][childId]", "#{topic_country}_#{topic_currency}_#{item_id}")
                      elsif item_role == "parent"
                          event.set("[test][parentId]", "#{topic_country}_#{topic_currency}_#{item_id}")
                      end
                  end
              end
          '
      }
}