[AWSINTS-3506] feat(go-forwarder): add S3 pipeline (with multiline regex)

aws/logs_monitoring_go/cmd/forwarder/main.go

@@ -11,6 +11,7 @@ import (
 	"log/slog"
 
 	"github.com/DataDog/datadog-serverless-functions/aws/logs_monitoring_go/internal/config"
+	"github.com/DataDog/datadog-serverless-functions/aws/logs_monitoring_go/internal/forwarding"
 	"github.com/DataDog/datadog-serverless-functions/aws/logs_monitoring_go/internal/parsing"
 	"github.com/DataDog/datadog-serverless-functions/aws/logs_monitoring_go/internal/pipeline"
 
@@ -33,7 +34,9 @@ func handleRequest(cfg *config.Config) func(context.Context, json.RawMessage) er
 		invocationSource := parsing.DetectInvocationSource(event)
 		switch invocationSource {
 		case parsing.InvocationSourceCloudwatchLogs:
-			return pipeline.Run(ctx, event, cfg, parsing.HandleCloudwatchLogs)
+			return pipeline.Run(ctx, event, cfg, forwarding.CloudwatchStorage, parsing.HandleCloudwatchLogs)
+		case parsing.InvocationSourceS3:
+			return pipeline.Run(ctx, event, cfg, forwarding.S3Storage, parsing.HandleS3)
 		default:
 			slog.Error("unsupported invocation source", slog.String("source", invocationSource.String()))
 			return nil

aws/logs_monitoring_go/go.mod

@@ -11,14 +11,19 @@ require (
 )
 
 require (
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.12 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.20 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.20 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 // indirect
 	github.com/aws/aws-sdk-go-v2/service/kms v1.50.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.99.0 // indirect
 	github.com/aws/aws-sdk-go-v2/service/signin v1.0.8 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssm v1.68.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.30.13 // indirect

aws/logs_monitoring_go/go.sum

@@ -4,6 +4,8 @@ github.com/aws/aws-sdk-go-v2 v1.41.4 h1:10f50G7WyU02T56ox1wWXq+zTX9I1zxG46HYuG1h
 github.com/aws/aws-sdk-go-v2 v1.41.4/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o=
 github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY=
 github.com/aws/aws-sdk-go-v2 v1.41.5/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 h1:eBMB84YGghSocM7PsjmmPffTa+1FBUeNvGvFou6V/4o=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8/go.mod h1:lyw7GFp3qENLh7kwzf7iMzAxDn+NzjXEAGjKS2UOKqI=
 github.com/aws/aws-sdk-go-v2/config v1.32.12 h1:O3csC7HUGn2895eNrLytOJQdoL2xyJy0iYXhoZ1OmP0=
 github.com/aws/aws-sdk-go-v2/config v1.32.12/go.mod h1:96zTvoOFR4FURjI+/5wY1vc1ABceROO4lWgWJuxgy0g=
 github.com/aws/aws-sdk-go-v2/credentials v1.19.12 h1:oqtA6v+y5fZg//tcTWahyN9PEn5eDU/Wpvc2+kJ4aY8=
@@ -20,12 +22,22 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 h1:PEgGVtPoB6NTpPrBgq
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21/go.mod h1:p+hz+PRAYlY3zcpJhPwXlLC4C+kqn70WIHwnzAfs6ps=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 h1:qYQ4pzQ2Oz6WpQ8T3HvGHnZydA72MnLuFK9tJwmrbHw=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6/go.mod h1:O3h0IK87yXci+kg6flUKzJnWeziQUKciKrLjcatSNcY=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 h1:rWyie/PxDRIdhNf4DzRk0lvjVOqFJuNnO8WwaIRVxzQ=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22/go.mod h1:zd/JsJ4P7oGfUhXn1VyLqaRZwPmZwg44Jf2dS84Dm3Y=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 h1:5EniKhLZe4xzL7a+fU3C2tfUN4nWIqlLesfrjkuPFTY=
 github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7/go.mod h1:x0nZssQ3qZSnIcePWLvcoFisRXJzcTVvYpAAdYX8+GI=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 h1:JRaIgADQS/U6uXDqlPiefP32yXTda7Kqfx+LgspooZM=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13/go.mod h1:CEuVn5WqOMilYl+tbccq8+N2ieCy0gVn3OtRb0vBNNM=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.20 h1:2HvVAIq+YqgGotK6EkMf+KIEqTISmTYh5zLpYyeTo1Y=
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.20/go.mod h1:V4X406Y666khGa8ghKmphma/7C0DAtEQYhkq9z4vpbk=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3xgIJMSC8S6hEVq+38DcvUlgFY0FM6mSI5oto=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21/go.mod h1:r6+pf23ouCB718FUxaqzZdbpYFyDtehyZcmP5KL9FkA=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 h1:ZlvrNcHSFFWURB8avufQq9gFsheUgjVD9536obIknfM=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21/go.mod h1:cv3TNhVrssKR0O/xxLJVRfd2oazSnZnkUeTf6ctUwfQ=
 github.com/aws/aws-sdk-go-v2/service/kms v1.50.4 h1:PgD1y0ZagPokGIZPmejCBUySBzOFDN+leZxCOfb1OEQ=
 github.com/aws/aws-sdk-go-v2/service/kms v1.50.4/go.mod h1:FfXDb5nXrsoGgxsBFxwxr3vdHXheC2tV+6lmuLghhjQ=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.99.0 h1:hlSuz394kV0vhv9drL5lhuEFbEOEP1VyQpy15qWh1Pk=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.99.0/go.mod h1:uoA43SdFwacedBfSgfFSjjCvYe8aYBS7EnU5GZ/YKMM=
 github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.41.4 h1:9aZbO86sraeCIHHCpZhxwN9tnVy9POkSKzi4/TpT54A=
 github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.41.4/go.mod h1:cxiXDhEzIq7Xx1BtmC4lGBK3SwAZ79+EUWiKawYHo14=
 github.com/aws/aws-sdk-go-v2/service/signin v1.0.8 h1:0GFOLzEbOyZABS3PhYfBIx2rNBACYcKty+XGkTgw1ow=

aws/logs_monitoring_go/internal/config/config.go

@@ -9,22 +9,24 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"regexp"
 )
 
 var ForwarderVersion = "6.0"
 
 type Config struct {
-	APIKey     string
-	Site       string
-	IntakeURL  string
-	APIURL     string
-	LogLevel   string
-	UseFIPS    bool
-	Source     string
-	Host       string
-	CustomTags string
-	Scrubbing  ScrubbingConfig
-	Filtering  FilteringConfig
+	APIKey              string
+	Site                string
+	IntakeURL           string
+	APIURL              string
+	LogLevel            string
+	UseFIPS             bool
+	Source              string
+	Host                string
+	CustomTags          string
+	Scrubbing           ScrubbingConfig
+	Filtering           FilteringConfig
+	S3MultilineLogRegex *regexp.Regexp
 }
 
 type ScrubbingConfig struct {
@@ -58,7 +60,9 @@ func Load(ctx context.Context) (*Config, error) {
 }
 
 func loadConfig() *Config {
+	S3MultilineLogRegex := loadS3MultilineLogRegex()
 	site := envOrDefault("DD_SITE", "datadoghq.com")
+
 	return &Config{
 		Site:       site,
 		IntakeURL:  envOrDefault("DD_URL", "https://http-intake.logs."+site+"/api/v2/logs"),
@@ -78,7 +82,23 @@ func loadConfig() *Config {
 			IncludePattern: envOrDefault("INCLUDE_AT_MATCH", ""),
 			ExcludePattern: envOrDefault("EXCLUDE_AT_MATCH", ""),
 		},
+		S3MultilineLogRegex: S3MultilineLogRegex,
+	}
+}
+
+func loadS3MultilineLogRegex() *regexp.Regexp {
+	pattern := envOrDefault("DD_MULTILINE_LOG_REGEX_PATTERN", "")
+	if pattern == "" {
+		return nil
 	}
+
+	re, err := regexp.Compile(pattern)
+	if err != nil {
+		slog.Error("invalid multiline log pattern", slog.String("pattern", pattern), slog.Any("error", err))
+		return nil
+	}
+
+	return re
 }
 
 func (c Config) LogValue() slog.Value {
@@ -93,5 +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 != nil),
 	)
 }

aws/logs_monitoring_go/internal/config/config_test.go

@@ -120,3 +120,43 @@ func TestLoadConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestLoadS3MultilineLogRegex(t *testing.T) {
+	tests := map[string]struct {
+		env     string
+		wantNil bool
+	}{
+		"empty_pattern_returns_nil": {
+			env:     "",
+			wantNil: true,
+		},
+		"valid_pattern_returns_compiled_regex": {
+			env: `\d{4}-\d{2}-\d{2}`,
+		},
+		"invalid_pattern_returns_nil": {
+			env:     `[invalid`,
+			wantNil: true,
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			if tc.env != "" {
+				t.Setenv("DD_MULTILINE_LOG_REGEX_PATTERN", tc.env)
+			}
+
+			got := loadS3MultilineLogRegex()
+
+			if tc.wantNil {
+				if got != nil {
+					t.Fatalf("want nil, got `%v", got)
+				}
+				return
+			}
+
+			if got == nil {
+				t.Fatal("want non-nil, got nil")
+			}
+		})
+	}
+}

