logstash.conf

input {
    kafka {
        bootstrap_servers => "kafka1:19092,kafka2:19092,kafka3:19092"
        topics => ["access-log"]
        group_id => "logstash"
        consumer_threads => 3
        auto_offset_reset => "earliest"
        codec => "json"
        type => "kafka"
    }
}

filter {
    grok {
        match => { "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:logged_at}\] \"%{WORD:http_method} %{URIPATH:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:bytes} %{NUMBER:duration} \"%{DATA:referrer}\" \"%{DATA:user_agent}\"" }
    }
    if "_grokparsefailure" in [tags] {
        drop {}
    }
    date {
        match => [ "logged_at", "dd/MMM/yyyy:HH:mm:ss Z" ]
        target => "logged_at"
    }
    mutate {
        remove_field => [ "message" ]
    }
}

output {
    elasticsearch {
        hosts => ["elasticsearch:9200"]
        index => "access-log-%{+YYYY-MM-dd}"
    }
}