nginx+lua+kafka 编写 在线日志上报系统

案例一

rewrite_by_lua ‘
--引入openresty自带的json处理对象
local cjson = require("cjson")
local producer = require "resty.kafka.producer"
-- 定义kafka broker地址，ip需要和kafka的host.name配置一致
local broker_list = {
{ host = "192.168.115.28", port = 9092 },
{ host = "192.168.115.29", port = 9092 },
{ host = "192.168.115.30", port = 9092 }
}
-- 定义json便于日志数据整理收集
local log_json = {}

local request_method = ngx.req.get_method
if "GET" == request_method then
-- 普通get请求
ngx.log(ngx.ERR,"不支持GET请求")
log_json["start_time"]=ngx.req.start_time()
log_json["header"]=ngx.req.raw_header()
log_json["uri"]=ngx.req.get_uri_args()
log_json["headers"]=ngx.req.get_headers()
log_json["body"]=ngx.req.read_body()
log_json["body_data"]=ngx.req.get_body_data()
elseif "POST" == request_method then
ngx.req.read_body()
local body_data = ngx.req.get_body_data() --body_data可是符合http协议的请求体，不是普通的字符串
--请求体的size大于nginx配置里的client_body_buffer_size，则会导致请求体被缓冲到磁盘临时文件里，client_body_buffer_size默认是8k或者16k
if not body_data then
ngx.log(ngx.WARN,"no request body found" )
end
log_json["start_time"]=ngx.req.start_time()
log_json["header"]=ngx.req.raw_header()
log_json["uri"]=ngx.req.get_uri_args()
log_json["post"]=ngx.req.get_post_args()
log_json["headers"]=ngx.req.get_headers()
log_json["body"]=ngx.req.read_body()
log_json["body_data"]=ngx.req.get_body_data()
end
ngx.log(ngx.INFO,"log_json",cjson.encode(log_json))
-- 转换json为字符串
local message = cjson.encode(log_json);
-- 定义kafka异步生产者
local bp = producer:new(broker_list, { producer_type = "async" })
-- 发送日志消息,send第二个参数key,用于kafka路由控制:
-- key为nill(空)时，一段时间向同一partition写入数据
-- 指定key，按照key的hash写入到对应的partition
local ok, err = bp:send("postman1", nil, message)

if not ok then
ngx.log(ngx.ERR, "kafka send err:", err)
return
end

‘;

案例二