aws/logs_monitoring_go/internal/forwarding/forwarding.go

@@ -19,17 +19,23 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-const numWorkers = 3
+const (
+	numWorkers        = 3
+	CloudwatchStorage = "cloudwatch"
+	S3Storage         = "s3"
+)
 
 type Forwarder struct {
-	config *config.Config
-	client *http.Client
+	config  *config.Config
+	client  *http.Client
+	storage string
 }
 
-func NewForwarder(config *config.Config, client *http.Client) Forwarder {
+func NewForwarder(config *config.Config, client *http.Client, storage string) Forwarder {
 	return Forwarder{
-		config: config,
-		client: client,
+		config:  config,
+		client:  client,
+		storage: storage,
 	}
 }
 
@@ -78,7 +84,7 @@ func (f Forwarder) send(ctx context.Context, body []byte) error {
 	req.Header.Set("Content-Encoding", "gzip")
 	req.Header.Set("DD-EVP-ORIGIN", "aws_forwarder")
 	req.Header.Set("DD-EVP-ORIGIN-VERSION", config.ForwarderVersion)
-	req.Header.Set("DD-STORAGE-TAG", "cloudwatch")
+	req.Header.Set("DD-STORAGE-TAG", f.storage)
 
 	resp, err := f.client.Do(req)
 	if err != nil {

aws/logs_monitoring_go/internal/forwarding/forwarding_test.go

@@ -23,6 +23,7 @@ func TestForward(t *testing.T) {
 
 	tests := map[string]struct {
 		statusCode int
+		storage    string
 		payloads   [][]byte
 		cancelCtx  bool
 		wantErr    bool
@@ -31,40 +32,52 @@ func TestForward(t *testing.T) {
 	}{
 		"single_message_accepted": {
 			statusCode: http.StatusAccepted,
+			storage:    CloudwatchStorage,
 			payloads:   [][]byte{[]byte("test payload")},
 			wantCalls:  1,
 		},
 		"multiple_messages_accepted": {
 			statusCode: http.StatusAccepted,
+			storage:    CloudwatchStorage,
 			payloads:   [][]byte{[]byte("first"), []byte("second"), []byte("third")},
 			wantCalls:  3,
 		},
 		"empty_channel": {
 			statusCode: http.StatusAccepted,
+			storage:    CloudwatchStorage,
 			payloads:   [][]byte{},
 			wantCalls:  0,
 		},
 		"server_returns_400": {
 			statusCode: http.StatusBadRequest,
+			storage:    CloudwatchStorage,
 			payloads:   [][]byte{[]byte("test payload")},
 			wantErr:    true,
 			wantErrMsg: "unexpected status from intake",
 			wantCalls:  1,
 		},
 		"server_returns_500": {
 			statusCode: http.StatusInternalServerError,
+			storage:    CloudwatchStorage,
 			payloads:   [][]byte{[]byte("test payload")},
 			wantErr:    true,
 			wantErrMsg: "unexpected status from intake",
 			wantCalls:  1,
 		},
 		"context_cancelled": {
 			statusCode: http.StatusAccepted,
+			storage:    CloudwatchStorage,
 			payloads:   [][]byte{[]byte("test payload")},
 			cancelCtx:  true,
 			wantErr:    true,
 			wantCalls:  0,
 		},
+		"s3_storage": {
+			statusCode: http.StatusAccepted,
+			storage:    S3Storage,
+			payloads:   [][]byte{[]byte("test payload")},
+			wantCalls:  1,
+		},
 	}
 
 	for name, tc := range tests {
@@ -91,6 +104,9 @@ func TestForward(t *testing.T) {
 				if got := req.Header.Get("DD-EVP-ORIGIN-VERSION"); got != config.ForwarderVersion {
 					t.Errorf("DD-EVP-ORIGIN-VERSION = %q, want %q", got, config.ForwarderVersion)
 				}
+				if got := req.Header.Get("DD-STORAGE-TAG"); got != tc.storage {
+					t.Errorf("DD-STORAGE-TAG = %q, want %q", got, tc.storage)
+				}
 
 				gr, err := gzip.NewReader(req.Body)
 				if err != nil {
@@ -111,10 +127,7 @@ func TestForward(t *testing.T) {
 			}))
 			defer server.Close()
 
-			f := NewForwarder(&config.Config{
-				IntakeURL: server.URL,
-				APIKey:    "test-api-key",
-			}, server.Client())
+			f := NewForwarder(&config.Config{IntakeURL: server.URL, APIKey: "test-api-key"}, server.Client(), tc.storage)
 
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()

aws/logs_monitoring_go/internal/model/s3.go

@@ -0,0 +1,25 @@
+// 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 model
+
+type S3LogEntry struct {
+	Message        string     `json:"message"`
+	Source         string     `json:"ddsource"`
+	SourceCategory string     `json:"ddsourcecategory"`
+	Service        string     `json:"service"`
+	Tags           Tags       `json:"ddtags"`
+	Metadata       S3Metadata `json:"aws"`
+}
+
+type S3Metadata struct {
+	Metadata
+	S3Context S3Context `json:"s3"`
+}
+
+type S3Context struct {
+	Bucket string `json:"bucket"`
+	Key    string `json:"key"`
+}