Dynamic field name with ruby

Elastic Stack Logstash
1

I have a curl response that looks like this.

{
"tables": [
{
"name": "PrimaryResult",
"columns": [
{
"name": "Category",
"type": "string"
},
{
"name": "count_",
"type": "long"
}
],
"rows": [
[
"Administrative",
111651142
],
[
"Recommendation",
5846
],
[
"Policy",
15664560
],
[
"Alert",
110135
],
[
"Security",
2101
],
[
"Autoscale",
27
]
]
}
]
}

I am able to extract individual events by using this configuration:

input {
  exec {
    command => "curl -X POST 'https://api.loganalytics.io/v1/workspaces/DEMO_WORKSPACE/query' -d '{\"query\": \"AzureActivity | summarize count() by Category\"}' -H 'x-api-key: DEMO_KEY' -H 'Content-Type: application/json'"
    interval => 60
    type => "json"
  }
}

filter{
  json { source => "message" }
  split { field => "[tables][0][rows]" }
  mutate {
    add_field => {
      "%{[tables][0][columns][0][name]}" => "%{[tables][0][rows][0]}"
      "%{[tables][0][columns][1][name]}" => "%{[tables][0][rows][1]}"
    }
    remove_field => [ "tables","message","command" ]
  }
}

output {
  stdout { codec => "rubydebug" }
}

How do I modify this so that it will add fields dynamically? I saw that this can be done through ruby filter but I'm not familiar with ruby.

2

What do you want the output to look like?

Also, why do you do the split?

3

Expected results would be something like this,

{
"count_" => "5846",
"type" => "json",
"Category" => "Recommendation",
"@version" => "1",
"@timestamp" => 2019-02-25T20:35:03.667Z,
"host" => "elk-stack-logstash"
}
{
"count_" => "110135",
"type" => "json",
"Category" => "Alert",
"@version" => "1",
"@timestamp" => 2019-02-25T20:35:03.667Z,
"host" => "elk-stack-logstash"
}
{
"count_" => "2099",
"type" => "json",
"Category" => "Security",
"@version" => "1",
"@timestamp" => 2019-02-25T20:35:03.667Z,
"host" => "elk-stack-logstash"
}
{
"count_" => "27",
"type" => "json",
"Category" => "Autoscale",
"@version" => "1",
"@timestamp" => 2019-02-25T20:35:03.667Z,
"host" => "elk-stack-logstash"
}

Each row is one event and the value of it is matched to its column name.

I used split to split each row as a separate event.

4

That's what your existing filter produces. What is the point of implementing in Ruby?

5

Because there's a possibility that there will be more columns that what I define. So I wanted to use ruby to add fields dynamically.

6

So you want to include [2], [3] etc. if they exist?

7

Yes. So I wont need to manually define in the case that there would be more columns.

8

Try

    split { field => "[tables][0][rows]" }
    ruby {
        code => '
            rows = event.get("[tables][0][rows]")
            cols = event.get("[tables][0][columns]")
            rows.each_index { |i|
                v = rows[i]
                k = cols[i]["name"]
                event.set(k,v)
            }
        '
    }

Error handling is left as an exercise for the reader.

9

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