From f9bf51c6fcbe0b4a776d1b00c7b8df22ec782e7e Mon Sep 17 00:00:00 2001 From: Nic Date: Tue, 13 Oct 2026 08:15:04 +0000 Subject: [PATCH 1/2] fix(consul): run the catalog and health watchers as independent timers connect() spawned the catalog and health watchers as light threads, waited for the first one and killed both. resty.consul reads the response body through resty.http, which does that in a child coroutine. When the killed watcher was reading a body, ngx.thread.kill does not cancel that child's pending socket read; its completion later resumes connect() at an unrelated yield point, e.g. the receive of the following catalog fetch, which then fails with "bad argument #1 to 'str_sub' (string expected, got boolean)". Each index now has its own watcher timer, and one run is one blocking query, so nothing is killed and every run frees its memory when it ends. The watcher that sees a change runs the round itself: validate the result, then fetch all services, backing off on errors as before. Rounds never overlap: a watcher that sees a change while another round runs parks, and is re-armed together with the round's watcher when it ends, so it queries again against the updated indexes. --- apisix/discovery/consul/client.lua | 10 +- apisix/discovery/consul/init.lua | 136 ++++++++++------------- t/discovery/consul-watch-race.t | 168 +++++++++++++++++++++++++++++ 3 files changed, 233 insertions(+), 81 deletions(-) create mode 100644 t/discovery/consul-watch-race.t diff --git a/apisix/discovery/consul/client.lua b/apisix/discovery/consul/client.lua index c5afd20e8e6e..a86f6356c3b9 100644 --- a/apisix/discovery/consul/client.lua +++ b/apisix/discovery/consul/client.lua @@ -144,10 +144,13 @@ end -- ─── blocking query watchers ────────────────────────────────────────── +-- watch_catalog / watch_health run one blocking query. They return the watch +-- type and the new index (an error index on failure), or nothing when the +-- index did not change. + function _M.watch_catalog(consul_server) local client = resty_consul:new(get_opts(consul_server, true)) - ::RETRY:: local watch_result, watch_err = client:get(consul_server.consul_watch_catalog_url) local watch_error_info = (watch_err ~= nil and watch_err) or ((watch_result ~= nil and watch_result.status ~= 200) @@ -166,7 +169,7 @@ function _M.watch_catalog(consul_server) local random_delay = math_random(default_random_range) log.info("watch catalog has no change, re-watch consul after ", random_delay, " seconds") core_sleep(random_delay) - goto RETRY + return end return watch_type_catalog, watch_result.headers['X-Consul-Index'] @@ -176,7 +179,6 @@ end function _M.watch_health(consul_server) local client = resty_consul:new(get_opts(consul_server, false)) - ::RETRY:: local watch_result, watch_err = client:get(consul_server.consul_watch_health_url) local watch_error_info = (watch_err ~= nil and watch_err) or ((watch_result ~= nil and watch_result.status ~= 200) @@ -195,7 +197,7 @@ function _M.watch_health(consul_server) local random_delay = math_random(default_random_range) log.info("watch health has no change, re-watch consul after ", random_delay, " seconds") core_sleep(random_delay) - goto RETRY + return end return watch_type_health, watch_result.headers['X-Consul-Index'] diff --git a/apisix/discovery/consul/init.lua b/apisix/discovery/consul/init.lua index 2277e93586cc..282b8a03b633 100644 --- a/apisix/discovery/consul/init.lua +++ b/apisix/discovery/consul/init.lua @@ -31,10 +31,7 @@ local json_delay_encode = core.json.delay_encode local process = require("ngx.process") local ngx_worker_id = ngx.worker.id local exiting = ngx.worker.exiting -local thread_spawn = ngx.thread.spawn -local thread_wait = ngx.thread.wait -local thread_kill = ngx.thread.kill -local math_random = math.random +local pcall = pcall local is_http = ngx.config.subsystem == "http" local _M = { @@ -55,7 +52,6 @@ local nodes_cache = core.lrucache.new({ }) local default_skip_services = {"consul"} -local default_random_range = 5 local function get_dict() @@ -224,83 +220,30 @@ end -- ─── polling loop ───────────────────────────────────────────────────── -local function check_keepalive(reg, consul_server, retry_delay) - if reg.stop_flag then - return - end - - if exiting() then - return - end - - if consul_server.keepalive then - local ok, err = ngx_timer_at(0, _M.connect, reg, consul_server, retry_delay) - if not ok then - log.error("create ngx_timer_at got error: ", err) - return - end - else - -- self-rescheduling poll: use timer.at instead of timer.every - -- so stop_flag can actually halt future wakeups - local ok, err = ngx_timer_at(consul_server.fetch_interval, - _M.connect, reg, consul_server, retry_delay) - if not ok then - log.error("create ngx_timer_at got error: ", err) - return - end - end -end +local watch -function _M.connect(premature, reg, consul_server, retry_delay) - if premature or reg.stop_flag then - return - end - - local catalog_thread, spawn_catalog_err = thread_spawn(consul_client.watch_catalog, - consul_server) - if not catalog_thread then - local random_delay = math_random(default_random_range) - log.error("failed to spawn thread watch catalog: ", spawn_catalog_err, - ", retry connecting consul after ", random_delay, " seconds") - core_sleep(random_delay) - - check_keepalive(reg, consul_server, retry_delay) +local function arm_watch(reg, consul_server, watch_fn, delay) + if reg.stop_flag or exiting() then return end - local health_thread, err = thread_spawn(consul_client.watch_health, consul_server) - if not health_thread then - thread_kill(catalog_thread) - local random_delay = math_random(default_random_range) - log.error("failed to spawn thread watch health: ", err, - ", retry connecting consul after ", random_delay, " seconds") - core_sleep(random_delay) - - check_keepalive(reg, consul_server, retry_delay) - return + local ok, err = ngx_timer_at(delay, watch, reg, consul_server, watch_fn) + if not ok then + log.error("create ngx_timer_at got error: ", err) end +end - local thread_wait_ok, watch_type, index = thread_wait(catalog_thread, health_thread) - thread_kill(catalog_thread) - thread_kill(health_thread) - if not thread_wait_ok then - local random_delay = math_random(default_random_range) - log.error("failed to wait thread: ", watch_type, ", retry connecting consul after ", - random_delay, " seconds") - core_sleep(random_delay) - - check_keepalive(reg, consul_server, retry_delay) - return - end +-- One round: validate the watch result, then fetch all services, backing off on +-- errors. +local function handle_change(reg, consul_server, watch_type, index) if not consul_client.watch_result_is_valid(tonumber(watch_type), tonumber(index), consul_server.catalog_index, consul_server.health_index) then - retry_delay = consul_client.get_retry_delay(retry_delay) + local retry_delay = consul_client.get_retry_delay(consul_server.retry_delay) + consul_server.retry_delay = retry_delay log.warn("get all svcs got err, retry connecting consul after ", retry_delay, " seconds") core_sleep(retry_delay) - - check_keepalive(reg, consul_server, retry_delay) return end @@ -318,12 +261,11 @@ function _M.connect(premature, reg, consul_server, retry_delay) }) if fetch_err then - retry_delay = consul_client.get_retry_delay(retry_delay) + local retry_delay = consul_client.get_retry_delay(consul_server.retry_delay) + consul_server.retry_delay = retry_delay log.warn("get all svcs got err: ", fetch_err, ", retry connecting consul after ", retry_delay, " seconds") core_sleep(retry_delay) - - check_keepalive(reg, consul_server, retry_delay) return end @@ -345,7 +287,44 @@ function _M.connect(premature, reg, consul_server, retry_delay) end consul_client.update_index(consul_server, new_catalog_index, new_health_index) - check_keepalive(reg, consul_server, retry_delay) +end + + +-- Each index has its own watcher timer, and one run is one blocking query: no +-- watcher is ever killed mid-read, and every run frees its memory when it ends. +-- A watcher that sees a change runs the round itself. Rounds never overlap: a +-- watcher that sees a change while another round runs parks, and is re-armed +-- with the round's watcher when it ends, so it queries again against the +-- updated indexes. +watch = function(premature, reg, consul_server, watch_fn) + if premature or reg.stop_flag then + return + end + + local watch_type, index = watch_fn(consul_server) + if not watch_type then + arm_watch(reg, consul_server, watch_fn, 0) + return + end + + if consul_server.in_round then + consul_server.parked[watch_fn] = true + return + end + + consul_server.in_round = true + local ok, err = pcall(handle_change, reg, consul_server, watch_type, index) + consul_server.in_round = false + if not ok then + log.error("failed to handle consul watch result: ", err) + end + + local delay = consul_server.keepalive and 0 or consul_server.fetch_interval + for parked_fn in pairs(consul_server.parked) do + arm_watch(reg, consul_server, parked_fn, delay) + end + consul_server.parked = {} + arm_watch(reg, consul_server, watch_fn, delay) end @@ -437,9 +416,12 @@ function _M.start_registry(reg) log.info("consul_server_list: ", json_delay_encode(consul_servers_list, true)) for _, server in ipairs(consul_servers_list) do - local ok, timer_err = ngx_timer_at(0, _M.connect, reg, server) - if not ok then - error("create consul got error: " .. timer_err) + server.parked = {} + for _, watch_fn in ipairs({consul_client.watch_catalog, consul_client.watch_health}) do + local ok, timer_err = ngx_timer_at(0, watch, reg, server, watch_fn) + if not ok then + error("create consul got error: " .. timer_err) + end end end end diff --git a/t/discovery/consul-watch-race.t b/t/discovery/consul-watch-race.t new file mode 100644 index 000000000000..921077efd356 --- /dev/null +++ b/t/discovery/consul-watch-race.t @@ -0,0 +1,168 @@ +# +# 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'; + +repeat_each(1); +log_level('info'); +no_root_location(); +no_shuffle(); + + +add_block_preprocessor(sub { + my ($block) = @_; + + # mock consul server: both indexes change together every 3 seconds. The + # catalog watch answers 0.3s after a change while the health watch streams a + # large chunked body for about a second, and a non-blocking catalog read + # takes 1.5s, so the health body is still being read when the catalog watch + # returns and finishes while the following fetch is waiting for its response. + my $http_config = $block->http_config // <<_EOC_; + + server { + listen 18507; + + location /v1/catalog/services { + content_by_lua_block { + local index = math.floor(ngx.now() / 3) + local requested = tonumber(ngx.req.get_uri_args().index) + if requested then + local deadline = ngx.now() + 60 + while index <= requested and ngx.now() < deadline do + ngx.sleep(0.05) + index = math.floor(ngx.now() / 3) + end + ngx.sleep(0.3) + else + ngx.sleep(1.5) + end + ngx.header["X-Consul-Index"] = index + ngx.header.content_type = "application/json" + ngx.say('{"consul":[],"service_a":[]}') + } + } + + location /v1/health/state/any { + content_by_lua_block { + local index = math.floor(ngx.now() / 3) + local requested = tonumber(ngx.req.get_uri_args().index) + ngx.header.content_type = "application/json" + if not requested then + ngx.header["X-Consul-Index"] = index + ngx.say('[]') + return + end + + local deadline = ngx.now() + 60 + while index <= requested and ngx.now() < deadline do + ngx.sleep(0.05) + index = math.floor(ngx.now() / 3) + end + ngx.header["X-Consul-Index"] = index + local check = '{"Node":"node1","CheckID":"check","Status":"passing",' + .. '"Notes":"' .. string.rep("x", 2000) .. '"},' + ngx.print("[") + for _ = 1, 100 do + ngx.print(check) + ngx.flush(true) + ngx.sleep(0.01) + end + ngx.print('{"Node":"node1","CheckID":"last","Status":"passing"}]') + } + } + + location /v1/health/service/service_a { + content_by_lua_block { + ngx.header["X-Consul-Index"] = math.floor(ngx.now() / 3) + ngx.header.content_type = "application/json" + ngx.say('[{"Node":{"Node":"node1","Address":"127.0.0.1"},' + .. '"Service":{"ID":"service_a1","Service":"service_a",' + .. '"Address":"127.0.0.1","Port":30511},"Checks":[]}]') + } + } + } + + server { + listen 30511; + + location /hello { + content_by_lua_block { + ngx.say("server 1") + } + } + } +_EOC_ + + $block->set_value("http_config", $http_config); +}); + +our $yaml_config = <<_EOC_; +apisix: + node_listen: 1984 +deployment: + role: data_plane + role_data_plane: + config_provider: yaml +discovery: + consul: + servers: + - "http://127.0.0.1:18507" + timeout: + connect: 1000 + read: 3000 + wait: 60 + keepalive: true +_EOC_ + +run_tests(); + +__DATA__ + +=== TEST 1: a watcher still reading its response body must not break the following fetch +--- yaml_config eval: $::yaml_config +--- apisix_yaml +routes: + - + uri: /hello + upstream: + service_name: service_a + discovery_type: consul + type: roundrobin +#END +--- config + location /t { + content_by_lua_block { + ngx.sleep(6) + + local http = require("resty.http") + local httpc = http.new() + local res, err = httpc:request_uri("http://127.0.0.1:1984/hello") + if not res then + ngx.say("request failed: ", err) + return + end + ngx.status = res.status + ngx.print(res.body) + } + } +--- timeout: 10 +--- request +GET /t +--- response_body +server 1 +--- no_error_log +[error] +got boolean From 8c4d6801e9cf1cc04043d0558c39ff4ebcdbb9ad Mon Sep 17 00:00:00 2001 From: Nic Date: Tue, 13 Oct 2026 09:04:11 +0000 Subject: [PATCH 2/2] test(consul): leave more headroom in the watcher race test timeout --- t/discovery/consul-watch-race.t | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/t/discovery/consul-watch-race.t b/t/discovery/consul-watch-race.t index 921077efd356..276c96b9a635 100644 --- a/t/discovery/consul-watch-race.t +++ b/t/discovery/consul-watch-race.t @@ -158,7 +158,7 @@ routes: ngx.print(res.body) } } ---- timeout: 10 +--- timeout: 15 --- request GET /t --- response_body