setup 'bro' to log into elasticsearch

Archive note, 8 October 2026. This is the 2015 Bro/Logstash/Elasticsearch stack and configuration syntax as used then, not a current installation guide. Before trusting a new pipeline, compare a raw input record to the indexed document: field types, source timestamp and timezone, parsed timestamp, log type, and parse failures. A dashboard rendering is not sufficient validation.

A recurring question on the bro mailling list is, "how do you setup bro to ship logs to elasticsearch?"
I'll explain how I've setup my bro network monitor and hopefully it will be useful to others as well.

Setup bro to log in json format:

There are others recipes where you can use grok to parse the text logs to json-ify it, but I find it easier and more stable if the application can log directly into json.

In you local.bro add the following lines:

redef LogAscii::json_timestamps = JSON::TS_ISO8601;
@load tuning/json-logs

restart bro with broctl restart --clean and you should be set.

Setup logstash input to feed on the logs:

Note that a few assumptions are made here.
I like to organise my logstash conf in different files, named [input, output, filter]-<purpose>.conf, this make it easy for templating and debugging. Off course you can always stuff it into a gigantic file, that entirely up to you :-) and in the output-default.conf, you will need to change the cluster name to whatever name your cluster has. you will also need to change the input path to match yours and so on ...
final gotcha; on Debian you will need to apt-get install logstash-contrib to have the translate plugin.

In /etc/logstash/conf.d/input-bro.conf, we specify the source files, the type and that the logs are in json.

input {
    file {
       type => 'bro_logs' 
       path => [ '/opt/spool/manager/*.log' ]
        codec => "json"
    }
}

conn_state_full with the full text based on the content of conn_state, the grok will add a field named bro_type with the type of bro logs (conn, weird, dns, ssl ...) based on the file name, we could do that in the input part as well by giving all the file name and adding a specific type but I'm lazy and I always forget one file so ...

filter {
        if [type] == "bro_logs" {
                date { 
                        match => [ "ts", "ISO8601" ]
                }
                translate { field => "conn_state" destination => "conn_state_full" dictionary => [ "S0", "Attempt", "S1", "Established", "S2", "Originator close only", "S3", "Responder close only", "SF", "SYN/FIN completion", "REJ", "Rejected", "RSTO", "Originator aborted", "RSTR", "Responder aborted", "RSTOS0", "Originator SYN +  RST", "RSTRH", "Responder SYN ACK + RST", "SH", "Originator SYN + FIN", "SHR", "Responder SYN ACK + FIN", "OTH", "Midstream traffic" ] }
                grok {
                        match => { "path" => ".*\/(?<bro_type>[a-zA-Z0-9]+)\.log$"} 
                } 
        }
}

output { 
        elasticsearch {
                cluster => 'elasticsearch'
        }
        #stdout { codec => rubydebug } 
}

Conclusion:

By now, you should have logs in json and logstash shipping them to elasticsearch, you can now start kibana and visualize our bro logs:

pretty pictures ^_^