magicdude4eva/logstash.conf

## 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 => ####
    }
}
	################################################################################
	## 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 => ####
	}
	}