Logstash Demo¶
实际业务日志解析案例¶
Demo 1:Nginx日志解析¶
1.1 解析流程
Filebeat –> Kafka –> Logstash –> Elasticsearch
1.2 Nginx日志配置要求
1.2.1 统一Nginx日志格式
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for" "$request_time" "$http_host"';
1.2.2 字段说明
| 字段名 | 含义 | | ——————– | ———————— | | remote_addr | 客户端请求IP地址 | | remote_user | 用户 | | time_local | 时间戳 | | request | 请求内容 | | status | HTTP状态码 | | body_bytes_sent | 发送的数据包大小 | | http_referer | 来源页面链接 | | http_user_agent | 用户客户端信息 | | http_x_forwarded_for | 反向代理时用户实际IP地址 | | request_time | 请求响应时间 | | http_host | 请求域名或者IP |
1.2.3 Nginx日志配置规范
1.2.3.1 日志文件名以 ${program_name}_access.log 命名,存放路径 /opt/nginx/logs
1.2.3.2 凌晨日志切割,存放在/opt/nginx/logs/archive
1.2.4 Nginx日志解析规则
%{COMBINEDAPACHELOG} \"%{IP:x_forwarded_for}\" %{QS:request_time} %{QS:http_host}
1.3 Filebeat配置参考
输出到Kafka:https://www.elastic.co/guide/en/beats/filebeat/current/kafka-output.html
filebeat.inputs:
- type: log
# Change to true to enable this input configuration.
enabled: true
# Paths that should be crawled and fetched. Glob based paths.
paths:
- /opt/nginx/logs/dmblapp_access.log
tags: ["dmblapp"]
fields:
serverip: 192.168.2.5 # 新增当前服务器IP字段
- type: log
# Change to true to enable this input configuration.
enabled: true
# Paths that should be crawled and fetched. Glob based paths.
paths:
- /opt/nginx/logs/lhb_access.log
tags: ["lhb"]
fields:
serverip: 192.168.2.5
output.kafka:
# initial brokers for reading cluster metadata
hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
# message topic selection + partitioning
topic: nginx_access_log
required_acks: 1
compression: gzip
max_message_bytes: 1000000
version: 0.10.0.0
1.4 Kafka配置说明
Kafka版本建议选择0.11以上。使用KafkaManager管理Kafka集群,Topic创建规则不使用点,如nginx.access调整为nginx_access。副本数和Broker数相同,分区数是Broker整数倍。
1.5 Logstash配置参考
input {
kafka {
bootstrap_servers => "kafka1:9092 kafka2:9092 kafka3:9092"
group_id => "nginx_access_consumer" # 多台消费必须统一group_id
topics => ["nginx_access_log"]
consumer_threads => 3
auto_offset_reset => "earliest"
enable_auto_commit => "true"
codec => "json" # json
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG} \"%{IP:x_forwarded_for}\" %{QS:request_time} %{QS:http_host}"}
}
date {
match => ["timestamp","dd/MMM/YYYY:HH:mm:ss +0800"]
}
geoip { # IP地址解析
source => "x_forwarded_for"
target => "geoip"
database => "/etc/logstash/GeoLite2-City.mmdb"
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
mutate {
convert => [ "[geoip][coordinates]", "float"]
}
useragent { # 客户端信息解析
target => "useragent"
source => "agent"
}
}
output {
elasticsearch {
hosts => ["es1:9200","es2:9200","es3:9200"]
index => "nginx_access-%{+yyyy.MM.dd}"
}
}
1.6 Kibana搜索与展示
根据index在Kibana上建立nginx_access-*,然后在Discover中可以查看具体字段解析以及搜索内容。