[AWSINTS-3450] feat(go-forwarder): add CloudwatchLogEntry creation logic by ndakkoune · Pull Request #1094 · DataDog/datadog-serverless-functions

ndakkoune · 2026-04-01T16:06:08Z

What does this PR do?

Add (Cloudwatch)LogEntry creation logic (source detection, overriding tags, etc.)

Motivation

Testing Guidelines

Any for now, quick feedback before writing the tests.

Additional Notes

Except from the CONTROL_MESSAGE skipping, the behavior is roughly the same as the Python one.

Types of changes

Bug fix
New feature
Breaking change
Misc (docs, refactoring, dependency upgrade, etc.)

Check all that apply

This PR's description is comprehensive
This PR contains breaking changes that are documented in the description
This PR introduces new APIs or parameters that are documented and unlikely to change in the foreseeable future
This PR impacts documentation, and it has been updated (or a ticket has been logged)
This PR's changes are covered by the automated tests
This PR collects user input/sensitive content into Datadog
This PR passes the integration tests (ask a Datadog member to run the tests)
This PR passes the unit tests
This PR passes the installation tests (ask a Datadog member to run the tests)

ndakkoune · 2026-04-01T16:09:15Z

+	if data.MessageType == "CONTROL_MESSAGE" {
+		return
+	}


Not present in the Python implementation. the control message are probing messages from cloudwatch to verify that the lambda is reachable => noise.

ViBiOh · 2026-04-02T08:15:14Z