server {
listen 80;
server_name localhost;
location /favicon.ico {
root html;
index index.html index.htm;
}
location / {
proxy_connect_timeout 8;
proxy_send_timeout 8;
proxy_read_timeout 8;
proxy_buffer_size 4k;
proxy_buffers 512 8k;
proxy_busy_buffers_size 8k;
proxy_temp_file_write_size 64k;
proxy_next_upstream http_500 http_502 http_503 http_504 error timeout invalid_header;
root html;
index index.html index.htm;
proxy_pass http://rc;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
# 使用log_by_lua 包含lua代码,因为log_by_lua指令运行在请求最后且不影响proxy_pass机制
rewrite_by_lua ‘
-- 引入lua所有api
local cjson = require "cjson"
local producer = require "resty.kafka.producer"
-- 定义kafka broker地址，ip需要和kafka的host.name配置一致
local broker_list = {
{ host = "10.10.78.52", port = 9092 },
}
-- 定义json便于日志数据整理收集
local log_json = {}
log_json["uri"]=ngx.var.uri
log_json["args"]=ngx.var.args
log_json["host"]=ngx.var.host
log_json["request_body"]=ngx.var.request_body
log_json["remote_addr"] = ngx.var.remote_addr
log_json["remote_user"] = ngx.var.remote_user
log_json["time_local"] = ngx.var.time_local
log_json["status"] = ngx.var.status
log_json["body_bytes_sent"] = ngx.var.body_bytes_sent
log_json["http_referer"] = ngx.var.http_referer
log_json["http_user_agent"] = ngx.var.http_user_agent
log_json["http_x_forwarded_for"] = ngx.var.http_x_forwarded_for
log_json["upstream_response_time"] = ngx.var.upstream_response_time
log_json["request_time"] = ngx.var.request_time
-- 转换json为字符串
local message = cjson.encode(log_json);
-- 定义kafka异步生产者
local bp = producer:new(broker_list, { producer_type = "async" })
-- 发送日志消息,send第二个参数key,用于kafka路由控制:
-- key为nill(空)时，一段时间向同一partition写入数据
-- 指定key，按照key的hash写入到对应的partition
local ok, err = bp:send("test1", nil, message)

if not ok then
ngx.log(ngx.ERR, "kafka send err:", err)
return
end
‘;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
}

参考：https://blog.csdn.net/u011239989/article/details/52239785

#user nobody;
worker_processes 1;

#error_log logs/error.log;
#error_log logs/error.log notice;
error_log logs/error.log info;

pid logs/nginx.pid;

events {
use epoll;
worker_connections 1024;
}

http {
include mime.types;
default_type application/octet-stream;

log_format main ‘$remote_addr - $remote_user [$time_local] "$request" ‘
‘$status $body_bytes_sent "$http_referer" ‘
‘"$http_user_agent" "$http_x_forwarded_for"‘;

access_log logs/access.log main;

sendfile on;
tcp_nopush on;

#keepalive_timeout 0;
keepalive_timeout 65;

#gzip on;
lua_package_path "lua-resty-kafka/lib/?.lua;;";
client_body_buffer_size 128m;
server {
listen 80;
server_name localhost;
location / {
root html;
index index.html index.htm;
}
location ^~ /api/getuserinfo {
rewrite_by_lua ‘
-- 引入lua所有api
local cjson = require("cjson")
local producer = require "resty.kafka.producer"
-- 定义kafka broker地址，ip需要和kafka的host.name配置一致
local broker_list = {
{ host = "192.168.115.28", port = 9092 },
{ host = "192.168.115.29", port = 9092 },
{ host = "192.168.115.30", port = 9092 }
}
-- 定义json便于日志数据整理收集
local log_json = {}
ngx.req.read_body()
log_json["start_time"]=ngx.req.start_time()
log_json["header"]=ngx.req.raw_header()
log_json["uri"]=ngx.req.get_uri_args()
log_json["post"]=ngx.req.get_post_args()
log_json["headers"]=ngx.req.get_headers()
log_json["body_data"]=ngx.req.get_body_data()
log_json["uri"]=ngx.var.uri
log_json["args"]=ngx.var.args
log_json["host"]=ngx.var.host
log_json["request_body"]=ngx.var.request_body
log_json["remote_addr"] = ngx.var.remote_addr
log_json["remote_user"] = ngx.var.remote_user
log_json["time_local"] = ngx.var.time_local
log_json["status"] = ngx.var.status
log_json["body_bytes_sent"] = ngx.var.body_bytes_sent
log_json["http_referer"] = ngx.var.http_referer
log_json["http_user_agent"] = ngx.var.http_user_agent
log_json["http_x_forwarded_for"] = ngx.var.http_x_forwarded_for
log_json["upstream_response_time"] = ngx.var.upstream_response_time
log_json["request_time"] = ngx.var.request_time
-- 转换json为字符串
local message = cjson.encode(log_json);
-- 定义kafka异步生产者
local bp = producer:new(broker_list, { producer_type = "async" })
-- 发送日志消息,send第二个参数key,用于kafka路由控制:
-- key为nill(空)时，一段时间向同一partition写入数据
-- 指定key，按照key的hash写入到对应的partition
local ok, err = bp:send("postman1", nil, message)

if not ok then
ngx.log(ngx.ERR, "kafka send err:", err)
return
end
‘;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
}
}

原文地址：https://www.cnblogs.com/net2817/p/10601824.html

时间： 2024-10-10 16:09:48

nginx+lua+kafka 编写 在线日志上报系统的相关文章

ELK+kafka构建日志收集系统

ELK+kafka构建日志收集系统 原文  http://lx.wxqrcode.com/index.php/post/101.html 背景: 最近线上上了ELK,但是只用了一台Redis在中间作为消息队列,以减轻前端es集群的压力,Redis的集群解决方案暂时没有接触过,并且Redis作为消息队列并不是它的强项:所以最近将Redis换成了专业的消息信息发布订阅系统Kafka, Kafka的更多介绍大家可以看这里: 传送门 ,关于ELK的知识网上有很多的哦, 此篇博客主要是总结一下目前线上这个

flume + kafka + sparkStreaming + HDFS 构建实时日志分析系统

一.Hadoop配置安装 注意:apache提供的hadoop-2.6.0的安装包是在32位操作系统编译的,因为hadoop依赖一些C++的本地库, 所以如果在64位的操作上安装hadoop-2.4.1就需要重新在64操作系统上重新编译 1.修改Linux主机名 2.修改IP 3.修改主机名和IP的映射关系 ######注意######如果你们公司是租用的服务器或是使用的云主机(如华为用主机.阿里云主机等) /etc/hosts里面要配置的是内网IP地址和主机名的映射关系 4.关闭防火墙 5.s

scribe、chukwa、kafka、flume日志系统对比

scribe.chukwa.kafka.flume日志系统对比 1. 背景介绍许多公司的平台每天会产生大量的日志(一般为流式数据,如,搜索引擎的pv,查询等),处理 这些日志需要特定的日志系统,一般而言,这些系统需要具有以下特征:(1) 构建应用系统和分析系统的桥梁,并将它们之间的关联解耦:(2) 支持近实时的在线分析系统和类似于Hadoop之类的离线分析系统:(3) 具有高可扩展性.即:当数据量增加时,可以通过增加节点进行水平扩展. 本文从设计架构,负载均衡,可扩展性和容错性等方面对比了当今开

ELKR分布式搭建nginx日志分析系统

ELKR分布式搭建nginx日志分析系统 一.名词介绍 1.什么是elk ELK 其实并不是一款软件,而是一整套解决方案,是三个软件产品的首字母缩写,Elasticsearch,Logstash 和 Kibana.这三款软件都是开源软件,通常是配合使用. 2.Elasticsearch 2.1.Elasticsearch介绍 Elasticsearch 是一个实时的分布式搜索和分析引擎,它可以用于全文搜索,结构化搜索以及分析.它是一个建立在全文搜索引擎 Apache Lucene 基础上的搜索引

Kafka+Zookeeper+Filebeat+ELK 搭建日志收集系统

ELK ELK目前主流的一种日志系统,过多的就不多介绍了 Filebeat收集日志,将收集的日志输出到kafka,避免网络问题丢失信息 kafka接收到日志消息后直接消费到Logstash Logstash将从kafka中的日志发往elasticsearch Kibana对elasticsearch中的日志数据进行展示 image 环境介绍: 软件版本: - Centos 7.4 - java 1.8.0_45 - Elasticsearch 6.4.0 - Logstash 6.4.0 - F

第11天 日志收集系统kafka库实战

本节主要内容: 1. 日志收集系统设计2. 日志客户端开发 1. 项目背景    a. 每个系统都有日志,当系统出现问题时,需要通过日志解决问题    b. 当系统机器比较少时,登陆到服务器上查看即可满足    c. 当系统机器规模巨大,登陆到机器上查看几乎不现实 2. 解决方案    a. 把机器上的日志实时收集,统一的存储到中心系统    b. 然后再对这些日志建立索引,通过搜索即可以找到对应日志    c. 通过提供界面友好的web界面,通过web即可以完成日志搜索 3. 面临的问题   

使用nginx lua实现网站统计中的数据收集

使用nginx lua实现网站统计中的数据收集 导读 网站数据统计分析工具是各网站站长和运营人员经常使用的一种工具,常用的有 谷歌分析.百度统计和腾讯分析等等.所有这些统计分析工具的第一步都是网站访问数据的收集.目前主流的数据收集方式基本都是基于javascript的.在此简要分析数据收集的原理,并按照步骤,带领大家一同搭建一个实际的数据收集系统. 数据收集原理分析 简单来说,网站统计分析工具需要收集到用户浏览目标网站的行为(如打开某网页.点击某按钮.将商品加入购物车等)及行为附加数据(如某下单

开源日志分析系统ELK平台搭建部署

开源日志分析系统ELK平台搭建部署 一.前言 日志主要包括系统日志.应用程序日志和安全日志.系统运维和开发人员可以通过日志了解服务器软硬件信息.检查配置过程中的错误及错误发生的原因.经常分析日志可以了解服务器的负荷,性能安全性,从而及时采取措施纠正错误. 通常,日志被分散的储存不同的设备上.如果你管理数十上百台服务器,你还在使用依次登录每台机器的传统方法查阅日志.这样是不是感觉很繁琐和效率低下.当务之急我们使用集中化的日志管理,例如:开源的syslog,将所有服务器上的日志收集汇总. 集中化管理

ELK日志分析系统的应用

收集和分析日志是应用开发中至关重要的一环,互联网大规模.分布式的特性决定了日志的源头越来越分散, 产生的速度越来越快,传统的手段和工具显得日益力不从心.在规模化场景下,grep.awk 无法快速发挥作用,我们需要一种高效.灵活的日志分析方式,可以给故障处理,问题定位提供更好的支持.基于全文搜索引擎 Lucene 构建的 ELKstack 平台,是目前比较流行的日志收集方解决方案. ELK系统的部署按照官方文档操作即可,相关资料也很多,这篇文章更多的关注三个组件的设计和实现,帮助大家了解这个流行的