Add marshal processor

docs/processors.md

@@ -16,6 +16,7 @@ Below is a list of supported processors with links to their documentation pages.
 | Log DeDuplication Processor             | [logdeduplicationprocessor](../processor/logdeduplicationprocessor/README.md) |
 | Logs Transform Processor                | [logstransform](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.108.0/processor/logstransformprocessor/README.md) |
 | Lookup Processor                        | [lookupprocessor](../processor/lookupprocessor/README.md) |
+| Marshal Processor                          | [marshalprocessor](../processor/marshalprocessor/README.md) |
 | Mask Processor                          | [maskprocessor](../processor/maskprocessor/README.md) |
 | Memory Limiter Processor                | [memorylimiterprocessor](https://github.com/open-telemetry/opentelemetry-collector/blob/v0.108.0/processor/memorylimiterprocessor/README.md) |
 | Metric Extract Processor                | [metricextract](../processor/metricextractprocessor/README.md) |

factories/processors.go

@@ -20,6 +20,7 @@ import (
 	"github.com/observiq/bindplane-agent/processor/logcountprocessor"
 	"github.com/observiq/bindplane-agent/processor/logdeduplicationprocessor"
 	"github.com/observiq/bindplane-agent/processor/lookupprocessor"
+	"github.com/observiq/bindplane-agent/processor/marshalprocessor"
 	"github.com/observiq/bindplane-agent/processor/maskprocessor"
 	"github.com/observiq/bindplane-agent/processor/metricextractprocessor"
 	"github.com/observiq/bindplane-agent/processor/metricstatsprocessor"
@@ -66,6 +67,7 @@ var defaultProcessors = []processor.Factory{
 	logdeduplicationprocessor.NewFactory(),
 	logstransformprocessor.NewFactory(),
 	lookupprocessor.NewFactory(),
+	marshalprocessor.NewFactory(),
 	maskprocessor.NewFactory(),
 	memorylimiterprocessor.NewFactory(),
 	metricextractprocessor.NewFactory(),

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/observiq/bindplane-agent/processor/logcountprocessor v1.60.0
 	github.com/observiq/bindplane-agent/processor/logdeduplicationprocessor v1.60.0
 	github.com/observiq/bindplane-agent/processor/lookupprocessor v1.60.0
+	github.com/observiq/bindplane-agent/processor/marshalprocessor v1.60.0
 	github.com/observiq/bindplane-agent/processor/maskprocessor v1.60.0
 	github.com/observiq/bindplane-agent/processor/metricextractprocessor v1.60.0
 	github.com/observiq/bindplane-agent/processor/metricstatsprocessor v1.60.0
@@ -797,6 +798,8 @@ replace github.com/observiq/bindplane-agent/processor/throughputmeasurementproce
 
 replace github.com/observiq/bindplane-agent/processor/samplingprocessor => ./processor/samplingprocessor
 
+replace github.com/observiq/bindplane-agent/processor/marshalprocessor => ./processor/marshalprocessor
+
 replace github.com/observiq/bindplane-agent/processor/maskprocessor => ./processor/maskprocessor
 
 replace github.com/observiq/bindplane-agent/processor/logcountprocessor => ./processor/logcountprocessor

processor/marshalprocessor/README.md

@@ -0,0 +1,192 @@
+# Marshal Processor
+
+This processor is used to marshal parsed logs into JSON or KV format.
+
+This processor is intended to be wrapped into the Marshal processor in Bindplane.
+
+NOTE: XML support is in progress and not yet available.
+
+## Supported pipelines
+
+- Logs
+
+## How It Works
+
+1. This processor expects its input to contain a parsed log body. It will marshal the body fields only; if additional fields from the log are desired, they must first be moved to the body.
+
+2. The body can be marshaled to string-encoded JSON or KV.
+
+   - For KV:
+     - Fields will be converted to "key1=value1 key2=value2 key3=value3..." if no separators are configured
+     - The parsed fields should be flattened first so that every key is at the top level
+     - If fields are not flattened, the nested fields will be converted to "nested=[k1=v1,k2=v2]..." if no map separators are configured
+     - If any key or value contains characters that conflict with the separators, they will be wrapped in `"` and any `"` inside them will be escaped
+     - Arrays will simply be stringified
+
+3. The output of this processor will be the same as the input, but with a modified log body. Any body incompatible with the marshal type will be unchanged.
+
+## Configuration
+
+| Field                 | Type   | Default | Description                                   |
+| --------------------- | ------ | ------- | --------------------------------------------- |
+| marshal_to            | string | ""      | The format to marshal into. Can be JSON or KV |
+| kv_separator          | rune   | "="     | The separator between key and value           |
+| kv_pair_separator     | rune   | " "     | The separator between KV pairs                |
+| map_kv_separator      | rune   | "="     | The separator between nested KV pairs         |
+| map_kv_pair_separator | rune   | ","     | The separator between nested KV pairs         |
+
+## Example Config for JSON
+
+```yaml
+receivers:
+  otlp:
+processors:
+  transform:
+  marshal:
+    marshal_to: "JSON"
+exporters:
+  chronicle:
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [transform, marshal]
+      exporters: [chronicle]
+```
+
+## Example config for KV
+
+```yaml
+receivers:
+  otlp:
+processors:
+  transform:
+  marshal:
+    marshal_to: "KV"
+    kv_separator: ","
+    kv_pair_separator: ":"
+exporters:
+  chronicle:
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [transform, marshal]
+      exporters: [chronicle]
+```
+
+## Example Output
+
+The parsed body will be replaced by a marshaled body. All other fields are untouched.
+
+### 1. Nested parsed body to JSON
+
+In the example below, "bindplane-otel-attributes" represents attributes that have been moved to the body.
+
+#### Parsed body
+
+```
+"body": {
+    "kvlistValue": {
+        "values": [
+            { "key": "severity", "value": { "doubleValue": 155 } },
+            {
+                "key": "nested",
+                "value": {
+                    "kvlistValue": {
+                        "values": [
+                            { "key": "n2", "value": { "doubleValue": 2 } },
+                            { "key": "n1", "value": { "doubleValue": 1 } }
+                        ]
+                    }
+                }
+            },
+            { "key": "name", "value": { "stringValue": "test" } },
+            {
+                "key": "bindplane-otel-attributes",
+                "value": {
+                    "kvlistValue": {
+                        "values": [
+                            {
+                                "key": "baba",
+                                "value": { "stringValue": "you" }
+                            },
+                            {
+                                "key": "host",
+                                "value": { "stringValue": "myhost" }
+                            }
+                        ]
+                    }
+                }
+            }
+        ]
+    }
+},
+
+```
+
+#### JSON output
+
+```
+"body": {
+    "stringValue": {
+        {
+            "bindplane-otel-attributes":
+                {
+                    "baba":"you",
+                    "host":"myhost"
+                },
+            "name":"test",
+            "nested":
+                {
+                    "n1":1,
+                    "n2":2
+                },
+            "severity":155
+        }
+    }
+}
+```
+
+### 2: Flattened parsed body to KV with default separators
+
+In the example below, flattening has already been done on the "nested" field and the "bindplane-otel-attributes" field.
+
+#### Parsed flattened body
+
+```
+"body": {
+    "kvlistValue": {
+        "values": [
+            { "key": "severity", "value": { "doubleValue": 155 } },
+            {
+                "key": "nested-n1",
+                "value": { "doubleValue": 1 }
+            },
+            {
+                "key": "nested-n2",
+                "value": { "doubleValue": 2 }
+            },
+            { "key": "name", "value": { "stringValue": "test" } },
+            {
+                "key": "bindplane-otel-attributes-baba",
+                "value": { "stringValue": "you" }
+            },
+            {
+                "key": "bindplane-otel-attributes-host",
+                "value": { "stringValue": "myhost" }
+            }
+        ]
+    }
+}
+```
+
+#### KV output
+
+```
+"body": {
+    "stringValue": {
+        bindplane-otel-attributes-baba=you bindplane-otel-attributes-host=myhost name=test nested-n1=1 nested-n2=2 severity=155
+    }
+}
+```

processor/marshalprocessor/config.go

@@ -0,0 +1,82 @@
+// Copyright  observIQ, Inc.
+//
+// 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 marshalprocessor provides a processor that marshals logs to a specified format.
+package marshalprocessor
+
+import (
+	"errors"
+	"strings"
+
+	"go.opentelemetry.io/collector/component"
+	"go.uber.org/multierr"
+)
+
+var errInvalidMarshalTo = errors.New("marshal_to must be JSON or KV")
+var errXMLNotSupported = errors.New("XML not yet supported")
+var errKVSeparatorsEqual = errors.New("kv_separator and kv_pair_separator must be different")
+var errMapKVSeparatorsEqual = errors.New("map_kv_separator and map_kv_pair_separator must be different")
+
+const (
+	defaultMarshalTo          = "JSON"
+	defaultKVSeparator        = '='
+	defaultKVPairSeparator    = ' '
+	defaultMapKVSeparator     = '='
+	defaultMapKVPairSeparator = ','
+)
+
+// Config is the configuration for the processor
+type Config struct {
+	MarshalTo          string `mapstructure:"marshal_to"` // MarshalTo is either JSON or KV
+	KVSeparator        rune   `mapstructure:"kv_separator"`
+	KVPairSeparator    rune   `mapstructure:"kv_pair_separator"`
+	MapKVSeparator     rune   `mapstructure:"map_kv_separator"`
+	MapKVPairSeparator rune   `mapstructure:"map_kv_pair_separator"`
+}
+
+// Validate validates the processor configuration
+func (cfg Config) Validate() error {
+	var errs error
+
+	// Validate MarshalTo choice
+	switch strings.ToUpper(cfg.MarshalTo) {
+	case "JSON":
+	case "XML":
+		errs = multierr.Append(errs, errXMLNotSupported)
+	case "KV":
+		// Validate KV separators, which must be different from each other
+		if cfg.KVSeparator == cfg.KVPairSeparator && cfg.KVSeparator != 0 {
+			errs = multierr.Append(errs, errKVSeparatorsEqual)
+		}
+		// Validate Map KV separators, which must be different from each other (but can match KV separators)
+		if cfg.MapKVSeparator == cfg.MapKVPairSeparator && cfg.MapKVSeparator != 0 {
+			errs = multierr.Append(errs, errMapKVSeparatorsEqual)
+		}
+	default:
+		errs = multierr.Append(errs, errInvalidMarshalTo)
+	}
+
+	return errs
+}
+
+// createDefaultConfig returns the default config for the processor.
+func createDefaultConfig() component.Config {
+	return &Config{
+		MarshalTo:          defaultMarshalTo,
+		KVSeparator:        defaultKVSeparator,
+		KVPairSeparator:    defaultKVPairSeparator,
+		MapKVSeparator:     defaultMapKVSeparator,
+		MapKVPairSeparator: defaultMapKVPairSeparator,
+	}
+}