pkg/logs/processor: use custom marshaler for json payload message

Change jsonPayload Message field type to use encoding.TextMarshaler/TextUnmarshaler
to avoid allocation due to conversion of valid UTF-8 bytes to string in the common case.

```
goos: darwin
goarch: arm64
pkg: github.com/DataDog/datadog-agent/pkg/logs/processor
cpu: Apple M4 Max
                              │    HEAD~1    │                HEAD                 │
                              │    sec/op    │    sec/op     vs base               │
JSONEncoder_Encode/valid-16     563.1n ± 17%   543.7n ± 12%  -3.45% (p=0.009 n=10)
JSONEncoder_Encode/invalid-16   567.0n ±  1%   550.0n ±  1%  -3.00% (p=0.000 n=10)
geomean                         565.0n         546.8n        -3.22%

                              │    HEAD~1    │                HEAD                 │
                              │     B/op     │     B/op      vs base               │
JSONEncoder_Encode/valid-16     1.127Ki ± 0%   1.017Ki ± 0%  -9.79% (p=0.000 n=10)
JSONEncoder_Encode/invalid-16   1.127Ki ± 0%   1.017Ki ± 0%  -9.79% (p=0.000 n=10)
geomean                         1.127Ki        1.017Ki       -9.79%

                              │   HEAD~1    │                HEAD                │
                              │  allocs/op  │ allocs/op   vs base                │
JSONEncoder_Encode/valid-16     10.000 ± 0%   9.000 ± 0%  -10.00% (p=0.000 n=10)
JSONEncoder_Encode/invalid-16   10.000 ± 0%   9.000 ± 0%  -10.00% (p=0.000 n=10)
geomean                          10.00        9.000       -10.00%
```

Author: AlexanderYastrebov

cmd/agent/subcommands/analyzelogs/command_test.go

@@ -154,7 +154,7 @@ Auto-discovery IDs:
 		err := json.Unmarshal(msg.GetContent(), &parsedMessage)
 		assert.NoError(t, err)
 
-		assert.Equal(t, parsedMessage.Message, expectedOutput[i])
+		assert.Equal(t, parsedMessage.Message.String(), expectedOutput[i])
 	}
 
 	launcher.Stop()
@@ -178,7 +178,7 @@ func TestRunAnalyzeLogsInvalidConfig(t *testing.T) {
       - type: exclude_at_match
         name: exclude_random
         pattern: "datadog-agent"
-      
+
 `, tempLogFile.Name())
 	tempConfigFile := CreateTestFile(tempDir, "config.yaml", invalidConfig)
 	assert.NotNil(t, tempConfigFile)

pkg/logs/processor/encoder.go

@@ -7,6 +7,7 @@
 package processor
 
 import (
+	"bytes"
 	"strings"
 	"unicode/utf8"
 
@@ -18,6 +19,36 @@ type Encoder interface {
 	Encode(msg *message.Message, hostname string) error
 }
 
+type ValidUtf8Bytes []byte
+
+func (msg ValidUtf8Bytes) MarshalText() (text []byte, err error) {
+	if utf8.Valid(msg) {
+		return msg, nil
+	}
+
+	var buf bytes.Buffer
+	buf.Grow(len(msg))
+
+	for len(msg) > 0 {
+		r, size := utf8.DecodeRune(msg)
+		// in case of invalid utf-8, DecodeRune returns (utf8.RuneError, 1)
+		// and since RuneError is the same as unicode.ReplacementChar
+		// no need to handle the error explicitly
+		buf.WriteRune(r)
+		msg = msg[size:]
+	}
+	return buf.Bytes(), nil
+}
+
+func (msg *ValidUtf8Bytes) UnmarshalText(text []byte) error {
+	*msg = bytes.Clone(text)
+	return nil
+}
+
+func (msg ValidUtf8Bytes) String() string {
+	return string(msg)
+}
+
 // toValidUtf8 ensures all characters are UTF-8.
 func toValidUtf8(msg []byte) string {
 	if utf8.Valid(msg) {

pkg/logs/processor/encoder_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/DataDog/agent-payload/v5/pb"
+
 	"github.com/DataDog/datadog-agent/comp/logs/agent/config"
 	"github.com/DataDog/datadog-agent/pkg/logs/message"
 	"github.com/DataDog/datadog-agent/pkg/logs/sources"

pkg/logs/processor/json.go

@@ -27,13 +27,13 @@ type jsonEncoder struct{}
 
 // JSON representation of a message.
 type jsonPayload struct {
-	Message   string `json:"message"`
-	Status    string `json:"status"`
-	Timestamp int64  `json:"timestamp"`
-	Hostname  string `json:"hostname"`
-	Service   string `json:"service"`
-	Source    string `json:"ddsource"`
-	Tags      string `json:"ddtags"`
+	Message   ValidUtf8Bytes `json:"message"`
+	Status    string         `json:"status"`
+	Timestamp int64          `json:"timestamp"`
+	Hostname  string         `json:"hostname"`
+	Service   string         `json:"service"`
+	Source    string         `json:"ddsource"`
+	Tags      string         `json:"ddtags"`
 }
 
 // Encode encodes a message into a JSON byte array.
@@ -48,7 +48,7 @@ func (j *jsonEncoder) Encode(msg *message.Message, hostname string) error {
 	}
 
 	encoded, err := json.Marshal(jsonPayload{
-		Message:   toValidUtf8(msg.GetContent()),
+		Message:   ValidUtf8Bytes(msg.GetContent()),
 		Status:    msg.GetStatus(),
 		Timestamp: ts.UnixNano() / nanoToMillis,
 		Hostname:  hostname,

pkg/logs/processor/json_serverless_init.go

@@ -26,13 +26,13 @@ type jsonServerlessInitEncoder struct {
 
 // JSON representation of a message for serverless-init.
 type jsonServerlessInitPayload struct {
-	Message   string `json:"message"`
-	Status    string `json:"status"`
-	Timestamp int64  `json:"timestamp"`
-	Hostname  string `json:"hostname"`
-	Service   string `json:"service,omitempty"`
-	Source    string `json:"ddsource"`
-	Tags      string `json:"ddtags"`
+	Message   ValidUtf8Bytes `json:"message"`
+	Status    string         `json:"status"`
+	Timestamp int64          `json:"timestamp"`
+	Hostname  string         `json:"hostname"`
+	Service   string         `json:"service,omitempty"`
+	Source    string         `json:"ddsource"`
+	Tags      string         `json:"ddtags"`
 }
 
 // Encode encodes a message into a JSON byte array.
@@ -52,7 +52,7 @@ func (j *jsonServerlessInitEncoder) Encode(msg *message.Message, hostname string
 	}
 
 	encoded, err := json.Marshal(jsonServerlessInitPayload{
-		Message:   toValidUtf8(msg.GetContent()),
+		Message:   ValidUtf8Bytes(msg.GetContent()),
 		Status:    msg.GetStatus(),
 		Timestamp: ts.UnixNano() / nanoToMillis,
 		Hostname:  hostname,