Port move operator (#271)

* Add move operator and the doc for it

* Update changelog

* Force Ci-check

* Force Ci

* Update init_common

* Force retest

* Force Test

* Fix change-log

Co-authored-by: Daniel Jaglowski <[email protected]>

cmd/stanza/init_common.go

@@ -27,6 +27,7 @@ import (
 	_ "github.com/observiq/stanza/operator/builtin/transformer/hostmetadata"
 	_ "github.com/observiq/stanza/operator/builtin/transformer/k8smetadata"
 	_ "github.com/observiq/stanza/operator/builtin/transformer/metadata"
+	_ "github.com/observiq/stanza/operator/builtin/transformer/move"
 	_ "github.com/observiq/stanza/operator/builtin/transformer/noop"
 	_ "github.com/observiq/stanza/operator/builtin/transformer/ratelimit"
 	_ "github.com/observiq/stanza/operator/builtin/transformer/recombine"

docs/operators/move.md

@@ -0,0 +1,283 @@
+## `move` operator
+
+The `move` operator moves (or renames) a field from one location to another.
+
+It's configured by passing 'to' and 'from' fields.
+
+### Configuration Fields
+
+| Field      | Default          | Description                                                                                                                                                                                                                              |
+| ---        | ---              | ---                                                                                                                                                                                                                                      |
+| `id`       | `move`    | A unique identifier for the operator                                                                                                                                                                                                     |
+| `output`   | Next in pipeline | The connected operator(s) that will receive all outbound entries                                                                                                                                                                         |
+| `from`      | required       | The [field](/docs/types/field.md)  to move the value out of.   
+| `to`      | required       | The [field](/docs/types/field.md)  to move the value into.
+| `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. |
+
+Example usage: 
+
+Rename value
+```yaml
+- type: move
+    from: key1
+    to: key3
+```
+
+<table>
+<tr><td> Input Entry</td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "key1": "val1",
+    "key2": "val2"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "key3": "val1",
+    "key2": "val2"
+  }
+}
+```
+
+</td>
+</tr>
+</table>
+<hr>
+
+Move a value from the record to resource
+
+```yaml
+- type: move
+    from: uuid
+    to: $resoruce.uuid
+```
+
+<table>
+<tr><td> Input Entry</td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "uuid": "091edc50-d91a-460d-83cd-089a62937738"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { 
+    "uuid": "091edc50-d91a-460d-83cd-089a62937738"
+  },
+  "labels": { },  
+  "record": { }
+}
+```
+
+</td>
+</tr>
+</table>
+
+<hr>
+
+Move a value from the record to labels
+
+```yaml
+- type: move
+    from: ip
+    to: $labels.ip
+```
+
+<table>
+<tr><td> Input Entry</td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "ip": "8.8.8.8"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { 
+    "ip": "8.8.8.8"
+  },  
+  "record": { }
+}
+```
+
+</td>
+</tr>
+</table>
+
+<hr>
+
+Replace the record with an individual value nested within the record
+```yaml
+- type: move
+    from: log
+    to: $record
+```
+
+<table>
+<tr><td> Input Entry</td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "log": "The log line"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": "The log line"
+}
+```
+
+</td>
+</tr>
+</table>
+
+<hr>
+
+Remove a layer from the record
+```yaml
+- type: move
+    from: wrapper
+    to: $record
+```
+
+<table>
+<tr><td> Input Entry</td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "wrapper": {
+      "key1": "val1",
+      "key2": "val2",
+      "key3": "val3"
+    }
+  }
+}
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "key1": "val1",
+    "key2": "val2",
+    "key3": "val3"
+  }
+}
+```
+
+</td>
+</tr>
+</table>
+
+<hr>
+
+Merge a layer to the record
+```yaml
+- type: move
+    from: wrapper
+    to: $record
+```
+
+<table>
+<tr><td> Input Entry</td> <td> Output Entry </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "wrapper": {
+      "key1": "val1",
+      "key2": "val2",
+      "key3": "val3"
+    },
+      "key4": "val1",
+      "key5": "val2",
+      "key6": "val3"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "resource": { },
+  "labels": { },  
+  "record": {
+    "key1": "val1",
+    "key2": "val2",
+    "key3": "val3",
+    "key4": "val1",
+    "key5": "val2",
+    "key6": "val3"
+  }
+}
+```
+
+</td>
+</tr>
+</table>
+

operator/builtin/transformer/move/move.go

@@ -0,0 +1,69 @@
+package move
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/observiq/stanza/entry"
+	"github.com/observiq/stanza/operator"
+	"github.com/observiq/stanza/operator/helper"
+)
+
+func init() {
+	operator.Register("move", func() operator.Builder { return NewMoveOperatorConfig("") })
+}
+
+// NewMoveOperatorConfig creates a new move operator config with default values
+func NewMoveOperatorConfig(operatorID string) *MoveOperatorConfig {
+	return &MoveOperatorConfig{
+		TransformerConfig: helper.NewTransformerConfig(operatorID, "move"),
+	}
+}
+
+// MoveOperatorConfig is the configuration of a move operator
+type MoveOperatorConfig struct {
+	helper.TransformerConfig `mapstructure:",squash" yaml:",inline"`
+	From                     entry.Field `mapstructure:"from" yaml:"from"`
+	To                       entry.Field `mapstructure:"to" yaml:"to"`
+}
+
+// Build will build a Move operator from the supplied configuration
+func (c MoveOperatorConfig) Build(context operator.BuildContext) ([]operator.Operator, error) {
+	transformerOperator, err := c.TransformerConfig.Build(context)
+	if err != nil {
+		return nil, err
+	}
+
+	if c.To == entry.NewNilField() || c.From == entry.NewNilField() {
+		return nil, fmt.Errorf("move: missing to or from field")
+	}
+
+	moveOperator := &MoveOperator{
+		TransformerOperator: transformerOperator,
+		From:                c.From,
+		To:                  c.To,
+	}
+
+	return []operator.Operator{moveOperator}, nil
+}
+
+// MoveOperator is an operator that moves a field's value to a new field
+type MoveOperator struct {
+	helper.TransformerOperator
+	From entry.Field
+	To   entry.Field
+}
+
+// Process will process an entry with a move transformation.
+func (p *MoveOperator) Process(ctx context.Context, entry *entry.Entry) error {
+	return p.ProcessWith(ctx, entry, p.Transform)
+}
+
+// Transform will apply the move operation to an entry
+func (p *MoveOperator) Transform(e *entry.Entry) error {
+	val, exist := p.From.Delete(e)
+	if !exist {
+		return fmt.Errorf("move: field does not exist")
+	}
+	return p.To.Set(e, val)
+}