fix(healthcheck): read shm status so workers do not keep unhealthy nodes

apisix/balancer.lua

@@ -75,6 +75,20 @@ local function fetch_all_nodes(upstream)
 end
 
 
+-- Include shm health in the picker cache key. checker.status_ver only bumps
+-- when this worker receives a resty.events update, so a shm flip from another
+-- worker would otherwise keep serving a picker that still contains the
+-- unhealthy node (apache/apisix#13888).
+local function health_cache_ver(upstream, checker)
+    if not checker then
+        return "x"
+    end
+    local host = upstream.checks and upstream.checks.active and upstream.checks.active.host
+    local port = upstream.checks and upstream.checks.active and upstream.checks.active.port
+    return healthcheck_manager.node_status_ver(checker, upstream.nodes, host, port)
+end
+
+
 local function create_health_status(upstream, checker)
     local nodes = upstream.nodes
     local host = upstream.checks and upstream.checks.active and upstream.checks.active.host
@@ -135,7 +149,8 @@ local function fetch_health_status(upstream, checker, key, version)
         return nil
     end
 
-    local health_status = lrucache_health_status(key, version .. "#" .. checker.status_ver,
+    local health_status = lrucache_health_status(key,
+                                                 version .. "#" .. health_cache_ver(upstream, checker),
                                                  create_health_status, upstream, checker)
     if not health_status or health_status.all_unhealthy then
         return nil
@@ -304,7 +319,7 @@ local function pick_server(route, ctx)
     end
 
     if checker and up_conf.type ~= "chash" then
-        version = version .. "#" .. checker.status_ver
+        version = version .. "#" .. health_cache_ver(up_conf, checker)
     end
 
     -- the same picker will be used in the whole request, especially during the retry

apisix/healthcheck_manager.lua

@@ -20,6 +20,7 @@ local pcall   = pcall
 local exiting      = ngx.worker.exiting
 local pairs    = pairs
 local tostring = tostring
+local table_concat = table.concat
 local core = require("apisix.core")
 local config_local   = require("apisix.core.config_local")
 local resource = require("apisix.resource")
@@ -250,12 +251,46 @@ function _M.fetch_checker(resource_path, resource_ver)
 end
 
 
+-- resty.healthcheck INTERNAL_STATES: 1=healthy, 2=unhealthy,
+-- 3=mostly_healthy, 4=mostly_unhealthy. Routing treats 1 and 3 as usable.
+local SHM_HEALTHY = {
+    [1] = true,
+    [3] = true,
+}
+
+
+-- SHM is the cross-worker source of truth. get_target_status reads a
+-- worker-local cache filled asynchronously via resty.events, so a worker
+-- that missed the event (or whose checker was created after another worker
+-- already flipped the node) keeps routing to an unhealthy node
+-- (apache/apisix#13888).
+local function fetch_shm_target_status(checker, ip, port, hostname)
+    local shm = checker.shm
+    local prefix = checker.TARGET_STATE
+    if not shm or not prefix or not ip or not port then
+        return nil
+    end
+    hostname = hostname or ip
+    local key = prefix .. ":" .. ip .. ":" .. tostring(port) .. ":" .. hostname
+    local state = shm:get(key)
+    if state == nil then
+        return nil
+    end
+    return SHM_HEALTHY[state] == true
+end
+
+
 function _M.fetch_node_status(checker, ip, port, hostname)
     -- check if the checker is valid
     if not checker or checker.dead then
         return true
     end
 
+    local shm_ok = fetch_shm_target_status(checker, ip, port, hostname)
+    if shm_ok ~= nil then
+        return shm_ok
+    end
+
     local ok, err = checker:get_target_status(ip, port, hostname)
     if err == "target not found" then
         -- get_target_status reads a worker-local cache that resty.healthcheck fills
@@ -273,6 +308,24 @@ function _M.fetch_node_status(checker, ip, port, hostname)
 end
 
 
+-- Cache-key fragment that changes when shm (or the local status_ver) changes,
+-- so a picker cached against a stale worker-local view is rebuilt.
+function _M.node_status_ver(checker, nodes, host, port)
+    if not checker or not nodes then
+        return "x"
+    end
+    local n = #nodes
+    local parts = core.table.new(n + 1, 0)
+    parts[1] = tostring(checker.status_ver or 0)
+    for i = 1, n do
+        local node = nodes[i]
+        parts[i + 1] = _M.fetch_node_status(checker, node.host, port or node.port, host)
+                       and "1" or "0"
+    end
+    return table_concat(parts, ":")
+end
+
+
 local function add_working_pool(resource_path, resource_ver, checker, checks)
     working_pool[resource_path] = {
         version = resource_ver,

t/node/healthcheck-stale-worker-cache.t

@@ -0,0 +1,200 @@
+#
+# 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();
+worker_connections(256);
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: fetch_node_status reads shm, not the worker-local event cache
+# resty.healthcheck.get_target_status reads a per-worker cache filled by
+# resty.events. Active checks run on one worker and write shm first; a peer
+# that missed the event (or whose checker was created after the flip) keeps
+# a stale "healthy" cache. Routing must consult shm (apache/apisix#13888).
+--- config
+    location /t {
+        content_by_lua_block {
+            local healthcheck = require("resty.healthcheck")
+            local hm = require("apisix.healthcheck_manager")
+
+            local checks = {
+                active = {
+                    type = "http",
+                    http_path = "/status",
+                    healthy = { interval = 100, successes = 1 },
+                    unhealthy = { interval = 100, http_failures = 2 },
+                },
+            }
+
+            local checker, err = healthcheck.new({
+                name = "upstream#test-stale-worker-cache",
+                shm_name = "upstream-healthcheck",
+                events_module = "resty.events",
+                checks = checks,
+            })
+            if not checker then
+                ngx.say("new failed: ", err)
+                return
+            end
+
+            local ok
+            ok, err = checker:add_target("127.0.0.1", 1980, nil, true)
+            if not ok then
+                ngx.say("add_target failed: ", err)
+                checker:stop()
+                return
+            end
+
+            -- this worker's local cache still says healthy
+            assert(checker:get_target_status("127.0.0.1", 1980, "127.0.0.1") == true,
+                   "local cache should start healthy")
+
+            -- another worker marked the node unhealthy in shm without an event
+            local key = checker.TARGET_STATE .. ":127.0.0.1:1980:127.0.0.1"
+            assert(checker.shm:set(key, 2)) -- 2 = unhealthy
+
+            local healthy = hm.fetch_node_status(checker, "127.0.0.1", 1980, "127.0.0.1")
+            checker:stop()
+            ngx.say(healthy and "healthy" or "unhealthy")
+        }
+    }
+--- request
+GET /t
+--- response_body
+unhealthy
+
+
+
+=== TEST 2: priority failover stops sending traffic once shm marks the primary unhealthy
+# Same gap as TEST 1, through the balancer: after the picker is cached on
+# status_ver (which only bumps on a local event), a shm-only flip must still
+# rebuild the picker and fail over to the backup node.
+--- config
+    location /t {
+        content_by_lua_block {
+            local t = require("lib.test_admin").test
+            local http = require("resty.http")
+            local healthcheck = require("resty.healthcheck")
+
+            local code, body = t('/apisix/admin/routes/1',
+                ngx.HTTP_PUT,
+                [[{
+                    "uri": "/server_port",
+                    "upstream": {
+                        "type": "roundrobin",
+                        "nodes": [
+                            {"host": "127.0.0.1", "port": 1980, "weight": 1, "priority": 0},
+                            {"host": "127.0.0.1", "port": 1981, "weight": 1, "priority": -1}
+                        ],
+                        "checks": {
+                            "active": {
+                                "http_path": "/status",
+                                "healthy": {
+                                    "interval": 100,
+                                    "successes": 1
+                                },
+                                "unhealthy": {
+                                    "interval": 100,
+                                    "http_failures": 2
+                                }
+                            }
+                        }
+                    }
+                }]]
+            )
+            if code >= 300 then
+                ngx.say(body)
+                return
+            end
+
+            local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/server_port"
+            local function hit()
+                local httpc = http.new()
+                local res, err = httpc:request_uri(uri, {method = "GET", keepalive = false})
+                if not res then
+                    return nil, err
+                end
+                return res.body
+            end
+
+            -- enqueue checker creation and wait for the timer to publish it
+            local port, err = hit()
+            if not port then
+                ngx.say("warmup failed: ", err)
+                return
+            end
+            ngx.sleep(2)
+
+            -- picker is now cached against this worker's still-healthy local view
+            port, err = hit()
+            if port ~= "1980" then
+                ngx.say("expected primary 1980 before shm flip, got ", port or err)
+                return
+            end
+
+            local name = "upstream#/apisix/routes/1"
+            local list = healthcheck.get_target_list(name, "upstream-healthcheck")
+            if not list then
+                ngx.say("no shm target list")
+                return
+            end
+
+            local shm = ngx.shared["upstream-healthcheck"]
+            local flipped = false
+            for _, target in ipairs(list) do
+                if tonumber(target.port) == 1980 then
+                    local key = "lua-resty-healthcheck:" .. name .. ":state:"
+                                .. target.ip .. ":" .. target.port .. ":" .. target.hostname
+                    assert(shm:set(key, 2)) -- unhealthy, no event
+                    flipped = true
+                end
+            end
+            if not flipped then
+                ngx.say("primary not in shm target list")
+                return
+            end
+
+            local ports = {}
+            for i = 1, 8 do
+                port, err = hit()
+                if not port then
+                    ngx.say("request ", i, " failed: ", err)
+                    return
+                end
+                ports[#ports + 1] = port
+            end
+
+            for i, p in ipairs(ports) do
+                if p ~= "1981" then
+                    ngx.say("request ", i, " still hit ", p, " after shm unhealthy")
+                    return
+                end
+            end
+            ngx.say("passed")
+        }
+    }
+--- request
+GET /t
+--- response_body
+passed
+--- timeout: 10