output.conf

input {
  redis { 
    host => "localhost"
	type => "redis"
	data_type => "list"
	key => "logstash"
  }
}

filter {
  grok { 
    type => "cdn"
    pattern => "%{IPORHOST:message_ip} %{IPORHOST:message_hostname} - %{HTTPDATE:message_date} \"%{WORD:message_httpmethod} %{URI:message_uri} HTTP/%{NUMBER:message_httpversion}\" %{NUMBER:message_responsecode}"
  }

kv { field_split => "&?" }

mutate { remove => [ "@message" ]
		remove => [ "@source" ]
		remove => [ "@source_path" ]
		remove => [ "@source_host" ]		
	}

date {
	match => [ "message_date", "dd/MMM/YYYY:HH:mm:ss Z" ]
  }
}

output { 
	stdout { debug => true debug_format => "json"}
	elasticsearch_http { 
		host => "localhost"
		port => "9200"
		flush_size => 100
#		index => "logstash"
	}
}