Support cross-region SNS topic subscription

Author: kaorimatz

app/controllers/barbeque/sns_subscriptions_controller.rb

@@ -44,6 +44,12 @@ def destroy
   private
 
   def fetch_sns_topic_arns
-    Barbeque::SNSSubscriptionService.sns_client.list_topics.topics.map(&:topic_arn)
+    if Barbeque.config.sns_regions.empty?
+      Aws::SNS::Client.new.list_topics.topics.map(&:topic_arn)
+    else
+      Barbeque.config.sns_regions.flat_map do |region|
+        Aws::SNS::Client.new(region: region).list_topics.topics.map(&:topic_arn)
+      end
+    end
   end
 end

app/models/barbeque/sns_subscription.rb

@@ -7,5 +7,9 @@ class SNSSubscription < ApplicationRecord
     validates :topic_arn,
       uniqueness: { scope: :job_queue, message: 'should be set with only one queue' },
       presence: true
+
+    def region
+      /\Aarn:aws:sns:([a-z0-9-]+):/.match(topic_arn)[1]
+    end
   end
 end

app/services/barbeque/sns_subscription_service.rb

@@ -6,10 +6,6 @@ def self.sqs_client
     @sqs_client ||= Aws::SQS::Client.new
   end
 
-  def self.sns_client
-    @sns_client ||= Aws::SNS::Client.new
-  end
-
   # @param [Barbeque::SNSSubscription] sns_subscription
   # @return [Boolean] `true` if succeeded to subscribe
   def subscribe(sns_subscription)
@@ -45,10 +41,6 @@ def sqs_client
     Barbeque::SNSSubscriptionService.sqs_client
   end
 
-  def sns_client
-    Barbeque::SNSSubscriptionService.sns_client
-  end
-
   # @param [Barbeque::SNSSubscription] sns_subscription
   def update_sqs_policy!(sns_subscription)
     attrs = sqs_client.get_queue_attributes(
@@ -98,6 +90,7 @@ def subscribe_topic!(sns_subscription)
     )
     queue_arn = sqs_attrs.attributes['QueueArn']
 
+    sns_client = Aws::SNS::Client.new(region: sns_subscription.region)
     sns_client.subscribe(
       topic_arn: sns_subscription.topic_arn,
       protocol: 'sqs',
@@ -113,6 +106,8 @@ def unsubscribe_topic!(sns_subscription)
     )
     queue_arn = sqs_attrs.attributes['QueueArn']
 
+    sns_client = Aws::SNS::Client.new(region: sns_subscription.region)
+
     subscriptions = sns_client.list_subscriptions_by_topic(
       topic_arn: sns_subscription.topic_arn,
     )

lib/barbeque/config.rb

@@ -3,7 +3,7 @@
 
 module Barbeque
   class Config
-    attr_accessor :exception_handler, :executor, :executor_options, :sqs_receive_message_wait_time, :maximum_concurrent_executions, :runner_wait_seconds
+    attr_accessor :exception_handler, :executor, :executor_options, :sqs_receive_message_wait_time, :maximum_concurrent_executions, :runner_wait_seconds, :sns_regions
 
     def initialize(options = {})
       options.each do |key, value|
@@ -27,6 +27,7 @@ module ConfigBuilder
       # nil means unlimited
       'maximum_concurrent_executions' => nil,
       'runner_wait_seconds' => 10,
+      'sns_regions' => [],
     }
 
     def config

spec/controllers/barbeque/job_definitions_controller_spec.rb

@@ -191,8 +191,8 @@
       let(:subscription_arn) { 'arn:aws:sns:ap-northeast-1:012345678912:barbeque-spec:01234567-89ab-cdef-0123-456789abcdef' }
 
       before do
-        allow(Barbeque::SNSSubscriptionService).to receive(:sns_client).and_return(sns_client)
         allow(Barbeque::SNSSubscriptionService).to receive(:sqs_client).and_return(sqs_client)
+        allow(Aws::SNS::Client).to receive(:new).with(region: 'ap-northeast-1').and_return(sns_client)
 
         allow(sqs_client).to receive(:get_queue_attributes).
           with(queue_url: sns_subscription.job_queue.queue_url, attribute_names: ['QueueArn']).

spec/controllers/barbeque/sns_subscriptions_controller_spec.rb

@@ -5,7 +5,7 @@
 
   before do
     allow(Barbeque::SNSSubscriptionService).to receive(:sqs_client).and_return(sqs_client)
-    allow(Barbeque::SNSSubscriptionService).to receive(:sns_client).and_return(sns_client)
+    allow(Aws::SNS::Client).to receive(:new).with(region: 'ap-northeast-1').and_return(sns_client)
   end
 
   describe '#create' do

spec/factories/sns_subscription.rb

@@ -1,6 +1,6 @@
 FactoryBot.define do
   factory :sns_subscription, class: Barbeque::SNSSubscription do
-    sequence(:topic_arn) { |n| "arn:aws:sns:ap-northest-1:123456789012/Topic-#{n}" }
+    sequence(:topic_arn) { |n| "arn:aws:sns:ap-northeast-1:123456789012/Topic-#{n}" }
     job_queue
     job_definition
   end