Copilot commented on code in PR #14032:
URL: https://github.com/apache/apisix/pull/14032#discussion_r4256506790
##########
apisix/discovery/consul/init.lua:
##########
@@ -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
Review Comment:
`watch_result_is_valid` returns false both for error indexes (`index <= 0`)
and for 'no longer new' indexes (`index == current_index`). In the new
two-timer design, it’s realistic for a watcher to return an index that has
already been committed by the other watcher’s completed round; treating that as
an error will incorrectly back off/sleep and slow discovery. Split the
handling: back off only when `index <= 0` (or other real error), and treat
`index == current_index` as a no-op (re-arm without changing `retry_delay` or
sleeping). Consider updating the log message accordingly so it doesn't report
an error for stale/no-op results.
##########
apisix/discovery/consul/init.lua:
##########
@@ -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
Review Comment:
The PR description says: 'if a timer can't be created, only that watcher
stops instead of the whole loop', but the initial timer creation still calls
`error(...)`, which will abort startup for the worker (and likely take down
discovery entirely). To match the intended behavior, replace the `error(...)`
with `log.error(...)` and continue (or only disable that specific watcher),
similar to how `arm_watch` logs and returns when `ngx_timer_at` fails.
##########
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)
Review Comment:
This test relies on several wall-clock sleeps (mock server timing +
`ngx.sleep(6)`) but only allows a 10s overall timeout, leaving a relatively
small margin on slower/loaded CI workers. To reduce flakes, consider increasing
`--- timeout` (or reducing/eliminating fixed sleeps by making the mock trigger
index changes deterministically via an explicit request/flag).
##########
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
Review Comment:
This test relies on several wall-clock sleeps (mock server timing +
`ngx.sleep(6)`) but only allows a 10s overall timeout, leaving a relatively
small margin on slower/loaded CI workers. To reduce flakes, consider increasing
`--- timeout` (or reducing/eliminating fixed sleeps by making the mock trigger
index changes deterministically via an explicit request/flag).
##########
apisix/discovery/consul/client.lua:
##########
@@ -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
Review Comment:
`watch_catalog`/`watch_health` now conceptually run 'one blocking query per
timer run', but they still `core_sleep(random_delay)` on the no-change path.
That keeps the timer handler coroutine alive during the sleep and delays
responsiveness to `stop_flag`/worker exit checks until after the sleep
completes. A more timer-friendly approach is to avoid sleeping inside the watch
function and instead re-arm the timer with `random_delay` (e.g., by returning
the delay to the caller or letting `watch()` decide the next delay), so the run
ends promptly and stop/exiting checks happen between queries as described.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
