nginx向kafka转发日志，每小时切割一次日志

user  root root;

worker_processes  8;
worker_cpu_affinity 00000001 00000010 00000100 00001000 00010000 00100000 01000000 1000000;

error_log /home/openresty/nginx/logs/error.log crit;

worker_rlimit_nofile 65535;

events {
    accept_mutex on;
    multi_accept on;
    use epoll;
    worker_connections  65535;
}


http {
    include       mime.types;
    default_type  application/octet-stream;
    proxy_buffering off;
    charset UTF-8;
    log_escape_non_ascii off;
    
    server_names_hash_bucket_size 128;
    client_header_buffer_size 2k;
    large_client_header_buffers 4 4k;
    client_max_body_size 8m;
    sendfile on;
    access_log off;
    gzip off;
    keepalive_requests 8192;

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request"'
    #access_log  logs/access.log  main;
    access_log off;
    log_format  postdata  '$request_body|';
    tcp_nopush     on;
    #keepalive_timeout  0;
    tcp_nodelay on;
    #gzip  on;
    client_header_timeout 12;
    keepalive_timeout 15;
    send_timeout 10;
    open_file_cache max=100000 inactive=20s;
    open_file_cache_valid 30s;
    open_file_cache_min_uses 2;
    open_file_cache_errors on;
#lua调用的文件路径
    lua_package_path "/home/openresty/lualib/kafka/?.lua;;";
    resolver 127.0.0.1;

    map $host $fmt_localtime {
                 default '';
                }
   server {
        listen       8007;
        server_name  172.1.11.18;
        client_max_body_size  10m;
        client_body_buffer_size  128k;
        proxy_connect_timeout 90;
        proxy_send_timeout 90;
        proxy_read_timeout 90;
        proxy_buffer_size 4k;
        proxy_buffers 4 32k;
        proxy_busy_buffers_size 64k;
        proxy_temp_file_write_size 64k;       
        server_tokens off;
        set $resp_body "";
#取时间，对日志进行每小时切割
        location / {
        if ($time_iso8601 ~ "^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})")
          {
            set $year $1;
            set $month $2;
            set $day $3;
            set $hour $4;
            #set $minutes $5;
            #set $seconds $6;
          }
        lua_need_request_body on;
        fastcgi_keep_conn on;
        content_by_lua '
                local resp_body = ngx.req.get_body_data();
                local producer = require "resty.kafka.producer"
#kafka集群
                   local broker_list = {
                         { host = "172.2.0.3", port = 9092 },
                         { host = "172.2.0.4", port = 9092 },
                          { host = "172.2.0.5", port = 9092 },
                        }
#向KAFKA转发日志               
            local bp = producer:new(broker_list, { producer_type = "async" })
             local ok, err = bp:send("apkmsg", nil, resp_body.."|")
    
                 if not ok then
                       ngx.log(ngx.ERR, err)
                       return ngx.exit(200)
                  end
                  ngx.say("success")

            ';

     access_log  /data/new_collect/log/$year.$month.$day.$hour.evt postdata;

}
        location /nginx_status {
                stub_status on;
                access_log off;
        }
        
        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }
    }
}