fix(healthchecks) ensure health posted via cluster_events can be read

kong/api/routes/upstreams.lua

@@ -84,7 +84,9 @@ local function post_health(is_healthy)
     end
 
     local health = is_healthy and 1 or 0
-    local packet = ("%s|%d|%d|%s"):format(ip, port, health, self.upstream.name)
+    local packet = ("%s|%d|%d|%s|%s"):format(ip, port, health,
+                                             self.upstream.id,
+                                             self.upstream.name)
     cluster_events:broadcast("balancer:post_health", packet)
 
     return responses.send_HTTP_NO_CONTENT()

kong/runloop/balancer.lua

@@ -799,7 +799,7 @@ end
 
 --------------------------------------------------------------------------------
 -- Update health status and broadcast to workers
--- @param upstream a table with upstream data
+-- @param upstream a table with upstream data: must have `name` and `id`
 -- @param hostname target hostname
 -- @param port target port
 -- @param is_healthy boolean: true if healthy, false if unhealthy

kong/runloop/handler.lua

@@ -336,9 +336,10 @@ return {
 
       -- manual health updates
       cluster_events:subscribe("balancer:post_health", function(data)
-        local ip, port, health, name = data:match("([^|]+)|([^|]+)|([^|]+)|(.*)")
+        local pattern = "([^|]+)|([^|]+)|([^|]+)|([^|]+)|(.*)"
+        local ip, port, health, id, name = data:match(pattern)
         port = tonumber(port)
-        local upstream = { name = name }
+        local upstream = { id = id, name = name }
         local ok, err = balancer.post_health(upstream, ip, port, health == "1")
         if not ok then
           log(ERR, "failed posting health of ", name, " to workers: ", err)

spec/02-integration/05-proxy/09-balancer_spec.lua

@@ -351,9 +351,9 @@ do
     assert.same(200, api_send("PATCH", "/upstreams/" .. upstream_name, data))
   end
 
-  get_upstream_health = function(upstream_name)
+  get_upstream_health = function(upstream_name, forced_port)
     local path = "/upstreams/" .. upstream_name .."/health"
-    local status, body = api_send("GET", path)
+    local status, body = api_send("GET", path, nil, forced_port)
     if status == 200 then
       return body
     end
@@ -428,20 +428,22 @@ local function truncate_relevant_tables(db, dao)
 end
 
 
-local function poll_wait_health(upstream_name, localhost, port, value)
+local function poll_wait_health(upstream_name, host, port, value, admin_port)
   local hard_timeout = 300
   local expire = ngx.now() + hard_timeout
   while ngx.now() < expire do
-    local health = get_upstream_health(upstream_name)
+    local health = get_upstream_health(upstream_name, admin_port)
     if health then
       for _, d in ipairs(health.data) do
-        if d.target == localhost .. ":" .. port and d.health == value then
+        if d.target == host .. ":" .. port and d.health == value then
           return
         end
       end
     end
     ngx.sleep(0.01) -- poll-wait
   end
+  assert(false, "timed out waiting for " .. host .. ":" .. port " in " ..
+                upstream_name .. " to become " .. value)
 end
 
 
@@ -490,14 +492,18 @@ for _, strategy in helpers.each_strategy() do
 
     describe("#healthchecks (#cluster)", function()
 
+      -- second node ports are Kong test ports + 10
+      local proxy_port_1 = 9000
+      local admin_port_1 = 9001
+      local proxy_port_2 = 9010
+      local admin_port_2 = 9011
+
       setup(function()
-        -- start a second Kong instance (ports are Kong test ports + 10)
+        -- start a second Kong instance
         helpers.start_kong({
           database   = strategy,
-          admin_listen = "127.0.0.1:9011",
-          proxy_listen = "127.0.0.1:9010",
-          proxy_listen_ssl = "127.0.0.1:9453",
-          admin_listen_ssl = "127.0.0.1:9454",
+          admin_listen = "127.0.0.1:" .. admin_port_2,
+          proxy_listen = "127.0.0.1:" .. proxy_port_2,
           prefix = "servroot2",
           log_level = "debug",
           db_update_frequency = 0.1,
@@ -514,24 +520,24 @@ for _, strategy in helpers.each_strategy() do
 
           it("does not perform health checks when disabled (#3304)", function()
 
-            local old_rv = get_router_version(9011)
+            local old_rv = get_router_version(admin_port_2)
 
             local upstream_name = add_upstream()
             local port = add_target(upstream_name, localhost)
             local api_host = add_api(upstream_name)
 
-            wait_for_router_update(old_rv, localhost, 9010, 9011)
+            wait_for_router_update(old_rv, localhost, proxy_port_2, admin_port_2)
 
             -- server responds, then fails, then responds again
             local server = http_server(localhost, port, { 20, 20, 20 })
 
             local seq = {
-              { port = 9010, oks = 10, fails = 0, last_status = 200 },
-              { port = 9000, oks = 10, fails = 0, last_status = 200 },
-              { port = 9010, oks = 0, fails = 10, last_status = 500 },
-              { port = 9000, oks = 0, fails = 10, last_status = 500 },
-              { port = 9010, oks = 10, fails = 0, last_status = 200 },
-              { port = 9000, oks = 10, fails = 0, last_status = 200 },
+              { port = proxy_port_2, oks = 10, fails = 0, last_status = 200 },
+              { port = proxy_port_1, oks = 10, fails = 0, last_status = 200 },
+              { port = proxy_port_2, oks = 0, fails = 10, last_status = 500 },
+              { port = proxy_port_1, oks = 0, fails = 10, last_status = 500 },
+              { port = proxy_port_2, oks = 10, fails = 0, last_status = 200 },
+              { port = proxy_port_1, oks = 10, fails = 0, last_status = 200 },
             }
             for i, test in ipairs(seq) do
               local oks, fails, last_status = client_requests(10, api_host, "127.0.0.1", test.port)
@@ -546,6 +552,31 @@ for _, strategy in helpers.each_strategy() do
             assert.same(20, server_fails)
 
           end)
+
+          it("propagates posted health info", function()
+
+            local old_rv = get_router_version(admin_port_2)
+
+            local upstream_name = add_upstream({
+              healthchecks = healthchecks_config {}
+            })
+            local port = add_target(upstream_name, localhost)
+
+            wait_for_router_update(old_rv, localhost, proxy_port_2, admin_port_2)
+
+            local health1 = get_upstream_health(upstream_name, admin_port_1)
+            local health2 = get_upstream_health(upstream_name, admin_port_2)
+
+            assert.same("HEALTHY", health1.data[1].health)
+            assert.same("HEALTHY", health2.data[1].health)
+
+            post_target_endpoint(upstream_name, localhost, port, "unhealthy")
+
+            poll_wait_health(upstream_name, localhost, port, "UNHEALTHY", admin_port_1)
+            poll_wait_health(upstream_name, localhost, port, "UNHEALTHY", admin_port_2)
+
+          end)
+
         end)
       end
     end)