fix(consul): do not read the response body in the watcher threads

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)

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