This is an automated email from the ASF dual-hosted git repository.
spacewander pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new bbbdf58 refactor: the parent of upstream should point to its original
src (#3287)
bbbdf58 is described below
commit bbbdf58d555ee89144b6923e751deabf7e0bbf15
Author: 罗泽轩 <spacewander...@gmail.com>
AuthorDate: Fri Jan 15 00:43:39 2021 -0600
refactor: the parent of upstream should point to its original src (#3287)
Signed-off-by: spacewander <spacewander...@gmail.com>
---
apisix/core/table.lua | 45 ++++++---
apisix/http/service.lua | 1 +
apisix/plugin.lua | 7 +-
apisix/plugins/example-plugin.lua | 2 +-
apisix/plugins/traffic-split.lua | 4 +-
apisix/router.lua | 1 +
apisix/stream/plugins/mqtt-proxy.lua | 2 +-
apisix/upstream.lua | 39 +++-----
t/node/healthcheck.t | 120 +++++++++++++++++++++++-
t/node/healthcheck2.t | 175 ++++++++++++++++++++++++++++++++++-
10 files changed, 342 insertions(+), 54 deletions(-)
diff --git a/apisix/core/table.lua b/apisix/core/table.lua
index e314b39..d0bc1b2 100644
--- a/apisix/core/table.lua
+++ b/apisix/core/table.lua
@@ -96,25 +96,44 @@ function _M.setmt__gc(t, mt)
end
-local function deepcopy(orig)
- local orig_type = type(orig)
- if orig_type ~= 'table' then
- return orig
- end
+local deepcopy
+do
+ local function _deepcopy(orig, copied)
+ -- prevent infinite loop when a field refers its parent
+ copied[orig] = true
+ -- If the array-like table contains nil in the middle,
+ -- the len might be smaller than the expected.
+ -- But it doesn't affect the correctness.
+ local len = #orig
+ local copy = new_tab(len, nkeys(orig) - len)
+ for orig_key, orig_value in pairs(orig) do
+ if type(orig_value) == "table" and not copied[orig_value] then
+ copy[orig_key] = _deepcopy(orig_value, copied)
+ else
+ copy[orig_key] = orig_value
+ end
+ end
- -- If the array-like table contains nil in the middle,
- -- the len might be smaller than the expected.
- -- But it doesn't affect the correctness.
- local len = #orig
- local copy = new_tab(len, nkeys(orig) - len)
- for orig_key, orig_value in pairs(orig) do
- copy[orig_key] = deepcopy(orig_value)
+ return copy
end
- return copy
+
+ local copied_recorder = {}
+
+ function deepcopy(orig)
+ local orig_type = type(orig)
+ if orig_type ~= 'table' then
+ return orig
+ end
+
+ local res = _deepcopy(orig, copied_recorder)
+ _M.clear(copied_recorder)
+ return res
+ end
end
_M.deepcopy = deepcopy
+
local ngx_null = ngx.null
local function merge(origin, extend)
for k,v in pairs(extend) do
diff --git a/apisix/http/service.lua b/apisix/http/service.lua
index a23924d..9279dc3 100644
--- a/apisix/http/service.lua
+++ b/apisix/http/service.lua
@@ -79,6 +79,7 @@ local function filter(service)
service.value.upstream.nodes = new_nodes
end
+ service.value.upstream.parent = service
core.log.info("filter service: ", core.json.delay_encode(service))
end
diff --git a/apisix/plugin.lua b/apisix/plugin.lua
index 9928cb5..5445cfc 100644
--- a/apisix/plugin.lua
+++ b/apisix/plugin.lua
@@ -401,11 +401,8 @@ local function merge_service_route(service_conf,
route_conf)
local route_upstream = route_conf.value.upstream
if route_upstream then
new_conf.value.upstream = route_upstream
-
- if route_upstream.checks then
- route_upstream.parent = route_conf
- end
-
+ -- when route's upstream override service's upstream,
+ -- the upstream.parent still point to the route
new_conf.value.upstream_id = nil
new_conf.has_domain = route_conf.has_domain
end
diff --git a/apisix/plugins/example-plugin.lua
b/apisix/plugins/example-plugin.lua
index e270a5c..83722ff 100644
--- a/apisix/plugins/example-plugin.lua
+++ b/apisix/plugins/example-plugin.lua
@@ -99,7 +99,7 @@ function _M.access(conf, ctx)
local matched_route = ctx.matched_route
upstream.set(ctx, up_conf.type .. "#route_" .. matched_route.value.id,
- ctx.conf_version, up_conf, matched_route)
+ ctx.conf_version, up_conf)
return
end
diff --git a/apisix/plugins/traffic-split.lua b/apisix/plugins/traffic-split.lua
index d0bdec7..a94f6dd 100644
--- a/apisix/plugins/traffic-split.lua
+++ b/apisix/plugins/traffic-split.lua
@@ -240,14 +240,16 @@ local function set_upstream(upstream_info, ctx)
local ok, err = upstream.check_schema(up_conf)
if not ok then
+ core.log.error("failed to validate generated upstream: ", err)
return 500, err
end
local matched_route = ctx.matched_route
+ up_conf.parent = matched_route
local upstream_key = up_conf.type .. "#route_" ..
matched_route.value.id .. "_" ..upstream_info.vid
core.log.info("upstream_key: ", upstream_key)
- upstream.set(ctx, upstream_key, ctx.conf_version, up_conf, matched_route)
+ upstream.set(ctx, upstream_key, ctx.conf_version, up_conf)
return
end
diff --git a/apisix/router.lua b/apisix/router.lua
index 0ae2481..0f824e5 100644
--- a/apisix/router.lua
+++ b/apisix/router.lua
@@ -64,6 +64,7 @@ local function filter(route)
route.value.upstream.nodes = new_nodes
end
+ route.value.upstream.parent = route
core.log.info("filter route: ", core.json.delay_encode(route))
end
diff --git a/apisix/stream/plugins/mqtt-proxy.lua
b/apisix/stream/plugins/mqtt-proxy.lua
index b533430..f7cfd88 100644
--- a/apisix/stream/plugins/mqtt-proxy.lua
+++ b/apisix/stream/plugins/mqtt-proxy.lua
@@ -173,7 +173,7 @@ function _M.preread(conf, ctx)
local matched_route = ctx.matched_route
upstream.set(ctx, up_conf.type .. "#route_" .. matched_route.value.id,
- ctx.conf_version, up_conf, matched_route)
+ ctx.conf_version, up_conf)
return
end
diff --git a/apisix/upstream.lua b/apisix/upstream.lua
index 096f81e..856f44d 100644
--- a/apisix/upstream.lua
+++ b/apisix/upstream.lua
@@ -33,7 +33,7 @@ local lrucache_checker = core.lrucache.new({
local _M = {}
-local function set_directly(ctx, key, ver, conf, parent)
+local function set_directly(ctx, key, ver, conf)
if not ctx then
error("missing argument ctx", 2)
end
@@ -46,24 +46,22 @@ local function set_directly(ctx, key, ver, conf, parent)
if not conf then
error("missing argument conf", 2)
end
- if not parent then
- error("missing argument parent", 2)
- end
ctx.upstream_conf = conf
ctx.upstream_version = ver
ctx.upstream_key = key
- ctx.upstream_healthcheck_parent = parent
+ ctx.upstream_healthcheck_parent = conf.parent
return
end
_M.set = set_directly
-local function create_checker(upstream, healthcheck_parent)
+local function create_checker(upstream)
if healthcheck == nil then
healthcheck = require("resty.healthcheck")
end
+ local healthcheck_parent = upstream.parent
local checker, err = healthcheck.new({
name = "upstream#" .. healthcheck_parent.key,
shm_name = "upstream-healthcheck",
@@ -85,27 +83,18 @@ local function create_checker(upstream, healthcheck_parent)
end
end
- if upstream.parent then
- core.table.insert(upstream.parent.clean_handlers, function ()
- core.log.info("try to release checker: ", tostring(checker))
- checker:clear()
- checker:stop()
- end)
-
- else
- core.table.insert(healthcheck_parent.clean_handlers, function ()
- core.log.info("try to release checker: ", tostring(checker))
- checker:clear()
- checker:stop()
- end)
- end
+ core.table.insert(healthcheck_parent.clean_handlers, function ()
+ core.log.info("try to release checker: ", tostring(checker))
+ checker:clear()
+ checker:stop()
+ end)
core.log.info("create new checker: ", tostring(checker))
return checker
end
-local function fetch_healthchecker(upstream, healthcheck_parent, version)
+local function fetch_healthchecker(upstream, version)
if not upstream.checks then
return
end
@@ -115,8 +104,7 @@ local function fetch_healthchecker(upstream,
healthcheck_parent, version)
end
local checker = lrucache_checker(upstream, version,
- create_checker, upstream,
- healthcheck_parent)
+ create_checker, upstream)
return checker
end
@@ -150,7 +138,7 @@ function _M.set_by_route(route, api_ctx)
end
set_directly(api_ctx, up_conf.type .. "#upstream_" .. tostring(up_conf),
- api_ctx.conf_version, up_conf, route)
+ api_ctx.conf_version, up_conf)
local nodes_count = up_conf.nodes and #up_conf.nodes or 0
if nodes_count == 0 then
@@ -158,7 +146,7 @@ function _M.set_by_route(route, api_ctx)
end
if nodes_count > 1 then
- local checker = fetch_healthchecker(up_conf, route,
api_ctx.upstream_version)
+ local checker = fetch_healthchecker(up_conf, api_ctx.upstream_version)
api_ctx.up_checker = checker
end
@@ -219,6 +207,7 @@ function _M.init_worker()
upstream.value.nodes = new_nodes
end
+ upstream.value.parent = upstream
core.log.info("filter upstream: ",
core.json.delay_encode(upstream))
end,
})
diff --git a/t/node/healthcheck.t b/t/node/healthcheck.t
index 1056a53..d9bacb8 100644
--- a/t/node/healthcheck.t
+++ b/t/node/healthcheck.t
@@ -218,10 +218,10 @@ GET /t
--- response_body
[{"count":12,"port":"1980"}]
--- grep_error_log eval
-qr/unhealthy .* for '.*'/
+qr/\([^)]+\) unhealthy .* for '.*'/
--- grep_error_log_out
-unhealthy TCP increment (1/2) for 'foo.com(127.0.0.1:1970)'
-unhealthy TCP increment (2/2) for 'foo.com(127.0.0.1:1970)'
+(upstream#/apisix/routes/1) unhealthy TCP increment (1/2) for
'foo.com(127.0.0.1:1970)'
+(upstream#/apisix/routes/1) unhealthy TCP increment (2/2) for
'foo.com(127.0.0.1:1970)'
--- timeout: 10
@@ -795,3 +795,117 @@ GET /t
qr/expected 65536 to be smaller than 65535/
--- error_code chomp
400
+
+
+
+=== TEST 18: set route + upstream (two upstream node: one healthy + one
unhealthy)
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/upstreams/1',
+ ngx.HTTP_PUT,
+ [[{
+ "type": "roundrobin",
+ "nodes": {
+ "127.0.0.1:1980": 1,
+ "127.0.0.1:1970": 1
+ },
+ "checks": {
+ "active": {
+ "http_path": "/status",
+ "host": "foo.com",
+ "healthy": {
+ "interval": 1,
+ "successes": 1
+ },
+ "unhealthy": {
+ "interval": 1,
+ "http_failures": 2
+ }
+ }
+ }
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ ngx.say(body)
+ return
+ end
+
+ local code, body = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "uri": "/server_port",
+ "upstream_id": 1
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- request
+GET /t
+--- response_body
+passed
+--- grep_error_log eval
+qr/^.*?\[error\](?!.*process exiting).*/
+--- grep_error_log_out
+
+
+
+=== TEST 19: hit routes, ensure the checker is bound to the upstream
+--- config
+ location /t {
+ content_by_lua_block {
+ local http = require "resty.http"
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port
+ .. "/server_port"
+
+ do
+ local httpc = http.new()
+ local res, err = httpc:request_uri(uri, {method = "GET",
keepalive = false})
+ end
+
+ ngx.sleep(2.5)
+
+ local ports_count = {}
+ for i = 1, 12 do
+ local httpc = http.new()
+ local res, err = httpc:request_uri(uri, {method = "GET",
keepalive = false})
+ if not res then
+ ngx.say(err)
+ return
+ end
+
+ ports_count[res.body] = (ports_count[res.body] or 0) + 1
+ end
+
+ local ports_arr = {}
+ for port, count in pairs(ports_count) do
+ table.insert(ports_arr, {port = port, count = count})
+ end
+
+ local function cmd(a, b)
+ return a.port > b.port
+ end
+ table.sort(ports_arr, cmd)
+
+ ngx.say(require("toolkit.json").encode(ports_arr))
+ ngx.exit(200)
+ }
+ }
+--- request
+GET /t
+--- response_body
+[{"count":12,"port":"1980"}]
+--- grep_error_log eval
+qr/\([^)]+\) unhealthy .* for '.*'/
+--- grep_error_log_out
+(upstream#/apisix/upstreams/1) unhealthy TCP increment (1/2) for
'foo.com(127.0.0.1:1970)'
+(upstream#/apisix/upstreams/1) unhealthy TCP increment (2/2) for
'foo.com(127.0.0.1:1970)'
+--- timeout: 10
diff --git a/t/node/healthcheck2.t b/t/node/healthcheck2.t
index 1b25cc8..1c802e2 100644
--- a/t/node/healthcheck2.t
+++ b/t/node/healthcheck2.t
@@ -19,8 +19,10 @@ use t::APISIX 'no_plan';
master_on();
repeat_each(1);
+log_level('info');
no_root_location();
no_shuffle();
+worker_connections(256);
add_block_preprocessor(sub {
my ($block) = @_;
@@ -37,11 +39,7 @@ _EOC_
}
if (!$block->request) {
- $block->set_value("request", "GET /hello");
- }
-
- if ((!defined $block->error_log) && (!defined $block->no_error_log)) {
- $block->set_value("no_error_log", "[error]");
+ $block->set_value("request", "GET /t");
}
});
@@ -66,4 +64,171 @@ upstreams:
#END
--- error_log
value should match only one schema, but matches both schemas 1 and 2
+--- request
+GET /hello
--- error_code: 502
+
+
+
+=== TEST 2: route + service
+--- apisix_yaml
+services:
+ - id: 1
+ upstream:
+ type: roundrobin
+ nodes:
+ "127.0.0.1:1980": 1
+ "127.0.0.1:1970": 1
+ checks:
+ active:
+ http_path: /status
+ host: foo.com
+ healthy:
+ interval: 1
+ successes: 1
+ unhealthy:
+ interval: 1
+ http_failures: 2
+routes:
+ -
+ service_id: 1
+ uri: /server_port
+#END
+--- config
+ location /t {
+ content_by_lua_block {
+ local http = require "resty.http"
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port
+ .. "/server_port"
+
+ do
+ local httpc = http.new()
+ local res, err = httpc:request_uri(uri, {method = "GET",
keepalive = false})
+ end
+
+ ngx.sleep(2.5)
+
+ local ports_count = {}
+ for i = 1, 12 do
+ local httpc = http.new()
+ local res, err = httpc:request_uri(uri, {method = "GET",
keepalive = false})
+ if not res then
+ ngx.say(err)
+ return
+ end
+
+ ports_count[res.body] = (ports_count[res.body] or 0) + 1
+ end
+
+ local ports_arr = {}
+ for port, count in pairs(ports_count) do
+ table.insert(ports_arr, {port = port, count = count})
+ end
+
+ local function cmd(a, b)
+ return a.port > b.port
+ end
+ table.sort(ports_arr, cmd)
+
+ ngx.say(require("toolkit.json").encode(ports_arr))
+ ngx.exit(200)
+ }
+ }
+--- response_body
+[{"count":12,"port":"1980"}]
+--- grep_error_log eval
+qr/\([^)]+\) unhealthy .* for '.*'/
+--- grep_error_log_out
+(upstream#/services/1) unhealthy TCP increment (1/2) for
'foo.com(127.0.0.1:1970)'
+(upstream#/services/1) unhealthy TCP increment (2/2) for
'foo.com(127.0.0.1:1970)'
+--- timeout: 10
+
+
+
+=== TEST 3: route override service
+--- apisix_yaml
+services:
+ - id: 1
+ upstream:
+ type: roundrobin
+ nodes:
+ "127.0.0.2:1980": 1
+ "127.0.0.2:1970": 1
+ checks:
+ active:
+ http_path: /status
+ host: foo.com
+ healthy:
+ interval: 1
+ successes: 1
+ unhealthy:
+ interval: 1
+ http_failures: 2
+routes:
+ -
+ service_id: 1
+ uri: /server_port
+ upstream:
+ type: roundrobin
+ nodes:
+ "127.0.0.1:1980": 1
+ "127.0.0.1:1970": 1
+ checks:
+ active:
+ http_path: /status
+ host: foo.com
+ healthy:
+ interval: 1
+ successes: 1
+ unhealthy:
+ interval: 1
+ http_failures: 2
+#END
+--- config
+ location /t {
+ content_by_lua_block {
+ local http = require "resty.http"
+ local uri = "http://127.0.0.1:"; .. ngx.var.server_port
+ .. "/server_port"
+
+ do
+ local httpc = http.new()
+ local res, err = httpc:request_uri(uri, {method = "GET",
keepalive = false})
+ end
+
+ ngx.sleep(2.5)
+
+ local ports_count = {}
+ for i = 1, 12 do
+ local httpc = http.new()
+ local res, err = httpc:request_uri(uri, {method = "GET",
keepalive = false})
+ if not res then
+ ngx.say(err)
+ return
+ end
+
+ ports_count[res.body] = (ports_count[res.body] or 0) + 1
+ end
+
+ local ports_arr = {}
+ for port, count in pairs(ports_count) do
+ table.insert(ports_arr, {port = port, count = count})
+ end
+
+ local function cmd(a, b)
+ return a.port > b.port
+ end
+ table.sort(ports_arr, cmd)
+
+ ngx.say(require("toolkit.json").encode(ports_arr))
+ ngx.exit(200)
+ }
+ }
+--- response_body
+[{"count":12,"port":"1980"}]
+--- grep_error_log eval
+qr/\([^)]+\) unhealthy .* for '.*'/
+--- grep_error_log_out
+(upstream#/routes/arr_1) unhealthy TCP increment (1/2) for
'foo.com(127.0.0.1:1970)'
+(upstream#/routes/arr_1) unhealthy TCP increment (2/2) for
'foo.com(127.0.0.1:1970)'
+--- timeout: 10
