logstash 编码转换

oswatch@xx:~/logstash> cat sendkafka.conf
input {
     file {
       path=>["/home/afe/controller/log/all.log"]
       type=>"pmts-99.99.162.148"
       codec=>plain{charset=>"GB2312"}
       }
}




filter {
       mutate{
              add_field => ["source","99.99.162.148"]
              add_field => ["filename",'/home/afe/controller/log/all.log']
  }
}






output {
kafka {


    bootstrap_servers => "99.99.34.43:9092,99.99.34.44:9092,99.99.34.52:9092"
    topic_id => "loginsight"
   }
 stdout {
     codec=>json
    }
}