[Filebeat] Replace copy_from with templated value

CHANGELOG.next.asciidoc

@@ -280,6 +280,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix bug in `httpjson` that prevented `first_event` getting updated. {pull}26407[26407]
 - Fix bug in the Syslog input that misparsed rfc5424 days starting with 0. {pull}26419[26419]
 - Do not close filestream harvester if an unexpected error is returned when close.on_state_change.* is enabled. {pull}26411[26411]
+- Fix Elasticsearch compatibility for modules that use `copy_from` in `set` processors. {issue}26629[26629]
 
 *Filebeat*
 

filebeat/module/elasticsearch/audit/ingest/pipeline-json.yml

@@ -156,7 +156,7 @@ processors:
       path: elasticsearch.audit
   - set:
       field: http.request.id
-      copy_from: elasticsearch.audit.request.id
+      value: '{{{elasticsearch.audit.request.id}}}'
       ignore_empty_value: true
   - dot_expander:
       field: cluster.name

filebeat/module/nginx/ingress_controller/ingest/pipeline.yml

@@ -73,7 +73,7 @@ processors:
       value: ""
   - set:
       field: http.request.id
-      copy_from: nginx.ingress_controller.http.request.id
+      value: '{{{nginx.ingress_controller.http.request.id}}}'
       ignore_empty_value: true
       ignore_failure: true
   - script:

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

@@ -914,40 +914,59 @@ processors:
             value: "success"
           - set: event.reason
             from: cyberarkpas.audit.reason
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Failed to enrich based on ID #{{{ event.code }}}: {{{_ingest.on_failure_message}}}'
 
       source: >
