nic-6443 commented on code in PR #14032:
URL: https://github.com/apache/apisix/pull/14032#discussion_r4256780419
##########
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:
This can't happen as described. `watch_catalog` / `watch_health` already
compare the returned `X-Consul-Index` with the stored index for their type and
return nothing when they're equal, so an index the other watcher's round
already committed comes back as "no change", not as a result: the watcher logs
the info line, sleeps 1-5s and queries again. There's no yield between that
comparison and the `watch_result_is_valid` call in `handle_change`, so the
stored index can't move in between, and an equal index never reaches the
backoff branch. A watcher that sees a change while the other round is still
running doesn't validate at all, it parks and queries again after the round. So
`watch_result_is_valid` only fails on the error indexes (or a 0 index), same as
on master.
##########
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:
I'd keep this one. The sleep is at most 5s and only on the no-change path,
while a single run already holds a blocking query for up to `wait` (60s by
default), so ending the run before the sleep doesn't change how fast
`stop_flag` takes effect in any meaningful way. Keeping the sleep inside the
watch functions also keeps their no-change behaviour exactly as it was before.
##########
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:
That's the same as before this PR: master also raises from `start_registry`
when the initial `connect` timer can't be created, and callers that need to
survive it wrap `start_registry` in `pcall`. The "only that watcher stops" bit
in the description is about re-arming a running watcher, not about startup. I'd
rather not change what a failed start looks like in this PR.
##########
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:
Fair point, bumped the timeout to 15s in 8c4d6801e so there's plenty of room
on a slow runner. I kept the time-based index bumps since the race needs the
catalog watch to return while the health body is still streaming, which is what
the mock timing sets up.
--
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]
