# logstash

input { 
    kafka {
        bootstrap_servers   => "kafka.host:9092"
        auto_offset_reset   => "latest"
        topics              => [ "bro-events" ]
        enable_auto_commit  => "true"
        codec               => "json"
        group_id            => "bro-events-gp"
        consumer_threads    => 20
        tags                => [ "kafka-beat" ]
    }
}

filter {
    if "kafka-beat" in [tags] {
        if [type] == "bro_http" {
            mutate {
                rename => { "tags" => "tags-orig" }
            }
        }
        mutate {
            rename          => { "host" => "beat_host" }
        }

        json {
            source          => "message"
        }

        if [type] == "bro_http" {
            mutate {
                rename => { "tags"      => "http-tags" }
                rename => { "tags-orig" => "tags" }
            }
        }

        mutate {
            add_tag         => [ "syslogng" ]
        }
    }
}