user nginx;

worker_processes auto;

error_log /var/log/nginx/error.log;

pid /run/nginx.pid;

include /usr/share/nginx/modules/*.conf;

events {

    worker_connections 1024;

}

http {
map $http_x_forwarded_for  $clientRealIp {

    ""    $remote_addr;

    ~^(?P<firstAddr>[0-9\.]+),?.*$    $firstAddr;

}

     log_format nginx_log_json '{"accessip_list":"$proxy_add_x_forwarded_for","client_ip":"$clientRealIp","http_host":"$host","@timestamp":"$time_iso8601","method":"$request_method","url":"$request_uri","status":"$status","http_referer":"$http_referer","body_bytes_sent":"$body_bytes_sent","request_time":"$request_time","http_user_agent":"$http_user_agent","total_bytes_sent":"$bytes_sent","server_ip":"$server_addr"}';

    access_log  /var/log/nginx/access.log  nginx_log_json;

    sendfile            on;

    tcp_nopush          on;

    tcp_nodelay         on;

    keepalive_timeout   65;

    types_hash_max_size 2048;

    include             /etc/nginx/mime.types;

    default_type        application/octet-stream;

    include /etc/nginx/conf.d/*.conf;

    server {

        listen       80 default_server;

        listen       [::]:80 default_server;

        server_name  _;

        root         /usr/share/nginx/html;

        include /etc/nginx/default.d/*.conf;

        location / {

        }

        error_page 404 /404.html;

            location = /40x.html {

        }

        error_page 500 502 503 504 /50x.html;

            location = /50x.html {

        }

    }

}

input {

      kafka {

            bootstrap_servers => "192.168.37.134:9092,192.168.37.135:9092,192.168.37.136:9092"

            topics => "nginx_log"         #指定输入源中需要从哪个topic中读取日志数据

            group_id => "logstash"

            codec => json {

                  charset => "UTF-8"       #将输入的json日志输出进行utf8格式编码

          }

             add_field => { "[@metadata][myid]" => "nginxaccess-log" }    #增加一个字段，用于标识和判断，与下方output输出相对应

        }

}

filter {

    if [@metadata][myid] == "nginxaccess-log" {

       mutate {

        gsub => ["message", "\\x", "\\x"]      #message字段，也就是日志的输出内容，插件作用就是讲message字段内容中UTF8单字节编码做替换处理，目的就是应对URL出现中文情况，防止乱码的出现

      }

     if ( 'method":"HEAD' in [message] ) {     #如果message字段中有HEAD请求，就会删除

       drop {}

     }

     json {        #启动json解码插件，因为输入的数据是复合的数据结构，只是一部分记录的是json格式

          source => "message"    #指定json格式字段，也就是message字段

        remove_field => "prospector"

        remove_field => "beat"

        remove_field => "source"

        remove_field => "input"

        remove_field => "offset"

        remove_field => "fields"

        remove_field => "host"

        remove_field => "@version"

    remove_field => "message"    #因为json格式中已经定义好了每个字段，所以输出也就是按照每个字段输出，不需要message字段了，这里直接移除

}

 }

}

output {

  if [@metadata][myid]=="nginxaccess-log" {    #当有多个输入源的时候，可根据不同的标识，指定不同的输出地址

   elasticsearch {

    hosts =>["192.168.37.134:9200","192.168.37.135:9200","192.168.37.136:9200"]

    index => "webnginx_log-%{+YYYY-MM-dd}"    #自定义指定索引名称

   }

 }

} 

[root@localhost etc]# nohup /usr/local/logstash/bin/logstash -f nginxlog.conf &    #后台运行Logstash事件