Port25 / PowertMTA Logstash / Graylog configuration

logstash.conf

################################################################################
## Port25 Logstash configuration
##
##  Logging configuration:
##
##  <acct-file /var/log/pmta/acct.csv>
##      delete-after 60d
##      move-interval 5m
##      max-size 500M
##      records d,b,r,t,tq,f,rb,rs
##  </acct-file>


## The input directive specifies the various log files
input {
   
   # Port 25 logs
    file {
    type => "port25"
    tags => "port25"
    sincedb_path => "/var/log/logstash/sincedb_port25_acct.db"
    sincedb_write_interval => 15
    path => [ "/var/log/pmta/acct*.csv" ]
   }

}


# We use a Grok filter for matching - see here: https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html
filter {

  ##############################################################################################################################
  # Parse Port25 logs
  if [type] == "port25" {
    if ([message] =~ "^(t|b|d|r|tq),") {
      csv {
          source => "message"
          columns => [ "logtype","timeLogged","timeQueued","orig","rcpt","orcpt","dsnAction","dsnStatus","dsnDiag","dsnMta","bounceCat","srcType","srcMta","dlvType","dlvSourceIp","dlvDestinationIp","dlvEsmtpAvailable","dlvSize","vmta","jobId","envId","queue","vmtaPool","repSourceIp","feedbackType","format","userAgent","reportingMta","reportedDomain","header_From","header_Return-Path","header_X-job","header_Subject","rcvSourceIp","rcvDestinationIp","dsnReportingMta" ]
          separator => ","
          convert => [ "dlvSize", "integer" ]
          add_tag => [ "pmta" ]
          skip_empty_columns => true
      }
      
      if ("pmta" in [tags]) {
        date { match => [ "timeLogged", "YYYY-MM-dd HH:mm:ssZ", "ISO8601" ] 
          target => "@timestamp"
        }
        
        date { match => ["timeQueued", "YYYY-MM-dd HH:mm:ssZ", "ISO8601" ]
        }
        
        if [format] ==  "jmrp" {
          mutate {   replace => [ "rcpt", "%{header_From}" ] }
        }
      
        if [rcpt] {
          mutate { lowercase => ["rcpt"]
            add_field => { "rcptdomain" => "%{rcpt}" }
          }

          mutate { gsub => [ 'rcptdomain', '.*@', ''] }
        }
        
        mutate {
          gsub => [ "dsnStatus", "-", "_", "bounceCat", "-", "_" ]
        }

        if [timeLogged] and [timeQueued] {
          ruby {
            init => "require 'time';"
            code => "begin; secondsToDeliver = 0.0; secondsToDeliver = Time.parse(event.get('timeLogged')).to_f - Time.parse(event.get('timeQueued')).to_f; event.set('dlvTimeTaken', secondsToDeliver); rescue Exception; event.set('dlvTimeTaken', 'parse_failure'); end"
          }
        }
        
      }
    } else {
      # powermta acct header, ignore
      drop {}
    }
  }

}

output {
    # Requires /usr/share/logstash/bin/logstash-plugin install logstash-output-gelf
    gelf {
		host => "###.###.###.###"
		port => ####
    }
}