chore(deps) bump resty.worker.events from 1.0.0 to 2.0.0

### Summary

#### 2.0.0, 16-September-2020

- BREAKING: the `post` function does not call `poll` anymore, making all events
  asynchronous. When an immediate treatment to an event is needed an explicit
  call to `poll` must be done.
- BREAKING: the `post_local` function does not immediately execute the
  event anymore, making all local events asynchronous. When an immediate
  treatment to an event is needed an explicit call to `poll` must be done.
- fix: prevent spinning at 100% CPU when during a reload the event-shm is
  cleared
- fix: improved logging in case of failure to write to shm (add payload size
  for troubleshooting purposes)
- fix: do not log the payload anymore, since it might expose sensitive data
  through the logs
- change: updated `shm_retries` default to 999
- change: changed timer loop to a sleep-loop (performance)
- fix: when re-configuring make sure callbacks table is initialized

kong-2.1.4-0.rockspec

@@ -30,7 +30,7 @@ dependencies = {
   "luasyslog == 1.0.0",
   "lua_pack == 1.0.5",
   "lua-resty-dns-client == 5.0.1",
-  "lua-resty-worker-events == 1.0.0",
+  "lua-resty-worker-events == 2.0.0",
   "lua-resty-mediador == 0.1.2",
   "lua-resty-healthcheck == 2.0.0",
   "lua-resty-cookie == 0.1.0",

kong/api/api_helpers.lua

@@ -257,6 +257,8 @@ local function parse_params(fn)
 
     local res, err = fn(self, ...)
 
+    kong.worker_events.poll()
+
     if err then
       kong.log.err(err)
       return ngx.exit(500)

kong/db/declarative/init.lua

@@ -809,7 +809,7 @@ do
     ok, err, default_ws = declarative.load_into_cache(entities, meta, hash, SHADOW)
     if ok then
       ok, err = kong.worker_events.post("declarative", "flip_config", default_ws)
-      if ok ~= "done" then
+      if ok then
         return nil, "failed to flip declarative config cache pages: " .. (err or ok)
       end
 

kong/init.lua

@@ -1298,6 +1298,8 @@ end
 
 
 function Kong.admin_content(options)
+  kong.worker_events.poll()
+
   local ctx = ngx.ctx
   if not ctx.workspace then
     ctx.workspace = kong.default_workspace
@@ -1374,6 +1376,7 @@ do
 
   function Kong.stream_config_listener()
     local sock, err = ngx.req.socket()
+
     if not sock then
       kong.log.crit("unable to obtain request socket: ", err)
       return

kong/pdk/response.lua

@@ -775,6 +775,10 @@ local function new(self, major_version)
     -- return kong.response.exit(200, "Success")
     -- ```
     function _RESPONSE.exit(status, body, headers)
+      if self.worker_events and ngx.get_phase() == "content" then
+        self.worker_events.poll()
+      end
+
       check_phase(rewrite_access_header)
 
       if ngx.headers_sent then
@@ -947,6 +951,10 @@ local function new(self, major_version)
   --
   -- return kong.response.error(403)
   function _RESPONSE.error(status, message, headers)
+    if self.worker_events and ngx.get_phase() == "content" then
+      self.worker_events.poll()
+    end
+
     check_phase(rewrite_access_header)
 
     if ngx.headers_sent then

spec/01-unit/09-balancer_spec.lua

@@ -1,5 +1,7 @@
 local utils = require "kong.tools.utils"
 local mocker = require "spec.fixtures.mocker"
+local we = require "resty.worker.events"
+
 
 local ws_id = utils.uuid()
 
@@ -549,6 +551,7 @@ for _, consistency in ipairs({"strict", "eventual"}) do
         }
         for _, t in ipairs(tests) do
           assert(balancer.post_health(upstream_ph, t.host, nil, t.port, t.health))
+          we.poll()
           local health_info = assert(balancer.get_upstream_health("ph"))
           local response = t.health and "HEALTHY" or "UNHEALTHY"
           assert.same(response,
@@ -586,12 +589,14 @@ for _, consistency in ipairs({"strict", "eventual"}) do
             port = 1111,
             host = {hostname = "localhost"},
           }}, 429)
+        we.poll()
         my_balancer.report_http_status({
           address = {
             ip = "127.0.0.1",
             port = 1111,
             host = {hostname = "localhost"},
           }}, 200)
+        we.poll()
         balancer.unsubscribe_from_healthcheck_events(cb)
         my_balancer.report_http_status({
           address = {

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, 0.8)
+    assert.is.near(2, duration, 1.6)
   end)
 end)

spec/02-integration/03-db/11-postgres-ro_spec.lua

@@ -53,6 +53,8 @@ for _, strategy in helpers.each_strategy() do
         body = assert.res_status(201, res)
         json = cjson.decode(body)
 
+        ngx.sleep(1)
+
         route_id = json.id
 
         res = assert(proxy_client:send({
@@ -70,6 +72,8 @@ for _, strategy in helpers.each_strategy() do
         }))
         assert.res_status(204, res)
 
+        ngx.sleep(1)
+
         res = assert(proxy_client:send({
           method  = "GET",
           path    = "/",
@@ -124,6 +128,8 @@ for _, strategy in helpers.each_strategy() do
         }))
         assert.res_status(201, res)
 
+        ngx.sleep(1)
+
         res = assert(proxy_client:send({
           method  = "GET",
           path    = "/",

spec/02-integration/04-admin_api/08-targets_routes_spec.lua

@@ -679,6 +679,7 @@ describe("Admin API #" .. strategy, function()
         end)
 
         it("checks every combination of valid and invalid upstream and target", function()
+          ngx.sleep(1)
           for i, u in ipairs({ utils.uuid(), "invalid", upstream.name, upstream.id }) do
             for j, t in ipairs({ utils.uuid(), "invalid:1234", wrong_target.id, target.target, target.id }) do
               for _, e in ipairs({ "healthy", "unhealthy" }) do
@@ -701,6 +702,7 @@ describe("Admin API #" .. strategy, function()
             path = target_path .. "/unhealthy"
           })
           assert.same(204, status, body)
+          ngx.sleep(1)
           status, body = assert(client_send {
             method = "GET",
             path = "/upstreams/" .. upstream.id .. "/health"
@@ -714,6 +716,7 @@ describe("Admin API #" .. strategy, function()
             path = target_path .. "/healthy"
           })
           assert.same(204, status)
+          ngx.sleep(1)
           status, body = assert(client_send {
             method = "GET",
             path = "/upstreams/" .. upstream.id .. "/health"
@@ -725,12 +728,14 @@ describe("Admin API #" .. strategy, function()
         end)
 
         it("flips the target status from HEALTHY to UNHEALTHY", function()
+          ngx.sleep(1)
           local status, body, json
           status = assert(client_send {
             method = "POST",
             path = target_path .. "/healthy"
           })
           assert.same(204, status)
+          ngx.sleep(1)
           status, body = assert(client_send {
             method = "GET",
             path = "/upstreams/" .. upstream.id .. "/health"
@@ -744,6 +749,7 @@ describe("Admin API #" .. strategy, function()
             path = target_path .. "/unhealthy"
           })
           assert.same(204, status)
+          ngx.sleep(1)
           status, body = assert(client_send {
             method = "GET",
             path = "/upstreams/" .. upstream.id .. "/health"

spec/02-integration/05-proxy/04-plugins_triggering_spec.lua

@@ -1190,6 +1190,8 @@ for _, strategy in helpers.each_strategy() do
 
             assert.res_status(201, res)
 
+            ngx.sleep(1)
+
             local res = assert(proxy_client:get("/status/400", {
               headers = {
                 ["Host"] = "runs-init-worker.org",

spec/02-integration/05-proxy/10-balancer/01-healthchecks_spec.lua

@@ -840,7 +840,7 @@ for _, strategy in helpers.each_strategy() do
                 })
 
                 -- wait for old healthchecks to stop
-                ngx.sleep(0.5)
+                ngx.sleep(1)
 
                 -- start server
                 local server1 = bu.http_server(localhost, port, { 1 })