Logstash and Bro IDS - Hi-Volume Logs are not "picked up"

[sherifeldeeb]

Bro IDS "bro.org" creates multiple log files (http, dns, conn ...etc.)

I created one .conf file for each log type.

Problem is: logstash doesn't consistently "pick up" the log files that has high volume (like http.log and conn.log) sometimes it does, most often it doesn't.

When logstash do "pick up" the file, everything is parsed correctly.

Please advise.

[eperry (Ed)]

what do you mean pickup and High Volume? I mean if you have just http.log it should be constantly tailing the file. Are you rotating and deleting files? And how often

What version of logstash are you using?

I would read though https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html
there are some advance settings that you will probably want to tweak.

Also, providing an example config and any logs would be helpful.

[sherifeldeeb]

Bro rotates files every one hour by default, I am pointing the path to the "current" file being written to ... logstash handles rotation correctly.

Logstash 5.2.2

I tried different output plugins (file, tcp) and they both work as expected! ... issue shows when specifying elasticsearch as an output (nothing unusual showing in ES logs except the very few CSV parse errors, and inability to convert something to boolean/IP due to log inconsistencies ..etc.)

I installed a fresh elasticsearch with everything as default, and pointed logstash to write to it ...

I guess I'm zooming in to issue, will get back with results ... thanks a lot.

(sample of conf file)

### INPUT BLOCK ###
input {
    file {
        type => "http"
        start_position => "beginning"

        # EDIT THIS LINE #
        path => "/usr/local/bro/logs/current/http.log"
    }
}

### FILTER BLOCK ###
filter {
    if [type] == "http" {
        if [message] =~ /^#/ {
            drop { }
        }

        csv {
            columns => ["ts", "uid", "id.orig_h", "id.orig_p", "id.resp_h", "id.resp_p", "trans_depth", "method", "host", "uri", "referrer", "version", "user_agent", "request_body_len", "response_body_len", "status_code", "status_msg", "info_code", "info_msg", "tags", "username", "password", "proxied", "orig_fuids", "orig_filenames", "orig_mime_types", "resp_fuids", "resp_filenames", "resp_mime_types"]
            separator => "	"
        }

        date {
            match => [ "ts", "UNIX" ] 
        }

        # for each IP address in the fields, get geoip info

        geoip {
                source => "id.orig_h"
                target => "geoip_id_orig_h"
            }

        geoip {
                source => "id.resp_h"
                target => "geoip_id_resp_h"
            }


        # convert all bro types to logstash types, and removing dots from field names.
        mutate {
                rename => { "id.orig_h" => "id_orig_h" }
                rename => { "id.orig_p" => "id_orig_p" }
                convert => { "id_orig_p" => "integer" }
                rename => { "id.resp_h" => "id_resp_h" }
                rename => { "id.resp_p" => "id_resp_p" }
                convert => { "id_resp_p" => "integer" }
                convert => { "trans_depth" => "integer" }
                convert => { "request_body_len" => "integer" }
                convert => { "response_body_len" => "integer" }
                convert => { "status_code" => "integer" }
                convert => { "info_code" => "integer" }
                split => { "orig_fuids" => "," }
                split => { "orig_filenames" => "," }
                split => { "orig_mime_types" => "," }
                split => { "resp_fuids" => "," }
                split => { "resp_filenames" => "," }
                split => { "resp_mime_types" => "," }

        }
    }
}

### OUTPUT BLOCK ###
output {
    if [type] == "http" {
        stdout {
            codec => "rubydebug"
        }
    }
}