support attributes transfer

@@ -49,10 +49,9 @@ Processors:
         - index
         - beat
         - docker_container
-        - offset
-        - prospector
-        - source
-        - stream
+    - Action: transfer
+      Field: message
+      Target: data
   Output:
     ElasticSearch:
       Hosts:

@@ -49,10 +49,9 @@ Processors:
         - index
         - beat
         - docker_container
-        - offset
-        - prospector
-        - source
-        - stream
+    - Action: transfer
+      Field: message
+      Target: data
   Output:
     ElasticSearch:
       Hosts:

@@ -24,9 +24,11 @@ type (
 	}
 
 	Filter struct {
-		Action     string      `json:",options=drop|remove_field"`
+		Action     string      `json:",options=drop|remove_field|transfer"`
 		Conditions []Condition `json:",optional"`
 		Fields     []string    `json:",optional"`
+		Field      string      `json:",optional"`
+		Target     string      `json:",optional"`
 	}
 
 	Processor struct {

@@ -34,6 +34,9 @@ Processors:
         - prospector
         - source
         - stream
+    - Action: transfer
+      Field: message
+      Target: data
   Output:
     ElasticSearch:
       Hosts:

@@ -5,6 +5,7 @@ import "github.com/tal-tech/go-stash/stash/config"
 const (
 	filterDrop         = "drop"
 	filterRemoveFields = "remove_field"
+	filterTransfer     = "transfer"
 	opAnd              = "and"
 	opOr               = "or"
 	typeContains       = "contains"
@@ -22,6 +23,8 @@ func CreateFilters(p config.Processor) []FilterFunc {
 			filters = append(filters, DropFilter(f.Conditions))
 		case filterRemoveFields:
 			filters = append(filters, RemoveFieldFilter(f.Fields))
+		case filterTransfer:
+			filters = append(filters, TransferFilter(f.Field, f.Target))
 		}
 	}
 

+package filter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRemoveFieldFilter(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  map[string]interface{}
+		fields []string
+		expect map[string]interface{}
+	}{
+		{
+			name: "remove field",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+			fields: []string{"b"},
+			expect: map[string]interface{}{
+				"a": "aa",
+			},
+		},
+		{
+			name: "remove field",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+			fields: []string{"c"},
+			expect: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			actual := RemoveFieldFilter(test.fields)(test.input)
+			assert.EqualValues(t, test.expect, actual)
+		})
+	}
+}

+package filter
+
+import (
+	jsoniter "github.com/json-iterator/go"
+)
+
+func TransferFilter(field, target string) FilterFunc {
+	return func(m map[string]interface{}) map[string]interface{} {
+		val, ok := m[field]
+		if !ok {
+			return m
+		}
+
+		s, ok := val.(string)
+		if !ok {
+			return m
+		}
+
+		var nm map[string]interface{}
+		if err := jsoniter.Unmarshal([]byte(s), &nm); err != nil {
+			return m
+		}
+
+		delete(m, field)
+		if len(target) > 0 {
+			m[target] = nm
+		} else {
+			for k, v := range nm {
+				m[k] = v
+			}
+		}
+
+		return m
+	}
+}

+package filter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTransferFilter(t *testing.T) {
+	tests := []struct {
+		name   string
+		input  map[string]interface{}
+		field  string
+		target string
+		expect map[string]interface{}
+	}{
+		{
+			name: "with target",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+			field:  "b",
+			target: "data",
+			expect: map[string]interface{}{
+				"a": "aa",
+				"data": map[string]interface{}{
+					"c": "cc",
+				},
+			},
+		},
+		{
+			name: "without target",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+			field: "b",
+			expect: map[string]interface{}{
+				"a": "aa",
+				"c": "cc",
+			},
+		},
+		{
+			name: "without field",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+			field: "c",
+			expect: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"}`,
+			},
+		},
+		{
+			name: "with not json",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"`,
+			},
+			field: "b",
+			expect: map[string]interface{}{
+				"a": "aa",
+				"b": `{"c":"cc"`,
+			},
+		},
+		{
+			name: "with not string",
+			input: map[string]interface{}{
+				"a": "aa",
+				"b": map[string]interface{}{"c": "cc"},
+			},
+			field: "b",
+			expect: map[string]interface{}{
+				"a": "aa",
+				"b": map[string]interface{}{"c": "cc"},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			actual := TransferFilter(test.field, test.target)(test.input)
+			assert.EqualValues(t, test.expect, actual)
+		})
+	}
+}