Split a JSON array into individual documents in Logstash

Elastic Stack Logstash
1

I'm sorry to ask this honestly.. normally I'm a decently smart fella.. but I can't seem to grasp this. I have some JSON coming that I need to split into individual documents.


{
  "lan": {
    "inkbps": 756.2703471377937,
    "outkbps": 16469.254210648105,
    "interface": "em1"
  },
  "wan": {
    "inkbps": 12432.741640620181,
    "outkbps": 613.0682382407641,
    "interface": "pppoe0",
    "friendlyiface": "wan",
    "name": "internet_pppoe",
    "status": "okay",
    "monitorip": "8.8.8.8",
    "sourceip": "17.1.1.1",
    "delay": "53.561ms",
    "loss": "1%"
  },
  "opt2": {
    "inkbps": 33.28968847276036,
    "outkbps": 26.285475058869885,
    "interface": "ovpnc2",
    "friendlyiface": "opt2",
    "name": "pia_vpn_vpnv4",
    "status": "okay",
    "monitorip": "10.17.1.1",
    "sourceip": "10.17.1.2",
    "delay": "",
    "loss": ""
  }
}

It seems pretty simple based on all the examples I've found on this site and others.. and here's what I've ended up with..


filter
    {
            if [log_type] == "edge_utilization"
            {
                    json
                    {
                            source => "[messages]"
                    }
                    split
                    {
                            field => "[messages]"
                    }
            }
    }

Which then just throws _grokparsefailure, _split_type_failure

I've tried a bunch of different iterations, based on suggestions and articles I've found.. but even if I get the tags to go away, it still never splits the nodes into individual documents.

2

I can't say anything about the grok failure as you didn't include the configuration that has caused that. But your split problem is probably due to the fact that this is a hash, not an array. Using a ruby filter to change it might help?
event.set("messages", event.get("messages").values)

If it doesn't work, please post the ruby debug output, so we'll know exactly what your current result looks like. And have a look at the Logstash logs. Wasn't the _split_type_failure accompanied by a warning?

3

Hey.. thanks so much for the response. I'm not quite sure I understand.. where do I put that? If it helps, here is my entire pipeline:


input
{
	udp
	{
		port => 514
		codec => plain {charset => "ISO-8859-1"}
	}

	tcp 
	{
		port => 5040
	}

}

filter
{
	mutate
	{
		add_field => {"[raw_log]" => "%{[message]}"}
	}
}


filter
{
	if [log_type] == "edge_utilization"
	{
		json
		{
			source => "[message]"
		}
		split
		{
			field => "[message]"
		}
	}
}

output
{
	elasticsearch
	{
		hosts => ["http://localhost:9200"]
		index => "syslog-%{+YYYY.MM.dd}"
		document_id => "%{[fingerprint]}"	
	}
}
4 
ruby {
  code => 'event.set("message", event.get("message").values)'
}

before your split filter.

5

I did the following:


filter
{
        if [log_type] == "edge_utilization"
        {
                json
                {
                        source => "[message]"
                }

                ruby 
                {
                       code => 'event.set("message", event.get("message").values)'
                }

                split
                {
                        field => "[message]"
                }
        }
}

Which then results in a _rubyexception

Did I misunderstand where I needed to insert that line?

6

As I said, the logs and rubydebug output would be of interest. You can barely debug something based on tags. The data might look different from than I think.

7

Heh.. well, thanks anyway

8

I don't know what to make of that answer . You don't want to provide more information?