案例篇：ELFK收集nginx日志
前言
前面我们搭建好了ELFK架构，也对配置稍复杂的logstash进行了一些简单的实践，现在来尝试收集一下nginx日志吧
准备前工作
我们首先在nginx中定义一个自己想要的格式，如：
log_format access_ty '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'$http_user_agent $http_x_forwarded_for $request_time $upstream_addr $upstream_status $upstream_response_time';
在conf中定义访问日志时来引用access_ty这个自定义名字即可
access_log /var/log/nginx/access_test.log access_ty;
访问一下，得出日志数据如下，这个就作为我们的测试数据了：
100.100.201.135 - - [20/Mar/2023:17:40:12 +0800] "GET / HTTP/1.1" 404 0 "-" Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36 - 0.001 192.168.1.132:8005 404 0.000
思路
- Filebeat定义nginx日志路径，可固定也可通配符。将收集到的内容 发送给logstash进行过滤处理
- Logstash接收到filebeat发来的日志，进入到过滤阶段 通过grok正则将字段进行分解，可自定义某些冗余字段对其删除
- 处理后将message发给es，es中创建对应的索引
- kibana创建与es索引对应的
index patterns在界面可进行查看
步骤
将测试数据写入/opt/nginx.log中
echo '100.100.201.135 - - [20/Mar/2023:17:40:12 +0800] "GET / HTTP/1.1" 404 0 "-" Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36 - 0.001 192.168.1.132:8005 502 0.000'>>/opt/nginx.log
一.Filebeat配置
vim /opt/filebeat/filebeat.yaml
filebeat.inputs:
- type: log
input_type: log
paths:
- /opt/*.log
tags: ["nginx", "access"]
fields:
logIndex: nginx
docType: nginx_access
fields_under_root: true
tail_files: false
output.logstash:
enabled: true
hosts: ["192.168.2.113:5044"]
二.Logstash配置
注意：这个grok正则配置要根据自己的格式来，这里用的都是内置封装好的正则。调试的方法可以去kibana或者grok在线网站去将 测试数据和grok规则 调试匹配好再写到配置文件中 ，节省时间。
%{IPV4:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:request_method} %{DATA:request_path} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:response_size} \"%{DATA:http_referer}\" %{DATA:user_agent} - %{NUMBER:request_time:float} %{IPV4:upstream_ip}:%{NUMBER:upstream_port:int} %{NUMBER:upstream_response_code:int} %{NUMBER:upstream_response_time:float}
vim /opt/logstash/config/logstash.conf
input {
beats {
port => 5044
client_inactivity_timeout => 36000 #filebeat连接logstash超时时间
}
}
filter {
# if [type] == "nginx_access"{
grok {
match => { "message" => "%{IPV4:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:request_method} %{DATA:request_path} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:response_size} \"%{DATA:http_referer}\" %{DATA:user_agent} - %{NUMBER:request_time:float} %{IPV4:upstream_ip}:%{NUMBER:upstream_port:int} %{NUMBER:upstream_response_code:int} %{NUMBER:upstream_response_time:float}" }
}
date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
target => "@timestamp" #将匹配到的timestamp写入到@timestap
remove_field => [ "timestamp" ] #匹配到结果以后去除timestamp字段
}
# }
}
output {
elasticsearch {
hosts => ["http://192.168.2.114:9200","http://192.168.2.115:9200","http://192.168.2.116:9200"]
index => "nginx-logs-%{+YYYY.MM.dd}"
document_type => "%{[@metadata][type]}"
user => "logstash_server"
password => "123456"
}
# stdout { codec => rubydebug } #在输出控制台调试
}
启动logstash
cd /opt/logstash/config && ../bin/logstash -f logstash.conf
整个过程可以都打开Filebeat和logstash日志来实时查看。
三.Kibana配置并查看
如果上面日志都无报错，kibana上查看es索引应该是有了的。在对应es的索引创建kibana的index pattern
在写入一次测试数据。再次查看，成功收集到了
遇到的问题
Q1：
保证正则没问题的情况下，写入测试数据后，filebeat收集到日志了，但是es上并没有创建nginx-logs-03-20这个日期的索引，因为我在logstash配置里定义es块的时候写明了是当天日期作为标识，结果并没有。
原因与解决：
写入测试数据的时候，其中的日志时间并不是当天，而是[17/Mar/2023:17:40:12 +0800]，17号，所以将这个日志信息写入到了nginx-logs-03-17这个索引中，因为测试阶段都是固定的数据，所以把日志改成当天的来模拟就好了，正式用的时候都自动收集实时信息，不会出现这个问题的。
Q2:
日志消息写入到了es，但是kibana查看不到？
原因与解决：
通过查es索引得知已写入到了es中，那问题可能就在kibana上，解决办法是将kibana仪表盘右上角时间先设定为this year即可看到。
原因为收集的日志数据中如，[20/Mar/2023:17:40:12 +0800]为@timestamp ，kibana是根据这个来显示日志的，所以我们要让收集时间和@timestamp相同。所以logstash中配置以下内容：
date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
target => "@timestamp" #将匹配到的timestamp写入到@timestap
remove_field => [ "timestamp" ] #匹配到结果以后去除timestamp字段
}
重启Logstash配置后删除es中索引和kibana对应的index pattern再次查看
总结
了解elfk的工作流，出现问题，一步步排查，通过日志，每个阶段每个阶段的来查看日志是否能写进这一阶段。
查看是否能写到es中，在看是否能过滤成功。调试filebeat---logstash时多通过控制输入/输出台来调试 可大幅提升效率
控制台输出调试：stdout { codec => rubydebug }
