feat (pkg/stanza): add sanitize operator to replace invalid UTF-8 cha… (open-telemetry#42029)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

WIth `utf-8-raw` encoding, there may be invalid UTF-8 characters in the
final output, which breaks compatibility.

So let's add an operator to sanitize invalid UTF-8 characters, then
users can use it after recombine operator. For example:

```yaml
encoding: utf-8-raw
operators:
  - type: container
  - type: recombine
    combine_field: body
    is_first_entry: body matches "^[0-9]+-[0-9]+-[0-9]+"
    max_log_size: 128kb
    combine_with: "\n"
    source_identifier: attributes["log.file.path"]
  - type: sanitize # newly operator to sanitize invalid UTF-8 characters.
    field: body
```

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#42028 

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Added unit tests, and tested in our k8s cluter.

<!--Describe the documentation added.-->
#### Documentation
Added docs files in `pkg/stanza/docs/operators`

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Andrzej Stencel <[email protected]>

Author: h0cheung

.chloggen/stanza_sanitize.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add `sanitize_utf8` operator to replace invalid UTF-8 characters.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42028]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

pkg/stanza/adapter/register.go

@@ -30,5 +30,6 @@ import (
 	_ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/remove"
 	_ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/retain"
 	_ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/router"
+	_ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/sanitizeutf8"
 	_ "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/unquote"
 )

pkg/stanza/docs/operators/README.md

@@ -45,5 +45,6 @@ General purpose:
 - [remove](./remove.md)
 - [retain](./retain.md)
 - [router](./router.md)
+- [sanitize_utf8](./sanitize_utf8.md)
 - [unquote](./unquote.md)
 - [assign_keys](./assign_keys.md)

pkg/stanza/docs/operators/sanitize_utf8.md

@@ -0,0 +1,25 @@
+## `sanitize_utf8` Operator
+
+The `sanitize_utf8` operator is used to replace invalid UTF-8 characters in a specified field of a log entry. This is useful
+for ensuring that log data is properly encoded and can be processed by downstream systems.
+
+Invalid UTF-8 byte sequences will be replaced with `\uFFFD` (`�`).
+
+### Configuration Fields
+
+| Field      | Default          | Description                                                                                                                                                                                                                           |
+|------------|------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `id`       | `sanitize_utf8`  | A unique identifier for the operator.                                                                                                                                                                                                 |
+| `output`   | Next in pipeline | The connected operator(s) that will receive all outbound entries.                                                                                                                                                                     |
+| `field`    | `body`           | The [field](../types/field.md) to sanitize. This must be a string field.                                                                                                                                                              |
+| `on_error` | `send`           | The behavior of the operator if it encounters an error. See [on_error](../types/on_error.md).                                                                                                                                         |
+| `if`       |                  | An [expression](../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 Configurations
+
+#### Simple Configuration for String Body
+
+```yaml
+- type: sanitize_utf8
+  field: body
+```

pkg/stanza/operator/transformer/sanitizeutf8/config_test.go

@@ -0,0 +1,39 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizeutf8
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/operatortest"
+)
+
+func TestUnmarshal(t *testing.T) {
+	operatortest.ConfigUnmarshalTests{
+		DefaultConfig: NewConfig(),
+		TestsFile:     filepath.Join(".", "testdata", "config.yaml"),
+		Tests: []operatortest.ConfigUnmarshalTest{
+			{
+				Name:               "default",
+				ExpectUnmarshalErr: false,
+				Expect:             NewConfig(),
+			},
+			{
+				Name:               "sanitize_utf8_with_field",
+				ExpectUnmarshalErr: false,
+				Expect: func() *Config {
+					cfg := NewConfig()
+					cfg.Field = entry.NewBodyField()
+					return cfg
+				}(),
+			},
+			{
+				Name:               "sanitize_utf8_with_invalid_field",
+				ExpectUnmarshalErr: true,
+			},
+		},
+	}.Run(t)
+}

pkg/stanza/operator/transformer/sanitizeutf8/package_test.go

@@ -0,0 +1,14 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizeutf8
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}

pkg/stanza/operator/transformer/sanitizeutf8/testdata/config.yaml

@@ -0,0 +1,8 @@
+default:
+  type: sanitize_utf8
+sanitize_utf8_with_field:
+  type: sanitize_utf8
+  field: body
+sanitize_utf8_with_invalid_field:
+  type: sanitize_utf8
+  field: invalid_field

pkg/stanza/operator/transformer/sanitizeutf8/transformer.go

