You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@apisix.apache.org by to...@apache.org on 2020/11/30 12:23:15 UTC
[apisix] 01/01: fix: fixed the non effective config update in
http-logger
This is an automated email from the ASF dual-hosted git repository.
tokers pushed a commit to branch fix/batch-processor
in repository https://gitbox.apache.org/repos/asf/apisix.git
commit add5906f87c9131213ef6221ceac1f9595ac3bea
Author: Alex Zhang <zc...@gmail.com>
AuthorDate: Mon Nov 30 20:06:20 2020 +0800
fix: fixed the non effective config update in http-logger
---
apisix/plugins/http-logger.lua | 46 +++++++++++--
apisix/plugins/tcp-logger.lua | 2 +-
t/plugin/http-logger.t | 146 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 187 insertions(+), 7 deletions(-)
diff --git a/apisix/plugins/http-logger.lua b/apisix/plugins/http-logger.lua
index 7a79102..1b2de40 100644
--- a/apisix/plugins/http-logger.lua
+++ b/apisix/plugins/http-logger.lua
@@ -21,19 +21,22 @@ local core = require("apisix.core")
local http = require("resty.http")
local url = require("net.url")
local plugin = require("apisix.plugin")
+
local ngx = ngx
local tostring = tostring
local pairs = pairs
-local ipairs = ipairs
+local ipairs = ipairs
local str_byte = string.byte
-
+local timer_at = ngx.timer.at
local plugin_name = "http-logger"
+local stale_timer_running = false
local buffers = {}
local lru_log_format = core.lrucache.new({
ttl = 300, count = 512
})
+local plugin_conf
local schema = {
type = "object",
@@ -92,6 +95,8 @@ local function send_http_data(conf, log_message)
local host = url_decoded.host
local port = url_decoded.port
+ core.log.info("sending a batch logs to ", conf.uri)
+
if ((not port) and url_decoded.scheme == "https") then
port = 443
elseif not port then
@@ -169,6 +174,23 @@ local function gen_log_format(metadata)
end
+-- remove stale objects from the memory after timer expires
+local function remove_stale_objects(premature)
+ if premature then
+ return
+ end
+
+ for key, batch in ipairs(buffers) do
+ if #batch.entry_buffer.entries == 0 and #batch.batch_to_process == 0 then
+ core.log.debug("removing batch processor stale object, route id:", tostring(key))
+ buffers[key] = nil
+ end
+ end
+
+ stale_timer_running = false
+end
+
+
function _M.log(conf, ctx)
local metadata = plugin.plugin_metadata(plugin_name)
core.log.info("metadata: ", core.json.delay_encode(metadata))
@@ -199,6 +221,17 @@ function _M.log(conf, ctx)
entry.route_id = "no-matched"
end
+ if not stale_timer_running then
+ -- run the timer every 30 mins if any log is present
+ timer_at(1800, remove_stale_objects)
+ stale_timer_running = true
+ end
+
+ -- always cache the latest plugin conf in the module to avoid the
+ -- closure (method `func`) references to the old conf, in case of
+ -- plugin configuration update, the closure cannot sense the change.
+ plugin_conf = conf
+
local log_buffer = buffers[entry.route_id]
if log_buffer then
@@ -209,14 +242,15 @@ function _M.log(conf, ctx)
-- Generate a function to be executed by the batch processor
local func = function(entries, batch_max_size)
local data, err
- if conf.concat_method == "json" then
+
+ if plugin_conf.concat_method == "json" then
if batch_max_size == 1 then
data, err = core.json.encode(entries[1]) -- encode as single {}
else
data, err = core.json.encode(entries) -- encode as array [{}]
end
- elseif conf.concat_method == "new_line" then
+ elseif plugin_conf.concat_method == "new_line" then
if batch_max_size == 1 then
data, err = core.json.encode(entries[1]) -- encode as single {}
else
@@ -233,14 +267,14 @@ function _M.log(conf, ctx)
else
-- defensive programming check
- err = "unknown concat_method " .. (conf.concat_method or "nil")
+ err = "unknown concat_method " .. (plugin_conf.concat_method or "nil")
end
if not data then
return false, 'error occurred while encoding the data: ' .. err
end
- return send_http_data(conf, data)
+ return send_http_data(plugin_conf, data)
end
local config = {
diff --git a/apisix/plugins/tcp-logger.lua b/apisix/plugins/tcp-logger.lua
index ced5f8f..cf8dc99 100644
--- a/apisix/plugins/tcp-logger.lua
+++ b/apisix/plugins/tcp-logger.lua
@@ -23,7 +23,7 @@ local buffers = {}
local ngx = ngx
local tcp = ngx.socket.tcp
local ipairs = ipairs
-local stale_timer_running = false;
+local stale_timer_running = false
local timer_at = ngx.timer.at
local schema = {
diff --git a/t/plugin/http-logger.t b/t/plugin/http-logger.t
index 0b5619f..d007fbb 100644
--- a/t/plugin/http-logger.t
+++ b/t/plugin/http-logger.t
@@ -637,3 +637,149 @@ GET /t
done
--- no_error_log
[error]
+
+
+=== TEST 17: check plugin configuration updating
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body1 = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "plugins": {
+ "http-logger": {
+ "uri": "http://127.0.0.1:1982/hello",
+ "batch_max_size": 1,
+ "max_retry_count": 1,
+ "retry_delay": 2,
+ "buffer_duration": 2,
+ "inactive_timeout": 2
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1982": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/opentracing"
+ }]],
+ [[{
+ "node": {
+ "value": {
+ "plugins": {
+ "http-logger": {
+ "uri": "http://127.0.0.1:1982/hello",
+ "batch_max_size": 1,
+ "max_retry_count": 1,
+ "retry_delay": 2,
+ "buffer_duration": 2,
+ "inactive_timeout": 2
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1982": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/opentracing"
+ },
+ "key": "/apisix/routes/1"
+ },
+ "action": "set"
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ ngx.say("fail")
+ return
+ end
+
+ local code, _, body2 = t("/opentracing", "GET")
+ if code >= 300 then
+ ngx.status = code
+ ngx.say("fail")
+ return
+ end
+
+ local code, body3 = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "plugins": {
+ "http-logger": {
+ "uri": "http://127.0.0.1:1982/hello1",
+ "batch_max_size": 1,
+ "max_retry_count": 1,
+ "retry_delay": 2,
+ "buffer_duration": 2,
+ "inactive_timeout": 2
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1982": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/opentracing"
+ }]],
+ [[{
+ "node": {
+ "value": {
+ "plugins": {
+ "http-logger": {
+ "uri": "http://127.0.0.1:1982/hello1",
+ "batch_max_size": 1,
+ "max_retry_count": 1,
+ "retry_delay": 2,
+ "buffer_duration": 2,
+ "inactive_timeout": 2
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "127.0.0.1:1982": 1
+ },
+ "type": "roundrobin"
+ },
+ "uri": "/opentracing"
+ },
+ "key": "/apisix/routes/1"
+ },
+ "action": "set"
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ ngx.say("fail")
+ return
+ end
+
+ local code, _, body4 = t("/opentracing", "GET")
+ if code >= 300 then
+ ngx.status = code
+ ngx.say("fail")
+ return
+ end
+
+ ngx.print(body1)
+ ngx.print(body2)
+ ngx.print(body3)
+ ngx.print(body4)
+ }
+ }
+--- request
+GET /t
+--- wait: 0.5
+--- response_body
+passedopentracing
+passedopentracing
+--- grep_error_log eval
+qr/sending a batch logs to http:\/\/127.0.0.1:1982\/hello\d?/
+--- grep_error_log_out
+sending a batch logs to http://127.0.0.1:1982/hello
+sending a batch logs to http://127.0.0.1:1982/hello1