Helper script to read logs from s3 bucket.

log_select.rb

#!/usr/bin/env ruby

require "aws-sigv4"
require "aws-sdk-core"
require "active_support/all"
require "aws-eventstream"

require "excon"

BUCKET_NAME = ARGV[0]
key=ARGV[1].chomp
if key.empty?
  puts "error with params"
end

FILE_NAME=key
AWS_REGION = "ap-northeast-1"
OBJECT_KEY = "/#{FILE_NAME}?select&select-type=2"
HOST = "https://#{BUCKET_NAME}.s3-#{AWS_REGION}.amazonaws.com"
URL = "#{HOST}#{OBJECT_KEY}"
QUERY = ARGV[2] || "Select * from S3Object limit 20"
puts "QUERY: #{QUERY}"
XML_BODY = <<-DOC
<?xml version="1.0" encoding="UTF-8"?>
<SelectRequest>
    <Expression>#{QUERY}</Expression>
    <ExpressionType>SQL</ExpressionType>
    <InputSerialization>
        <CompressionType>GZIP</CompressionType>
        <CSV>
            <FileHeaderInfo>IGNORE</FileHeaderInfo>
            <RecordDelimiter>\n</RecordDelimiter>
            <FieldDelimiter>\t</FieldDelimiter>
            <QuoteCharacter></QuoteCharacter>
            <QuoteEscapeCharacter></QuoteEscapeCharacter>
            <Comments>#</Comments>
        </CSV>
    </InputSerialization>
    <OutputSerialization>
        <CSV>
            <QuoteFields>ASNEEDED</QuoteFields>
            <RecordDelimiter>\n</RecordDelimiter>
            <FieldDelimiter>\t</FieldDelimiter>
            <QuoteCharacter>'</QuoteCharacter>
            <QuoteEscapeCharacter>'</QuoteEscapeCharacter>
        </CSV>
    </OutputSerialization>
    <RequestProgress>
        <Enabled>FALSE</Enabled>
    </RequestProgress>
</SelectRequest>
DOC

signer = Aws::Sigv4::Signer.new(
  service: 's3',
  region: AWS_REGION,
  credentials_provider: Aws::SharedCredentials.new
)

headers = {
  "Content-Type"=>"text/xml",
}.merge(Excon.defaults[:headers])

signature = signer.sign_request(
  http_method: 'POST',
  url: URL,
  headers: headers,
  body: XML_BODY
)

conn = Excon.new(URL, omit_default_port: true)
decoder = Aws::EventStream::Decoder.new(format: false)

streamer = lambda do |chunk, remaining_bytes, total_bytes|
  msg, eof = decoder.decode_chunk(chunk)
  if msg&.headers && msg.headers[":event-type"]&.value == "Records"
    puts msg.payload.read
  else
  #  puts msg.headers.inspect
  #  puts msg.payload.read
  end
end

excon_headers = headers.merge(signature.headers)
excon_headers.delete("host")
response = conn.post(body: XML_BODY,
                     headers: excon_headers,
                     response_block: streamer)