feat(go-forwarder): add SQS as failed event storage

[ndakkoune]

This PR adds SQS as a failed event storage.

A refactor was needed on storing.Storage, which previously was not abstract enough to support diverse storage services. A new storing.Batch struct has been created to represent failed batches from the forwarder. Batch.DeleteHandle is a unique identifier of the underlying stored batch mechanism (key for S3, receiptHandler for SQS), used to delete the batch if the retry succeeds.

Diverging behaviors from the Python behavior are :

• Using sqs.SendMessageBatch, a batch equivalent of sqs.SendMessage to minimize API calls on Store().
• retry_prefix which would equals to either logs, metrics or traces, and stored as a message attribute is deleted since we know only deals with logs.
• function_prefix which would equal to the sha1 of the forwarder's arn and stored as a message attribute is deleted. It was used to differentiate between message sent from the forwarder, from other the consumers. Sharing the same SQS queue between unrelated use-cases is a bad design and would highly enhance API costs by receiving noisy messages.

Side notes :

• We now have DD_STORE_FAILED_EVENTS, DD_S3_BUCKET_NAME and DD_SQS_QUEUE_URL which wraps the same logic around the retry mechanism: storing failed event. Imo, we should only keep one of these (since is one or the other), or a new DD_FAILED_EVENT_STORAGE which would directly take either an S3 bucket name or an SQS queue URL, if such identification is feasible.

[datadog-official]

✨ Fix all issues with BitsAI

⚠️ Warnings

🚦 1 Pipeline job failed

Go Forwarder | test     

Useful? React with 👍 / 👎

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: 4d9ae86 | Docs | Datadog PR Page | Give us feedback!}

[ndakkoune]

The maximum SQS message size is 1MiB.
However, we have to take all the attributes of sqs.SendMessageInput into account, so we're forced to keep an overhead.

[ge0Aja]

discussed IRL let's try the two channel approach

[ge0Aja]

isn't the key known to the getter ? why are we returning it with the batch ?

[ndakkoune]

We would have to return either return (body io.Reader, storageTag string, err error) or manually add the storedBatch.DeleteKey=... inside the loop.
Since we are forced to encapsulate the GetObject call into a function (since we have to defer.Close() the out.Body), I found it more simpler to encapsulate the whole batching logic inside getBatch instead to additionally set the DeleteKey inside the loop.

[ndakkoune]

I had three solution to reuse the Batcher.Start() method :

1. Keep the old functional option pattern and change the in channel from model.LogEntry to any. Too much restrictive: we would lose type safety and change the Handler signature.
2. Keep the old functional option pattern, make the Batcher and the Option both generic. Too much verbose, e.g. batching.WithMaxItemSize[model.LogEntry](1*1024*1024) three times and it does not provide any value since only the Start method leverage genericity.
3. Delete the old functional option pattern, use a config struct directly and make the batcher generic.

Opted for 3.

[ge0Aja]

lgtm i trust you on the comments