Flatten the Json output

[rohanrajnv]

Hi Everyone,
I need some help to flatten the Json File that i get from wazuh I tried to use the Lift in the config and it is still not working. The parsing is also not working.

this is the Log that I get from wazuh:

{"timestamp":"2023-08-12T13:44:09.451+0000","rule":{"level":3,"description":"Successful sudo to ROOT executed.","id":"5402","mitre":{"id":["T1548.003"],"tactic":["Privilege Escalation","Defense Evasion"],"technique":["Sudo and Sudo Caching"]},"firedtimes":97386,"mail":false,"groups":["syslog","sudo"],"pci_dss":["10.2.5","10.2.2"],"gpg13":["7.6","7.8","7.13"],"gdpr":["IV_32.2"],"hipaa":["164.312.b"],"nist_800_53":["AU.14","AC.7","AC.6"],"tsc":["CC6.8","CC7.2","CC7.3"]},"agent":{"id":"037","name":"mzab01","ip":"192.168.49.99"},"manager":{"name":"OSHW01"},"id":"13.1","full_log":"Aug 12 17:44:08 mzab01 sudo: test : TTY=unknown ; PWD=/ ; USER=root ; COMMAND=/usr/bin/cp /waldb/172.16.171.110-O.xml /waldb/172.16.171.110-T.xml","predecoder":{"program_name":"sudo","timestamp":"Aug 12 17:44:08","hostname":"mzab01"},"decoder":{"parent":"sudo","name":"sudo","ftscomment":"First time user executed the sudo command"},"data":{"srcuser":"test","dstuser":"root","tty":"unknown","pwd":"/","command":"/usr/bin/cp /waldb/172.16.171.110-O.xml /waldb/172.16.171.110-T.xml"},"location":"/var/log/auth.log"}

But I want to change to
{
"timestamp": "2023-08-12T13:44:09.451+0000",
"rule.level": 3,
"rule.description": "Successful sudo to ROOT executed.",
"rule.id": "5402",
"rule.mitre.id[0]": "T1548.003",
"rule.mitre.tactic[0]": "Privilege Escalation",
"rule.mitre.tactic[1]": "Defense Evasion",
"rule.mitre.technique[0]": "Sudo and Sudo Caching",
"rule.firedtimes": 97386,
"rule.mail": false,
"rule.groups[0]": "syslog",
"rule.groups[1]": "sudo",
"rule.pci_dss[0]": "10.2.5",
"rule.pci_dss[1]": "10.2.2",
"rule.gpg13[0]": "7.6",
"rule.gpg13[1]": "7.8",
"rule.gpg13[2]": "7.13",
"rule.gdpr[0]": "IV_32.2",
"rule.hipaa[0]": "164.312.b",
"rule.nist_800_53[0]": "AU.14",
"rule.nist_800_53[1]": "AC.7",
"rule.nist_800_53[2]": "AC.6",
"rule.tsc[0]": "CC6.8",
"rule.tsc[1]": "CC7.2",
"rule.tsc[2]": "CC7.3",
"agent.id": "037",
"agent.name": "mzab01",
"agent.ip": "192.168.49.99",
"manager.name": "OSHW01",
"id": "13.1",
"full_log": "Aug 12 17:44:08 mzab01 sudo: test : TTY=unknown ; PWD=/ ; USER=root ; COMMAND=/usr/bin/cp /waldb/172.16.171.110-O.xml /waldb/172.16.171.110-T.xml",
"predecoder.program_name": "sudo",
"predecoder.timestamp": "Aug 12 17:44:08",
"predecoder.hostname": "mzab01",
"decoder.parent": "sudo",
"decoder.name": "sudo",
"decoder.ftscomment": "First time user executed the sudo command",
"data.srcuser": "test",
"data.dstuser": "root",
"data.tty": "unknown",
"data.pwd": "/",
"data.command": "/usr/bin/cp /waldb/172.16.171.110-O.xml /waldb/172.16.171.110-T.xml",
"location": "/var/log/auth.log"
}
Any Help would be really great

[KeelMatrix]

The core filters won't get you there: record_transformer, parser and friends work on the keys you declare, and fluentd has no core flatten filter that walks arbitrary nesting — so there's no config-side shortcut for what you were trying. The flatten itself is mechanical, so it's cleaner to do it with jq inside exec_filter and let fluentd keep doing routing and buffering.

flatten.jq, which produces exactly the shape in your post (dotted object paths, arrays indexed):

def flatname: reduce .[] as $s (""; if ($s|startswith("[")) then . + $s elif . == "" then $s else . + "." + $s end);
def flatten($p):
  if   type == "object" then reduce (to_entries[]) as $e ({}; . + ($e.value | flatten($p + [$e.key])))
  elif type == "array"  then reduce (to_entries[]) as $e ({}; . + ($e.value | flatten($p + ["[\($e.key)]"])))
  else { ($p | flatname): . }
  end;
flatten([])

On your Wazuh record it produces keys like rule.level, rule.description, rule.mitre.id[0], rule.mitre.tactic[1], rule.groups[0], agent.id, predecoder.program_name and data.srcuser — one flat object per event, no depth limit.

Wiring it in:

<match wazuh.**>
  @type exec_filter
  command jq -c -f /etc/fluent/flatten.jq
  <format>
    @type json
  </format>
  <parse>
    @type json
  </parse>
  <buffer>
    @type memory
    flush_interval 1s
  </buffer>
</match>

exec_filter is part of fluentd core (lib/fluent/plugin/out_exec_filter.rb). Both its input and output formats default to TSV, so setting <format> and <parse> to json is the part that matters. It spawns the child process per chunk, so on a busy Wazuh stream raise num_children or move the flatten step upstream of the log shipper.

If you'd rather not put jq in the data path, the same walk in Python:

def flatten(obj, prefix="", out=None):
    out = {} if out is None else out
    if isinstance(obj, dict):
        for k, v in obj.items():
            flatten(v, f"{prefix}.{k}" if prefix else k, out)
    elif isinstance(obj, list):
        for i, v in enumerate(obj):
            flatten(v, f"{prefix}[{i}]", out)
    else:
        out[prefix] = obj
    return out

import json
print(json.dumps(flatten(record)))

One more thing, since flat JSON is usually a step towards a table: for CSV the header has to be a union across events (different events carry different array lengths, so different key sets). Flatten with jq, then let a union-aware writer decide the columns:

$ jq -c -f flatten.jq wazuh.log | mlr --ijson --ocsv unsparsify > wazuh.csv

unsparsify buffers the stream to compute the union of keys, so it suits batch conversion, not an unbounded live stream.

Disclosure: I work on jsonnorm.com, an open JSON→CSV converter — log-shaped NDJSON with arrays is the case it's built around, and the array policy (keep [n] columns vs one row per element) is a visible setting rather than an accident: https://jsonnorm.com/convert/ndjson-jsonl-to-csv/?src=ghdisc