Add assume_sns option

Add support for SQS messages arriving via SNS. This is commonly used to
fan out S3 events.

Because SNS wraps the messages, we need to dive one layer deeper into
the SQS body.

Signed-off-by: Corry Haines <[email protected]>

Author: tabletcorry

lib/fluent/plugin/in_s3.rb

@@ -104,6 +104,8 @@ def initialize
       config_param :wait_time_seconds, :integer, default: 20
       desc "Polling error retry interval."
       config_param :retry_error_interval, :integer, default: 300
+      desc "Assume messages as SNS notifications"
+      config_param :assume_sns, :bool, default: false
     end
 
     desc "Tag string"
@@ -182,6 +184,10 @@ def run
           begin
             body = Yajl.load(message.body)
             log.debug(body)
+            if @assume_sns
+              next unless body["Message"]
+              body = Yajl.load(body["Message"])
+            end
             next unless body["Records"] # skip test queue
 
             process(body)