Logstash Config

input {
    beats {
        id => "api_beats"
        client_inactivity_timeout => 1200
        port => 11000
    }
}

filter {
    # Tag event based on its document type
    if [fields][document_type] == "requests" {
        mutate {
            add_tag => [ "apiRequestEvent" ]
            remove_field => [ "[fields][document_type]" ]
        }
    } else if [fields][document_type] == "errors" {
        mutate {
            add_tag => [ "errorEvent" ]
            remove_field => [ "[fields][document_type]" ]
        }
    } else if [fields][document_type] == "soaprequests" {
        mutate {
            add_tag => [ "soapRequestEvent" ]
            remove_field => [ "[fields][document_type]" ]
        }
    } else {
        mutate {
            add_tag => [ "unknownEvent" ]
        }
    }

    # Ensure the fields.environment field exists
    # as this field is used to push event to correct index
    if ![fields][environment] {
        mutate {
            add_field => { "[fields][environment]" => "unknown" }
        }
    } else if [fields][environment] == "" {
        mutate {
            add_field => { "[fields][environment]" => "unknown" }
        }
    }

    # Perform api request message parsing
    if "apiRequestEvent" in [tags] {
        grok {
            id => "apiRequestParsing"
            match => { "message" => "%{TIMESTAMP_ISO8601:RequestTime}\t%{DATE:RequestDate}\t%{DATA:APIKey}\t%{DATA:APIMethod}\t%{DATA:RequestUrl}\t%{DATA:RequestBody}\t%{DATA:WidgetID}\t%{INT:ElapsedMilliseconds}\t%{DATA:ClientIP}\t%{DATA:OtherFields}\t%{GREEDYDATA:MachineName}" }
            remove_field => [ "RequestDate", "message" ]
            tag_on_failure => [ "_grokparsefailure", "apiRequestParsingFailure" ]
        }

        # Parse RequestUrl field into its relevant pieces
        if [RequestUrl] =~ /.+/ {
            grok {
                id => "apiRequestUrlParsing"
                match => { "RequestUrl" => "%{GREEDYDATA:RequestUrl_Root}\?%{GREEDYDATA:RequestUrl_QueryString}" }
                tag_on_failure => [ "_grokparsefailure", "requestUrlParsingFailure" ]
            }

            kv {
                source => "RequestUrl_QueryString"
                field_split => "&"
                prefix => "RequestUrl_"
                transform_key => "lowercase"
                include_keys => [ "format", "mnetid", "trackid" ]
            }

            mutate {
                rename => { "RequestUrl_format" => "RequestUrl_Format" }
                rename => { "RequestUrl_mnetid" => "RequestUrl_MnetID" }
                rename => { "RequestUrl_trackid" => "RequestUrl_MnetID" }
            }

            kv {
                source => "RequestUrl_QueryString"
                field_split => "&"
                prefix => "RequestUrl_QueryString_"
                transform_key => "lowercase"
            }
        }

        # Parse OtherFields field into its relevant pieces
        if [OtherFields] =~ /.+/ {
            kv {
                source => "OtherFields"
                field_split => ";"
                prefix => "OtherFields_"
                include_keys => [ "HttpMethod", "PartnerName" ]
            }

            mutate {
                rename => { "OtherFields_HttpMethod" => "RequestUrl_HttpMethod" }
                rename => { "OtherFields_PartnerName" => "PartnerName" }
            }

            # Backfill APIKey as PartnerName when PartnerName is not present
            if ![PartnerName] {
                mutate {
                    copy => { "APIKey" => "PartnerName" }
                }
            }
        }

        # Lowercase APIMethod field to resolve mixed casing issues
        mutate {
            lowercase => [ "APIMethod" ]
        }
    }

    # Perform soap request message parsing
    if "soapRequestEvent" in [tags] {
        # grok doesn't seem to go past \n so convert to EOL first
        mutate {
            gsub => [ "message", "\n", "EOL" ]
        }

        grok {
            id => "soapRequestParsing"
            match => { "message" => "----- SoapRequest at %{DATESTAMP:RequestTime} to %{URI:RequestURL}(?<Line1>.*)EOL(?<RequestXML>.*)EOL----- SoapResponse at %{DATESTAMP:ResponseTime}(?<Line3>.*)EOL(?<ResponseXML>.*)EOL" }
            remove_field => [ "Line1", "Line3", "message" ]
            tag_on_failure => [ "_grokparsefailure", "soapRequestParsingFailure" ]
        }

        # Save host as MachineName as its not included in message
        mutate {
            copy => { "host" => "MachineName" }
        }
    }

    # Perform error message parsing
    if "errorEvent" in [tags] {
        grok {
            id => "errorParsing"
            match => { "message" => "%{TIMESTAMP_ISO8601:RequestTime}\t%{DATE:RequestDate}\t%{DATA:APIKey}\t%{DATA:APIService}\t%{WORD:APIServiceMethod}\t%{DATA:ErrorMessage}\t%{GREEDYDATA:StackTrace}\t%{INT:ErrorCode}\t%{IP:ClientIP}\t%{GREEDYDATA:MachineName}" }
            remove_field => [ "RequestDate", "message" ]
            tag_on_failure => [ "_grokparsefailure", "errorParsingFailure" ]
        }
    }

    # Remove unused default event fields
    mutate {
        remove_field => [ "host", "input_type", "offset", "type" ]
    }

    # Save current timestamp as ProcessedTimestamp only if there wasn't a grok parse failure
    if "_grokparsefailure" not in [tags] and "unknownEvent" not in [tags] {
        mutate {
            rename => { "@timestamp" => "ProcessedTimestamp" }
        }
    }

    # Set RequestTime as new timestamp and its own RequestTimestamp field
    if [RequestTime] =~ /.+/ {
        date {
            match => [ "RequestTime", "yyyy-MM-dd HH:mm:ss", "MM-dd-yyyy HH:mm:ss.SSS" ]
            timezone => "US/Pacific"
            target => "@timestamp"
        }

        date {
            match => [ "RequestTime", "yyyy-MM-dd HH:mm:ss", "MM-dd-yyyy HH:mm:ss.SSS" ]
            timezone => "US/Pacific"
            target => "RequestTimestamp"
            remove_field => [ "RequestTime" ]
        }
    }

    # Set ResponseTime as its own ResponseTimestamp field
    if [ResponseTime] =~ /.+/ {
        date {
            match => [ "ResponseTime", "MM-dd-yyyy HH:mm:ss.SSS" ]
            timezone => "US/Pacific"
            target => "ResponseTimestamp"
            remove_field => [ "ResponseTime" ]
        }

        # Calculate elapsed time
        ruby {
            code => 'event.set("RequestDuration", event.get("ResponseTimestamp") - event.get("RequestTimestamp"))'
        }
    }

    # Perform ClientIP lookup
    if [ClientIP] =~ /.+/ {
        geoip {
            source => "ClientIP"
        }
    }
}

output {
    # Change output index depending upon event tag
    if "apiRequestEvent" in [tags] {
        elasticsearch {
            hosts => [ "localhost:9200" ]
            id => "api-requests_elasticsearch"
            index => "api-%{[fields][environment]}-apirequests_write"
            template_name => "api"
        }
    } else if "soapRequestEvent" in [tags] {
        elasticsearch {
            hosts => [ "localhost:9200" ]
            id => "api-soaprequests_elasticsearch"
            index => "api-%{[fields][environment]}-soaprequests_write"
            template_name => "api"
        }
    } else {
        # Submit errorEvents and unknownEvents to same index
        elasticsearch {
            hosts => [ "localhost:9200" ]
            id => "api-errors_elasticsearch"
            index => "api-%{[fields][environment]}-errors_write"
            template_name => "api"
        }
    }
}