From e32cfa5311ab1c8774c455157852b079cc9d026c Mon Sep 17 00:00:00 2001 From: Nic Date: Thu, 8 Oct 2026 06:27:03 +0000 Subject: [PATCH] fix(consul): do not read the response body in the watcher threads connect() spawns the catalog and health watchers, waits for the first one and kills 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)". The watchers only need the status and the X-Consul-Index header. Read them with resty.http request(), which does all its I/O on the calling coroutine, and close the connection without reading the body. --- apisix/discovery/consul/client.lua | 63 ++++++----- t/discovery/consul-watch-race.t | 168 +++++++++++++++++++++++++++++ 2 files changed, 204 insertions(+), 27 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..9e6f2974c902 100644 --- a/apisix/discovery/consul/client.lua +++ b/apisix/discovery/consul/client.lua @@ -114,41 +114,51 @@ function _M.sort_nodes(nodes, sort_type) end --- ─── resty.consul options ───────────────────────────────────────────── +-- ─── blocking query watchers ────────────────────────────────────────── -local function get_opts(consul_server, is_catalog) - local opts = { - host = consul_server.host, - port = consul_server.port, - connect_timeout = consul_server.connect_timeout, - read_timeout = consul_server.read_timeout, - default_args = { - token = consul_server.token, - } - } - if not consul_server.keepalive then - return opts +-- The watchers only need the status and the X-Consul-Index header, so they read +-- them with resty.http request() and close the connection without reading the +-- body. resty.http reads a body in a child coroutine, and connect() kills the +-- watcher that has not returned yet: ngx.thread.kill does not cancel a read +-- pending in a child coroutine, and its completion would later resume connect() +-- at an unrelated yield point. +local function watch_request(consul_server, url, index) + local httpc = http.new() + httpc:set_timeout(consul_server.connect_timeout) + local ok, err = httpc:connect(consul_server.host, consul_server.port) + if not ok then + return nil, err end - opts.default_args.wait = consul_server.wait_timeout - - if is_catalog then - opts.default_args.index = consul_server.catalog_index + local query + if consul_server.keepalive then + -- blocking query: wait plus the 1/16 jitter Consul may add, plus read timeout + local wait = consul_server.wait_timeout * 1000 + httpc:set_timeout(wait + wait / 16 + consul_server.read_timeout) + query = {index = index, wait = consul_server.wait_timeout .. "s"} else - opts.default_args.index = consul_server.health_index + httpc:set_timeout(consul_server.read_timeout) end - return opts + local res + res, err = httpc:request({ + path = "/v1" .. url, + query = query, + headers = {["X-Consul-Token"] = consul_server.token}, + }) + httpc:close() + if not res then + return nil, err + end + + return {status = res.status, headers = res.headers} end --- ─── blocking query watchers ────────────────────────────────────────── - 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_result, watch_err = watch_request(consul_server, + consul_server.consul_watch_catalog_url, consul_server.catalog_index) local watch_error_info = (watch_err ~= nil and watch_err) or ((watch_result ~= nil and watch_result.status ~= 200) and watch_result.status) @@ -174,10 +184,9 @@ 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_result, watch_err = watch_request(consul_server, + consul_server.consul_watch_health_url, consul_server.health_index) local watch_error_info = (watch_err ~= nil and watch_err) or ((watch_result ~= nil and watch_result.status ~= 200) and watch_result.status) 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