central.conf

input {
        beats
        {
               port => 5044
               ssl  => true
               ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder-remote.domain.com.crt"
               ssl_key => "/etc/pki/tls/private/logstash-forwarder-remote.domain.com.key"
        }
}

filter
{
    if ([message] =~ "^(b|rb|rs),")
        {
        csv
        {
                separator => ","
                columns => [
                        "type",
                        "timeLogged",
                        "timeQueued",
                        "orig",
                        "rcpt",
                        "orcpt",
                        "dsnAction",
                        "dsnStatus",
                        "dsnDiag",
                        "dsnMTA",
                        "bounceCat",
                        "srcType",
                        "srcMTA",
                        "dlvType",
                        "dlvSourceIp",
                        "dlvDestinationIp",
                        "dlvEsmtpAvailable",
                        "dlvSize",
                        "vmta",
                        "jobId",
                        "envId",
                        "queue",
                        "vmtaPool",
                        "messageID",
                        "EUID",
                        "from_header",
                        "subject",
                        "header_X-Data-Rating",
                        "header_X-Contact-Score",
                        "dsnReportingMTA"
                ]
        }
        mutate
        {
                remove_field => ["message"]
        }
        mutate
        {
             convert => [ "header_X-Data-Rating" , "integer" ]
        }
        mutate
        {
             convert => [ "header_X-Contact-Score" , "integer" ]
        }
        mutate
        {
                lowercase => ["rcpt"]
        }
        date
        {
                match => [ "timeQueued", "yyyy-MM-dd HH:mm:ss+0200" ]
                target => "timeQueued"
        }
        date
        {
                match => [ "timeLogged", "yyyy-MM-dd HH:mm:ss+0200" ]
                target => "timeLogged"
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "orig" , "%{EMAILADDRESSORIG}" ]
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "rcpt" , "%{EMAILADDRESSRCPT}" ]
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "dsnMTA" , "%{DSNMTAHOST}" ]
                tag_on_failure => []
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "messageID" , "%{MESSAGEID}" ]
                tag_on_failure => []
        }
        mutate
        {
                convert => [ "dlvSize" , "integer" ]
        }
        mutate
        {
                gsub => ["from_header", "[<>]", ""]
        }
        grok
        {
                patterns_dir => ["/etc/logstash/patterns"]
                match => ["from_header", "%{EMAIL}"]
                tag_on_failure => []
        }
        mutate
        {
                remove_field => [ "messageID", "from_header" ]
        }
        if "_grokparsefailure" in [tags]
        {
            drop {}
        }

        } else if ([message] =~ "^(t|d|r),") {
            csv
        {
                separator => ","
                columns => [
                        "type",
                        "timeLogged",
                        "timeQueued",
                        "orig",
                        "rcpt",
                        "orcpt",
                        "dsnAction",
                        "dsnStatus",
                        "dsnDiag",
                        "dsnMTA",
                        "bounceCat",
                        "srcType",
                        "srcMTA",
                        "dlvType",
                        "dlvSourceIp",
                        "dlvDestinationIp",
                        "dlvEsmtpAvailable",
                        "dlvSize",
                        "vmta",
                        "jobId",
                        "envId",
                        "queue",
                        "vmtaPool",
                        "messageID",
                        "EUID",
                        "from_header",
                        "subject",
                        "header_X-Data-Rating",
                        "header_X-Contact-Score",
                        "dsnReportingMTA"
                ]
        }
        mutate
        {
                remove_field => ["message"]
        }
        mutate
        {
             convert => [ "header_X-Data-Rating" , "integer" ]
        }
        mutate
        {
             convert => [ "header_X-Contact-Score" , "integer" ]
        }
        mutate
        {
                lowercase => ["rcpt"]
        }
        date
        {
                match => [ "timeQueued", "yyyy-MM-dd HH:mm:ss+0200" ]
                target => "timeQueued"
        }
        date
        {
                match => [ "timeLogged", "yyyy-MM-dd HH:mm:ss+0200" ]
                target => "timeLogged"
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "orig" , "%{EMAILADDRESSORIG}" ]
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "rcpt" , "%{EMAILADDRESSRCPT}" ]
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "dsnMTA" , "%{DSNMTAHOST}" ]
                tag_on_failure => []
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "messageID" , "%{MESSAGEID}" ]
                tag_on_failure => []
        }
        mutate
        {
                convert => [ "dlvSize" , "integer" ]
        }
        mutate
        {
                gsub => ["from_header", "[<>]", ""]
        }
        grok
        {
                patterns_dir => ["/etc/logstash/patterns"]
                match => ["from_header", "%{EMAIL}"]
                tag_on_failure => []
        }
        mutate
        {
                remove_field => [ "messageID", "from_header" ]
        }
        if "_grokparsefailure" in [tags]
        {
            drop {}
        }

        } else if ([message] =~ "^(t|d|b|r),") {
         csv
        {
                separator => ","
                columns => [
                        "type",
                        "timeLogged",
                        "timeQueued",
                        "orig",
                        "rcpt",
                        "orcpt",
                        "dsnAction",
                        "dsnStatus",
                        "dsnDiag",
                        "dsnMTA",
                        "bounceCat",
                        "srcType",
                        "srcMTA",
                        "dlvType",
                        "dlvSourceIp",
                        "dlvDestinationIp",
                        "dlvEsmtpAvailable",
                        "dlvSize",
                        "vmta",
                        "jobId",
                        "envId",
                        "queue",
                        "vmtaPool",
                        "messageID",
                        "EUID",
                        "from_header",
                        "subject",
                        "header_X-Data-Rating",
                        "header_X-Contact-Score",
                        "dsnReportingMTA"
                ]
        }
        mutate
        {
                remove_field => ["message"]
        }
        mutate
        {
             convert => [ "header_X-Data-Rating" , "integer" ]
        }
        mutate
        {
             convert => [ "header_X-Contact-Score" , "integer" ]
        }
        mutate
        {
                lowercase => ["rcpt"]
        }
        date
        {
                match => [ "timeQueued", "yyyy-MM-dd HH:mm:ss+0200" ]
                target => "timeQueued"
        }
        date
        {
                match => [ "timeLogged", "yyyy-MM-dd HH:mm:ss+0200" ]
                target => "timeLogged"
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "orig" , "%{EMAILADDRESSORIG}" ]
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "rcpt" , "%{EMAILADDRESSRCPT}" ]
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "dsnMTA" , "%{DSNMTAHOST}" ]
                tag_on_failure => []
        }
        grok
        {
                patterns_dir => [ "/etc/logstash/patterns" ]
                match => [ "messageID" , "%{MESSAGEID}" ]
                tag_on_failure => []
        }
        mutate
        {
                convert => [ "dlvSize" , "integer" ]
        }
        mutate
        {
                gsub => ["from_header", "[<>]", ""]
        }
        grok
        {
                patterns_dir => ["/etc/logstash/patterns"]
                match => ["from_header", "%{EMAIL}"]
                tag_on_failure => []
        }
        mutate
        {
                remove_field => [ "messageID", "from_header" ]
        }
        if "_grokparsefailure" in [tags]
        {
            drop {}
        }
        }
}

output
{
        elasticsearch
        {
                hosts => "172.16.112.xx:9200"
                index => "pmta-%{+YYYY.MM.dd}"
        }
        stdout { codec => rubydebug }
}