Merge values in JSON (jdbc events logstash)

Elastic Stack Logstash
1

I have a JSON structure like this:

 {
  "details": [
{
 "place": "abc",
 "group": 3,
  "year": 2006,
  "id": 1304,
  "street": "xyz 14",
  "lf_number": "0118",
 "code": 4433,
 "name": "abc coorperation",
 "group2": 3817"
 "group1": 32"
 "postal_code": "22926",
 "status": 2
 },
 {
 "place": "cbc",
  "group": 2,
  "year": 2007,
 "id": 4983,
 "street": "mnc 14",
 "lf_number": "0145",
 "code": 4433,
 "name": "abc coorperation", 
 "group2": 3817,
 "group1": 32"
 "postalcode": "22926",
 "status": 2
 }
 ],
 "@timestamp": "2017-09-04",
"parent": {
 "child": [
 {
 "w_2": 0.5,
 "w_1": 0.1,
 "id": 14226,
 "name": "air"
  },
 {
  "w_2": null,
 "w_1": 91,
  "id": 25002,
 "name": "Water"
 }]
 },
 "p_name": "anacin",
 "@version": "1",
 "id": 28841
 }

I want to edit the details. This is the restructured output from logstash JDBC plugin using aggregate filter to create parent-child relation. So, this JSON is outcome of JDBC filter and aggregate filter from Logstash. Now, I want to merge the information in details here. I want to construct new fields.

  Field 1)  coorperations: (details.name | details.postal_code details.street ; details.name | details.postal_code details.street)
 
 Output:
   Coorperations: (abc coorperation |22926 xyz 14; abc coorperation | 22926 mnc 14)


 Field 2) code: (details.code; details.code)
 
 Output:
 code: (4433,4433)


 Field 3) access_code: (details.status-details.id-details.group1-details.group2-details.group(always two digit)/details.year(only last two digits); details.status-details.id-details.group1-details.group2-details.group(always two digit)/details.year(only last two digits))

 Output: access_code (2-32-3817-03-06; 2-32-3817-02-07)

How can I achieve this for all the values in details. Without touching other parts of the JSON.

2

No hints from anyone. I am still looking for answer. Just a clue. @fbaligand any input from your side.

3

As your input is a array containing structured objects (details), I don't see a way to do it with standard filters (like mutate).

I invite you to use ruby filter.
And in code, you make a "for" loop on all details.
For each detail, you get each field you need and put it in a result string. Then you set the result string as a new global field using event.set()

4

Thanks for your feedback again. I have guessed already that mutate might not work. But the ruby code i have tried so far also is not working. I am very new to it. Like, I just tried to test if I could rename a field in all the members of "details". This is not working. What is the problem?

