Capstone Example: Constructing a List of Complex Composite JSON Objects

In this example we will pick a more complex scenario ; Construct a JSON object consisting of  "action_type","target_ip" of the first session only, and the user integer ID.

     JSON Path of Source Fields ; 

          $.message.user{}.sessions}0].actions

Input Schema before Flattening

Input Schema after Flattening

Required Target Schema 

filter {
json { source => "message"   array_function => "split_columns"}


mutate { convert => { "user.id" => "string"} on_error =>"convertError"}
mutate {replace => {"userId"=>"%{user.id}"}}
mutate { convert => { "userId" => "string"} on_error =>"missingUserId"}


#Loop1
for i1, v1 in user.sessions {
    if "i1] == 0 {


        #Loop2
        for i2, v2 in v1.actions {
            mutate {replace => {"temp.action.userId"=>"%{userId}"}}
            mutate {replace => {"temp.action.dst"=>"%{v2.targetIP}"}}
            mutate {replace => {"temp.action.type"=>"%{v2.action_type}"}}
            mutate {replace => {"temp.action.query"=>"%{v2.query}"} on_error => "missingQueryField"}
            mutate {merge => {"listFirstSession"=>"temp"} on_error => "mergeError"}
            mutate {replace => {"temp"=>""} on_error => "errorClearingTempPlaceHolder"}
        }
    }
}
statedump {}
}

Snippet from statedump output:

{… 

 "listFirstSession": >

    {

      "action": {

        "dst": "10.0.0.10",

        "type": "login",

        "userId": "12345"

      }

    },

    {

      "action": {

        "dst": "10.0.0.10",

        "query": "weather",

        "type": "search",

        "userId": "12345"

      }

    },

    {

      "action": {

        "dst": "10.0.0.11",

        "type": "logout",

        "userId": "12345"

      }

    }

  ],

..}

Schematic for the transform

Summary: The core logic for mapping data involves transforming the input schema into a target schema with the following structure: listFirstSession(Repeated).action(composite).type, dst, userid, query(atomic).

We'll use a temporary variable, temp, to hold the structured data for each action within the first session. This variable will have a hierarchy mirroring the target: temp.action.type, temp.action.dst, temp.action.userid, and temp.action.query.

The input data has two levels of repeated fields: user.sessions and user.sessions.actions. To process all the actions, we'll implement two nested loops. The outer loop iterates through each session, and the inner loop iterates through the actions within each session.

Finally, because the query field in the input is optional and can appear multiple times, we'll include error handling to gracefully manage cases where it's absent."
 

1. 1. We begin by extracting and converting the $.user.id to an integer field              userId, as these fields are directly accessible and outside the scope of            the session loops.

   2. To access the 'actions', which are nested within                                             $.message.user.sessionsc].actionsn], we'll use two nested loops. The outer      loop iterates through each session, and the inner loop iterates through the      actions within the current session. The variable v2 will represent each              individual 'actions' object in the inner loop.

2. 3. To target only the first session, we introduce a conditional statement in            the outer loop: if i1 == 0. This ensures that the inner loop processing                actions are only executed for the session at index 0.

3. 4. Within the inner loop, we extract the mandatory fields 'action_type' and             'targetIP' , along with the common field ‘userId’ ’into a temporary variable        $.temp with hierarchy $.temp.action.type,dst,userId. For the optional                'query' field, we use a conditional check using on_error to implement an            operator similar to  if Exists v2.query.
4. 5. For each action processed in the loops, we populate a temporary                     placeholder $.temp.action with the following:

•         • $.temp.action.type ← v2.action_type

•         • $.temp.action.dst ← v2.targetIP

•         • $.temp.action.userId ← $.userId (This is the username captured                      earlier, common to all actions in the first session)

•         •$.temp.action.query ← v2.query ifExists, on_error :
           $.queryMissingFlag ← true (If 'query' exists, we map it; otherwise, we               set a flag).

6. 6. Once the $.temp.action structure is populated for a given action, we                append the top-level $.temp object to the $.listFirstSession repeated field      using merge to implement ; $.listFirstSession0] ← append ($.temp).
7. 7. After each merge operation, we clear the $.temp placeholder to ensure           that subsequent action mappings are independent.
   7.  
   8. Using the algorithm described above, we smoothly managed to capture AND map the fields in the target schema