@@ -0,0 +1,90 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizeutf8 // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/transformer/sanitizeutf8"
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"unicode/utf8"
+
+	"go.opentelemetry.io/collector/component"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
+)
+
+const (
+	operatorType           = "sanitize_utf8"
+	invalidCharacterMarker = string(utf8.RuneError) // Unicode replacement character
+)
+
+func init() {
+	operator.Register(operatorType, func() operator.Builder { return NewConfig() })
+}
+
+// NewConfig creates a new sanitize_utf8 config with default values
+func NewConfig() *Config {
+	return NewConfigWithID(operatorType)
+}
+
+// NewConfigWithID creates a new sanitize_utf8 config with default values
+func NewConfigWithID(operatorID string) *Config {
+	return &Config{
+		TransformerConfig: helper.NewTransformerConfig(operatorID, operatorType),
+		Field:             entry.NewBodyField(),
+	}
+}
+
+// Config is the configuration of a sanitize_utf8 operator
+type Config struct {
+	helper.TransformerConfig `mapstructure:",squash"`
+	Field                    entry.Field `mapstructure:"field"`
+}
+
+// Build creates a new Transformer from a config
+func (c *Config) Build(set component.TelemetrySettings) (operator.Operator, error) {
+	transformer, err := c.TransformerConfig.Build(set)
+	if err != nil {
+		return nil, fmt.Errorf("failed to build transformer config: %w", err)
+	}
+
+	return &Transformer{
+		TransformerOperator: transformer,
+		field:               c.Field,
+	}, nil
+}
+
+// Transformer is an operator that replaces invalid UTF-8 characters in a field
+type Transformer struct {
+	helper.TransformerOperator
+	field entry.Field
+}
+
+func (t *Transformer) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
+	return t.ProcessBatchWithTransform(ctx, entries, t.ProcessWith)
+}
+
+func (t *Transformer) Process(ctx context.Context, e *entry.Entry) error {
+	return t.TransformerOperator.ProcessWith(ctx, e, t.ProcessWith)
+}
+
+func (t *Transformer) ProcessWith(e *entry.Entry) error {
+	v, ok := t.field.Get(e)
+	if !ok {
+		return nil
+	}
+	s, ok := v.(string)
+	if !ok {
+		return fmt.Errorf("field '%s' is not a string, got %T", t.field.String(), v)
+	}
+	if utf8.ValidString(s) {
+		return nil
+	}
+	if err := t.field.Set(e, strings.ToValidUTF8(s, invalidCharacterMarker)); err != nil {
+		return fmt.Errorf("failed to set field %s: %w", t.field.String(), err)
+	}
+	return nil
+}

pkg/stanza/operator/transformer/sanitizeutf8/transformer_test.go

@@ -0,0 +1,161 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sanitizeutf8
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/testutil"
+)
+
+type processTestCase struct {
+	name   string
+	op     *Config
+	input  func() *entry.Entry
+	output func() *entry.Entry
+}
+
+func TestSanitizeUTF8(t *testing.T) {
+	now := time.Now()
+	cases := []processTestCase{
+		{
+			name: "normal_string_body",
+			op:   newCfgWithField("body"),
+			input: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is a normal string"
+				e.ObservedTimestamp = now
+				return e
+			},
+			output: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is a normal string"
+				e.ObservedTimestamp = now
+				return e
+			},
+		},
+		{
+			name: "invalid_utf8_body",
+			op:   newCfgWithField("body"),
+			input: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is an invalid utf8 string \xfe"
+				e.ObservedTimestamp = now
+				return e
+			},
+			output: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is an invalid utf8 string \uFFFD"
+				e.ObservedTimestamp = now
+				return e
+			},
+		},
+		{
+			name: "consecutive_invalid_utf8_body",
+			op:   newCfgWithField("body"),
+			input: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is an invalid utf8 string \xfe\xfe"
+				e.ObservedTimestamp = now
+				return e
+			},
+			output: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is an invalid utf8 string \uFFFD"
+				e.ObservedTimestamp = now
+				return e
+			},
+		},
+		{
+			name: "unconsecutive_invalid_utf8_body",
+			op:   newCfgWithField("body"),
+			input: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is an invalid utf8 string \xfe and another \xfe"
+				e.ObservedTimestamp = now
+				return e
+			},
+			output: func() *entry.Entry {
+				e := entry.New()
+				e.Body = "This is an invalid utf8 string \uFFFD and another \uFFFD"
+				e.ObservedTimestamp = now
+				return e
+			},
+		},
+		{
+			name: "normal_string_attribute",
+			op:   newCfgWithField("attributes.foo"),
+			input: func() *entry.Entry {
+				e := entry.New()
+				e.Attributes = map[string]any{
+					"foo": "This is a normal string",
+				}
+				e.ObservedTimestamp = now
+				return e
+			},
+			output: func() *entry.Entry {
+				e := entry.New()
+				e.Attributes = map[string]any{
+					"foo": "This is a normal string",
+				}
+				e.ObservedTimestamp = now
+				return e
+			},
+		},
+		{
+			name: "invalid_utf8_attribute",
+			op:   newCfgWithField("attributes.foo"),
+			input: func() *entry.Entry {
+				e := entry.New()
+				e.Attributes = map[string]any{
+					"foo": "This is an invalid utf8 string \xfe",
+				}
+				e.ObservedTimestamp = now
+				return e
+			},
+			output: func() *entry.Entry {
+				e := entry.New()
+				e.Attributes = map[string]any{
+					"foo": "This is an invalid utf8 string \uFFFD",
+				}
+				e.ObservedTimestamp = now
+				return e
+			},
+		},
+	}
+	for _, tc := range cases {
+		t.Run("BuildandProcess/"+tc.name, func(t *testing.T) {
+			cfg := tc.op
+			cfg.OutputIDs = []string{"fake"}
+			cfg.OnError = "drop"
+			op, err := cfg.Build(componenttest.NewNopTelemetrySettings())
+			require.NoError(t, err)
+
+			sanitizeUTF8 := op.(*Transformer)
+			fake := testutil.NewFakeOutput(t)
+			require.NoError(t, sanitizeUTF8.SetOutputs([]operator.Operator{fake}))
+			val := tc.input()
+			err = sanitizeUTF8.Process(t.Context(), val)
+			require.NoError(t, err)
+			fake.ExpectEntry(t, tc.output())
+		})
+	}
+}
+
+func newCfgWithField(field string) *Config {
+	entryField, err := entry.NewField(field)
+	if err != nil {
+		panic(fmt.Sprintf("failed to create field %s: %v", field, err))
+	}
+	config := NewConfigWithID("sanitize_utf8/test")
+	config.Field = entryField
+	return config
+}