kafka_pipeline_example.conf

input {
	kafka {
  	  zk_connect => "localhost:2181"
  	  topic_id => "input_log_topic"
    }
}

filter{
	grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }

    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
}

output {
	 kafka {
		bootstrap_servers => "localhost:9092"
		topic_id => "output_log_topic"
		batch_size => 1
	 }
	 stdout {
	     codec => rubydebug
	 }
}