前言

目前线上环境nginx日志一天10亿左右，日志已经实时放入kafka队列中，想实时监控分析，首先，理清我们的需求：

1.实时监控 reponse的情况
2.实时监控 request path 的流量变化
3.实时监控请求时间
4.从request 里面解析其中的参数，监控变化量

上面最难的是从request解析指定的字段出来做监控，这里必须经过一次解析后才能做统计

想了几种方案：

最先想到的是搭建ES集群，利用ES集群实现实时聚合搜索
利用现在流行的flink spark storm 等实时流处理方案
利用现有的presto ，实现presto on kafka

一、方案选取

根据我们现有的实际情况，不方便增加更多的机器，也没有人专门写代码，所以选择了第3种kafka -->  logstash --> kafka  --> presto

二、各项配置

1.logstash配置

配置文件如下：


# 我们的nginx日志是按照竖线分割的  xx|xx|x|x|
input {kafka {topics   => "nginx"group_id => "logstash"bootstrap_servers => "192.168.x.xx:9092"codec => "plain"}
}filter{# 提取需要的字段，不是每个字段都用到了mutate {split => { "message" => "|" }add_field => { "time"   => "%{[message][3]}" "request" => "%{[message][7]}" "response" => "%{[message][9]}" "req_length" => "%{[message][5]}" "bytes" => "%{[message][10]}" "up_addr" => "%{[message][17]}" }remove_field => [ "message" ]}#   path?a=xx&b=xx&uuid=xxx&c=xx
#   解析request ，这里是提取了uuid kv {source => "request"field_split => "&?"target => "kv"include_keys => ["uuid"]}# 提取path字段，不要后面的参数，我们想统计那个path访问最多mutate {split => {"request" => "?" }add_field => { "req_url" => "%{[request][0]}"  }remove_field => ["request"]}#使用nginx 日志时间替换timestampdate {match => [ "timet" , "[dd/MMM/YYYY:HH:mm:ss Z]"]timezone => "+08:00"target => "@timestamp"locale => "en"}
# 增加一个字段，将timestamp转换为long类型ruby{code => "event.set('unix_ms_time',(event.get('@timestamp').to_f.round(3)*1000).to_i)"}
}output{#stdout{}kafka {bootstrap_servers => "192.168.1.xx:9092"codec => jsontopic_id => "nginxlog"}
}

处理后的消息为：

{"req_len": "1096","response": "200","req_url": "/get/api","unix_ms_time": 1618912479000,"@timestamp": "2021-04-20T09:54:39.000Z","bytes": "46","time": "[20/Apr/2021:17:54:39 +0800]","up_addr": "192.168.16.62:8080","kv": {"uuid": "898767826552698"},"@version": "1"
}

经过logstash解析后，提取了我们需要的信息，大大减少了无用的信息，此时可以使用presto on kafka直接查询了

2.presto 增加kafka connector

参考：https://prestodb.io/docs/current/connector.html

在presto catalog 目录下增加 kafka.properties文件

connector.name=kafka
kafka.nodes=192.168.xx.xx:9092
kafka.table-names=nginxlog
kafka.hide-internal-columns=false

在presto的etc目录下新建kafka目录，在kafka目录下新建nginxlog.json

{"tableName": "nginxlog","schemaName": "default","topicName": "nginxlog","message": {"dataFormat": "json","fields": [{"name": "req_leng","mapping": "req_len","type": "VARCHAR"},{"name": "bytes","mapping": "bytes","type": "VARCHAR"},{"name": "response","mapping": "response","type": "VARCHAR"},{"name": "req_url","mapping": "req_url","type": "VARCHAR"},{"name": "time_ms","mapping": "unix_ms_time","type": "BIGINT"},{"name": "up_addr","mapping": "up_addr","type": "VARCHAR"},{"name": "kv","mapping": "kv","type": "VARCHAR"}]}
}

重启presto server

3. 分析监控

 bin/presto  --server localhost:8080    --catalog kafka --schema defaultColumn       |  Type   | Extra |                   Comment
