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