logstash GridFTP config

logstash-gridftp.conf

#Brock Palen brockp@umich.edu
#   5/2015
#   GridFTP config for Logstash

input {

#track the globus gridftp transfer logs
        file { 
                path => [ "/var/log/gridftp-go.log" ]
                sincedb_path => "/var/run/logstash-gridftp.sincedb"
                type => "gridftp-xfer"
                tags => [ "hpc" ]
        }


        #generator {
        #       message => "[11161] Wed Jul 16 14:29:16 2014 :: Transfer stats: DATE=20140716182916.007444 HOST=flux-xfer1.engin.umich.edu PROG=globus-gridftp-server NL.EVNT=FTP_INFO START=20140716182914.502772 USER=brockp FILE=/tmp/brockp/linux_x64_64_sfx.exe BUFFER=87380 BLOCK=262144 NBYTES=280590752 VOLUME=/ STREAMS=4 STRIPES=1 DEST=[141.212.30.10] TYPE=RETR CODE=226"
        #       type => "gridftp-xfer"
        #       count => 1
        #}
}

filter {
        if [type] == "gridftp-xfer" {

#       [15298] Wed Jul 16 11:08:42 2014 :: Transfer stats: DATE=20140716150842.633689 HOST=flux-xfer1.engin.umich.edu PROG=globus-gridftp-server NL.EVNT=FTP_INFO START=20140716150841.618581 USER=brockp FILE=/tmp/brockp/linux_x64_64_sfx.exe BUFFER=87380 BLOCK=262144 NBYTES=280590752 VOLUME=/ STREAMS=4 STRIPES=1 DEST=[141.212.30.10] TYPE=RETR CODE=226
#
                grok { 
                        match => [ "message", "Transfer stats:" ]
                }
                #some transfer stats messages are status messages and don't have all the data, reject these
                #[22304] Mon Sep  8 16:45:52 2014 :: Transfer stats: DATE=20140908204552.177237 HOSTS=1944 VOLUME=/ STREAMS=2 STRIPES=1 DEST=[141.212.30.10] TYPE=STOR CODE=226
                grok { 
                        match => [ "message", "NBYTES" ]
                }
                if "_grokparsefailure" in [tags] {
                        drop{}
                }
                grok { 
                        #we only want the Transfer stats logs
                        match => [ "message", "Transfer stats: %{GREEDYDATA:data}"]
                }
                kv {   
                        #now split the data part on kv pairs
                        source => "data"
                }

                #the DEST ip, is wrapped in brackets lets remove them
                #  "DEST" => "[190.192.23.240]",
                grok { 
                        match => [ "DEST", "%{IP:DESTIP}"]
                }
                #lets see where our transfers are going/coming
                geoip {
                        source => "DESTIP"
                }
                # lets change the start date to a the stamp,
                # DATE is the end, START is the start
                date { 
                        target => [ "start_date" ]
                        match => [ "START", "YYYYMMddHHmmss.SSSSSS" ]
                        timezone => "UTC"
                }
                date { 
                        target => [ "end_date" ]
                        match => [ "DATE", "YYYYMMddHHmmss.SSSSSS" ]
                        timezone => "UTC"
                }
                #calculate bandwidth in Mbps
                mutate {
                    convert => ['NBYTES', 'integer']
                }
        #       ruby {
        #               code => 'event["bandwidth"] = (event["NBYTES"]*8) / (event["end_date"]-event["start_date"])/1000/1000'
        #       }

                #tag 'oncampus' if it matches a on campus network range
                #http://www.itcom.itd.umich.edu/backbone/umnet/
                #Flint and Dearborn have own filter
                #Bio Station and other networks not geo local to ann arbor excluded
                cidr { 
                        add_tag => [ "oncampus" ]
                        address => [ "%{DESTIP}" ]
                        network => [ "35.0.0.0/16", "35.1.0.0/16", "35.2.0.0/16", "67.194.0.0/16", "141.211.0.0/16", "141.212.0.0/16", "141.213.0.0/17", "141.213.128.0/17", "141.214.0.0/16", "192.12.80.0/24", "192.231.253.0/24", "198.108.8.0/21", "198.111.224.0/22", "198.111.181.0/25", "207.75.144.0/20", "2607:F018::/32" ]
                }
                cidr { 
                        add_tag => [ "dearborn" ]
                        address => [ "%{DESTIP}" ]
                        network => [ "141.215.0.0/16" ]
                }
                cidr { 
                        add_tag => [ "flint" ]
                        address => [ "%{DESTIP}" ]
                        network => [ "141.216.0.0/16" ]
                }
        }
}

output {
#       stdout { codec => rubydebug }
        redis { host => "<redis host>" data_type => "list" key => "logstash" }
}