三歩あるけば物も忘れる

結局ShellScript頼みなので、ご参考までに・・・。

■処理の流れ
S3→SQS→Logstash→ShellScript→ElasticSearch
※S3イベント通知の設定はAws/SimpleQueueService/S3イベント通知を参照願います。

■処理イメージ図

・input：plugins-inputs-sqsでSQSからキューを取得
・filter：SQSからキューで取得できるRecords(fields)からkeyとsizeとbucketnameを抽出。※urldecodeは日本語ファイルを想定してです。
・output：sizeが0じゃなかったらkeyとbucketnameをShellScriptに渡しています。※sizeが0はフォルダ作成イベントなので除外。

input {
  sqs {
    queue => "{SQSのキュー名}"
    region => "ap-northeast-1"
  }
}
filter {
    split { field => "Records" }
    mutate {
        add_field => {
            "key" => "%{[Records][s3][object][key]}"
            "size" => "%{[Records][s3][object][size]}"
            "bucket" => "%{[Records][s3][bucket][name]}"
            }
        remove_field => [ "Records" ]
    }
    urldecode { field => "key" }
}
output {
    if !([size] == "0") {
       exec {
           command => "bash /tmp/logstash.sh %{key} %{bucket}"
       }
    }
    stdout{}
}

Aws/ElasticSearch/11.日本語ファイルを登録するの処理を流用し、前処理にAWSCLIコマンドでバケット名とキー名からファイルを取得しています。

#! /bin/bash
path="/tmp/filebeat/"
file_name=$(basename $1)
file_path=$path$file_name
aws s3api get-object --bucket $2 --key $1 $file_path

file=$(base64 $file_path | perl -pe 's/\n//g')
echo -e "{ \"index\" : { \"_index\" : \"{インデックス名}\", \"_type\" : \"_doc\", \"_id\" : \"{ID}\", \"pipeline\": \"{pipeline名}\" }\n{ \"@timestamp\" : \"`date +'%Y-%m-%dT%H:%M:%S.%NZ'`\", \"data\" : \"$file\" }" > input.json
curl -X POST -H 'Content-Type: application/json' '{Elasticsearchエンドポイント}/_bulk?pretty' --data-binary @input.json

rm -f $file_path

TemplateとPipelineは下記を参考に設定願います。
Aws/ElasticSearch/3.Template
Aws/ElasticSearch/4.Pipeline

・S3→SQS→Filebeat→Logstash→ElasticSearch
　→Aws/ElasticSearch/12.ディレクトリに置いたファイルを自動登録するの失敗事例と同じで実現を断念、やはりバイナリファイル向きではない。

・S3→SQS→Filebeat→Logstash→ShellScript→ElasticSearch
　→output.fileは、SQSのキュー毎にファイルが分けられない為Aws/ElasticSearch/12.ディレクトリに置いたファイルを自動登録するの成功事例の方法が使えなかった。