filter {
 if ([details]) {
ruby {
 code => "event.get('details').each {|k|
  k.rename('code') = k('secode')
   k.delete('code')
}"} } }
5

Two things :

Why don't you use the solution I provided to you here ?

6

yes, I am trying to do a different thing here to learn the ruby code syntax. As, I am trying to see if i loop-over correctly.

The solution you have mentioned earlier used aggregate filter syntax. Therefore, i am bit confused. It does not work with this case now.

I have used jdbc filter, jdbc streaming filter and then aggregate plugin. In the end, this is my output. Now, I want to aggregate again. Using the second aggregate filter does not help here. How can i access only array "details" here.

7

I guess "details" info is built using jdbc + aggregate filter, right ?
So in the previous topic, I gave you a solution to create a field named "all" that combines in one string, all informations details.
That is precisely what you need here.

The aggregate job is to do just after jdbc input, where each detail is a seperate event.
So in one aggregate filter, you can aggregate all details into one event, and also create this global aggregated string field with all details info.

8 
filter {
aggregate {
task_id => "%{s_id}"
code => "
map['parent'] ||= {}
map['s_id'] = event.get('s_id')
map['s_name'] = event.get('s_name')
map['details'] = event.get('details')
map['parent']['child'] ||= []
map['parent']['child'] << {'id' => event.get('id'),'name' => event.get('name'),'w_1' => event.get('w_1'),'w_2' => event.get('w_2')}
event.cancel()"
push_previous_map_as_event => true
timeout => 5
}
}

This is my aggregate filter definition at the moment. And the fields i want to combine are in details. How can i define here to operate only on "details"

9

OK, so if you add here my suggested code with sprintf method, you could do what you need.

10

It is not working this is the problem. I am not able to find out how can i access the values inside "details" properly.

This is what i tried so far:

filter {
 aggregate {
  task_id => "%{s_id}"
  code => "
map['parent'] ||= {}
map['id'] = event.get('id')
map['p_name'] = event.get('p_name')
map['details'] = event.get('details')
map['details']['cooperations'] ||= ''
map['details']['cooperations'] += ' ; ' unless map['details']['cooperations'].empty?
map['details']['cooperations'] += event.sprintf('%{name} | %{postal_code} %{street}')
event.set('cooperations', map['details']['cooperations'])
     map['parent']['child'] ||= []
     map['parent']['child'] << {'id' => event.get('id'),'name' => event.get('name'),'w_1' => event.get('w_1'),'w_2' => event.get('w_2')}
     event.cancel()

  "
push_previous_map_as_event => true
timeout => 5} }

Can you help in correcting the code here. This is what i am trying to figure out. How can i access only values inside the "details" and make new field inside it.

11

Could you provide event json content that arrives just before aggregate filter ?

12

{
"@timestamp" => 2017-09-19T06:51:37.659Z,
"details" => [
[0] {
"place" => "abc",
"group" => 2,
"year" => 2004,
"id" => 753,
"street" => "Kaiser Str. 30",
"lf_number" => "1511",
"code" => 200,
"name" => "abc coorperation",
"group2" => 3817,
"group1" => 8,
"postal_code" => "90763",
"status" => 2
},
[1] {
"place" => "abc",
"group" => 2,
"year" => 2005,
"id" => 2142,
"street" => "Kaiser Str. 30",
"lf_number" => "1628",
"code" => 200,
"name" => "abc cooperation",
"group2" => 3817,
"group1" => 32,
"postal_code" => "90763",
"status" => 2
}
],
"p_name" => "anacin",
"w_2" => nil,
"w_1" => 1.9,
"@version" => "1",
"s_id" => 24051,
"id" => 23225,
"name" => "water"
}
{
"@timestamp" => 2017-09-19T06:51:37.659Z,
"melder_details" => [
[0] {
"place" => "abc",
"group" => 2,
"year" => 2004,
"id" => 753,
"street" => "Kaiser Str. 30",
"lf_number" => "1511",
"code" => 200,
"name" => "abc coorperation",
"group2" => 3817,
"group1" => 8,
"postalcode" => "90763",
"status" => 2
},
[1] {
"place" => "abc",
"group" => 2,
"year" => 2005,
"id" => 2142,
"street" => "Kaiser Str. 30",
"lf_number" => "1628",
"code" => 200,
"name" => "abc cooperation",
"group2" => 3817,
"group1" => 32,
"postalcode" => "90763",
"status" => 2
}
],
"p_name" => "anacin",
"w_2" => nil,
"w_1" => 0.2,
"@version" => "1",
"s_id" => 24051,
"id" => 23319,
"name" => "air"
}

Here is how it looks like before aggregate filter. I aggregate on "s_id". So, this is parent (p_name, s_id) and child (w_1,w_2, id, name). The fields in details remain as such and i want to do "sprintf" on them.

13

I managed to print a new field to combine the postalcode from "details" using ruby filter. But, this does not work when i want to combine more fields.

  filter{
    ruby {
 code => "if(!event.get('[details][0]').nil?)
 event.set('[postalcode]',event.get('[details]').collect { |m| m['postalcode'] }.join(' '))
end"}
}
14

@fbaligand, can you suggest a solution?

15

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.