ELK+Filebeat+Kafka构建Nginx日志分析平台

爱被打了一巴掌 2023-01-13 05:42 127阅读 0赞

### 1 软件版本 ###

drwxr-xr-x. 11 tech root   205 4月  15 16:19 elasticsearch-7.1.0
    drwxr-xr-x.  6 tech root   224 4月  15 16:07 filebeat-7.1.0
    drwxr-xr-x.  7 root root   156 4月  15 16:09 kafka_2.13-2.7.0
    drwxr-xr-x. 13 tech root   246 4月  12 19:25 kibana-7.1.0-linux-x86_64
    drwxr-xr-x. 13 tech root   267 4月  12 19:39 logstash-7.1.0
    drwxr-xr-x.  9 1001 1001   204 4月  15 14:00 nginx-1.7.4

### 2 nginx配置 ###

#user  nobody;
    worker_processes  1;
    
    #error_log  logs/error.log;
    #error_log  logs/error.log  notice;
    #error_log  logs/error.log  info;
    
    #pid        logs/nginx.pid;
    
    
    events {
        worker_connections  1024;
    }
    
    
    http {
        include       mime.types;
        default_type  application/octet-stream;
    
        #log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
        #                  '$status $body_bytes_sent "$http_referer" '
        #                  '"$http_user_agent" "$http_x_forwarded_for"';
    
        #access_log  logs/access.log  main;
       
        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   /usr/local/java/elk/nginx-1.7.4/access.log  nginx_log_json;
    
        sendfile        on;
        #tcp_nopush     on;
    
        #keepalive_timeout  0;
        keepalive_timeout  65;
    
        #gzip  on;
    
        server {
            listen       80;
            server_name  localhost;
    
            #charset koi8-r;
    
            #access_log  logs/host.access.log  main;
    
            location / {
                root   html;
                index  index.html index.htm;
            }
    
            #error_page  404              /404.html;
    
            # redirect server error pages to the static page /50x.html
            #
            error_page   500 502 503 504  /50x.html;
            location = /50x.html {
                root   html;
            }
    
            # proxy the PHP scripts to Apache listening on 127.0.0.1:80
            #
            #location ~ \.php$ {
            #    proxy_pass   http://127.0.0.1;
            #}
    
            # pass the PHP scripts to FastCGI server listening on 127.0.0.1:9000
            #
            #location ~ \.php$ {
            #    root           html;
            #    fastcgi_pass   127.0.0.1:9000;
            #    fastcgi_index  index.php;
            #    fastcgi_param  SCRIPT_FILENAME  /scripts$fastcgi_script_name;
            #    include        fastcgi_params;
            #}
    
            # deny access to .htaccess files, if Apache's document root
            # concurs with nginx's one
            #
            #location ~ /\.ht {
            #    deny  all;
            #}
        }
    
    
        # another virtual host using mix of IP-, name-, and port-based configuration
        #
        #server {
        #    listen       8000;
        #    listen       somename:8080;
        #    server_name  somename  alias  another.alias;
    
        #    location / {
        #        root   html;
        #        index  index.html index.htm;
        #    }
        #}
    
    
        # HTTPS server
        #
        #server {
        #    listen       443 ssl;
        #    server_name  localhost;
    
        #    ssl_certificate      cert.pem;
        #    ssl_certificate_key  cert.key;
    
        #    ssl_session_cache    shared:SSL:1m;
        #    ssl_session_timeout  5m;
    
        #    ssl_ciphers  HIGH:!aNULL:!MD5;
        #    ssl_prefer_server_ciphers  on;
    
        #    location / {
        #        root   html;
        #        index  index.html index.htm;
        #    }
        #}
    
    }

说明：主要设置nginx的输出日志格式为json,输出到 /usr/local/java/elk/nginx-1.7.4/access.log

启动nginx:

/usr/local/nginx/sbin/nginx -c  /usr/local/java/elk/nginx-1.7.4/conf/nginx.conf

访问nginx 看到日志已经写入access.log

### 3 kafka启动 ###

# 启动zookeeper
    bin/zookeeper-server-start.sh config/zookeeper.properties
     
    # 启动kafka
    bin/kafka-server-start.sh config/server.properties
     
    # 查看beats输出的数据使用的topic是否创建
    bin/kafka-topics.sh --list --zookeeper localhost:2181
     
     
    # 消费主体的消息，需要把test topic换成filebeat output的topic
    bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
    This is a message
    This is another message

### 4 Filebeat配置 ###

