Update Remove operator to support removing the entire body, resource …

…or attributes (#112)

docs/operators/remove.md

@@ -8,7 +8,7 @@ The `remove` operator removes a field from a record.
 | ---        | ---              | ---                                                                                                                                                                                                                                      |
 | `id`       | `remove`    | A unique identifier for the operator                                                                                                                                                                                                     |
 | `output`   | Next in pipeline | The connected operator(s) that will receive all outbound entries                                                                                                                                                                         |
-| `field`      | required       | The [field](/docs/types/field.md) to remove.
+| `field`      | required       | The [field](/docs/types/field.md) to remove. if '$attributes' or '$resource' is specified, all fields of that type will be removed.
 | `on_error` | `send`           | The behavior of the operator if it encounters an error. See [on_error](/docs/types/on_error.md)                                                                                                                                          |
 | `if`       |                  | An [expression](/docs/types/expression.md) that, when set, will be evaluated to determine whether this operator should be used for the given entry. This allows you to do easy conditional parsing without branching logic with routers. |
 
@@ -175,6 +175,92 @@ Remove a value from resource
 }
 ```
 
+</td>
+</tr>
+</table>
+
+<hr>
+
+Remove all resource fields
+```yaml
+- type: remove 
+    field: $resource
+```
+
+<table>
+<tr><td> Input Entry </td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { 
+    "key1.0": "val",
+    "key2.0": "val"
+  },
+  "attributes": {  },  
+  "body": {
+    "key": "val"
+  },
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { },
+  "attributes": { },  
+  "body": { 
+    "key": "val"
+  }
+}
+```
+
+</td>
+</tr>
+</table>
+
+<hr>
+
+Remove all attributes
+```yaml
+- type: remove 
+    field: $attributes
+```
+
+<table>
+<tr><td> Input Entry </td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": {  },
+  "attributes": { 
+    "key1.0": "val",
+    "key2.0": "val"
+  },  
+  "body": {
+    "key": "val"
+  },
+}
+```
+
+</td>
+<td> 
+
+```json
+{
+  "resource": { },
+  "attributes": { },  
+  "body": { 
+    "key": "val"
+  }
+}
+```
+
 </td>
 </tr>
 </table>

entry/body_field.go

@@ -211,7 +211,7 @@ func (f BodyField) MarshalYAML() (interface{}, error) {
 func fromJSONDot(value string) BodyField {
 	keys := strings.Split(value, ".")
 
-	if keys[0] == "$" || keys[0] == bodyPrefix {
+	if keys[0] == "$" || keys[0] == BodyPrefix {
 		keys = keys[1:]
 	}
 
@@ -221,7 +221,7 @@ func fromJSONDot(value string) BodyField {
 // toJSONDot returns the JSON dot notation for a field.
 func toJSONDot(field BodyField) string {
 	if field.isRoot() {
-		return bodyPrefix
+		return BodyPrefix
 	}
 
 	containsDots := false
@@ -233,7 +233,7 @@ func toJSONDot(field BodyField) string {
 
 	var b strings.Builder
 	if containsDots {
-		b.WriteString(bodyPrefix)
+		b.WriteString(BodyPrefix)
 		for _, key := range field.Keys {
 			b.WriteString(`['`)
 			b.WriteString(key)

entry/entry_test.go

@@ -234,7 +234,7 @@ func TestFieldFromString(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			f, err := fieldFromString(tc.input)
+			f, err := NewField(tc.input)
 			if tc.expectedError {
 				require.Error(t, err)
 				return

entry/field.go

@@ -20,9 +20,9 @@ import (
 )
 
 const (
-	attributesPrefix = "$attributes"
-	resourcePrefix   = "$resource"
-	bodyPrefix       = "$body"
+	AttributesPrefix = "$attributes"
+	ResourcePrefix   = "$resource"
+	BodyPrefix       = "$body"
 )
 
 // Field represents a potential field on an entry.
@@ -47,7 +47,7 @@ func (f *Field) UnmarshalJSON(raw []byte) error {
 	if err != nil {
 		return err
 	}
-	*f, err = fieldFromString(s)
+	*f, err = NewField(s)
 	return err
 }
 
@@ -58,28 +58,28 @@ func (f *Field) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	if err != nil {
 		return err
 	}
-	*f, err = fieldFromString(s)
+	*f, err = NewField(s)
 	return err
 }
 
-func fieldFromString(s string) (Field, error) {
+func NewField(s string) (Field, error) {
 	split, err := splitField(s)
 	if err != nil {
 		return Field{}, fmt.Errorf("splitting field: %s", err)
 	}
 
 	switch split[0] {
-	case attributesPrefix:
+	case AttributesPrefix:
 		if len(split) != 2 {
 			return Field{}, fmt.Errorf("attributes cannot be nested")
 		}
 		return Field{AttributeField{split[1]}}, nil
-	case resourcePrefix:
+	case ResourcePrefix:
 		if len(split) != 2 {
 			return Field{}, fmt.Errorf("resource fields cannot be nested")
 		}
 		return Field{ResourceField{split[1]}}, nil
-	case bodyPrefix, "$":
+	case BodyPrefix, "$":
 		return Field{BodyField{split[1:]}}, nil
 	default:
 		return Field{BodyField{split}}, nil

entry/field_test.go

@@ -257,19 +257,19 @@ func TestSplitField(t *testing.T) {
 }
 
 func TestFieldFromStringInvalidSplit(t *testing.T) {
-	_, err := fieldFromString("$resource[test]")
+	_, err := NewField("$resource[test]")
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "splitting field")
 }
 
 func TestFieldFromStringWithResource(t *testing.T) {
-	field, err := fieldFromString(`$resource["test"]`)
+	field, err := NewField(`$resource["test"]`)
 	require.NoError(t, err)
 	require.Equal(t, "$resource.test", field.String())
 }
 
 func TestFieldFromStringWithInvalidResource(t *testing.T) {
-	_, err := fieldFromString(`$resource["test"]["key"]`)
+	_, err := NewField(`$resource["test"]["key"]`)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "resource fields cannot be nested")
 }

operator/builtin/transformer/remove/config_test.go

@@ -27,23 +27,47 @@ func TestGoldenConfig(t *testing.T) {
 			Name: "remove_body",
 			Expect: func() *RemoveOperatorConfig {
 				cfg := defaultCfg()
-				cfg.Field = entry.NewBodyField("nested")
+				cfg.Field = newBodyField("nested")
 				return cfg
 			}(),
 		},
 		{
 			Name: "remove_single_attribute",
 			Expect: func() *RemoveOperatorConfig {
 				cfg := defaultCfg()
-				cfg.Field = entry.NewAttributeField("key")
+				cfg.Field = newAttributeField("key")
 				return cfg
 			}(),
 		},
 		{
 			Name: "remove_single_resource",
 			Expect: func() *RemoveOperatorConfig {
 				cfg := defaultCfg()
-				cfg.Field = entry.NewResourceField("key")
+				cfg.Field = newResourceField("key")
+				return cfg
+			}(),
+		},
+		{
+			Name: "remove_entire_resource",
+			Expect: func() *RemoveOperatorConfig {
+				cfg := defaultCfg()
+				cfg.Field.allResource = true
+				return cfg
+			}(),
+		},
+		{
+			Name: "remove_entire_body",
+			Expect: func() *RemoveOperatorConfig {
+				cfg := defaultCfg()
+				cfg.Field.Field = entry.NewBodyField()
+				return cfg
+			}(),
+		},
+		{
+			Name: "remove_entire_attributes",
+			Expect: func() *RemoveOperatorConfig {
+				cfg := defaultCfg()
+				cfg.Field.allAttributes = true
 				return cfg
 			}(),
 		},
@@ -58,3 +82,18 @@ func TestGoldenConfig(t *testing.T) {
 func defaultCfg() *RemoveOperatorConfig {
 	return NewRemoveOperatorConfig("move")
 }
+
+func newBodyField(keys ...string) rootableField {
+	field := entry.NewBodyField(keys...)
+	return rootableField{Field: field}
+}
+
+func newResourceField(key string) rootableField {
+	field := entry.NewResourceField(key)
+	return rootableField{Field: field}
+}
+
+func newAttributeField(key string) rootableField {
+	field := entry.NewAttributeField(key)
+	return rootableField{Field: field}
+}

operator/builtin/transformer/remove/remove.go

@@ -27,18 +27,18 @@ func init() {
 	operator.Register("remove", func() operator.Builder { return NewRemoveOperatorConfig("") })
 }
 
-// NewRemoveOperatorConfig creates a new restructure operator config with default values
+// NewRemoveOperatorConfig creates a new remove operator config with default values
 func NewRemoveOperatorConfig(operatorID string) *RemoveOperatorConfig {
 	return &RemoveOperatorConfig{
 		TransformerConfig: helper.NewTransformerConfig(operatorID, "remove"),
 	}
 }
 
-// RemoveOperatorConfig is the configuration of a restructure operator
+// RemoveOperatorConfig is the configuration of a remove operator
 type RemoveOperatorConfig struct {
 	helper.TransformerConfig `mapstructure:",squash" yaml:",inline"`
 
-	Field entry.Field `mapstructure:"field"  json:"field" yaml:"field"`
+	Field rootableField `mapstructure:"field"  json:"field" yaml:"field"`
 }
 
 // Build will build a Remove operator from the supplied configuration
@@ -47,7 +47,8 @@ func (c RemoveOperatorConfig) Build(context operator.BuildContext) ([]operator.O
 	if err != nil {
 		return nil, err
 	}
-	if c.Field == entry.NewNilField() {
+
+	if c.Field.Field == entry.NewNilField() {
 		return nil, fmt.Errorf("remove: field is empty")
 	}
 
@@ -62,17 +63,27 @@ func (c RemoveOperatorConfig) Build(context operator.BuildContext) ([]operator.O
 // RemoveOperator is an operator that deletes a field
 type RemoveOperator struct {
 	helper.TransformerOperator
-	Field entry.Field
+	Field rootableField
 }
 
-// Process will process an entry with a restructure transformation.
+// Process will process an entry with a remove transformation.
 func (p *RemoveOperator) Process(ctx context.Context, entry *entry.Entry) error {
 	return p.ProcessWith(ctx, entry, p.Transform)
 }
 
 // Transform will apply the restructure operations to an entry
 func (p *RemoveOperator) Transform(entry *entry.Entry) error {
-	_, exist := entry.Delete(p.Field)
+	if p.Field.allAttributes {
+		entry.Attributes = nil
+		return nil
+	}
+
+	if p.Field.allResource {
+		entry.Resource = nil
+		return nil
+	}
+
+	_, exist := entry.Delete(p.Field.Field)
 	if !exist {
 		return fmt.Errorf("remove: field does not exist")
 	}