fix(balancer) use correct upstream object in event handler

In the upstream event handler, `create_balancer` was being called
with the object received via the event, which contains "id" and
"name" only, and not the entire entity table containing the rest
of the upstream fields. This caused it create a healthchecker with
an empty configuration (ignoring the user's configuration),
which then fell back into the lua-resty-healthcheck defaults.

This fix obtains the proper entity object from the id and passes
it to `create_balancer`.

A regression test is included, which spawns two Kong instances
and reproduces the error scenario described by @mengskysama.

Fixes #3304.

kong/core/balancer.lua

@@ -575,35 +575,50 @@ end
 --------------------------------------------------------------------------------
 -- Called on any changes to an upstream.
 -- @param operation "create", "update" or "delete"
--- @param upstream Upstream table with `id` and `name` fields
-local function on_upstream_event(operation, upstream)
+-- @param upstream_data table with `id` and `name` fields
+local function on_upstream_event(operation, upstream_data)
+  local upstream_id = upstream_data.id
+  local upstream_name = upstream_data.name
 
   if operation == "create" then
 
     singletons.cache:invalidate_local("balancer:upstreams")
 
+    local upstream = get_upstream_by_id(upstream_id)
+    if not upstream then
+      log(ERR, "upstream not found for ", upstream_id)
+      return
+    end
+
     local _, err = create_balancer(upstream)
     if err then
-      log(CRIT, "failed creating balancer for ", upstream.name, ": ", err)
+      log(CRIT, "failed creating balancer for ", upstream_name, ": ", err)
     end
 
   elseif operation == "delete" or operation == "update" then
 
     singletons.cache:invalidate_local("balancer:upstreams")
-    singletons.cache:invalidate_local("balancer:upstreams:" .. upstream.id)
-    singletons.cache:invalidate_local("balancer:targets:"   .. upstream.id)
+    singletons.cache:invalidate_local("balancer:upstreams:" .. upstream_id)
+    singletons.cache:invalidate_local("balancer:targets:"   .. upstream_id)
 
-    local balancer = balancers[upstream.id]
+    local balancer = balancers[upstream_id]
     if balancer then
       stop_healthchecker(balancer)
     end
 
     if operation == "delete" then
-      balancers[upstream.id] = nil
+      balancers[upstream_id] = nil
+
     else
+      local upstream = get_upstream_by_id(upstream_id)
+      if not upstream then
+        log(ERR, "upstream not found for ", upstream_id)
+        return
+      end
+
       local _, err = create_balancer(upstream, true)
       if err then
-        log(ERR, "failed recreating balancer for ", upstream.name, ": ", err)
+        log(ERR, "failed recreating balancer for ", upstream_name, ": ", err)
       end
     end
 

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

@@ -258,11 +258,13 @@ local function http_server(timeout, host, port, counts, test_log)
 end
 
 
-local function client_requests(n, headers)
+local function client_requests(n, headers, host, port)
   local oks, fails = 0, 0
   local last_status
   for _ = 1, n do
-    local client = helpers.proxy_client()
+    local client = (host and port)
+                   and helpers.http_client(host, port)
+                   or  helpers.proxy_client()
     local res = client:send {
       method = "GET",
       path = "/",
@@ -302,6 +304,19 @@ local function api_send(method, path, body)
 end
 
 
+local function file_contains(filename, searched)
+  local fd = assert(io.open(filename, "r"))
+  for line in fd:lines() do
+    if line:find(searched, 1, true) then
+      fd:close()
+      return true
+    end
+  end
+  fd:close()
+  return false
+end
+
+
 local localhosts = {
   ipv4 = "127.0.0.1",
   ipv6 = "[0000:0000:0000:0000:0000:0000:0000:0001]",
@@ -500,6 +515,112 @@ dao_helpers.for_each_dao(function(kong_config)
 
     end)
 
+    describe("#healthchecks (#cluster)", function()
+
+      before_each(function()
+
+        helpers.start_kong({
+          log_level = "debug",
+          db_update_frequency = 0.1,
+        })
+
+        -- start a second Kong instance (ports are Kong test ports + 10)
+        helpers.start_kong({
+          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",
+          prefix = "servroot2",
+          log_level = "debug",
+          db_update_frequency = 0.1,
+        })
+      end)
+
+      after_each(function()
+        helpers.stop_kong(nil, true)
+        helpers.stop_kong("servroot2", true, true)
+      end)
+
+      it("does not perform health checks when disabled (#3304)", function()
+
+        assert.same(201, api_send("POST", "/apis", {
+          strip_uri = true,
+          hosts = {
+              "balancer.test"
+          },
+          name = "test",
+          methods = {
+              "GET",
+              "HEAD"
+          },
+          http_if_terminated = true,
+          https_only = false,
+          retries = 1,
+          uris = {
+              "/"
+          },
+          preserve_host = false,
+          upstream_connect_timeout = 1000,
+          upstream_read_timeout = 3000,
+          upstream_send_timeout = 3000,
+          upstream_url = "http://test",
+        }))
+
+        assert.same(201, api_send("POST", "/upstreams", {
+          name = "test",
+        }))
+
+        assert.same(201, api_send("POST", "/upstreams/test/targets", {
+          target = "127.0.0.1:" .. PORT
+        }))
+
+        helpers.wait_until(function()
+          return file_contains("servroot2/logs/error.log", "balancer:targets")
+        end, 10)
+
+        local timeout = 10
+        -- server responds, then fails, then responds again
+        local server = http_server(timeout, localhost, PORT, { 20, 20, 20 })
+
+        local oks, fails, last_status = client_requests(10)
+        assert.same(10, oks)
+        assert.same(0, fails)
+        assert.same(200, last_status)
+
+        oks, fails, last_status = client_requests(10, nil, "127.0.0.1", 9010)
+        assert.same(10, oks)
+        assert.same(0, fails)
+        assert.same(200, last_status)
+
+        oks, fails, last_status = client_requests(10)
+        assert.same(0, oks)
+        assert.same(10, fails)
+        assert.same(500, last_status)
+
+        oks, fails, last_status = client_requests(10, nil, "127.0.0.1", 9010)
+        assert.same(0, oks)
+        assert.same(10, fails)
+        assert.same(500, last_status)
+
+        oks, fails, last_status = client_requests(10)
+        assert.same(10, oks)
+        assert.same(0, fails)
+        assert.same(200, last_status)
+
+        oks, fails, last_status = client_requests(10, nil, "127.0.0.1", 9010)
+        assert.same(10, oks)
+        assert.same(0, fails)
+        assert.same(200, last_status)
+
+        -- collect server results
+        local _, server_oks, server_fails = server:join()
+        assert.same(40, server_oks)
+        assert.same(20, server_fails)
+
+      end)
+
+    end)
+
     describe("#healthchecks", function()
       local upstream