# # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # use t::APISIX 'no_plan'; log_level('debug'); repeat_each(1); no_long_string(); no_root_location(); run_tests; __DATA__ === TEST 1: send invalid arguments for constructor --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local config = { max_retry_count = 2, batch_max_size = 1, retry_delay = 0, } local func_to_send = function(elements) return true end local log_buffer, err = Batch:new("", config) if log_buffer then log_buffer:push({hello='world'}) ngx.say("done") end if not log_buffer then ngx.say("failed") end } } --- request GET /t --- response_body failed --- wait: 0.5 === TEST 2: sanity --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local func_to_send = function(elements) return true end local config = { max_retry_count = 2, batch_max_size = 1, retry_delay = 0, } local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- error_log Batch Processor[log buffer] successfully processed the entries --- wait: 0.5 === TEST 3: batch processor timeout exceeded --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, inactive_timeout = 1 } local func_to_send = function(elements) return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- error_log Batch Processor[log buffer] buffer duration exceeded, activating buffer flush Batch Processor[log buffer] successfully processed the entries --- wait: 3 === TEST 4: batch processor batch max size exceeded --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, } local func_to_send = function(elements) return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- no_error_log Batch Processor[log buffer] buffer duration exceeded, activating buffer flush --- error_log Batch Processor[log buffer] batch max size has exceeded Batch Processor[log buffer] successfully processed the entries --- wait: 1 === TEST 5: first failed to process and second try success --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local core = require("apisix.core") local retry = false local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, } local func_to_send = function(elements) if not retry then retry = true return false end return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- error_log Batch Processor[log buffer] failed to process entries Batch Processor[log buffer] successfully processed the entries --- wait: 0.5 === TEST 6: Exceeding max retry count --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, } local func_to_send = function(elements) return false end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- no_error_log Batch Processor[log buffer] buffer duration exceeded, activating buffer flush --- error_log Batch Processor[log buffer] failed to process entries Batch Processor[log buffer] exceeded the max_retry_count --- wait: 0.5 === TEST 7: two batches --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local core = require("apisix.core") local count = 0 local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, } local func_to_send = function(elements) count = count + 1 core.log.info("batch[", count , "] sent") return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- no_error_log Batch Processor[log buffer] activating flush due to no activity --- error_log batch[1] sent batch[2] sent --- wait: 0.5 === TEST 8: batch processor retry count 0 and fail processing --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local config = { max_retry_count = 0, batch_max_size = 2, retry_delay = 0, } local func_to_send = function(elements) return false end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- no_error_log Batch Processor[log buffer] activating flush due to no activity --- error_log Batch Processor[log buffer] exceeded the max_retry_count --- wait: 0.5 === TEST 9: batch processor timeout exceeded --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, buffer_duration = 60, inactive_timeout = 1, } local func_to_send = function(elements) return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({hello='world'}) ngx.say("done") } } --- request GET /t --- response_body done --- error_log Batch Processor[log buffer] buffer duration exceeded, activating buffer flush Batch Processor[log buffer] successfully processed the entries --- wait: 3 === TEST 10: json encode and log elements --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local core = require("apisix.core") local config = { max_retry_count = 2, batch_max_size = 2, retry_delay = 0, } local func_to_send = function(elements) core.log.info(require("toolkit.json").encode(elements)) return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({msg='1'}) log_buffer:push({msg='2'}) log_buffer:push({msg='3'}) log_buffer:push({msg='4'}) ngx.say("done") } } --- request GET /t --- response_body done --- no_error_log Batch Processor[log buffer] activating flush due to no activity --- error_log [{"msg":"1"},{"msg":"2"}] [{"msg":"3"},{"msg":"4"}] --- wait: 0.5 === TEST 11: extend timer --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local core = require("apisix.core") local config = { max_retry_count = 1, batch_max_size = 3, retry_delay = 0, inactive_timeout = 1 } local func_to_send = function(elements) core.log.info(require("toolkit.json").encode(elements)) return true end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({msg='1'}) ngx.sleep(0.3) log_buffer:push({msg='2'}) log_buffer:push({msg='3'}) log_buffer:push({msg='4'}) ngx.say("done") } } --- request GET /t --- response_body done --- no_error_log Batch Processor[log buffer] activating flush due to no activity --- error_log Batch Processor[log buffer] extending buffer timer --- wait: 3 === TEST 12: partially consumed entries --- config location /t { content_by_lua_block { local Batch = require("apisix.utils.batch-processor") local core = require("apisix.core") local config = { max_retry_count = 1, batch_max_size = 3, retry_delay = 0, inactive_timeout = 1 } local func_to_send = function(elements) core.log.info(require("toolkit.json").encode(elements)) return false, "error after consuming single entry", 2 end local log_buffer, err = Batch:new(func_to_send, config) if not log_buffer then ngx.say(err) end log_buffer:push({msg='1'}) log_buffer:push({msg='2'}) log_buffer:push({msg='3'}) log_buffer:push({msg='4'}) ngx.say("done") } } --- request GET /t --- response_body done --- error_log [{"msg":"1"},{"msg":"2"},{"msg":"3"}] Batch Processor[log buffer] failed to process entries [2/3]: error after consuming single entry [{"msg":"2"},{"msg":"3"}] Batch Processor[log buffer] failed to process entries [1/2]: error after consuming single entry [{"msg":"4"}] --- wait: 2