fix(balancer) start only needed upstream update timers (#8694)

When Kong is using worker_consistency = eventual, a timer is used to update the internal balancer state on every worker_state_update_frequency seconds. Instead of using the same timer for all changes, the said timer was being started again on every balancer update, the previous ones were kept running, piling up the number of timers.

FTI-3274

kong/runloop/balancer/init.lua

@@ -25,15 +25,13 @@ local pairs = pairs
 local tostring = tostring
 local table = table
 local table_concat = table.concat
-local timer_at = ngx.timer.at
 local run_hook = hooks.run_hook
 local var = ngx.var
 
 
 local CRIT = ngx.CRIT
 local ERR = ngx.ERR
 local WARN = ngx.WARN
-local DEBUG = ngx.DEBUG
 local EMPTY_T = pl_tablex.readonly {}
 
 
@@ -189,14 +187,7 @@ local function init()
     end
   end
 
-  local _
-  local frequency = kong.configuration.worker_state_update_frequency or 1
-  _, err = timer_at(frequency, upstreams.update_balancer_state)
-  if err then
-    log(CRIT, "unable to start update proxy state timer: ", err)
-  else
-    log(DEBUG, "update proxy state timer scheduled")
-  end
+  upstreams.update_balancer_state()
 end
 
 

kong/runloop/balancer/upstreams.lua

@@ -20,6 +20,7 @@ local timer_at = ngx.timer.at
 
 
 local CRIT = ngx.CRIT
+local DEBUG = ngx.DEBUG
 local ERR = ngx.ERR
 
 local GLOBAL_QUERY_OPTS = { workspace = null, show_ws_id = true }
@@ -232,11 +233,13 @@ local function set_upstream_events_queue(operation, upstream_data)
 end
 
 
-function upstreams_M.update_balancer_state(premature)
+local update_balancer_state_running
+local function update_balancer_state_timer(premature)
   if premature then
     return
   end
 
+  update_balancer_state_running = true
 
   while upstream_events_queue[1] do
     local event  = upstream_events_queue[1]
@@ -250,14 +253,30 @@ function upstreams_M.update_balancer_state(premature)
   end
 
   local frequency = kong.configuration.worker_state_update_frequency or 1
-  local _, err = timer_at(frequency, upstreams_M.update_balancer_state)
+  local _, err = timer_at(frequency, update_balancer_state_timer)
   if err then
+    update_balancer_state_running = false
     log(CRIT, "unable to reschedule update proxy state timer: ", err)
   end
 
 end
 
 
+function upstreams_M.update_balancer_state()
+  if update_balancer_state_running then
+    return
+  end
+
+  local frequency = kong.configuration.worker_state_update_frequency or 1
+  local _, err = timer_at(frequency, update_balancer_state_timer)
+  if err then
+    log(CRIT, "unable to start update proxy state timer: ", err)
+  else
+    log(DEBUG, "update proxy state timer scheduled")
+  end
+end
+
+
 
 
 --------------------------------------------------------------------------------

spec/02-integration/04-admin_api/15-off_spec.lua

@@ -1113,3 +1113,97 @@ describe("Admin API #off with Unique Foreign #unique", function()
     -- assert.equal(references.data[1].unique_foreign.id, unique_reference.unique_foreign.id)
   end)
 end)
+
+describe("Admin API #off worker_consistency=eventual", function()
+
+  local client
+
+  lazy_setup(function()
+    assert(helpers.start_kong({
+      database = "off",
+      lmdb_map_size = LMDB_MAP_SIZE,
+      worker_consistency = "eventual",
+      worker_state_update_frequency = 0.1,
+    }))
+  end)
+
+  lazy_teardown(function()
+    helpers.stop_kong(nil, true)
+  end)
+
+  before_each(function()
+    client = assert(helpers.admin_client())
+  end)
+
+  after_each(function()
+    if client then
+      client:close()
+    end
+  end)
+
+  it("does not increase timer usage (regression)", function()
+    -- 1. configure a simple service
+    local res = assert(client:send {
+      method = "POST",
+      path = "/config",
+      body = helpers.unindent([[
+        _format_version: '1.1'
+        services:
+        - name: konghq
+          url: http://konghq.com
+          path: /
+        plugins:
+        - name: prometheus
+      ]]),
+      headers = {
+        ["Content-Type"] = "text/yaml"
+      },
+    })
+    assert.response(res).has.status(201)
+
+    -- 2. check the timer count
+    res = assert(client:send {
+      method  = "GET",
+      path    = "/metrics",
+    })
+    local res_body = assert.res_status(200, res)
+    local req1_pending_timers = assert.matches('kong_nginx_timers{state="pending"} %d+', res_body)
+    local req1_running_timers = assert.matches('kong_nginx_timers{state="running"} %d+', res_body)
+    req1_pending_timers = assert(tonumber(string.match(req1_pending_timers, "%d")))
+    req1_running_timers = assert(tonumber(string.match(req1_running_timers, "%d")))
+
+    -- 3. update the service
+    res = assert(client:send {
+      method = "POST",
+      path = "/config",
+      body = helpers.unindent([[
+        _format_version: '1.1'
+        services:
+        - name: konghq
+          url: http://konghq.com
+          path: /install#kong-community
+        plugins:
+        - name: prometheus
+      ]]),
+      headers = {
+        ["Content-Type"] = "text/yaml"
+      },
+    })
+    assert.response(res).has.status(201)
+
+    -- 4. check if timer count is still the same
+    res = assert(client:send {
+      method  = "GET",
+      path    = "/metrics",
+    })
+    local res_body = assert.res_status(200, res)
+    local req2_pending_timers = assert.matches('kong_nginx_timers{state="pending"} %d+', res_body)
+    local req2_running_timers = assert.matches('kong_nginx_timers{state="running"} %d+', res_body)
+    req2_pending_timers = assert(tonumber(string.match(req2_pending_timers, "%d")))
+    req2_running_timers = assert(tonumber(string.match(req2_running_timers, "%d")))
+
+    assert.equal(req1_pending_timers, req2_pending_timers)
+    assert.equal(req1_running_timers, req2_running_timers)
+  end)
+
+end)