This is an automated email from the ASF dual-hosted git repository.
nic-6443 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 6a3a76648d fix(consul): run the catalog and health watchers as
independent timers (#14032)
6a3a76648d is described below
commit 6a3a76648d5a017f82f0363b82b879f2a9935e85
Author: Nic <[email protected]>
AuthorDate: Wed Oct 14 10:38:42 2026 +0800
fix(consul): run the catalog and health watchers as independent timers
(#14032)
---
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(-)
diff --git a/apisix/discovery/consul/client.lua
b/apisix/discovery/consul/client.lua
index c5afd20e8e..a86f6356c3 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 2277e93586..282b8a03b6 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 0000000000..276c96b9a6
--- /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: 15
+--- request
+GET /t
+--- response_body
+server 1
+--- no_error_log
+[error]
+got boolean