-------------------+---------+-------+---------------------------------------------req_len           | varchar |       |                                             bytes             | varchar |       |                                             response          | varchar |       |                                             req_url           | varchar |       |                                             time_ms           | bigint  |       |                                             up_addr           | varchar |       |                                             kv                | varchar |       |                                             _partition_id     | bigint  |       | Partition Id                                _partition_offset | bigint  |       | Offset for the message within the partition _message_corrupt  | boolean |       | Message data is corrupt                     _message          | varchar |       | Message text                                _message_length   | bigint  |       | Total number of message bytes               _key_corrupt      | boolean |       | Key data is corrupt                         _key              | varchar |       | Key text                                    _key_length       | bigint  |       | Total number of key bytes                   _timestamp        | bigint  |       | Offset Timestamp
(16 rows)

短横线开头的是默认的字段
_timestamp 表示了进入kafka的时间
_partition_id 表示在哪个分区

这样，如果我们就可以使用sql 直接进行统计分析了

-- 求response的分布
select count(*) as total,response  from nginxlog where  _timestamp > xx and _timestamp  < xx and _partition_id =x  group by response order by total desc;
-- 求一段时间内访问path分布
select count(*) as total,req_url from nginxlog where  _timestamp > xx and _timestamp  < xx and _partition_id =x  group by req_url  order by total desc;

总结

这里我们的nginx日志每秒才1W多条，然后kafka分了10partition，我们只取一个分区进行抽样分析也是很快的，当nginx 日志更大的时候，可能用presto查询就没有那么快了，到时候可能需要ES集群，splunk ，Loki 等专门的日志软件才能满足需求了，大家有更好的方案可以一起讨论讨论。

presto + kafka + logstash 实时监控分析nginx日志相关推荐

CentOS 7.2下ELK分析Nginx日志生产实战(高清多图)
注:本文系原创投稿本文以api.mingongge.com.cn域名为测试对象进行统计,日志为crm.mingongge.com.cn和risk.mingongge.com.cn请求之和(此二者域名 ...
python 正则分析nginx日志
有个需求要分析nginx日志,也懒得去研究logstach之类的开源工具,干脆直接写一个脚本,自己根据需求来实现: 先看日志格式:我们跟别人的不太一样,所以没办法了: 12.195.166.35 [1 ...
python分析nginx日志_利用python分析nginx日志
最近在学习python,写了个脚本分析nginx日志,练练手.写得比较粗糙,但基本功能可以实现. 脚本功能:查找出当天访问次数前十位的IP,并获取该IP来源,并将分析结果发送邮件到指定邮箱. 实现前两 ...
通过filebeat、logstash、rsyslog采集nginx日志的几种方式
由于nginx功能强大,性能突出,越来越多的web应用采用nginx作为http和反向代理的web服务器.而nginx的访问日志不管是做用户行为分析还是安全分析都是非常重要的数据源之一.如何有效便捷的 ...
elk平台分析nginx日志的基本搭建
一.elk套件介绍 ELK 由 ElasticSearch . Logstash 和 Kiabana 三个开源工具组成.官方网站: https://www.elastic.co/products El ...
linux nginx 日志查看,查看nginx日志_Linux系统怎么分析Nginx日志
摘要腾兴网为您分享:Linux系统怎么分析Nginx日志,追书神器,有信,虚拟机,天气预报等软件知识,以及云软件,东方财富股票软件,扑飞,名片扫描王,微信加人软件安卓,微主题,每日金股,电脑硬盘检测 ...
使用GoAccess分析Nginx日志
下载 GoAccess 的源代码.编译和安装: http://www.goaccess.io/download # wget http://tar.goaccess.io/goaccess-0.9.6 ...
python分析nginx日志
利用python脚本分析nginx日志内容,默认统计ip.访问url.状态,可以通过修改脚本统计分析其他字段. 一.脚本运行方式 python count_log.py -f med.xxxx.com ...
awk分析nginx日志里面的接口响应时间
2019独角兽企业重金招聘Python工程师标准>>> 最近,有客户反应客户端卡,老板集合技术人员开会讨论,找出慢的原因,由此产生了分析nginx响应时间,由于线上环境nginx日志 ...

presto + kafka + logstash 实时监控分析nginx日志

文章目录

前言