logstash.conf

input
{
    file
    {
        path => "{PATH}"
        start_position => "beginning"
        sincedb_path => "/dev/null"
        file_completed_action => "log"
        file_completed_log_path => "/root/logstash/ingest.log"
        mode => "read"
    }
}

filter
{

    mutate
    {
        gsub => ["message", "[\u0000]", ""]
    }

    json
    {
        source => "message"
    }

    fingerprint
    {
        source => [ "message" ]
        concatenate_sources => true
        method => "MD5"
        key => "fingerprint"
        base64encode => true
        target => "[@metadata][fingerprint]"
    }

    mutate
    {
        remove_field => ["message"]
    }

    if [sourceIPAddress]
    {
        geoip
        {
            source => "sourceIPAddress"
            fields => ["city_name", "country_code2", "country_name", "latitude", "longitude"]
            target => "ActorIP-Location"
        }

        mutate
        {
            rename =>
            {
                "[ActorIP-Location][latitude]" => "[ActorIP-Location][geo][lat]"
                "[ActorIP-Location][longitude]" => "[ActorIP-Location][geo][lon]"
            }
        }
    }

 }

 output
{
    elasticsearch
    {
        hosts => "{URL}"
        document_id => "%{[@metadata][fingerprint]}"
        index => "aws-cloudtrail"
        user => "elastic"
        password => "{PASS}"
    }
}