logstash收集日志案例

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
  1. 为了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="{&quot;clientip&quot;:&quot;%h&quot;,&quot;ClientUser&quot;:&quot;%l&quot;,&quot;authenticated&quot;:&quot;%u&quot;,&quot;AccessTime&quot;: &quot;%t&quot;,&quot;method&quot;:&quot;%r&quot;,&quot;status&quot;:&quot;%s&quot;,&quot;SendBytes&quot;:&quot;%b&quot;,&quot;Query?string&quot;:&quot;%q&quot;,    &quot;partner&quot;:&quot;%{Referer}i&quot;,&quot;AgentVersion&quot;:&quot;%{User-Agent}i&quot;}" />
    ...
    
    # &quot; 表示双引号
    
    [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中的属性名
  2. 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}"
            }
        }
    }
  3. 重启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 9889

B:服务端

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