SQS Source wrap up

[jlambatl]

Question

There are many questions related to the SQS source, and since I am currently investigating SQS, I figured it was worth consolidating them into a single summary question.

At the moment, we have several very busy SQS queues, with many billions of events per day. vector is handling this with ease (awesome work to everyone who's worked on the source/sink ❤️ )

Concurrency

The existing concurrency model has some room for improvement. When the SQS queue is very busy, all of the nodes' concurrency is running well, and we may see a slight increase in the max message age. However, when the message rate is slower, the number of consumers per host remains constant. This results in the CloudWatch metric NumberOfEmptyReceives increasing. When many nodes have a high concurrency level configured, this can result in unnecessary costs and wasted compute.

I've had a look at how this is handled in other sources or sinks, and there are several approaches. The current implementation of the SQS sink uses an adaptive concurrency model that appears to be based on round-trip time. A similar approach could be used to dynamically adjust the concurrency based on the message attribute SentTimestamp, which is already being extracted in get_timestamp().

A MessageAgeMonitor would need to be implemented to dynamically track and manage concurrency. The message age thresholds could be configured in the source, with a max concurrency value that defaults to the number of cores or a configured value. An alternative to strictly configured values is to use an exponentially weighted moving average of message ages. This is probably a more user-friendly approach, but it may need a broader discussion, as there are many possible approaches.

Another common approach across sinks is the common::backoff::ExponentialBackoff crate/module/library. Implementing this would be less work; however, I wanted to reach out to discuss and see what the maintainers prefer.

Message attributes

There are many different discussions and issues that cover this; it seems to be something that isn't outright rejected and is looking for someone to implement it. While I am sure that many people overload the use of attributes, one use case that is useful is encoding the source request_id or similar application metadata in the attributes and leaving the original message intact. Sometimes this is used for routing decisions because the message doesn't need to be deserialised or parsed.

Has anyone put any time into thinking about a preferred implementation for handling message attributes? My naive thinking has been to use the dropped_fields ObjectMap in the EventMetadata, but from reading issues and comments, that might not be the right choice.

SQS DLQ

SQS supports DLQ, in the scenario where receiving messages from external "untrusted" sources is possible, messages may not be well formed, or may have become corrupted (rare). Has there been any discussion about adding DLQ support for messages that can't be processed correctly the first x times?

These are items that we are in a position to resolve through sending in a PR, particularly the concurrency item as it's currently an item showing up in our aws bills. These would be submitted as separate smaller PR's based on whatever path forward can be agreed to (or not).

Thanks in advance,

James.

Vector Config

No response

Vector Logs

No response

[pront]

Thanks, James. Separate issues and PRs make sense for these changes.

• Concurrency: Let's continue the design discussion in #13322. Your workload and cost measurements would help. The design should cover both keeping up with busy queues and reducing empty receives when traffic drops; message age alone cannot guide scaling down when the queue is empty.
• Message attributes: #18877 tracks this request. Source metadata accessible from VRL seems like a suitable approach, with behavior defined for both log namespaces. dropped_fields serves a different purpose: retaining values removed from events by semantic meaning.
• DLQ: SQS redrive policies can move repeatedly received messages to a DLQ, provided Vector leaves them undeleted. I tested Vector 0.60.0 against a local mock SQS endpoint and found that a message rejected by JSON decoding was still deleted, even with acknowledgements enabled. Please open a separate issue for this case so we can discuss how decoding failures should be handled.

Please don't hesitate to followup in this thread if you have further questions.

[jlambatl]

@pront, thanks for the response. I've taken the opportunity to submit a PR for the adaptive batching similar to GCP, and will review the message attributes items when I have more time to think about what the implementation would need to look like.