function var_dump(data, max_level, prefix)   
	if type(prefix) ~= "string" then   
		prefix = ""  
	end   
	if type(data) ~= "table" then   
		dump_html(prefix .. tostring(data))   
	else  
		dump_html(tostring(data))   
		if max_level ~= 0 then   
			local prefix_next = prefix .. "    "  
			dump_html(prefix .. "{")   
			for k,v in pairs(data) do   
				dump_html(prefix_next .. k .. " = ") 
				if type(v) ~= "table" or (type(max_level) == "number" and max_level <= 1) then   
					dump_html(v)   
				else  
					if max_level == nil then   
						var_dump(v, nil, prefix_next)   
					else  
						var_dump(v, max_level - 1, prefix_next)   
					end   
				end   
			end   
			dump_html(prefix .. "}")   
		end   
	end   
end  


function dump_html(str)
	if str ~= nil then
		ngx.header.content_type = "text/html"
        ngx.say(str)
	end
end

input {
#    beats {
#        host => "0.0.0.0"
#        port => 5400
#    }

stdin { }
}

filter {
 grok {
   patterns_dir => "/etc/logstash/patterns"
   #match => [ "message" , "%{NGINXACCESS}"]
   match => [ "message" , "%{DATA:timestr} \[%{DATA:error_level}\] (?<nginx_message>(.|\r|\n)*)(?:, client: %{IPORHOST:clientip})(?:, server: %{IPORHOST:nginx_server})(?:, request: \"%{DATA:nginx_request}\")?(?:, upstream: \"%{DATA:nginx_upstream}\")?(?:, host: \"%{DATA:nginx_host}\")?(?:, referrer: \"%{DATA:nginx_referrer}\")?"]
 }

      if [http_x_forwarded_for] == "-" or [http_x_forwarded_for] == "null" {
         mutate {
            update => { "http_x_forwarded_for" => "" }
         }
      }

      if [referer] == "-" or [referer] == "null" {
         mutate {
            update => { "referer" => "" }
         }
      }

    geoip {
      source => "clientip"
    }

    useragent {
      source => "agent"
      target => "agent_fields"
    }

  date {
    match => [ "timestr", "yyyy/MM/dd HH:mm:ss" ]
    timezone => "Asia/Shanghai"
    #target => "newtimestr"
    #locale => "en"
  }

        ruby {
                code => "event.set('index_day', event.get('@timestamp').time.localtime.strftime('%Y%m%d'))"
        }


}

output {
 elasticsearch {
   hosts => ["127.0.0.1:9200"]
   #index => "tek-%{+YYYY.MM.dd}"
   index => "tek-%{index_day}"
   document_type => "nginx_logs"
   template_name => "ta"
 }
 stdout { codec => rubydebug }
}

防止一天的日志分在了两个index中

        ruby {
                code => "event.set('index_day', event.get('@timestamp').time.localtime.strftime('%Y%m%d'))"
        }