logstash_s3.conf

input {
  s3 {
    #add_field => ... # hash (optional), default: {}
    #backup_to_bucket => ... # string (optional), default: nil
    #backup_to_dir => ... # string (optional), default: nil
    bucket => "-" # string (required)
    #codec => ... # codec (optional), default: "plain"
    credentials => ["-", "-"] # array (optional), default: nil
    #debug => true # boolean (optional), default: false
    #delete => ... # boolean (optional), default: false
    #interval => ... # number (optional), default: 60
    prefix => "static_logs" # string (optional), default: nil
    #region => "cn-north-1" # string (optional), default: "us-east-1"
    region_endpoint => "cn-north-1"
    sincedb_path => "/mnt2/.sincedb*" # string (optional), default: nil
    #tags => ... # array (optional)
    type => "s3_access_log" # string (optional)
   }
}
filter {
    if [type] == "s3_access_log" {
        grok {
            match => [ "message", "%{S3_ACCESS_LOG}" ]
        }
        kv {
            source => "req_para"
            field_split => "&?"
            prefix => "req_para_"
           }
        date {
            match => [ "timestamp", "dd/MMM/YYY:HH:mm:ss Z" ]
        }
    }
}
output {
    elasticsearch { 
                embedded => true
                index => "s3_access_log-%{+YYYY.MM.dd-HH}" 
    }
}