logstash收集日志案例
本文最后更新于：2023年12月5日 晚上
tomcat 日志
tomcat 默认的格式比较简单，包含 ip、date、method、url、status code 等信息，例如：
10.0.0.1 - - [06/Mar/2021:09:26:09 +0800] "GET / HTTP/1.1" 200 11156
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /tomcat.svg HTTP/1.1" 200 67795
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /tomcat.css HTTP/1.1" 200 5542
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /bg-nav.png HTTP/1.1" 200 1401
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /bg-middle.png HTTP/1.1" 200 1918
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /bg-upper.png HTTP/1.1" 200 3103
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /bg-button.png HTTP/1.1" 200 713
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /asf-logo-wide.svg HTTP/1.1" 200 27235
10.0.0.1 - - [06/Mar/2021:09:26:10 +0800] "GET /favicon.ico HTTP/1.1" 200 21630
10.0.0.1 - - [06/Mar/2021:09:26:12 +0800] "GET /manager/status HTTP/1.1" 403 3446
10.0.0.1 - - [06/Mar/2021:09:26:14 +0800] "GET /manager/html HTTP/1.1" 403 3446
10.0.0.1 - - [06/Mar/2021:09:26:19 +0800] "GET / HTTP/1.1" 200 11156
10.0.0.1 - - [06/Mar/2021:09:26:19 +0800] "GET /favicon.ico HTTP/1.1" 200 21630为了 kibana 能单独统计每个字段，需要日志记录成 json 格式，当然也可以是其他格式，只要有对应的 codec 插件可以解析就行
[root@elk2-ljk conf]$pwd /usr/local/tomcat/conf [root@elk2-ljk conf]$vim server.xml # 自定义日志格式 为json格式 ... <Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs" prefix="tomcat_access_log" suffix=".log" pattern="{"clientip":"%h","ClientUser":"%l","authenticated":"%u","AccessTime": "%t","method":"%r","status":"%s","SendBytes":"%b","Query?string":"%q", "partner":"%{Referer}i","AgentVersion":"%{User-Agent}i"}" /> ... # " 表示双引号 [root@elk2-ljk conf]$systemctl restart tomcat.service # 重启tomcat # 查看日志，已经成功修改为json格式 [root@elk2-ljk logs]$cat ../logs/tomcat_access_log.2021-03-06.log {"clientip":"10.0.0.1","ClientUser":"-","authenticated":"-","AccessTime":"[06/Mar/2021:12:50:56 +0800]","method":"GET / HTTP/1.1","status":"200","SendBytes": "11156","Query?string":"","partner":"-","AgentVersion":"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.190 Safari/537.36"} {"clientip":"10.0.0.1","ClientUser":"-","authenticated":"-","AccessTime":"[06/Mar/2021:12:50:56 +0800]","method":"GET /favicon.ico HTTP/1.1","status":"200", "SendBytes":"21630","Query?string":"","partner":"http://10.0.1.122:8080/","AgentVersion":"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.190 Safari/537.36"} {"clientip":"10.0.0.1","ClientUser":"-","authenticated":"-","AccessTime":"[06/Mar/2021:12:51:00 +0800]","method":"GET / HTTP/1.1","status":"200","SendBytes": "11156","Query?string":"","partner":"-","AgentVersion":"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.190 Safari/537.36"}百度搜索 “tomcat 日志格式” 或 “apache 日志格式”，下面是部分格式说明：
％a - 远程IP地址 ％A - 本地IP地址 ％b - 发送的字节数，不包括HTTP头，或“ - ”如果没有发送字节 ％B - 发送的字节数，不包括HTTP头 ％h - 远程主机名 ％H - 请求协议 ％l (小写的L)- 远程逻辑从identd的用户名（总是返回' - '） ％m - 请求方法 ％p - 本地端口 ％q - 查询字符串（在前面加上一个“？”如果它存在，否则是一个空字符串 ％r - 第一行的要求 ％s - 响应的HTTP状态代码 ％S - 用户会话ID ％t - 日期和时间，在通用日志格式 ％u - 远程用户身份验证 ％U - 请求的URL路径 ％v - 本地服务器名 ％D - 处理请求的时间（以毫秒为单位） ％T - 处理请求的时间（以秒为单位） ％I （大写的i） - 当前请求的线程名称 ％{XXX}i xxx代表传入的头(HTTP Request) ％{XXX}o xxx代表传出的响应头(Http Resonse) ％{XXX}c xxx代表特定的Cookie名 ％{XXX}r xxx代表ServletRequest属性名 ％{XXX}s xxx代表HttpSession中的属性名logstash 配置文件：
input { file { type => "tomcat-access-log" path => "/usr/local/tomcat/logs/tomcat_access_log.*.log" start_position => "end" stat_interval => 3 codec => "json" } } output { if [type] == "tomcat-access-log" { elasticsearch { hosts => ["10.0.1.121:9200"] index => "mytest-%{type}-%{+xxxx.ww}" } } }重启 logstash，注意要以 root 用户身份启动，否则无法采集数据
[root@elk2-ljk conf.d]$systemctl restart logstash.service
java 日志
基于 java 开发的应用，都会有 java 日志，java 日志会记录 java 的报错信息，但是一个报错会产生多行日志，例如：
为了方便观察，需要将一个报错的多行日志合并为一行，以 elasticsearch 为例：
input {
file {
type => "elasticsearch-java-log"
path => "/data/elasticsearch/logs/es-cluster.log"
start_position => "beginning"
stat_interval => 3
codec => multiline {
pattern => "^\["
negate => true
what => "previous"
}
}
}
output {
elasticsearch {
hosts => ["10.0.1.121:9200"]
index => "mytest-%{type}-%{+yyyy.MM}"
}
}查看 kibana：
nginx 访问日志
和采集 tomcat 日志类似，重点是把 nginx 日志修改为 json 格式
# 注意：log_format要写在server外面
[root@47105171233 vhost]$cat lujinkai.cn.conf
log_format access_json '{"@timestamp":"$time_iso8601",'
'"host":"$server_addr",'
'"clientip":"$remote_addr",'
'"size":$body_bytes_sent,'
'"responsetime":$request_time,'
'"upstreamtime":"$upstream_response_time",'
'"upstreamhost":"$upstream_addr",'
'"http_host":"$host",'
'"url":"$uri",'
'"domain":"$host",'
'"xff":"$http_x_forwarded_for",'
'"referer":"$http_referer",'
'"status":"$status"}';
server {
listen 80;
server_name lujinkai.cn;
access_log /data/wwwlogs/lujinkai.cn_nginx.log access_json;
rewrite / http://blog.lujinkai.cn permanent;
}
# 日志成功转为json格式
[root@47105171233 wwwlogs]$tail -f lujinkai.cn_nginx.log
{"@timestamp":"2021-03-06T18:20:13+08:00","host":"10.0.0.1","clientip":"113.120.245.191","size":162,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"lujinkai.cn","url":"/","domain":"lujinkai.cn","xff":"-","referer":"-","status":"301"}
{"@timestamp":"2021-03-06T18:20:13+08:00","host":"10.0.0.1","clientip":"113.120.245.191","size":162,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"lujinkai.cn","url":"/robots.txt","domain":"lujinkai.cn","xff":"-","referer":"-","status":"301"}TCP/UDP 日志
场景：没有安装 logstash 的服务器（A），向安装了 logstash 的服务器（B）发送日志信息，这个场景不多
实现：A 通过 nc 命令给 B 发送日志信息，B 监听本地的对应端口，接收数据
A：客户端
[root@elk2-ljk ~]$nc 10.0.1.121 9889B：服务端
input {
tcp {
port => 9889
type => "tcplog"
mode => "server"
}
}
output {
stdout {
codec => rubydebug
}
}通过 rsyslog 收集 haproxy 日志
有些设备无法安装 logstash，例如 路由器、交换机，但是厂家内置了 rsyslog 功能，这里以 haproxy 为例
# 1. haproxy.cfg 定义haproxy的日志设备为local6
log 127.0.0.1 local6 info
# 2. rsyslog.conf
module(load="imudp")
input(type="imudp" port="514")
module(load="imtcp")
input(type="imtcp" port="514")
local6.* @@10.0.1.122
# 3. 10.0.1.122主机监听本机的udp514端口，接收日志数据
input{
syslog {
type => "ststem-rsyslog"
}
}
output{
elasticsearch {
hosts => ["10.0.1.123:9200"]
index => "logstash-rsyslog-%{+YYYY.MM.dd}"
}
}filebeat 收集日志并写入 redis/kafka
考虑到 elasticsearch 的性能问题，通常不会直接往 elasticsearch 写日志，会加 redis 或 kafka 缓冲一下
- 有少数场景，需要收集多个不同格式的日志，例如有的是 syslog 格式，有的是 json 格式，所有的日志都 output 到 redis 的一个 list 中，因为 logstash 的 input 没有条件判断，只能配置一个 codec，所以 logstash 可以将不同的日志发送到 elasticsearch 的不同 index，却无法对不同的日志格式配置不同的 codec，这样的数据最后展示在 kibana 也没有意义，解决方法是在日志收集和日志缓存中间在加一个 logstash（可以共用日志提取及过滤的 logstash），将不同的日志转发到 redis 的不同 list
- 日志缓存如果用 redis，需要大内存，推荐 32G；如果用 kafka，16G 就够，因为 kafka 存储数据到磁盘
日志收集实战
从左向右看，当要访问 ELK 日志统计平台的时候，首先访问的是两台 nginx+keepalived 做的负载高可用，访问的地址是 keepalived 的 IP，当一台 nginx 代理服务器挂掉之后也不影响访问，然后 nginx 将请求转发到 kibana，kibana 再去 elasticsearch 获取数据，elasticsearch 是两台做的集群，数据会随机保存在任意一台 elasticsearch 服务器，redis 服务器做数据的临时保存，避免 web 服务器日志量过大的时候造成的数据收集与保存不一致导致的日志丢失，可以临时保存到 redis，redis 可以是集群，然后再由 logstash 服务器在非高峰时期从 redis 持续的取出即可，另外有一台 mysql 数据库服务器，用于持久化保存特定的数据，web 服务器的日志由 filebeat 收集之后发送给另外的一台 logstash，再有其写入到 redis 即可完成日志的收集，从图中可以看出，redis 服务器处于前端结合的最中间，其左右都要依赖于 redis 的正常运行，web 服务删个日志经过 filebeat 收集之后通过日志转发层的 logstash 写入到 redis 不同的 key 当中，然后提取层 logstash 再从 redis 将数据提取并安按照不同的类型写入到 elasticsearch 的不同 index 当中，用户最终通过 nginx 代理的 kibana 查看到收集到的日志的具体内容
通过坐标地图统计客户 IP 所在城市
https://www.elastic.co/guide/en/logstash/current/plugins-filters-geoip.html
日志写入数据库
写入数据库的目的是用于持久化保存重要数据，比如状态码、客户端 IP、客户端浏览器版本等等，用于后期按月做数据统计等
写个脚本从，定期从 elasticsearch 中获取数据，写入到 PostgreSQL