###################### Filebeat Configuration Example #########################
    
    # This file is an example configuration file highlighting only the most common
    # options. The filebeat.reference.yml file from the same directory contains all the
    # supported options with more comments. You can use it as a reference.
    #
    # You can find the full configuration reference here:
    # https://www.elastic.co/guide/en/beats/filebeat/index.html
    
    # For more available modules and options, please see the filebeat.reference.yml sample
    # configuration file.
    
    #=========================== Filebeat inputs =============================
    
    filebeat.inputs:
    
    # Each - is an input. Most options can be set at the input level, so
    # you can use different inputs for various configurations.
    # Below are the input specific configurations.
    
    - type: log
    
      # Change to true to enable this input configuration.
      enabled: true
    
      # Paths that should be crawled and fetched. Glob based paths.
      paths:
        - /usr/local/java/elk/nginx-1.7.4/access.log
        #- c:\programdata\elasticsearch\logs\*
    
      # Exclude lines. A list of regular expressions to match. It drops the lines that are
      # matching any regular expression from the list.
      #exclude_lines: ['^DBG']
    
      # Include lines. A list of regular expressions to match. It exports the lines that are
      # matching any regular expression from the list.
      #include_lines: ['^ERR', '^WARN']
    
      # Exclude files. A list of regular expressions to match. Filebeat drops the files that
      # are matching any regular expression from the list. By default, no files are dropped.
      #exclude_files: ['.gz$']
    
      # Optional additional fields. These fields can be freely picked
      # to add additional information to the crawled log files for filtering
      fields:
       log_topic : nginxlogs
      #  level: debug
      #  review: 1
    
      ### Multiline options
    
      # Multiline can be used for log messages spanning multiple lines. This is common
      # for Java Stack Traces or C-Line Continuation
    
      # The regexp Pattern that has to be matched. The example pattern matches all lines starting with [
      #multiline.pattern: ^\[
    
      # Defines if the pattern set under pattern should be negated or not. Default is false.
      #multiline.negate: false
    
      # Match can be set to "after" or "before". It is used to define if lines should be append to a pattern
      # that was (not) matched before or after or as long as a pattern is not matched based on negate.
      # Note: After is the equivalent to previous and before is the equivalent to to next in Logstash
      #multiline.match: after
    
    
    #============================= Filebeat modules ===============================
    
    filebeat.config.modules:
      # Glob pattern for configuration loading
      path: ${path.config}/modules.d/*.yml
    
      # Set to true to enable config reloading
      reload.enabled: false
    
      # Period on which files under path should be checked for changes
      #reload.period: 10s
    
    #==================== Elasticsearch template setting ==========================
    
    setup.template.settings:
      index.number_of_shards: 1
      #index.codec: best_compression
      #_source.enabled: false
    
    #================================ General =====================================
    
    # The name of the shipper that publishes the network data. It can be used to group
    # all the transactions sent by a single shipper in the web interface.
    name : 192.168.204.162
    # The tags of the shipper are included in their own field with each
    # transaction published.
    #tags: ["service-X", "web-tier"]
    
    # Optional fields that you can specify to add additional information to the
    # output.
    #fields:
    #  env: staging
    
    
    #============================== Dashboards =====================================
    # These settings control loading the sample dashboards to the Kibana index. Loading
    # the dashboards is disabled by default and can be enabled either by setting the
    # options here or by using the `setup` command.
    #setup.dashboards.enabled: false
    
    # The URL from where to download the dashboards archive. By default this URL
    # has a value which is computed based on the Beat name and version. For released
    # versions, this URL points to the dashboard archive on the artifacts.elastic.co
    # website.
    #setup.dashboards.url:
    
    #============================== Kibana =====================================
    
    # Starting with Beats version 6.0.0, the dashboards are loaded via the Kibana API.
    # This requires a Kibana endpoint configuration.
    setup.kibana:
    
      # Kibana Host
      # Scheme and port can be left out and will be set to the default (http and 5601)
      # In case you specify and additional path, the scheme is required: http://localhost:5601/path
      # IPv6 addresses should always be defined as: https://[2001:db8::1]:5601
      #host: "localhost:5601"
    
      # Kibana Space ID
      # ID of the Kibana Space into which the dashboards should be loaded. By default,
      # the Default Space will be used.
      #space.id:
    
    #============================= Elastic Cloud ==================================
    
    # These settings simplify using filebeat with the Elastic Cloud (https://cloud.elastic.co/).
    
    # The cloud.id setting overwrites the `output.elasticsearch.hosts` and
    # `setup.kibana.host` options.
    # You can find the `cloud.id` in the Elastic Cloud web UI.
    #cloud.id:
    
    # The cloud.auth setting overwrites the `output.elasticsearch.username` and
    # `output.elasticsearch.password` settings. The format is `<user>:<pass>`.
    #cloud.auth:
    
    #================================ Outputs =====================================
    
    # Configure what output to use when sending the data collected by the beat.
    
    
    
    #output.file:
      #path: "/usr/local/java/elk/filebeat-test"
      #filename: filebeat-test.log
      #rotate_every_kb: 10000
      #  #number_of_files: 7
      #    #permissions: 0600
    
    #-------------------------- Elasticsearch output ------------------------------
    #output.elasticsearch:
      # Array of hosts to connect to.
      #hosts: ["localhost:9200"]
    
      # Optional protocol and basic auth credentials.
      #protocol: "https"
      #username: "elastic"
      #password: "changeme"
    
    #----------------------------- Logstash #output --------------------------------
    #output.logstash:
      #The Logstash hosts
      #hosts: ["localhost:5044"]
    
      # Optional SSL. By default is off.
      # List of root certificates for HTTPS server verifications
      #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
    
      # Certificate for SSL client authentication
      #ssl.certificate: "/etc/pki/client/cert.pem"
    
      # Client Certificate Key
      #ssl.key: "/etc/pki/client/cert.key"
    
    output.kafka:
      #   initial brokers for reading cluster metadata
      hosts: ["127.0.0.1:9092"]
      #
      #     # message topic selection + partitioning
      topic: '%{[fields.log_topic]}'
      partition.round_robin:
        reachable_only: false
      
      required_acks: 1
      compression: gzip
      max_message_bytes: 1000000
    
    #================================ Processors =====================================
    
    # Configure processors to enhance or manipulate events generated by the beat.
    
    processors:
      - add_host_metadata: ~
      - add_cloud_metadata: ~
      #- drop_fields:
       #   fields: ["beat", "input", "source", "offset","prospector"]
    
    
    #================================ Logging =====================================
    
    # Sets log level. The default log level is info.
    # Available log levels are: error, warning, info, debug
    logging.level: debug
    
    # At debug level, you can selectively enable logging only for some components.
    # To enable all selectors use ["*"]. Examples of other selectors are "beat",
    # "publish", "service".
    #logging.selectors: ["*"]
    
    #============================== Xpack Monitoring ===============================
    # filebeat can export internal metrics to a central Elasticsearch monitoring
    # cluster.  This requires xpack monitoring to be enabled in Elasticsearch.  The
    # reporting is disabled by default.
    
    # Set to true to enable the monitoring reporter.
    #xpack.monitoring.enabled: false
    
    # Uncomment to send the metrics to Elasticsearch. Most settings from the
    # Elasticsearch output are accepted here as well. Any setting that is not set is
    # automatically inherited from the Elasticsearch output configuration, so if you
    # have the Elasticsearch output configured, you can simply uncomment the
    # following line.
    #xpack.monitoring.elasticsearch:
    
    #================================= Migration ==================================
    
    # This allows to enable 6.7 migration aliases
    #migration.6_to_7.enabled: true

配置Filebeat input采集的日志文件为nginx日志文件，output输出结果到kafka。

启动filebeat

./filebeat -e -c filebeat.yml

### 5 Logstash配置 ###

input {
        kafka {
            bootstrap_servers => "127.0.0.1:9092"
            topics => "nginxlogs"		#指定输入源中需要从哪个topic中读取数据，这里会自动新建一个名为nginxlogs的topic
            group_id => "logstash"
            codec => json {
               charset => "UTF-8"
            }
            add_field => { "[@metadata][myid]" => "nginxaccess-log" }   #增加一个字段，用于标识和判断，在output输出中会用到。
        }
    }
    
    
    
    filter {
        if [@metadata][myid] == "nginxaccess-log" {
         mutate {
            gsub => ["message", "\\x", "\\\x"]   #这里的message就是message字段，也就是日志的内容。这个插件的作用是将message字段内容中UTF-8单字节编码做替换处理，这是为了应对URL有中文出现的情况。
          }
          if ( 'method":"HEAD' in [message] ) {    #如果message字段中有HEAD请求，就删除此条信息。
               drop {}
          }
       	json{
               source=>"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"
    
    
            }
    
      
       }
    }
    
    
    output {
        if [@metadata][myid] == "nginxaccess-log" {
            elasticsearch {
                hosts => ["127.0.0.1:9200"]
                index => "nginx-logs-%{+YYYY.MM.dd}"   #指定Nginx日志在elasticsearch中索引的名称，这个名称会在Kibana中用到。索引的名称推荐以logstash开头，后面跟上索引标识和时间。
            }
        }
    }

logstash配置input插件从kafka采集数据，增加一个新的字段myid，做数据过滤，使用fileter插件做数据转换，使用json进行json转换并移除不必要的字段，使用output插件输出数据到elasticsearch。

### 6 启动Elasticsearch和kibana ###

./bin/elasticsearch
    ./bin/kibana

elasticsearch要在Logstash启动前启动。

在kibana上创建indexpatter，使用discovery进行查询indexpattern。