Grafana和Kibana绘制Nginx请求分布地图

最近一时兴起，想用地图展示我们服务(nginx)的请求分布情况，于是就有了这篇记录。

思路：

第一步：将nginx日志通过GeoLite2-City.mmdb地图库把IP转换成经纬度，传到ELK中；

第二步：通过kibana或grafana读取经纬度信息并绘制成地图。

其中：kibana使用Coordinate Map插件，高版本自带的有我这里用的是6.5版本；grafana使用Worldmap Panel插件，按grafana的 Worldmap Panel 步骤安装。

1、采集 nginx 日志到 ELK

1.1 nginx 配置 log_format

1
2
3

log_format  access  '$remote_addr - $remote_user [$time_local] "$request" '
                  '$status $body_bytes_sent "$http_referer" '
                  '"$http_user_agent" "$http_x_forwarded_for"';

别忘了重启nginx使配置生效

1	./nginx/sbin/nginx -s reload

1.2 filebeat 采集 nginx 日志

这里将采集的nginx日志转发到logstash，主要是为了聚合日志，然后通过logstash统一传到kafka进行过度后，再通过logstash写入到ES；这样当日志突增的时候不会因为logstash消费不过来而丢失日志。

cat filebeat/filebeat.yml

filebeat.prospectors:

- input_type: log
  paths:
    - /work/admin/nginx/logs/39.97.130.241.log
    - /work/admin/nginx/logs/access.log
    - /work/admin/nginx/logs/*.com.log
    - /work/admin/nginx/logs/*.cn.log
  document_type: nginx-logs

output.logstash:
  hosts: ["logstash.feiersmart.local:5044"]
output.console:
  pretty: true

这里使用的还是filebeat-5.5.3的filebeat，如果是6X版本的配置可能稍有变化

启动filebeat，我这里使用supervisor管理服务

[program:filebeat]
command=/work/admin/filebeat/filebeat -e -c /work/admin/filebeat/filebeat.yml
directory=/work/admin/filebeat
redirect_stderr=true
autorestart=true

1.3 logstash 传到 kafka

cat logstash-nginx-logs.conf 

output {

    if [type] == "nginx-logs" {

        kafka {
            bootstrap_servers => "kafka1.feiersmart.local:9092,kafka2.feiersmart.local:9092,kafka3.feiersmart.local:9092"
            topic_id => "logstash-nginx-logs"
            codec => "json"
            compression_type => "gzip"
	    max_request_size => 10000120
        }
    }
}

配置好后，重启logstash服务

1.4 logstash 写到 ES

cat logstash-6.5.3-es/conf.d/logstash-nginx-logs.conf 

input {
    kafka {
        bootstrap_servers => "kafka1.feiersmart.local:9092,kafka2.feiersmart.local:9092,kafka3.feiersmart.local:9092"
        topics => ["logstash-nginx-logs"]
        codec => "json"
    }
}

filter {
  grok {
      match => {
        "message" => "%{COMBINEDAPACHELOG}"
      }
  }
  geoip {
      source => "clientip"
      target => "geoip"
      add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
      add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
  }
   mutate {
      convert => [ "[geoip][coordinates]", "float"]
  }
}

output {

    if [type] == "nginx-logs" {

        stdout {
            codec=>rubydebug
        }

        elasticsearch {
            hosts => ["bj-es1.feiersmart.local:9200","bj-es2.feiersmart.local:9200","bj-es3.feiersmart.local:9200"]
            index => "logstash-nginx-logs-%{+YYYY.MM.dd}"
            document_id => "%{type}-%{+YYYY.MM.dd}-%{host}-%{offset}"
        }
    }
}

grok:日格式匹配
geoip:
source:通过geoip处理的信息，这里我们为了通过ip获取经纬，所以选择clientip作为处理的对象
target:通过geoip解析出来的地址信息存储在什么字段里，这里指定为geoip
add_field：将解析出来的经纬存放在geoip.latitude和geoip.longitude中，并且有一个geoip.coordinates是生成的经纬点
mutate：定义格式为float
output：
输出到elasticsearch,host是elk的elasticsearch的IP，默认端口9200
设置index为logstash-nginx-logs-，这在Kibana中创建index的时候会用到

配置好后，重启logstash服务