[Filebeat] Make set processor with copy_from compatible with ES < 7.13

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 module pipeline compatability for Elasticsearch versions less than 7.13.0 in modules that use Ingest Node `set` processors with `copy_from`. {pull}26593[26593]
 
 *Filebeat*
 

filebeat/fileset/compatibility.go

@@ -50,6 +50,13 @@ var processorCompatibilityChecks = []processorCompatibility{
 		},
 		adaptConfig: deleteProcessor,
 	},
+	{
+		procType: "set",
+		checkVersion: func(esVersion *common.Version) bool {
+			return esVersion.LessThan(common.MustNewVersion("7.13.0"))
+		},
+		adaptConfig: replaceSetCopyFromValue,
+	},
 	{
 		procType: "set",
 		checkVersion: func(esVersion *common.Version) bool {
@@ -174,6 +181,22 @@ func replaceSetIgnoreEmptyValue(config map[string]interface{}, log *logp.Logger)
 	return false, nil
 }
 
+// replaceSetCopyFromValue replaces copy_from option with a 'value'.
+func replaceSetCopyFromValue(config map[string]interface{}, log *logp.Logger) (bool, error) {
+	copyFrom, ok := config["copy_from"].(string)
+	if !ok {
+		return false, nil
+	}
+
+	log.Debug("Removing unsupported 'copy_from' from set processor.")
+	delete(config, "copy_from")
+
+	value := "{{{" + copyFrom + "}}}"
+	log.Debugf("Adding 'value: %q' to replace 'copy_from' in set processor.", value)
+	config["value"] = value
+	return false, nil
+}
+
 // replaceAppendAllowDuplicates replaces allow_duplicates option with an if statement
 // so ES less than 7.10 will work.
 func replaceAppendAllowDuplicates(config map[string]interface{}, log *logp.Logger) (bool, error) {

filebeat/fileset/compatibility_test.go

@@ -315,6 +315,69 @@ func TestReplaceSetIgnoreEmptyValue(t *testing.T) {
 	}
 }
 
+func TestReplaceSetCopyFromValue(t *testing.T) {
+	setWithCopyFrom := map[string]interface{}{
+		"processors": []interface{}{
+			map[string]interface{}{
+				"set": map[string]interface{}{
+					"field":     "rule.name",
+					"copy_from": "panw.panos.ruleset",
+				},
+			},
+		}}
+
+	cases := []struct {
+		name          string
+		esVersion     *common.Version
+		content       map[string]interface{}
+		expected      map[string]interface{}
+		isErrExpected bool
+	}{
+		{
+			name:      "ES < 7.13.0",
+			esVersion: common.MustNewVersion("7.12.2"),
+			content:   setWithCopyFrom,
+			expected: map[string]interface{}{
+				"processors": []interface{}{
+					map[string]interface{}{
+						"set": map[string]interface{}{
+							"field": "rule.name",
+							"value": "{{{panw.panos.ruleset}}}",
+						},
+					},
+				},
+			},
+			isErrExpected: false,
+		},
+		{
+			name:      "ES == 7.13.0",
+			esVersion: common.MustNewVersion("7.13.0"),
+			content:   setWithCopyFrom,
+			expected:  setWithCopyFrom,
+		},
+		{
+			name:      "ES > 7.13.0",
+			esVersion: common.MustNewVersion("8.0.0"),
+			content:   setWithCopyFrom,
+			expected:  setWithCopyFrom,
+		},
+	}
+
+	for _, test := range cases {
+		test := test
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+			err := adaptPipelineForCompatibility(*test.esVersion, "foo-pipeline", test.content, logp.NewLogger(logName))
+			if test.isErrExpected {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, test.expected, test.content, test.name)
+			}
+		})
+	}
+}
+
 func TestReplaceAppendAllowDuplicates(t *testing.T) {
 	cases := []struct {
 		name          string