-        def clone(def val) {
-            return val instanceof List? new ArrayList(val) : val;
+        def clone(def ref) {
+          if (ref == null) return ref;
+          if (ref instanceof Map) {
+            ref = ref.entrySet().stream().collect(
+                    Collectors.toMap(
+                      e -> e.getKey(),
+                      e -> clone(e.getValue())
+                    )
+                  );
+          } else if (ref instanceof List) {
+            ref = ref.stream().map(e -> clone(e)).collect(
+                    Collectors.toList()
+                  );
+          }
+          return ref;
         }
         def read_field(def map, String name) {
           if (map == null || !(map instanceof Map)) return null;
           int pos = name.indexOf(".");
           return pos == -1? map[name]
                           : read_field(map[name.substring(0, pos)], name.substring(pos+1));
         }
+        boolean set_field(Map map, String name, def value) {
+          int pos = name.indexOf(".");
+          if (pos == -1) {
+            map[name] = clone(value);
+            return true;
+          }
+          String key  = name.substring(0, pos),
+                 path = name.substring(pos+1);
+          if (!map.containsKey(key)) {
+            map[key] = new HashMap();
+          }
+          map = map[key];
+          return map instanceof Map? set_field(map, path, value)
+                                   : false;
+        }
         String msgID = ctx.event?.code;
         def actions = params.get(msgID);
         if (actions == null) return;
-        List values = new ArrayList();
         for (def item : actions) {
           def val = item.value;
           if (val == null && (val = read_field(ctx, item.from)) == null || val == "") continue;
-          values.add([
-            "to": item.set,
-            "value": clone(val)
-          ]);
+          if (!set_field(ctx, item.set, val)) {
+            throw new Exception("Failed to set field " + item.set);
+          }
         }
-        if (!values.isEmpty()) ctx._tmp["values"] = values;
-
-  - foreach:
-      field: _tmp.values
-      ignore_missing: true
-      processor:
-        set:
-          field: '{{{_ingest._value.to}}}'
-          copy_from: '_ingest._value.value'
-          ignore_empty_value: true
-          override: true
 
   #
   # Force event.outcome: unknown in case it gets a value other than one of the allowed.
@@ -994,7 +1013,7 @@ processors:
       on_failure:
         - set:
             field: source.domain
-            copy_from: source.address
+            value: '{{{source.address}}}'
   - convert:
       field: destination.address
       target_field: destination.ip
@@ -1003,7 +1022,7 @@ processors:
       on_failure:
         - set:
             field: destination.domain
-            copy_from: destination.address
+            value: '{{{destination.address}}}'
 
   #
   # Populate related.ip

x-pack/filebeat/module/panw/panos/ingest/pipeline.yml

@@ -462,10 +462,30 @@ processors:
         ANY: '.*'
       if: 'ctx?.file?.name != null && ctx?.file?.name != ""'
 
-  - set:
-      field: user
-      copy_from: source.user
-      if: "ctx?.source?.user != null"
+  - script:
+      lang: painless
+      description: Copy source.user to user
+      source: >
+        def clone(def ref) {
+          if (ref == null) return ref;
+          if (ref instanceof Map) {
+            ref = ref.entrySet().stream().collect(
+                    Collectors.toMap(
+                      e -> e.getKey(),
+                      e -> clone(e.getValue())
+                    )
+                  );
+          } else if (ref instanceof List) {
+            ref = ref.stream().map(e -> clone(e)).collect(
+                    Collectors.toList()
+                  );
+          }
+          return ref;
+        }
+        def u = ctx?.source?.user;
+        if (u != null) {
+          ctx["user"] = clone(u);
+        }
 
   - append:
       field: related.user

x-pack/filebeat/module/threatintel/abuseurl/ingest/pipeline.yml

@@ -45,7 +45,7 @@ processors:
     remove_if_successful: true
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
 - rename:
     field: threatintel.abuseurl.host

x-pack/filebeat/module/threatintel/anomali/ingest/pipeline.yml

@@ -84,7 +84,7 @@ processors:
     if: ctx?.threatintel?.indicator?.type == 'url'
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
 - rename:
     field: _tmp.threatvalue

x-pack/filebeat/module/threatintel/anomalithreatstream/ingest/pipeline.yml

@@ -265,7 +265,7 @@ processors:
 
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
 
 - rename:

x-pack/filebeat/module/threatintel/misp/ingest/pipeline.yml

@@ -139,7 +139,7 @@ processors:
 
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
     if: "ctx?.threatintel?.indicator?.type == 'url' && ctx?.threatintel?.misp?.attribute?.type != 'uri'"
 

x-pack/filebeat/module/threatintel/otx/ingest/pipeline.yml

@@ -87,7 +87,7 @@ processors:
     if: ctx?.threatintel?.indicator?.type == 'url'
 - set:
     field: threatintel.indicator.url.full
-    copy_from: threatintel.indicator.url.original
+    value: '{{{threatintel.indicator.url.original}}}'
     ignore_empty_value: true
     if: "ctx?.threatintel?.otx?.type == 'URL'"
 

x-pack/filebeat/module/threatintel/recordedfuture/ingest/pipeline.yml

@@ -122,7 +122,7 @@ processors:
 
   - set:
       field: json.ip_range
-      copy_from: json.entity.name
+      value: '{{{json.entity.name}}}'
       if: 'ctx.json.entity?.type == "IpAddress" && ctx.json.entity.name != null && ctx.json.entity.name.contains("/")'
   - set:
       field: json.ip_range
@@ -134,7 +134,7 @@ processors:
       if: 'ctx.threatintel?.indicator?.type == "ipv6-addr" && ctx.json.entity.name != null && !ctx.json.entity.name.contains("/")'
   - set:
       field: json.ip_range
-      copy_from: json.entity.name
+      value: '{{{json.entity.name}}}'
       if: 'ctx.json.entity?.type == "IpAddress" && ctx.json.entity.name != null && ctx.json.entity.name.contains("/")'
 
   - rename:

x-pack/filebeat/module/zoom/webhook/ingest/meeting.yml

@@ -51,7 +51,7 @@ processors:
     if: ctx?.url?.original != null
 - set:
     field: url.full
-    copy_from: url.original
+    value: '{{{url.original}}}'
     ignore_failure: true
     if: ctx?.url?.original != null
 #