feat(timerng) integrate the lua-resty-timer-ng (#8912)

* fix(plugin_servers) bypass timer library
* feat(AdminAPI) add `/timer` route
* feat(conf) add new config `legacy_timer_mechanism` (boolean)
* chore(rockspec) add `resty.timerng` as dependency
* tests(*) add `wait_timers_*()` for helper
* tests(plugin/zipkin) fix
* tests(unit/db/cache_warmup) fix
* tests(integration/cmd/quit) fix
* tests(integration/hybrid_mode/ocsp) fix
* tests(plugin/rate-limiting/access) fix
* tests(integration/db/query-semaphore) fix
* tests(integration/admin/timers) add tests for `http://kong_admin/timers`
* tests(integration/cmd/start_stop) fix
* tests(integration/proxy/router) fix
* docs(CHANGELOG) update

Co-authored-by: Mayo <[email protected]>
Co-authored-by: Wangchong Zhou <[email protected]>

CHANGELOG.md

@@ -198,6 +198,18 @@
 
 ### Additions
 
+#### Performance
+- Do not register unnecessary event handlers on Hybrid mode Control Plane
+nodes [#8452](https://github.com/Kong/kong/pull/8452).
+- Use the new timer library to improve performance,
+  except for the plugin server.
+  [8912](https://github.com/Kong/kong/pull/8912)
+
+#### Admin API
+
+- Added a new API `/timers` to get the timer statistics.
+  [8912](https://github.com/Kong/kong/pull/8912)
+
 #### Core
 
 - Added `cache_key` on target entity for uniqueness detection.
@@ -279,12 +291,6 @@ a restart (e.g., upon a plugin server crash).
   instead of `proxy_error_log` [8583](https://github.com/Kong/kong/pull/8583)
 
 
-### Additions
-
-#### Performance
-- Do not register unnecessary event handlers on Hybrid mode Control Plane
-nodes [#8452](https://github.com/Kong/kong/pull/8452).
-
 ## [2.8.0]
 
 ### Deprecations

kong-2.8.0-0.rockspec

@@ -41,6 +41,7 @@ dependencies = {
   "lua-resty-ipmatcher == 0.6.1",
   "lua-resty-acme == 0.8.0",
   "lua-resty-session == 3.10",
+  "lua-resty-timer-ng == 0.1.0",
 }
 build = {
   type = "builtin",

kong/api/routes/kong.lua

@@ -185,4 +185,9 @@ return {
       return kong.response.exit(200, copy)
     end
   },
+  ["/timers"] = {
+    GET = function (self, db, helpers)
+      return kong.response.exit(200, _G.timerng_stats())
+    end
+  }
 }

kong/globalpatches.lua

@@ -68,6 +68,50 @@ return function(options)
   end
 
 
+  do
+    _G.native_timer_at = ngx.timer.at
+    _G.native_timer_every = ngx.timer.every
+
+    local timerng
+
+    if options.cli or options.rbusted then
+      timerng = require("resty.timerng").new({
+        min_threads = 16,
+        max_threads = 32,
+      })
+
+      timerng:start()
+
+    else
+      timerng = require("resty.timerng").new()
+
+      -- TODO rename
+      _G.timerng_start = function (debug)
+        timerng:start()
+        timerng:set_debug(debug)
+      end
+
+    end
+
+    _G.ngx.timer.at = function (delay, callback, ...)
+      return timerng:at(delay, callback, ...)
+    end
+
+    _G.ngx.timer.every = function (interval, callback, ...)
+      return timerng:every(interval, callback, ...)
+    end
+
+    -- TODO rename
+    _G.timerng_stats = function ()
+      return timerng:stats({
+        verbose = true,
+        flamegraph = true,
+      })
+    end
+
+  end
+
+
 
   do  -- implement a Lua based shm for: cli (and hence rbusted)
 

kong/init.lua

@@ -592,6 +592,7 @@ function Kong.init_worker()
   -- duplicated seeds.
   math.randomseed()
 
+  _G.timerng_start(kong.configuration.log_level == "debug")
 
   -- init DB
 

kong/runloop/plugin_servers/init.lua

@@ -8,7 +8,6 @@ local kong = kong
 local ngx_var = ngx.var
 local coroutine_running = coroutine.running
 local get_plugin_info = proc_mgmt.get_plugin_info
-local ngx_timer_at = ngx.timer.at
 local subsystem = ngx.config.subsystem
 
 --- keep request data a bit longer, into the log timer
@@ -250,7 +249,7 @@ local function build_phases(plugin)
           response_status = ngx.status,
         }
 
-        ngx_timer_at(0, function()
+        _G.native_timer_at(0, function()
           local co = coroutine_running()
           save_for_later[co] = saved
 
@@ -322,7 +321,7 @@ function plugin_servers.start()
 
   for _, server_def in ipairs(proc_mgmt.get_server_defs()) do
     if server_def.start_command then
-      ngx_timer_at(0, pluginserver_timer, server_def)
+      _G.native_timer_at(0, pluginserver_timer, server_def)
     end
   end
 end

spec/01-unit/01-db/08-cache_warmup_spec.lua

@@ -1,4 +1,5 @@
 local cache_warmup = require("kong.cache.warmup")
+local helpers = require("spec.helpers")
 
 
 local function mock_entity(db_data, entity_name, cache_key)
@@ -211,9 +212,15 @@ describe("cache_warmup", function()
 
     cache_warmup._mock_kong(kong)
 
+    local runs_old = _G.timerng_stats().sys.runs
+
     assert.truthy(cache_warmup.execute({"my_entity", "services"}))
 
-    ngx.sleep(0) -- yield so that async DNS caching happens
+    -- waiting async DNS cacheing
+    helpers.wait_until(function ()
+      local runs = _G.timerng_stats().sys.runs
+      return runs_old < runs
+    end)
 
     -- `my_entity` isn't a core entity; lookup is on client cache
     assert.same(kong.cache:get("111").bbb, 222)
@@ -264,9 +271,15 @@ describe("cache_warmup", function()
 
     cache_warmup._mock_kong(kong)
 
+    local runs_old = _G.timerng_stats().sys.runs
+
     assert.truthy(cache_warmup.execute({"my_entity", "services"}))
 
-    ngx.sleep(0.001) -- yield so that async DNS caching happens
+    -- waiting async DNS cacheing
+    helpers.wait_until(function ()
+      local runs = _G.timerng_stats().sys.runs
+      return runs_old < runs
+    end)
 
     -- `my_entity` isn't a core entity; lookup is on client cache
     assert.same(kong.cache:get("111").bbb, 222)
@@ -279,6 +292,7 @@ describe("cache_warmup", function()
 
     -- skipped IP entry
     assert.same({ "example.com", "example.test" }, dns_queries)
+
   end)
 
 

spec/02-integration/02-cmd/02-start_stop_spec.lua

@@ -13,7 +13,9 @@ describe("kong start/stop #" .. strategy, function()
   end)
   after_each(function()
     helpers.kill_all()
+    os.execute("rm -rf " .. helpers.test_conf.prefix .. "/worker_events.sock")
   end)
+
   lazy_teardown(function()
     helpers.clean_prefix()
   end)
@@ -178,6 +180,10 @@ describe("kong start/stop #" .. strategy, function()
   end)
 
   describe("verbose args", function()
+    after_each(function ()
+      os.execute("rm -rf " .. helpers.test_conf.prefix .. "/worker_events.sock")
+    end)
+
     it("accepts verbose --v", function()
       local _, _, stdout = assert(helpers.kong_exec("start --v --conf " .. helpers.test_conf_path))
       assert.matches("[verbose] prefix in use: ", stdout, nil, true)
@@ -226,18 +232,23 @@ describe("kong start/stop #" .. strategy, function()
   end)
 
   describe("/etc/hosts resolving in CLI", function()
-    it("resolves #cassandra hostname", function()
-      assert(helpers.kong_exec("start --vv --run-migrations --conf " .. helpers.test_conf_path, {
-        cassandra_contact_points = "localhost",
-        database = "cassandra"
-      }))
-    end)
-    it("resolves #postgres hostname", function()
-      assert(helpers.kong_exec("start --conf " .. helpers.test_conf_path, {
-        pg_host = "localhost",
-        database = "postgres"
-      }))
-    end)
+    if strategy == "cassandra" then
+      it("resolves #cassandra hostname", function()
+        assert(helpers.kong_exec("start --vv --run-migrations --conf " .. helpers.test_conf_path, {
+          cassandra_contact_points = "localhost",
+          database = "cassandra"
+        }))
+      end)
+
+    elseif strategy == "postgres" then
+      it("resolves #postgres hostname", function()
+        assert(helpers.kong_exec("start --conf " .. helpers.test_conf_path, {
+          pg_host = "localhost",
+          database = "postgres"
+        }))
+      end)
+    end
+
   end)
 
   -- TODO: update with new error messages and behavior

spec/02-integration/02-cmd/08-quit_spec.lua

@@ -31,6 +31,6 @@ describe("kong quit", function()
     assert(helpers.kong_exec("quit --wait 2 --prefix " .. helpers.test_conf.prefix))
     ngx.update_time()
     local duration = ngx.now() - start
-    assert.is.near(2, duration, 1.6)
+    assert.is.near(2, duration, 2.5)
   end)
 end)

spec/02-integration/03-db/09-query-semaphore_spec.lua

@@ -33,14 +33,18 @@ describe("#postgres Postgres query locks", function()
   end)
 
   it("results in query error failing to acquire resource", function()
+    local wait_timers_ctx = helpers.wait_timers_begin()
+
     local res = assert(client:send {
       method = "GET",
       path = "/slow-resource?prime=true",
       headers = { ["Content-Type"] = "application/json" }
     })
     assert.res_status(204 , res)
 
-    -- make a request that would run a query while no resources are available
+    -- wait for zero-delay timer
+    helpers.wait_timers_end(wait_timers_ctx, 0.5)
+
     res = assert(client:send {
       method = "GET",
       path = "/slow-resource",
@@ -49,5 +53,6 @@ describe("#postgres Postgres query locks", function()
     local body = assert.res_status(500 , res)
     local json = cjson.decode(body)
     assert.same({ error = "error acquiring query semaphore: timeout" }, json)
+
   end)
 end)

spec/02-integration/04-admin_api/20-timers_spec.lua

@@ -0,0 +1,54 @@
+local helpers = require "spec.helpers"
+local cjson = require "cjson"
+
+
+for _, strategy in helpers.each_strategy() do
+
+describe("Admin API[#" .. strategy .. "]" , function()
+local client
+
+    lazy_setup(function()
+        helpers.get_db_utils(strategy)
+
+        assert(helpers.start_kong({
+        database = strategy,
+        nginx_conf = "spec/fixtures/custom_nginx.template",
+        }))
+
+        client = helpers.admin_client()
+    end)
+
+    teardown(function()
+        if client then
+            client:close()
+        end
+        helpers.stop_kong()
+    end)
+
+    it("/timers", function ()
+        local res = assert(client:send {
+            method = "GET",
+            path = "/timers",
+            headers = { ["Content-Type"] = "application/json" }
+        })
+
+        local body = assert.res_status(200 , res)
+        local json = cjson.decode(body)
+
+        assert(type(json.flamegraph.running) == "string")
+        assert(type(json.flamegraph.pending) == "string")
+        assert(type(json.flamegraph.elapsed_time) == "string")
+
+        assert(type(json.sys.total) == "number")
+        assert(type(json.sys.runs) == "number")
+        assert(type(json.sys.running) == "number")
+        assert(type(json.sys.pending) == "number")
+        assert(type(json.sys.waiting) == "number")
+
+        assert(type(json.timers) == "table")
+
+    end)
+
+end)
+
+end

spec/02-integration/05-proxy/02-router_spec.lua

@@ -72,6 +72,9 @@ local function insert_routes(bp, routes)
     local cfg = bp.done()
     local yaml = declarative.to_yaml_string(cfg)
     local admin_client = helpers.admin_client()
+
+    local wait_timers_ctx = helpers.wait_timers_begin()
+
     local res = assert(admin_client:send {
       method  = "POST",
       path    = "/config",
@@ -85,9 +88,12 @@ local function insert_routes(bp, routes)
     assert.res_status(201, res)
     admin_client:close()
 
+    -- wait for timers finish
+    helpers.wait_timers_end(wait_timers_ctx, 0.5)
   end
 
-  ngx.sleep(0.01)  -- temporary wait
+  ngx.sleep(0.01)  -- temporary wait for worker events
+
   return routes
 end