Separate array into individual events

Draco3

I have a JSON file that I am ingesting into CRIBL. I have created a pipeline with the following functions “JSON Unroll” which is then followed by the “Parser” function with the following settings (Operation mode = extract; Type=JSON Object, Source Field = _raw).

My single event now has a new field created called “records:” with what appears to be multiple arrays for each record (there are in excess of 250) with this

records:
{}0:
a account_id: xxxxxxxxxx
a account_name: xxxxxxxx
a owner: xxxxxxxxx
a cost: xxxxxxxxxx
a group: xxxxxxxxxxxx
a reporter: xxxxxxxxxxxxx
{}1:
a account_id: yyyyyyyyyyy
a account_name: yyyyyyyyyyy
a owner: yyyyyyyyyyy
a cost: yyyyyyyyyyy
a group: yyyyyyyyyyy
a reporter: yyyyyyyyyyy
{}2:
a account_id: zzzzzzzzzzzzz
a account_name: zzzzzzzzzzzzz
a owner: zzzzzzzzzzzzz
a cost: zzzzzzzzzzzzz
a group: zzzzzzzzzzzzz
a reporter: zzzzzzzzzzzzz

etc…

I am running CRIBL version 3.5.1. I am trying to separate each of these records into an individual event. So in my example above {}0 with its 6 fields would be 1 event, {}1 with its 6 fields would be the 2nd event etc.

I have tried using various functions but can’t seem to get this to work. Would like to understand what the best way would be to do this.

Draco3

Thanks to CRIBL support, it was identified that the function unroll was required and this did the job.

Draco3

Thanks to CRIBL support, it was identified that the function unroll was required and this did the job.

Ahmed Kira

You can also do this with an Event Breaker rule and associate it to the source. This is common with many events from Azure, AWS Cloudtrail and AWS Kinesis sources. In the event breaker rule, select the event type of json_array. And in this case, the JSON array field would be records. Reference the out-of-box AWS Ruleset for an example.

Here's also an event breaker you can import:

{
  "minRawLength": 256,
  "lib": "custom",
  "id": "Azure NSG Flow Logs",
  "rules": [
    {
      "condition": "true",
      "type": "json_array",
      "timestampAnchorRegex": "/^/",
      "timestamp": {
        "type": "auto",
        "length": 150
      },
      "timestampTimezone": "local",
      "timestampEarliest": "-420weeks",
      "timestampLatest": "+1week",
      "maxEventBytes": 5120000,
      "disabled": false,
      "jsonExtractAll": false,
      "eventBreakerRegex": "/[\\n\\r]+(?!\\s)/",
      "jsonArrayField": "records",
      "name": "JSON Records"
    }
  ],
  "description": "Event breaker for JSON records array with records field"
}