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..276c96b9a635 --- /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