Port key value parser from Stanza

docs/operators/key_value_parser.md

@@ -0,0 +1,178 @@
+## `key_value_parser` operator
+
+The `key_value_parser` operator parses the string-type field selected by `parse_from` into key value pairs. All values are of type string.
+
+### Configuration Fields
+
+| Field            | Default             | Description                                                                                                                                                                                                                               |
+| ---              | ---                 | ---                                                                                                                                                                                                                                       |
+| `id`             | `key_value_parser`  | A unique identifier for the operator.                                                                                                                                                                                                     |
+| `delimiter`      | `=`                 | The delimiter used for splitting a value into a key value pair.                                                                                                                                                                           |
+| `pair_delimiter` |                     | The delimiter used for seperating key value pairs, defaults to whitespace.                                                                                                                                                                |
+| `output`         | Next in pipeline    | The connected operator(s) that will receive all outbound entries.                                                                                                                                                                         |
+| `parse_from`     | `body`              | A [field](/docs/types/field.md) that indicates the field to be parsed into key value pairs.                                                                                                                                               |
+| `parse_to`       | `body`              | A [field](/docs/types/field.md) that indicates the field to be parsed as into key value pairs.                                                                                                                                            |
+| `preserve_to`    |                     | Preserves the unparsed value at the specified [field](/docs/types/field.md).                                                                                                                                                              |
+| `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.  |
+| `timestamp`      | `nil`               | An optional [timestamp](/docs/types/timestamp.md) block which will parse a timestamp field before passing the entry to the output operator.                                                                                               |
+| `severity`       | `nil`               | An optional [severity](/docs/types/severity.md) block which will parse a severity field before passing the entry to the output operator.                                                                                                  |
+
+
+### Example Configurations
+
+#### Parse the field `message` into key value pairs
+
+Configuration:
+```yaml
+- type: key_value_parser
+  parse_from: message
+```
+
+<table>
+<tr><td> Input body </td> <td> Output body </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "message": "name=stanza"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "name": "stanza"
+  }
+}
+```
+
+</td>
+</tr>
+</table>
+
+#### Parse the field `message` into key value pairs, using a non default delimiter
+
+Configuration:
+```yaml
+- type: key_value_parser
+  parse_from: message
+  delimiter: ":"
+```
+
+<table>
+<tr><td> Input body </td> <td> Output body </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "message": "name:stanza"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "name": "stanza"
+  }
+}
+```
+
+#### Parse the field `message` into key value pairs, using a non default pair delimiter
+
+Configuration:
+```yaml
+- type: key_value_parser
+  parse_from: message
+  pair_delimiter: "!"
+```
+
+<table>
+<tr><td> Input body </td> <td> Output body </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "message": "name=stanza ! org=otel      ! group=dev"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "name": "stanza",
+    "org": "otel",
+    "group": "dev"
+  }
+}
+```
+
+</td>
+</tr>
+</table>
+
+#### Parse the field `message` as key value pairs, and parse the timestamp
+
+Configuration:
+```yaml
+- type: key_value_parser
+  parse_from: message
+  timestamp:
+    parse_from: seconds_since_epoch
+    layout_type: epoch
+    layout: s
+```
+
+<table>
+<tr><td> Input body </td> <td> Output body </td></tr>
+<tr>
+<td>
+
+```json
+{
+  "timestamp": "",
+  "body": {
+    "message": "name=stanza seconds_since_epoch=1136214245"
+  }
+}
+```
+
+</td>
+<td>
+
+```json
+{
+  "timestamp": "2006-01-02T15:04:05-07:00",
+  "body": {
+    "name": "stanza"
+  }
+}
+```
+
+</td>
+</tr>
+</table>

go.mod

@@ -24,6 +24,7 @@ require (
 )
 
 require (
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/influxdata/go-syslog/v3 v3.0.1-0.20210608084020-ac565dc76ba6
 	go.uber.org/multierr v1.8.0
 )
@@ -39,6 +40,7 @@ require (
 	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/googleapis/gnostic v0.5.6 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect

go.sum

@@ -181,6 +181,10 @@ github.com/googleapis/gnostic v0.5.5/go.mod h1:7+EbHbldMins07ALC74bsA81Ovc97Dwqy
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=

operator/parser/keyvalue/config_test.go

@@ -0,0 +1,122 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package keyvalue
+
+import (
+	"testing"
+
+	"github.com/open-telemetry/opentelemetry-log-collection/entry"
+	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper"
+	"github.com/open-telemetry/opentelemetry-log-collection/operator/helper/operatortest"
+)
+
+func TestKVParserConfig(t *testing.T) {
+	cases := []operatortest.ConfigUnmarshalTest{
+		{
+			Name:   "default",
+			Expect: defaultCfg(),
+		},
+		{
+			Name: "parse_from_simple",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				cfg.ParseFrom = entry.NewBodyField("from")
+				return cfg
+			}(),
+		},
+		{
+			Name: "parse_to_simple",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				cfg.ParseTo = entry.NewBodyField("log")
+				return cfg
+			}(),
+		},
+		{
+			Name: "on_error_drop",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				cfg.OnError = "drop"
+				return cfg
+			}(),
+		},
+		{
+			Name: "timestamp",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				parseField := entry.NewBodyField("timestamp_field")
+				newTime := helper.TimeParser{
+					LayoutType: "strptime",
+					Layout:     "%Y-%m-%d",
+					ParseFrom:  &parseField,
+				}
+				cfg.TimeParser = &newTime
+				return cfg
+			}(),
+		},
+		{
+			Name: "severity",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				parseField := entry.NewBodyField("severity_field")
+				severityField := helper.NewSeverityParserConfig()
+				severityField.ParseFrom = &parseField
+				mapping := map[interface{}]interface{}{
+					"critical": "5xx",
+					"error":    "4xx",
+					"info":     "3xx",
+					"debug":    "2xx",
+				}
+				severityField.Mapping = mapping
+				cfg.SeverityParserConfig = &severityField
+				return cfg
+			}(),
+		},
+		{
+			Name: "preserve_to",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				preserve := entry.NewBodyField("aField")
+				cfg.PreserveTo = &preserve
+				return cfg
+			}(),
+		},
+		{
+			Name: "delimiter",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				cfg.Delimiter = ";"
+				return cfg
+			}(),
+		},
+		{
+			Name: "pair_delimiter",
+			Expect: func() *KVParserConfig {
+				cfg := defaultCfg()
+				cfg.PairDelimiter = ";"
+				return cfg
+			}(),
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.Name, func(t *testing.T) {
+			tc.Run(t, defaultCfg())
+		})
+	}
+}
+
+func defaultCfg() *KVParserConfig {
+	return NewKVParserConfig("key_value_parser")
+}