OpenResty+Lua+Kafka收集日志

Lua 是一个小巧的脚本语言。Lua 脚本可以很容易的被 C、C++ 代码调用，也可以反过来调用 C、C++ 的函数，这使得 Lua 在应用程序中可以被广泛应用。不仅仅作为扩展脚本，也可以作为普通的配置文件，代替 XML、ini 等文件格式，并且更容易理解和维护。

Lua 由标准 C 编写而成，代码简洁优美，几乎在所有操作系统和平台上都可以编译、运行。一个完整的 Lua 解释器不过200k，在所有脚本引擎中，Lua 的速度是最快的。这一切都决定了 Lua 是作为嵌入式脚本的最佳选择。

#### 场景描述

对于线上大流量服务或者需要上报日志的 Nginx 服务，每天会产生大量的日志，这些日志非常有价值。可用于计数上报、用户行为分析、接口质量、性能监控等需求。但传统 Nginx 记录日志的方式数据会散落在各自 Nginx 上，而且大流量日志本身对磁盘也是一种冲击。

我们需要把这部分 Nginx 日志统一收集汇总起来，收集过程和结果需要满足如下需求:

- 支持不同业务获取数据，如监控业务，数据分析统计业务，推荐业务等。
- 数据实时性
- 高性能保证

#### 技术方案

得益于 OpenResty 和 Kafka 的高性能，我们可以非常轻量高效的实现当前需求。

方案描述：
1、线上请求打向 Nginx 后，使用lua完成日志整理:如统一日志格式，过滤无效请求，分组等。
2、根据不同业务的 Nginx 日志,划分不同的 topic。
3、Lua 实现 producter 异步发送到 Kafka 集群。
4、对不同日志感兴趣的业务组实时消费获取日志数据。

#### 相关技术

openresty: http://openresty.org
kafka: http://kafka.apache.org
lua-resty-kafka: https://github.com/doujiang24/lua-resty-kafka

#### 配置

1、nginx.conf

```plaintext
user  www;
worker_processes  auto;
worker_rlimit_nofile 100000;

events {
    worker_connections  102400;
    multi_accept on;
    use epoll;
}

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  /var/log/nginx/access.log  main;

resolver 8.8.8.8;
    # resolver 127.0.0.1 valid=3600s;

sendfile        on;
    keepalive_timeout  65;
    underscores_in_headers on;
    gzip  on;

include /opt/openresty/nginx/conf/conf.d/common.conf;
}
```

2、common.conf

```plaintext
# api
lua_package_path "/opt/openresty/lualib/resty/kafka/?.lua;;";
lua_package_cpath "/opt/openresty/lualib/?.so;;";

lua_shared_dict ngx_cache 128m;  # cache
lua_shared_dict cache_lock 100k; # lock for cache

server {
    listen       80;
    server_name  192.168.0.66;
    root         html;
    lua_need_request_body on; # 打开获取消息体的开关，以便能获取到消息体

access_log /var/log/nginx/message.access.log  main;
    error_log  /var/log/nginx/message.error.log  notice;

location = /lzp/message {
        lua_code_cache on;
        charset utf-8;
        default_type 'application/json';
        content_by_lua_file "/opt/openresty/nginx/lua/testMessage_kafka.lua";
    }
}
```

3、testMessage_kafka.lua

```lua
-- require需要resty.kafka.producer的lua脚本，没有会报错
local producer = require("resty.kafka.producer")

-- 转换json为字符串需要用到
local cjson = require("cjson.safe")

-- kafka的集群信息，单机也是可以的
local broker_list = {
    {host = "192.168.0.66", port = 9092},
}

-- 定义最终kafka接受到的数据是怎样的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

-- 定义kafka同步生产者，也可设置为异步 async
-- -- 注意！！！当设置为异步时，在测试环境需要修改batch_num，默认是200条，若大不到200条kafka端接受不到消息
-- -- encode()将log_json日志转换为字符串
-- -- 发送日志消息，send配套之第一个参数topic:
-- -- 发送日志消息，send配套之第二个参数key，用于kafka路由控制:
-- -- key为nil(空)时，一段时间向同一partition写入数据
-- -- 指定key，按照key的hash写入到对应的partition

-- -- batch_num修改为1方便测试
local bp = producer:new(broker_list, { producer_type = "async", batch_num = 1 })
-- local bp = producer:new(broker_list, { producer_type = "async" })
-- 转换json为字符串
local sendMsg = cjson.encode(log_json)
-- 发送日志消息,send第二个参数key,用于kafka路由控制:
-- key为nil(空)时，一段时间向同一partition写入数据
-- 指定key，按照key的hash写入到对应的partition
local ok, err = bp:send("testMessage", nil, sendMsg)
if not ok then
   ngx.log(ngx.ERR, 'kafka send err:', err)
elseif ok then
   ngx.say("the message send successful")
else
   ngx.say("未知错误")
end
```

#### 总结
使用 Openresty+Lua+Kafka 就可以将用户的埋点数据实时采集到 Kafka 集群中，并且 Openresty 是基于 Nginx 的，而 Nginx 能处理上万的并发量，所以即使用户的数据在短时间内激增，这套架构也能轻松的应对，不会导致集群崩溃。另一方面，若数据过多导致集群的超负荷，我们也可以随时加多一台机器，非常方便。

另外一个小小的拓展：若业务数据非常多，需要发送到不同的 topic 中，我们也不用编写多个脚本，而是可以联系后端在 json 格式里面加一个字段，这个字段的值就是 topic 的名称。我们只需要编写一个通用脚本，解析 json 数据将 topic 名称拿出来就可以了。

OpenResty+Lua+Kafka收集日志

相关文章

热门标签

热门文章