+	"github.com/aws/aws-lambda-go/lambdacontext"
+)
+
+func parseCloudwatchLogs(ctx context.Context, event json.RawMessage, cfg *config.Config, out chan<- model.CloudwatchLogEntry) {


💬 suggestion: ‏The naming of the function is a bit confusing. Usually a parser is a "pure" function, you give an input, you get an output and an error.

Here, the error are logged directly and we output to the chan. I think we should make this function "pure" (easier to test, because currently untested) and let the Parse function that call it deal with the error and forwarding to the chan

Edit: I see below that we have the log entry in memory and we stream them, so maybe just an intermediate function handleCloudwatchLogs that read the parse, and then stream

Great. I kept the channel approach with only one goroutine for cloudwatch. The S3 Handler will probably trigger more goroutines and leverage streaming.

ViBiOh · 2026-04-02T08:17:49Z

+		select {
+		case out <- entry:
+		case <-ctx.Done():
+			return
+		}


🥜 nitpick: ‏While it "would" work as expected, it's not deterministic. Have a look at the internal repository, there is a SafeSender method you can copy/paste in an util package here

ViBiOh · 2026-04-02T08:24:10Z

+	if strings.HasPrefix(logStream, "states/") {
+		source = "stepfunction"
+	}


💬 suggestion: ‏ You can move this after the source override and change to a definitive return (doing a fast-exit)

ViBiOh · 2026-04-02T08:25:51Z

+	if lc, ok := lambdacontext.FromContext(ctx); ok {
+		metadata.InvokedFunctionARN = lc.InvokedFunctionArn
+	} else {
+		slog.Warn("failed lambda context loading")


❓ question: ‏What's the actionable thing to do there? We should either enrich with some field or remove the log

This one is a safety net. The only ways the else case is hit is (1) when the code is not ran from AWS Lambda (e.g. testing) or (2) someone passes a plain context.Background(). It should NEVER be hit in production, and if it does, the default string value "" will be sent in the backend (we would notice it with a status:skipped).

You should change the logged message to this explanation.

My point is: if a customer see that in 6 months, the log should explain the situation clearly.

ndakkoune · 2026-04-09T14:07:37Z

+	"strings"
+)
+
+type Tags []string


This little workaround is necessary due to constraints in the backend side.

json.MarshalJSON converts a slice to a JSON array (expected behavior), but the backend needs a string of comma-separated key:value pairs, hence this custom MarshalJSON.

…dler return the error and update main with source detection + pipeline trigger

ViBiOh · 2026-04-13T09:39:40Z

+		case parsing.InvocationSourceCloudwatchLogs:
+			return pipeline.Run(ctx, event, cfg, parsing.HandleCloudwatchLogs)
+		default:
+			slog.Error("unsupported invocation source")


⚠️ issue: ‏ You should store the dectected invocation source as a variable and put in the log. In current scheme, the log is not actionable / hard du debug.

ViBiOh · 2026-04-13T09:45:07Z

+					return err
+				}
+				if !ok {
+					return nil


⚠️ issue: ‏There is a concurrency issue in this function.

We have 3 workers. In any worker encounter an error, it exits and stop, stopping to dequeue from the input channel.

If the 3 workers fail, the in chan will be full, nothing will dequeue it, introducing a deadlock for the upstream goroutine.

The only call of Forward is here but the ctx is not managed by the errgroup.Group. So if the goroutine fails, there is no signal sent to other goroutine to stop.

We should fix the context cancellation in the pipeline file and maybe leave a TODO here to handle the retry mechanism (later in the migration).

ViBiOh · 2026-04-13T09:48:25Z

+	if lc, ok := lambdacontext.FromContext(ctx); ok {
+		metadata.InvokedFunctionARN = lc.InvokedFunctionArn
+	} else {
+		slog.Warn("failed lambda context loading")


You should change the logged message to this explanation.

My point is: if a customer see that in 6 months, the log should explain the situation clearly.

ViBiOh · 2026-04-14T12:42:41Z

+func HandleCloudwatchLogs(ctx context.Context, event json.RawMessage, cfg *config.Config, out chan<- model.CloudwatchLogEntry) error {
+	logEntries, err := parseCloudwatchLogs(ctx, event, cfg)
+	if err != nil {
+		slog.Error("failed parse cloudwatch logs", slog.Any("error", err))


❓ question: ‏We don't return after the parsing error. Should we return the error and let the caller log it, or return nil.

The logEntries below will very likely to be nil/empty, so we will return nil in the end. I think we should return the error. If we can't parse the cloudwatch message, it's very likely to be a broad failure and not pretend everything was fine if you don't look at the logs.

I'm going for the return err.

ViBiOh · 2026-04-14T12:50:23Z

+	var source string
+	if strings.Contains(logStream, "_CloudTrail_") {
+		source = "cloudtrail"
+	} else {
+		source = getSourceFromLogGroup(strings.ToLower(logGroup))
+	}
+	return source


Same here regarding "fast-exit", if you have your answer, exit the function directly. This avoid nesting instruction and give a clear execution path of the function.

ViBiOh · 2026-04-14T12:56:52Z

 )

-type invocationSource int
+//go:generate stringer -type InvocationSource -trimprefix InvocationSource


👏 praise: ‏ 👍

ViBiOh · 2026-04-14T12:57:47Z

+	var service string
+
+	if cfg.CustomTags != "" {
+		for _, tag := range strings.Split(cfg.CustomTags, ",") {


I think your code editor will suggest to use SplitSeq

ViBiOh · 2026-04-14T12:59:18Z

+	cfg *config.Config,
+	handler func(context.Context, json.RawMessage, *config.Config, chan<- T) error,
+) error {
+	g, ctx := errgroup.WithContext(ctx) // Fragile because if one goroutine fails, all stages will stop gracefully


📝 note: ‏ I don't think it's "fragile" in this case. It's the graceful shutdown management and ensure that there won't be any deadlock of one goroutine quitting listening to a chan , leaving the others stuck

I wrote this comment as a self memory note for the future, where we probably won stop all the goroutines if, for example, a forwarding worker fails.

Co-authored-by: Vincent Boutour <vincent.boutour@datadoghq.com>

…1106)

ndakkoune · 2026-04-15T11:53:54Z

-				if service != "" {
-					continue
+				if service == "" {
+					service = tag[8:]


More explicit with constants

ndakkoune requested a review from a team as a code owner April 1, 2026 16:06

github-actions bot added the aws label Apr 1, 2026

ndakkoune commented Apr 1, 2026

View reviewed changes

ViBiOh self-assigned this Apr 2, 2026

ViBiOh reviewed Apr 2, 2026

View reviewed changes

ndakkoune requested a review from ViBiOh April 2, 2026 15:54

ndakkoune force-pushed the nabil.dakkoune/AWSINTS-3450 branch from 173fefa to 26ba416 Compare April 8, 2026 09:04

ndakkoune commented Apr 9, 2026

View reviewed changes

ndakkoune force-pushed the nabil.dakkoune/AWSINTS-3450 branch from f7ac191 to 493ec22 Compare April 10, 2026 16:01

ndakkoune added 6 commits April 13, 2026 10:27

[AWSINTS-3450] feat(go-forwarder): add CloudwatchLogEntry creation logic

363def9

Add SafeXXX in concurrent.go, pipeline.go with generics, make the Han…

62971ce

…dler return the error and update main with source detection + pipeline trigger

propagate error

87f74ed

refactor: add Tags type to prevent Unmarshal to create an array

1757e45

add unmarshal

d760674

concurrent forawrding + update pipeline

4dbd0e9

ndakkoune force-pushed the nabil.dakkoune/AWSINTS-3450 branch from dd5ed1c to 4dbd0e9 Compare April 13, 2026 08:29

ndakkoune added 2 commits April 13, 2026 10:31

refactor(go-forwarder): Tags struct to marshall properly (#1105)

8795b6b

Use pointer field, but value struct.

c4d9c81

ViBiOh reviewed Apr 13, 2026

View reviewed changes

update with vincent comments

d05ffdb

ndakkoune requested a review from ViBiOh April 14, 2026 09:08

ViBiOh reviewed Apr 14, 2026

View reviewed changes

ndakkoune and others added 4 commits April 14, 2026 15:31

Update aws/logs_monitoring_go/internal/parsing/cloudwatchlogs.go

9dd1b5a

Co-authored-by: Vincent Boutour <vincent.boutour@datadoghq.com>

Update aws/logs_monitoring_go/internal/parsing/cloudwatchlogs.go

d543cfa

Co-authored-by: Vincent Boutour <vincent.boutour@datadoghq.com>

splitseq and error message clarification

5349a82

[AWSINTS-3493] fix(go-forwarder): add ddtags extraction from message (#…

aba33a6

…1106)

ndakkoune requested a review from ViBiOh April 14, 2026 14:36

fix missing dd-storage-tag header

d9eb163

ndakkoune commented Apr 15, 2026

View reviewed changes

ViBiOh approved these changes Apr 15, 2026

View reviewed changes

ndakkoune merged commit 0ffcb57 into nabil.dakkoune/go-forwarder Apr 15, 2026
10 checks passed

ndakkoune deleted the nabil.dakkoune/AWSINTS-3450 branch April 15, 2026 12:00

Conversation

ndakkoune commented Apr 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What does this PR do?

Motivation

Testing Guidelines

Additional Notes

Types of changes

Check all that apply

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ndakkoune Apr 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ndakkoune commented Apr 1, 2026 •

edited

Loading

ndakkoune Apr 2, 2026 •

edited

Loading