logstash.conf

input {
  kafka {
    bootstrap_servers => "broker:9092"
    topics => ["topics"]
    codec => json
  }
}

filter {
        grok {
            match => {
               source => "/(?<logname>[^/]+)\.log$"
            }
        }
    }
    date {
        match => ["timestamp", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSZZ"]
    }
}

output {
#    stdout { codec => rubydebug }
    if [input][type] == "log" {
        file {
                path => "/opt/log_bak/%{+yyyy}/%{+MM}/%{+dd}/%{[fields][module_name]}/%{[fields][log_type]}-%{logname}.log"
                codec => line {
                        format => "%{message}"
                }
                flush_interval => 5
        }
    }
    if [input][type] == "docker" {
        file {
                path => "/opt/log_bak/%{+yyyy}/%{+MM}/%{+dd}/%{[fields][module_name]}/%{[fields][module_name]}-%{stream}.log"
                codec => line {
                        format => "%{message}"
                }
        flush_interval => 5
        }
    }
}