Reindex Cloudtrail logs in ElasticSearch, for Kibana usage

reindex_cloudtrail.rb

require 'aws-sdk'
require 'json'
require 'uri'

# Your current cloudtrail S3 preffix
BASE="AWSLogs/xxxxxxxxxxxxxxx/CloudTrail/ap-southeast-1"

BUCKET_NAME="bucket_name"
AWS_ACCESS_ID = 'XXXXXXXXXXXXXX'
AWS_SECRET_KEY = 'xxxxxxxxxxxxxxxxxxxxx'

def gunzip(data)
  sio = StringIO.new(data)
  gz = Zlib::GzipReader.new(sio)
  read_data = gz.read
  gz.close
  read_data
end

Aws.config.update({
  region: 'ap-southeast-1',
  credentials: Aws::Credentials.new(AWS_ACCESS_ID, AWS_SECRET_KEY),
})


@resource = Aws::S3::Resource.new
@bucket = @resource.bucket(BUCKET_NAME)

DAY=86400
time = Time.now - DAY

def by_date(time)
  time_str = time.strftime("%Y/%m/%d")
  prefix = "#{BASE}/#{time_str}"

  @json = []
  @bucket.objects(prefix: prefix).each do |object|
    puts object.key
    object = @bucket.object(object.key)
    object.get({response_target: "/tmp/test.json.gz"})
    data = gunzip(File.read "/tmp/test.json.gz")
    json = JSON.load(data)
    @json += json["Records"] if json
  end;nil

  logstash_date = time.strftime("%Y.%m.#{time.day}")

  @all = File.open('all', 'w')
  @json.each do |json|
    date = json["eventTime"]
    @all.write %Q{{ "index" : { "_index" : "logstash-#{logstash_date}", "_type" : "fluentd", "_timestamp" : "#{date}" } }}
    @all.write "\n"
    @all.write json.to_json
    @all.write "\n"
  end
  @all.close

  puts "-------------------"
  puts "Processing #{@json.size} requests for #{logstash_date}"
  `curl -s -XPOST localhost:9200/_bulk --data-binary @all; echo`
end

# How many days to go back.
# Index the last 200 days, starting from yesterday
(1..200).each do |i|
  time = Time.now - DAY*i
  puts '------------------'
  puts time
  by_date(time)
end