ELK专题：Day3——添加自定义字段并通过Geoip插件识别IP地址的地理位置
1. 前言
我们在上一篇文章《ELK专题：Day2——Logstash配置Nginx日志分析》中围绕着Web业务日志采集这个场景初步完成了必要的ELK Stack调试，但我们只是完成了一个雏形。在实际的运维场景中，还需要进行一些必要的补充，我们将会在后面继续探讨这个主题。
2. 场景及解决方案
2.1 添加字段
在现实的运维工作中，我们一套业务都会有多个环境，一般最少会分为开发环境（dev）、线上测试环境（test）、生产环境（live）。虽然我们可以通过logstash中自动生成的agent.hostname字段去区分日志来源，但再考虑到其他的需求或场景，如：区分同一台服务器中的日志类型、对同一个日志文件里面的个别内容做差异化处理等，我们需要给日志增加字段。
2.1.1 在filebeat中添加字段
2.1.1.1 filebeat配置示例
我们可以在filebeat的配置文件中使用fields配置项增加字段，fields的内容可以是字符串、数组或者字典。配置示例如下：
1 | filebeat.inputs: |
2.1.1.2 logstash对自定义字段的处理
在filebeat的配置中添加了自定义的字段之后，当日志内容进入到logstash，会体现为fields.nginx_log_type： access这样的形式，想要在logstash中显式地对自定义字段进行过滤和匹配，需要配置如下：
1 | ... |
关于logstash pipeline配置文件中对于if的使用，参考文档：https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals
2.1.2 在logstash中添加字段
logstash的pipeline配置里面支持通过插件mutate添加或者修改字段，我们使用配置示例如下：
1 | input { stdin { } } |
我们可以看到，在插入字段的时候，我们还可以使用变量去对字段内容进行修改，测试结果如下：
1 | { |
2.2 识别IP地址的地理位置
在常见的业务分析场景里面，我们往往需要对访问来源进行统计，比如说查出网站内访问量最高的文章，或者找出访问量最密集的时间点，甚至只是简单地统计页面打开的速度，我们都可以通过前面我们已经做好的日志内容检索去入手。但如果我们还想知道访问来源，在访问来源中统计出哪个省份或者城市的用户最多，就需要对IP地址进行识别了。
Logstash提供了插件geoip，通过GeoLite2自动识别IP地址所在的区域，并自动添加需要的字段。示例配置如下：
1 | input {} |
需要注意的是，必须要先完成日志内容的识别后，向geoip插件提供ip地址或者域名信息，提供到geoip的source字段，地址才可以被正确识别。同时通过target配置项指定geoip的识别结果会被组织到一个命名为geoip的字段中。
初次使用geoip的时候，可能需要等待几分钟的时候，待GeoLite2的数据库完成初始化，才可以正常工作。
经过地理位置识别后，返回的结果示例如下：
1 | { |
参考文档：https://www.elastic.co/guide/en/logstash/7.14/plugins-filters-geoip.html
2.3 Pattern集中管理
在上一篇博文中，我们在pipeline配置文件中使用grok插件直接配置pattern完成了对日志内容的识别：
1 | input { stdin { } } |
这样的配置方式存在着不易维护的问题，当我们对同一种日志格式有多个pipeline配置文件的时候，我们每次改动日志格式，都需要修改多个pipeline配置文件，而且这种配置方式也使配置文件显得过于凌乱。
我们可以通过如下方式去统一维护各个pattern：
创建文件
/etc/logstash/pattern.d/mypattern在文件
mypattern中放置我们需要的pattern，内容如下：1
NGINXCOMBINEDLOG %{IPORHOST:remote_addr} - %{DATA:remote_user} \[%{HTTPDATE:time_local}\] \"%{WORD:request_method} %{DATA:uri} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:body_sent_bytes} \"%{DATA:http_referrer}\" \"%{DATA:http_user_agent}\"
把pipeline配置文件修改成如下形式：
1
2
3
4
5
6
7
8
9input { }
filter {
grok {
patterns_dir => [ "/etc/logstash/pattern.d" ]
match => { "message" => "%{NGINXCOMBINEDLOG}" }
}
...
}
output { }
这样，当我们需要修改pattern的时候，只需要修改一个文件，就可以在多个pipeline中生效了。
除此之外，logstash的开发者为我们提供了很多常见的日志格式的pattern，我们可以直接下载引用：
https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns
patterns_dir参考文档：https://www.elastic.co/guide/en/logstash/7.14/plugins-filters-grok.html#plugins-filters-grok-patterns_dir
2.4 整合配置文件
综合我们上面提到的几点优化，最后我们使用的配置文件如下：
filebeat配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28logging.level: info
logging.to_files: true
logging.files:
path: /var/log/filebeat
name: filebeat
keepfiles: 7
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/nginx/hexo_access.log
fields:
env: test
nginx_log_type: access
- type: log
enabled: true
paths:
- /var/log/nginx/hexo_error.log
fields:
env: test
nginx_log_type: error
setup.template.settings:
index.number_of_shards: 1
output.logstash:
hosts: ["192.168.0.211:5400"]logstash pipeline配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28input {
beats {
host => "0.0.0.0"
port => 5400
}
}
filter {
if [fields][nginx_log_type] == "access" {
grok {
patterns_dir => ["/etc/logstash/pattern.d"]
match => { "message" => "%{NGINXCOMBINEDLOG}" }
}
date {
match => [ "time_local", "dd/MMM/yyyy:HH:mm:ss Z" ]
}
geoip {
source => "remote_addr"
}
}
}
output {
elasticsearch {
hosts => ["192.168.0.212:9200"]
index => "rc_index_pattern-%{+YYYY.MM.dd}"
}
}
修改配置文件后，重启服务使配置生效。
Tips:
logstash动态加载配置文件方法：
kill -SIGHUP ${logstash_pid}
3. 总结
在上面的实践中，我们对logstash和filebeat的配置完成了少量的优化，使其更接近我们实际的生产场景。经过优化后，我们可以在kibana中通过字段field.env区分生产环境和测试环境的流量。在后面，我们就可以开始对我们的网站进行业务分析了。
点击展开完整json
1 | { |