x-pack/filebeat/module/mysqlenterprise: fix handling of streaming dat…

…a sent as partial array object

MySQL send its audit logs as parts of an infinitely long JSON array and
so separates each line of the logs with a comma. We don't know that we
are in an array since the first line of the log may not have been sent
to us, so remove the trailing comma to treat each element of the partial
array object as an object in a JSON stream.

CHANGELOG.next.asciidoc

@@ -112,6 +112,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Fix base for UDP and TCP queue metrics and UDP drops metric. {pull}35123[35123]
 - Sanitize filenames for request tracer in httpjson and cel inputs. {pull}35143[35143]
 - decode_cef processor: Fix ECS output by making `observer.ip` into an array of strings instead of string. {issue}35140[35140] {pull}35149[35149]
+- Fix handling of MySQL audit logs with strict JSON parser. {issue}35158[35158] {pull}35160[35160]
 
 *Heartbeat*
 

x-pack/filebeat/module/mysqlenterprise/audit/ingest/pipeline.yml

@@ -3,9 +3,24 @@ processors:
 - set:
     field: event.ingested
     value: '{{_ingest.timestamp}}'
+- set:
+    field: event.original
+    copy_from: message
+- script:
+    description: Trim trailing commas.
+    # MySQL sends audit logs as parts of a single infinite JSON array
+    # rather than as a JSON stream, and so has comma separators. We
+    # don't have the array open token, so remove the commas.
+    lang: painless
+    source:
+      ctx.message = ctx.message.substring(0, ctx.message.length() - 1);
+    if: ctx.message instanceof String && ctx.message.endsWith(',')
 - json:
     field: message
     target_field: mysqlenterprise.audit
+- remove:
+    field: message
+    ignore_missing: true
 - remove:
     field: '@timestamp'
     ignore_missing: true