Resolve Credentials issues and queue URL issues

README.rdoc

@@ -21,25 +21,23 @@ Read events from Amazon SQS.
 
       type sqs
 
-      # url as https://sqs.us-west-2.amazonaws.com/123456789012/myqueue
-      sqs_url {queue_url}
-
-      # following attribute is required if you don't declare a sqs_url
-      queue_name {queue_instance_key}
+      # following attribute is required
+      queue_name {queue_name}
 
       # following attributes are required if you don't use IAM Role nor export credentials to ENV
 
       aws_key_id {your_aws_key_id}
       aws_sec_key {your_aws_secret_key}
 
+      # following attributes are required if you use FIFO queue
+
+      message_group_id {string}
+
       # following attributes are optional
 
       create_queue {boolean}
-      region {your_region}
-
-      # following attributes are required if you use FIFO queue
 
-      message_group_id {string}
+      region {your_region}
 
       ### region list ###
       # Asia Pacific (Tokyo) [Default] : ap-northeast-1
@@ -53,6 +51,7 @@ Read events from Amazon SQS.
       delay_seconds {delivery_delay_seconds}
 
       include_tag {boolean}
+
       tag_property_name {tag's property name in json}
 
 
@@ -66,7 +65,7 @@ Read events from Amazon SQS.
 
       # following attribute is required
 
-      sqs_url {queue_url}
+      queue_name {queue_name}
 
       # following attributes are required if you don't use IAM Role nor export credentials to ENV
 

lib/fluent/plugin/in_sqs.rb

@@ -11,7 +11,8 @@ class SQSInput < Input
     config_param :aws_sec_key, :string, default: nil, secret: true
     config_param :tag, :string
     config_param :region, :string, default: 'ap-northeast-1'
-    config_param :sqs_url, :string, default: nil
+    config_param :queue_name, :string, default: nil
+    config_param :queue_owner_aws_account_id, :string, default: nil
     config_param :receive_interval, :time, default: 0.1
     config_param :max_number_of_messages, :integer, default: 10
     config_param :wait_time_seconds, :integer, default: 10
@@ -21,12 +22,6 @@ class SQSInput < Input
 
     def configure(conf)
       super
-
-      Aws.config = {
-        access_key_id: @aws_key_id,
-        secret_access_key: @aws_sec_key,
-        region: @region
-      }
     end
 
     def start
@@ -36,11 +31,19 @@ def start
     end
 
     def client
-      @client ||= Aws::SQS::Client.new(stub_responses: @stub_responses)
+      @client ||= Aws::SQS::Client.new(
+        access_key_id: @aws_key_id,
+        secret_access_key: @aws_sec_key,
+        region: @region,
+        stub_responses: @stub_responses
+      )
     end
 
     def queue
-      @queue ||= Aws::SQS::Resource.new(client: client).queue(@sqs_url)
+      @queue ||= Aws::SQS::Queue.new(
+        url: client.get_queue_url(queue_name: @queue_name, queue_owner_aws_account_id: @queue_owner_aws_account_id).queue_url,
+        client: client
+      )
     end
 
     def shutdown

lib/fluent/plugin/out_sqs.rb

@@ -18,7 +18,6 @@ class SQSOutput < Output
     config_param :aws_key_id, :string, default: nil, secret: true
     config_param :aws_sec_key, :string, default: nil, secret: true
     config_param :queue_name, :string, default: nil
-    config_param :sqs_url, :string, default: nil
     config_param :create_queue, :bool, default: true
     config_param :region, :string, default: 'ap-northeast-1'
     config_param :delay_seconds, :integer, default: 0
@@ -34,38 +33,37 @@ def configure(conf)
       compat_parameters_convert(conf, :buffer, :inject)
       super
 
-      if (!@queue_name.nil? && @queue_name.end_with?('.fifo')) || (!@sqs_url.nil? && @sqs_url.end_with?('.fifo'))
+      if (!@queue_name.nil? && @queue_name.end_with?('.fifo'))
         raise Fluent::ConfigError, 'message_group_id parameter is required for FIFO queue' if @message_group_id.nil?
       end
-
-      Aws.config = {
-        access_key_id: @aws_key_id,
-        secret_access_key: @aws_sec_key,
-        region: @region
-      }
     end
 
     def client
-      @client ||= Aws::SQS::Client.new
-    end
-
-    def resource
-      @resource ||= Aws::SQS::Resource.new(client: client)
+      @client ||= Aws::SQS::Client.new(
+        access_key_id: @aws_key_id,
+        secret_access_key: @aws_sec_key,
+        region: @region,
+        stub_responses: @stub_responses
+      )
     end
 
     def queue
       return @queue if @queue
 
-      @queue = if @create_queue && @queue_name
-                 resource.create_queue(queue_name: @queue_name)
-               else
-                 @queue = if @sqs_url
-                            resource.queue(@sqs_url)
-                          else
-                            resource.get_queue_by_name(queue_name: @queue_name)
-                          end
-               end
+      begin
+        sqs_url = client.get_queue_url(queue_name: @queue_name, queue_owner_aws_account_id: @queue_owner_aws_account_id).queue_url
+      rescue Aws::SQS::Errors::NonExistentQueue => e
+        if @create_queue
+          sqs_url = client.create_queue(queue_name: @queue_name).queue_url
+        else
+          raise
+        end
+      end
 
+      @queue = Aws::SQS::Queue.new(
+        url: sqs_url,
+        client: client
+      )
       @queue
     end