系统环境：

• ELK 版本：7.6.2
• Kubernetes 版本：v1.20.0

在收集Nginx的访问日志后，通过kafka丢到logstash，然后经过过滤丢到es，最后在kibana上展示，通过Grafana更美化的展示图表

ELK Kafka搭建

可参考之前博客k8s搭建，这里就不过多叙述

Nginx日志格式配置

请保证 nginx 使用该字段，名称如果有修改，grafana 模板需要做一定修改

log_format main
    '{"@timestamp":"$time_iso8601",'
    '"host":"$hostname",'
    '"server_ip":"$server_addr",'
    '"client_ip":"$remote_addr",'
    '"xff":"$http_x_forwarded_for",'
    '"domain":"$host",'
    '"url":"$uri",'
    '"referer":"$http_referer",'
    '"args":"$args",'
    '"upstreamtime":"$upstream_response_time",'
    '"responsetime":"$request_time",'
    '"request_method":"$request_method",'
    '"status":"$status",'
    '"size":"$body_bytes_sent",'
    '"request_body":"$request_body",'
    '"request_length":"$request_length",'
    '"protocol":"$server_protocol",'
    '"upstreamhost":"$upstream_addr",'
    '"file_dir":"$request_filename",'
    '"http_user_agent":"$http_user_agent"'
  '}';

ELK配置

filebeat 配置

filebeat.inputs:                  
- type: log                        
  access:
  enabled: true                    
  #日志是json开启这个
  json.keys_under_root: true       
  json.overwrite_keys: true        
  max_bytes: 20480              
  paths:
    - /var/log/nginx/access.log    # 监控nginx的access日志

  fields:         #额外的字段
    source: nginx           #自定义source字段,用于es建议索引

#自定义es的索引需要把ilm设置为false
setup.ilm.enabled: false

output.kafka:            #输出到kafka
  enabled: true          #该output配置是否启用
  hosts: ["10.0.0.x:31090", "10.0.0.x:31091", "10.0.0.x:31092"]  #kafk节点列表
  topic: "elk-%{[fields.source]}"   #kafka会创建该topic
  partition.hash:
    reachable_only: true #是否只发往可达分区
  compression: gzip      #压缩
  max_message_bytes: 1000000  #Event最大字节数,默认1000000,应小于等于kafka broker message.max.bytes值
  required_acks: 1       #kafka ack等级
  worker: 1              #kafka output的最大并发数
  bulk_max_size: 2048    #单次发往kafka的最大事件数
#输出所有日志到file,默认true,达到日志文件大小限制时,日志文件会自动限制替换
logging.to_files: tr
#如果一个文件在某个时间段内没有发生过更新，则关闭监控的文件handle,默认1h
close_older: 30m         
#这个选项关闭一个文件,当文件名称的变化,在window建议为true
force_close_files: false 
#没有新日志采集后多长时间关闭文件句柄,默认5分钟,设置成1分钟,加快文件句柄关闭
close_inactive: 1m
#传输了3h后没有传输完成的话就强行关闭文件句柄,这个配置项是解决以上案例问题的key point
close_timeout: 3h
#默认值是0表示不清理,在运行一段时间后,registry会变大,可能会带来问题
clean_inactive: 72h
#设置了clean_inactive后就需要设置ignore_older且要保证ignore_older<clean_inactive
ignore_older: 70h
#限制一个CPU核心,避免过多抢占业务资源
max_procs: 1
#存储于内存队列的事件数，排队发送 (默认4096)
queue.mem.events: 256 
queue.mem.flush.min_events: 128

logstash 配置

input {                                        #输入组件
    kafka {                                    #从kafka消费数据
        bootstrap_servers => ["10.0.0.x:31090,10.0.0.x:31090,10.0.0.x:31090"]
        topics_pattern => "elk-.*"             #使用正则匹配topic
        codec => "json"                        #数据格式
        consumer_threads => 3                  #消费线程数量
        decorate_events => true                #可向事件添加Kafka元数据，比如主题、消息大小的选项，这将向logstash事件中添加一个名为kafka的字段
        auto_offset_reset => "latest"          #自动重置偏移量到最新的偏移量
        group_id => "logstash-groups1"         #消费组ID，多个有相同group_id的logstash实例为一个消费组
        client_id => "logstash1"               #客户端ID
        fetch_max_wait_ms => "1000"            #指当没有足够的数据立即满足fetch_min_bytes时，服务器在回答fetch请求之前将阻塞的最长时间
  }
}
    
filter {
  geoip {
    #multiLang => "zh-CN"
    target => "geoip"
    source => "client_ip"
    database => "/usr/share/logstash/data/GeoLite2-City/GeoLite2-City.mmdb"
    add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
    add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
    #去掉显示geoip显示的多余信息
    remove_field => ["[geoip][latitude]", "[geoip][longitude]", "[geoip][country_code]", "[geoip][country_code2]", "[geoip][country_code3]", "[geoip][timezone]", "[geoip][continent_code]", "[geoip][region_code]"]
  }
  mutate {
    convert => [ "size", "integer" ]
    convert => [ "status", "integer" ]
    convert => [ "responsetime", "float" ]
    convert => [ "upstreamtime", "float" ]
    convert => [ "[geoip][coordinates]", "float" ]
    #过滤filebeat没用的字段,这里过滤的字段要考虑好输出到es的，否则过滤了就没法做判断
    remove_field => [ "ecs","agent","host","cloud","@version","input","logs_type" ]
  }
  #根据http_user_agent来自动处理区分用户客户端系统与版本
  useragent {
    source => "http_user_agent"
    target => "ua"
    #过滤useragent没用的字段
    remove_field => [ "[ua][minor]","[ua][major]","[ua][build]","[ua][patch]","[ua][os_minor]","[ua][os_major]" ]
  }
}


output {                           
    elasticsearch {
        hosts => ["elasticsearch:9200"]
        index => "logstash-%{[fields][source]}-%{+YYYY-MM-dd}"    
    }
}

GeoLite2-City插件需要自行安装，这里提供下载传送门

Grafana Nginx 图表

grafana里添加es数据源，配置相对应索引，添加11190模板即可，打开地图可能不会显示，那是因为图片URL访问不到导致，解决方法如下：

cd /var/lib/grafana/plugins/
#备份如下文件
grafana-worldmap-panel/module.js
grafana-worldmap-panel/module.js.map
 
sed -i 's/https:\/\/cartodb-basemaps{s}.global.ssl.fastly.net\/light_all\/{z}\/{x}\/{y}.png/http:\/\/{s}.basemaps.cartocdn.com\/light_all\/{z}\/{x}\/{y}.png/' module.js
sed -i 's/https:\/\/cartodb-basemaps{s}.global.ssl.fastly.net\/light_all\/{z}\/{x}\/{y}.png/http:\/\/{s}.basemaps.cartocdn.com\/light_all\/{z}\/{x}\/{y}.png/' module.js.map
 
sed -i 's/https:\/\/cartodb-basemaps-{s}.global.ssl.fastly.net\/dark_all\/{z}\/{x}\/{y}.png/http:\/\/{s}.basemaps.cartocdn.com\/dark_all\/{z}\/{x}\/{y}.png/' module.js \
sed -i 's/https:\/\/cartodb-basemaps-{s}.global.ssl.fastly.net\/dark_all\/{z}\/{x}\/{y}.png/http:\/\/{s}.basemaps.cartocdn.com\/dark_all\/{z}\/{x}\/{y}.png/' module.js.map 

#重启grafana即可生效