fix(go-forwarder): add scanner to handle delimiters

Author: ndakkoune

aws/logs_monitoring_go/internal/config/config.go

@@ -113,6 +113,6 @@ func (c Config) LogValue() slog.Value {
 		slog.Bool("customScrubbing", c.Scrubbing.CustomRule != ""),
 		slog.Bool("includeFilter", c.Filtering.IncludePattern != ""),
 		slog.Bool("excludeFilter", c.Filtering.ExcludePattern != ""),
-		slog.Bool("multilineRegex", c.S3MultilineLogRegex.String() != ""),
+		slog.Bool("multilineRegex", c.S3MultilineLogRegex != nil),
 	)
 }

aws/logs_monitoring_go/internal/model/cloudwatchlogs.go

@@ -18,7 +18,7 @@ type CloudwatchLogEntry struct {
 }
 
 type CloudwatchMetadata struct {
-	ForwarderMetadata
+	Metadata
 	Logs CloudwatchLogsContext `json:"awslogs"`
 }
 

aws/logs_monitoring_go/internal/model/forwarder.go

@@ -15,7 +15,7 @@ type S3LogEntry struct {
 }
 
 type S3Metadata struct {
-	ForwarderMetadata
+	Metadata
 	S3Context S3Context `json:"s3"`
 }
 

aws/logs_monitoring_go/internal/model/s3.go

@@ -21,13 +21,13 @@ import (
 )
 
 type s3RecordContext struct {
-	forwarderMetadata model.ForwarderMetadata
-	tags              model.Tags
-	source            string
-	service           string
-	bucket            string
-	key               string
-	multilineRegex    *regexp.Regexp
+	metadata       model.Metadata
+	tags           model.Tags
+	source         string
+	service        string
+	bucket         string
+	key            string
+	multilineRegex *regexp.Regexp
 }
 
 func HandleS3(ctx context.Context, event json.RawMessage, cfg *config.Config, out chan<- model.S3LogEntry) error {
@@ -41,13 +41,16 @@ func HandleS3(ctx context.Context, event json.RawMessage, cfg *config.Config, ou
 		return fmt.Errorf("create S3 client: %w", err)
 	}
 
-	forwarderMetadata := model.GetForwarderMetadata(ctx)
+	forwarderMetadata, err := model.GetMetadata(ctx)
+	if err != nil {
+		return err
+	}
 
 	for _, record := range s3Event.Records {
 		bucket := record.S3.Bucket.Name
 		key := record.S3.Object.URLDecodedKey
 
-		tags, service := getTagsAndService(*cfg)
+		tags, service := getTagsAndService(cfg)
 		source := getS3Source(cfg.Source, key)
 		if service == "" {
 			service = source
@@ -78,8 +81,9 @@ func processS3Record(ctx context.Context, client S3APIClient, out chan<- model.S
 
 	var buf strings.Builder
 	scanner := bufio.NewScanner(body)
+	scanner.Split(split)
 	for scanner.Scan() {
-		line := scanner.Text()
+		line := strings.ToValidUTF8(scanner.Text(), "")
 		if rc.multilineRegex != nil {
 			if rc.multilineRegex.MatchString(line) {
 				if buf.Len() > 0 {
@@ -124,7 +128,7 @@ func makeS3Entry(rc s3RecordContext, message string) model.S3LogEntry {
 
 	ddtags = append(ddtags, "service:"+entryService)
 	metadata := model.S3Metadata{
-		ForwarderMetadata: rc.forwarderMetadata,
+		Metadata: rc.metadata,
 		S3Context: model.S3Context{
 			Bucket: rc.bucket,
 			Key:    rc.key,

aws/logs_monitoring_go/internal/parsing/s3.go

@@ -0,0 +1,28 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-Present Datadog, Inc.
+
+package parsing
+
+import "bytes"
+
+func split(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+
+	if i := bytes.IndexAny(data, "\n\r\f"); i >= 0 {
+		j := i
+		for j < len(data) && (data[j] == '\n' || data[j] == '\r' || data[j] == '\f') {
+			j++
+		}
+		return j, data[:i], nil
+	}
+
+	if atEOF {
+		return len(data), data, nil
+	}
+
+	return 0, nil, nil
+}

aws/logs_monitoring_go/internal/parsing/scanner.go

@@ -0,0 +1,56 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2026-Present Datadog, Inc.
+
+package parsing
+
+import (
+	"bufio"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestSplit(t *testing.T) {
+	t.Parallel()
+
+	tests := map[string]struct {
+		input string
+		want  []string
+	}{
+		"empty_string":      {input: "", want: nil},
+		"plain_string":      {input: "hello", want: []string{"hello"}},
+		"new_lines":         {input: "a\nb\nc", want: []string{"a", "b", "c"}},
+		"trailing_new_line": {input: "a\nb\n", want: []string{"a", "b"}},
+		"crlf":              {input: "a\r\nb\r\nc", want: []string{"a", "b", "c"}},
+		"form_feed":         {input: "a\fb\fc", want: []string{"a", "b", "c"}},
+		"mixed_delimiters":  {input: "a\r\n\fb", want: []string{"a", "b"}},
+		"only_delimiters":   {input: "\n\r\f", want: nil},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			scanner := bufio.NewScanner(strings.NewReader(tc.input))
+			scanner.Split(split)
+
+			var got []string
+			for scanner.Scan() {
+				if text := scanner.Text(); text != "" {
+					got = append(got, text)
+				}
+			}
+
+			if err := scanner.Err(); err != nil {
+				t.Fatalf("unexpected scanner error: %v", err)
+			}
+
+			if diff := cmp.Diff(tc.want, got); diff != "" {
+				t.Errorf("mismatch (-want +got):\n%s", diff)
+			}
+		})
+	}
+}