Fanout support for AWS SQS

[rlaunch]

Hello,

I can see on in the transport comparison table that fan out is supported for SQS when using SimpleDB. I'm unable to find any further documentation on how to configure this other than to set the 'supports_fanout' parameter to True in the transport_options.

with Connection(
    "sqs://",
    transport_options={"supports_fanout": True}
) as conn:
    pass

Unfortunately, this does not work for me when using a Broadcast queue type, whereas it does in RabbitMQ. Is fanout still supported for SQS? I looked at using a topic exchange, too but have not been able to get this to work either.

If this is not supported, then I may be able to provide a PR to implement this functionality using AWS SNS instead of SimpleDB, as it's how our home-grown messaging library does this at the moment.

Any help is much appreciated,
Ryan

[auvipy]

@akosel can you weigh in, plz?

[rlaunch]

Apologies, just seen this issue #871, so it would seem that the docs are just out of date and the functionality has indeed been removed.

In that case, would a PR be welcomed that re-added this support, but using AWS SNS with AWS SQS queues as a subscribed target?

Best,
Ryan

[akosel]

SNS does seem like the right tool for implementing fanout with SQS.

Do you have a high-level plan for implementing it? It might help to talk through, just so we can clarify anything before putting in all the work!

[rlaunch]

Hi @akosel,

My plan to implement this would be:

• When a fanout exchange is defined:
  • Check if an existing AWS SNS topic is defined with the name of the exchange used (+ queue prefix if set)
  • If not present, create an AWS SNS topic with the name
  • Probably as a cleanup action, we should look over all existing subscriptions on the SNS topic and remove any that go to blackholes/SQS queues that no longer exist
• When a queue is defined using the fanout exchange
  • Check whether the created AWS SQS queue is already subscribed to the previously created SNS topic
  • If not, then subscribe the SQS queue to the SNS topic
• When a message is sent, we send the serialised message to SNS which handles the fanout to the subscribed SQS queues
• When the queue deletes, we remove the subscription from the SNS topic first

Best,
Ryan

[akosel]

Okay, awesome. That sounds great to me.

In my opinion, a PR is welcome, assuming no objections from @auvipy.

My extra thoughts/questions for your consideration:

• Cleanup timing: I'm unsure when we should scrub dead subscriptions (startup, periodic job, or every queue op). Maybe a startup is the best time, but I'm curious if you have thought about this.
• Race safety: Make the topic/subscribe calls idempotent. Multiple workers booting at once shouldn't collide/crash.
• Cost & quotas: I believe fanout means N copies per message, so just flag the extra spend/quota hit in the docs.
• IAM scope: To avoid mystery 403s, it might be helpful to spell out the exact perms needed in the docs (create/delete topic, subscribe/unsubscribe, etc.). However, I don't think we have this for SQS, so it's more of a nice-to-have.
• Logging: Log every topic/sub create/delete so ops can trace weirdness fast.

[rlaunch]

Hi @akosel,

Thanks for your feedback, that all sounds good. Please see responses inline below:

Cleanup timing: I'm unsure when we should scrub dead subscriptions (startup, periodic job, or every queue op). Maybe a startup is the best time, but I'm curious if you have thought about this.
Currently, in our home-grown module, we handle cleanup in two ways:

• On application shutdown, each application instance unsubscribes it's SQS queue from the SNS topic
• We have a scheduled task to look at the SNS topics, and gets a list of their subscriptions. We then go and check these to see if they are valid, and if not we remove them.

The second method catches any instances where the first may not have cleaned up correctly. I think it would be good to add this to the Kombu implementation; however, it's probably safer and less clunky to just have the cleanup run at every startup, rather than a scheduled task. We could expose a method to allow the cleanup to be called more regularly too.

I'm not cleaning up SNS topics at the moment, but we could do this by tagging the SNS topic with something like 'ManagedBy'='Kombu' so we can identify it as one the library has created. We could then clean this up if it has no subscriptions.

Race safety: Make the topic/subscribe calls idempotent. Multiple workers booting at once shouldn't collide/crash.

In my tests so far, SNS topic creation is pretty safe from this, as calling create topic multiple times just returns the existing topic without error. Other cleanup actions would carry a risk of races. Without any other external systems, my best idea to prevent this is just to handle errors gracefully and let AWS APIs handle their state. My opinion is a cleanup operation error should be handled and not raise an exception to the wider codebase, as it's not a critical issue. What are your thoughts on this?

Cost & quotas: I believe fanout means N copies per message, so just flag the extra spend/quota hit in the docs.

Yeah, the message would be sent to SNS instead of SQS, which would then place a copy of each message on each subscribed queue. According to the SNS pricing page, delivering messages to SQS is free. It's just the SQS service you pay for, which is mostly costed around API usage. I'll ensure to document this in the PR.

IAM scope: To avoid mystery 403s, it might be helpful to spell out the exact perms needed in the docs (create/delete topic, subscribe/unsubscribe, etc.). However, I don't think we have this for SQS, so it's more of a nice-to-have.

Sure thing, no problem.

Logging: Log every topic/sub create/delete so ops can trace weirdness fast.

No problem. Logging message delivery between SNS and SQS would have to be done with SNS delivery status, but as you say, we can log each time a topic or subscription is added or removed.

Best,
Ryan

[auvipy]

you can come with a draft PR

[akosel]

My opinion is a cleanup operation error should be handled and not raise an exception to the wider codebase, as it's not a critical issue. What are your thoughts on this?

I agree with that, and everything else you said makes sense to me.

[rlaunch]

Hey @akosel and @auvipy ,

I've created a PR (#2339) for my first pass at the changes.
If you're happy with the changes, I'll get the tests written and update the remaining documentation.

Any feedback welcome,
Best,
Ryan