[8.7](backport #35268) [Filebeat] Fix conditions to decode_json_fields and which pipeline to run

filebeat/module/kibana/log/config/log.yml

@@ -5,15 +5,16 @@ paths:
 {{ end }}
 exclude_files: [".gz$"]
 
+json.keys_under_root: false
+json.add_error_key: true
 processors:
-  # non-ECS: same as json.keys_under_root: false, allows compatibility with non-ecs logs.
-- decode_json_fields:
-    fields: [message]
-    target: 'json'
 - add_fields:
     target: ""
     fields:
       ecs.version: 1.12.0
     when:
-      not:
-        has_fields: ['ecs.version']
+      and:
+        - not:
+            has_fields: ['ecs.version']
+        - not:
+            has_fields: ['json.ecs.version']

filebeat/module/kibana/log/ingest/pipeline-7.yml

@@ -16,8 +16,6 @@ processors:
     target_field: '@timestamp'
 - remove:
     field: kibana.log.meta.@timestamp
-- remove:
-    field: message
 - rename:
     field: kibana.log.meta.message
     target_field: message

filebeat/module/kibana/log/ingest/pipeline-ecs.yml

@@ -6,31 +6,18 @@ processors:
 - set:
     copy_from: '@timestamp'
     field: event.created
-- rename:
-    field: message
-    target_field: _ecs_json_message
-    if: |-
-      def message = ctx.message;
-      return message != null
-          && message.startsWith('{')
-          && message.endsWith('}')
-          && message.contains('"@timestamp"')
-    ignore_missing: true
-- json:
-    field: _ecs_json_message
-    add_to_root: true
-    add_to_root_conflict_strategy: merge
-    allow_duplicate_keys: true
-    if: ctx.containsKey('_ecs_json_message')
-    on_failure:
-    - rename:
-        field: _ecs_json_message
-        target_field: message
-        ignore_missing: true
-    - set:
-        field: error.message
-        value: Error while parsing JSON
-        override: false
+- script:
+    lang: painless
+    if: 'ctx.json != null'
+    description: Merges filebeat generated fields with ECS log content
+    source: |-
+      ctx.json.keySet().each(key -> ctx.merge(key, ctx.json.get(key), (oldValue, newValue) -> {
+        if (newValue instanceof Map) {
+          newValue.putAll(oldValue);
+        }
+
+        return newValue;
+      }))
 - rename:
     field: http.request.headers
     target_field: kibana.log.meta.req.headers
@@ -47,6 +34,21 @@ processors:
     field: event.outcome
     value: failure
     if: 'ctx?.http?.response?.status_code != null && ctx.http.response.status_code >= 400'
+- script:
+    lang: painless
+    description: Overrides log entry with custom field values. Applicable when fields_under_root is true
+    if: "ctx.fields != null"
+    source: |-
+      ctx.fields.keySet().each(key -> ctx.merge(key, ctx.fields.get(key), (oldValue, newValue) -> {
+        if (oldValue instanceof Map) {
+          oldValue.putAll(newValue);
+        }
+
+        return oldValue;
+      }));
+- remove:
+    field: fields
+    ignore_missing: true
 - remove:
     field: json
     ignore_missing: true

filebeat/module/kibana/log/ingest/pipeline.yml

@@ -4,9 +4,10 @@ on_failure:
     field: error.message
     value: '{{ _ingest.on_failure_message }}'
 processors:
+# Ensures the correct pipeline is called regardless of whether `json` object is present or not
 - pipeline:
-    if: 'ctx?.json?.ecs?.version == null'
+    if: "ctx?.json?.type != null || ctx?.type != null"
     name: '{< IngestPipeline "pipeline-7" >}'
 - pipeline:
-    if: 'ctx?.json?.ecs?.version != null'
+    if: "ctx?.json?.ecs?.version != null || ctx?.ecs?.version != null"
     name: '{< IngestPipeline "pipeline-ecs" >}'