Split "message" Field using Ruby

[jakecramer17 (Jake Cramer)]

My input with the type "sym" contains a single string of multiple syslog logs separated by the '\r' character (which I'm guessing represents carriage return). I'm not focused on parsing each log yet. First, I need to separate them.

In this thread , magnusbaeck suggested using a ruby filter to rewrite the input string into an array of strings, then feed that array to a split filter.

I referenced some ruby code from here:

#string delimiter 
"hello".split('')               #=> ["h", "e", "l", "l", "o"]        
"hello".split('ll')             #=> ["he", "o"]  # regular expression delimiter 
"hello".split(//)               #=> ["h", "e", "l", "l", "o"]
"hello".split(/l+/)             #=> ["he", "o"]

Here is my filter configuration.

filter {
    if[type]=="syslog" {
        geoip {
            source => "[sourceAddress]"
        }
    }
    if[type]=="vpn" {
        grok {
            match => { "message" => "%{POSINT:message_id} <%{POSINT:syslog_pri}>%{POSINT:syslog_version} %{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: - - - %{TIMESTAMP_ISO8601:syslog_timestamp2} %{GREEDYDATA:syslog_message}" }
        }
    }
    if[type]=="sym" {
        ruby {
            code =>
                event.get("message").split('\r')
        }
#        mutate {
#            split => {"message" => "\r"}
#        }
    }
}

I don't know ruby code very well and am still in the process of understanding it. I'm exploring other options like mutate.

Can anyone tell me how to fix my syntax?
If anyone has another methodology to suggest, I'd be very grateful to find out!

[Badger]

@jakecramer17

ruby { code => event.get("message").split('\r') }

That should be

ruby {
    code => '
        event.set("message", event.get("message").split("\r"))
    '
}

However unless you have config.support_escapes enabled \r is not a carriage return. Instead you would use a literal ctrl+M (on UNIX you would type ctrl+V ctrl+M to enter that).

[jakecramer17 (Jake Cramer)]

Wow! Thank you so much!

"message" became an array containing each separation in its own index. Using the split filter separated each entry into its own event.

This is the configuration I am running now:

filter {
    if[type]=="syslog" {
        geoip {
            source => "[sourceAddress]"
        }
    }
    if[type]=="vpn" {
        grok {
            match => { "message" => "%{POSINT:message_id} <%{POSINT:syslog_pri}>%{POSINT:syslog_version} %{TIMESTAMP_ISO8601:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG:syslog_program}: - - - %{TIMESTAMP_ISO8601:syslog_timestamp2} %{GREEDYDATA:syslog_message}" }
        }
    }
    if[type]=="sym" {
        ruby {
            code => '
                event.set("message", event.get("message").split("\r"))
            '
        }
        split {
            field => "message"
        }
    }
}