cache-locks

kong/core/cluster.lua

@@ -1,6 +1,7 @@
 local singletons = require "kong.singletons"
 local cluster_utils = require "kong.tools.cluster"
 local cache = require "kong.tools.database_cache"
+local ngx_log = ngx.log
 
 local resty_lock
 local status, res = pcall(require, "resty.lock")
@@ -15,7 +16,7 @@ local ASYNC_AUTOJOIN_RETRIES = 20 -- Try for max a minute (3s * 20)
 local function create_timer(at, cb)
   local ok, err = ngx.timer.at(at, cb)
   if not ok then
-    ngx.log(ngx.ERR, "[cluster] failed to create timer: ", err)
+    ngx_log(ngx.ERR, "[cluster] failed to create timer: ", err)
   end
 end
 
@@ -25,25 +26,29 @@ local function async_autojoin(premature)
   -- If this node is the only node in the cluster, but other nodes are present, then try to join them
   -- This usually happens when two nodes are started very fast, and the first node didn't write his
   -- information into the datastore yet. When the second node starts up, there is nothing to join yet.
-  local lock = resty_lock:new("cluster_autojoin_locks", {
+  local lock, err = resty_lock:new("cluster_autojoin_locks", {
     exptime = ASYNC_AUTOJOIN_INTERVAL - 0.001
   })
+  if not lock then
+    ngx_log(ngx.ERR, "failed to init lock dictionary", err)
+    return
+  end
   local elapsed = lock:lock("async_autojoin")
   if elapsed and elapsed == 0 then
     -- If the current member count on this node's cluster is 1, but there are more than 1 active nodes in
     -- the DAO, then try to join them
     local count, err = singletons.dao.nodes:count()
     if err then
-      ngx.log(ngx.ERR, tostring(err))
+      ngx_log(ngx.ERR, tostring(err))
     elseif count > 1 then
       local members, err = singletons.serf:members()
       if err then
-        ngx.log(ngx.ERR, tostring(err))
+        ngx_log(ngx.ERR, tostring(err))
       elseif #members < 2 then
         -- Trigger auto-join
         local _, err = singletons.serf:autojoin(cluster_utils.get_node_name(singletons.configuration))
         if err then
-          ngx.log(ngx.ERR, tostring(err))
+          ngx_log(ngx.ERR, tostring(err))
         end
       else
         return -- The node is already in the cluster and no need to continue
@@ -74,12 +79,12 @@ local function send_keepalive(premature)
     local node_name = cluster_utils.get_node_name(singletons.configuration)
     local nodes, err = singletons.dao.nodes:find_all {name = node_name}
     if err then
-      ngx.log(ngx.ERR, tostring(err))
+      ngx_log(ngx.ERR, tostring(err))
     elseif #nodes == 1 then
       local node = nodes[1]
       local _, err = singletons.dao.nodes:update(node, node, {ttl=singletons.configuration.cluster.ttl_on_failure})
       if err then
-        ngx.log(ngx.ERR, tostring(err))
+        ngx_log(ngx.ERR, tostring(err))
       end
     end
   end

kong/core/reports.lua

@@ -107,9 +107,13 @@ end
 ping_handler = function(premature)
   if premature then return end
 
-  local lock = resty_lock:new("reports_locks", {
+  local lock, err = resty_lock:new("reports_locks", {
     exptime = ping_interval - 0.001
   })
+  if not lock then
+    log_error("failed to init lock dictionary: ", err)
+    return
+  end
 
   local elapsed, err = lock:lock("ping")
   if not elapsed then

kong/templates/nginx_kong.lua

@@ -34,6 +34,7 @@ lua_shared_dict cache ${{MEM_CACHE_SIZE}};
 lua_shared_dict reports_locks 100k;
 lua_shared_dict cluster_locks 100k;
 lua_shared_dict cluster_autojoin_locks 100k;
+lua_shared_dict cache_locks 100k;
 lua_shared_dict cassandra 1m;
 lua_shared_dict cassandra_prepared 5m;
 lua_socket_log_errors off;

kong/tools/database_cache.lua

@@ -1,5 +1,7 @@
+local resty_lock = require "resty.lock"
 local cjson = require "cjson"
 local cache = ngx.shared.cache
+local ngx_log = ngx.log
 
 local CACHE_KEYS = {
   APIS = "apis",
@@ -120,21 +122,46 @@ function _M.all_apis_by_dict_key()
 end
 
 function _M.get_or_set(key, cb)
+  local lock, err = resty_lock:new("cache_locks", {
+    exptime = 10,
+    timeout = 5
+  })
+  if not lock then
+    ngx_log(ngx.ERR, "failed to init lock dictionary: ", err)
+    return
+  end
+
   local value, err
-  -- Try to get
+
+  -- Try to get the value from the cache
+  value = _M.get(key)
+  if value then return value end
+
+  -- The value is missing, acquire a lock
+  local elapsed, err = lock:lock(key)
+  if not elapsed then
+    ngx_log(ngx.ERR, "failed to acquire cache lock: ", err)
+  end
+
+  -- Lock acquired. Since in the meantime another worker may have 
+  -- populated the value we have to check again
   value = _M.get(key)
   if not value then
     -- Get from closure
     value, err = cb()
-    if err then
-      return nil, err
-    elseif value then
+    if value then
       local ok, err = _M.set(key, value)
       if not ok then
-        ngx.log(ngx.ERR, err)
+        ngx_log(ngx.ERR, err)
       end
     end
   end
+
+  local ok, err = lock:unlock()
+  if not ok then
+    ngx_log(ngx.ERR, "failed to unlock: ", err)
+  end
+
   return value
 end
 

spec/02-integration/07-cache/database_cache_spec.lua

@@ -0,0 +1,78 @@
+local helpers = require "spec.helpers"
+local cjson = require "cjson"
+local meta = require "kong.meta"
+
+describe("Resolver", function()
+  local client
+  setup(function()
+
+    assert(helpers.dao.apis:insert {
+      request_host = "mockbin.com",
+      upstream_url = "http://mockbin.com"
+    })
+
+    assert(helpers.start_kong({
+      ["custom_plugins"] = "database-cache",
+      lua_package_path = "?/init.lua;./kong/?.lua;./spec/fixtures/?.lua"
+    }))
+
+    -- Add the plugin
+    local admin_client = helpers.admin_client()
+    local res = assert(admin_client:send {
+      method = "POST",
+      path = "/apis/mockbin.com/plugins/",
+      body = {
+        name = "database-cache"
+      },
+      headers = {
+        ["Content-Type"] = "application/json"
+      }
+    })
+    assert.res_status(201, res)
+    admin_client:close()
+  end)
+
+  teardown(function()
+    helpers.stop_kong()
+  end)
+
+  it("avoids dog-pile effect", function()
+    local function make_request(premature, sleep_time)
+      local client = helpers.proxy_client()
+      local res = assert(client:send {
+        method = "GET",
+        path = "/status/200?sleep="..sleep_time,
+        headers = {
+          ["Host"] = "mockbin.com"
+        }
+      })
+      res:read_body()
+      client:close()
+    end
+
+    assert(ngx.timer.at(0, make_request, 2))
+    assert(ngx.timer.at(0, make_request, 5))
+    assert(ngx.timer.at(0, make_request, 1))
+
+    helpers.wait_until(function()
+      local admin_client = helpers.admin_client()
+      local res = assert(admin_client:send {
+        method = "GET",
+        path = "/cache/invocations"
+      })
+      local body = res:read_body()
+      admin_client:close()
+      return cjson.decode(body).message == 3
+    end, 10)
+
+    -- Invocation are 3, but lookups should be 1
+    local admin_client = helpers.admin_client()
+    local res = assert(admin_client:send {
+      method = "GET",
+      path = "/cache/lookups"
+    })
+    local body = res:read_body()
+    admin_client:close()
+    assert.equal(1, cjson.decode(body).message)
+  end)
+end)

spec/fixtures/kong/plugins/database-cache/handler.lua

@@ -0,0 +1,35 @@
+local BasePlugin = require "kong.plugins.base_plugin"
+local cache = require "kong.tools.database_cache"
+
+local INVOCATIONS = "invocations"
+local LOOKUPS = "lookups"
+
+local DatabaseCacheHandler = BasePlugin:extend()
+
+DatabaseCacheHandler.PRIORITY = 1000
+
+function DatabaseCacheHandler:new()
+  DatabaseCacheHandler.super.new(self, "database-cache")
+end
+
+function DatabaseCacheHandler:init_worker()
+  DatabaseCacheHandler.super.init_worker(self)
+
+  cache.rawset(INVOCATIONS, 0)
+  cache.rawset(LOOKUPS, 0)
+end
+
+function DatabaseCacheHandler:access(conf)
+  DatabaseCacheHandler.super.access(self)
+
+  cache.get_or_set("pile_effect", function()
+    cache.incr(LOOKUPS, 1)
+    -- Adds some delay
+    ngx.sleep(tonumber(ngx.req.get_uri_args().sleep))
+    return true
+  end)
+
+  cache.incr(INVOCATIONS, 1)
+end
+
+return DatabaseCacheHandler

spec/fixtures/kong/plugins/database-cache/schema.lua

@@ -0,0 +1,3 @@
+return { 
+  fields = {}
+}