feat(cache) in-memory and shared dict caching strategy

kong-0.9.2-0.rockspec

@@ -27,7 +27,8 @@ dependencies = {
   "lua-resty-iputils == 0.2.1",
   "luacrypto == 0.3.2",
   "luasyslog == 1.0.0",
-  "lua_pack == 1.0.4"
+  "lua_pack == 1.0.4",
+  "lua-resty-worker-events == 0.2.0"
 }
 build = {
   type = "builtin",

kong/api/routes/cache.lua

@@ -30,4 +30,4 @@ return {
       return responses.send_HTTP_NOT_FOUND()
     end
   }
-}
+}

kong/api/routes/cluster.lua

@@ -1,5 +1,7 @@
 local singletons = require "kong.singletons"
 local responses = require "kong.tools.responses"
+local constants = require "kong.constants"
+local ev = require "resty.worker.events"
 
 local pairs = pairs
 local table_insert = table.insert
@@ -60,7 +62,9 @@ return {
       local message_t = self.params
 
       -- The type is always upper case
-      if message_t.type then
+      if not message_t or not message_t.type then
+        return responses.send_HTTP_BAD_REQUEST()
+      else
         message_t.type = string_upper(message_t.type)
       end
 
@@ -88,9 +92,9 @@ return {
       end
 
       -- Trigger event in the node
-      singletons.events:publish(message_t.type, message_t)
+      ev.post(constants.CACHE.CLUSTER, message_t.type, message_t)
 
       return responses.send_HTTP_OK()
     end
   }
-}
+}

kong/constants.lua

@@ -41,5 +41,8 @@ return {
   SYSLOG = {
     ADDRESS = "kong-hf.mashape.com",
     PORT = 61828
+  },
+  CACHE = {
+    CLUSTER = "cluster"
   }
-}
+}

kong/core/cluster.lua

@@ -55,11 +55,11 @@ local function async_autojoin(premature)
     end
 
     -- Create retries counter key if it doesn't exist
-    if not cache.get(cache.autojoin_retries_key()) then
-      cache.rawset(cache.autojoin_retries_key(), 0)
+    if not cache.sh_get(cache.autojoin_retries_key()) then
+      cache.sh_set(cache.autojoin_retries_key(), 0)
     end
 
-    local autojoin_retries = cache.incr(cache.autojoin_retries_key(), 1) -- Increment retries counter
+    local autojoin_retries = cache.sh_incr(cache.autojoin_retries_key(), 1) -- Increment retries counter
     if (autojoin_retries < ASYNC_AUTOJOIN_RETRIES) then
       create_timer(ASYNC_AUTOJOIN_INTERVAL, async_autojoin)
     end

kong/core/reports.lua

@@ -121,11 +121,11 @@ ping_handler = function(premature)
   elseif elapsed == 0 then
     send {
       signal = "ping",
-      requests = cache.get(cache.requests_key()) or 0,
+      requests = cache.sh_get(cache.requests_key()) or 0,
       unique_id = unique_str,
       database = singletons.configuration.database
     }
-    cache.rawset(cache.requests_key(), 0)
+    cache.sh_set(cache.requests_key(), 0)
   end
 
   create_ping_timer()
@@ -137,12 +137,12 @@ return {
   -----------------
   init_worker = function()
     if not enabled then return end
-    cache.rawset(cache.requests_key(), 0)
+    cache.sh_set(cache.requests_key(), 0)
     create_ping_timer()
   end,
   log = function()
     if not enabled then return end
-    cache.incr(cache.requests_key(), 1)
+    cache.sh_incr(cache.requests_key(), 1)
   end,
   -----------------
   -- custom methods
@@ -153,4 +153,4 @@ return {
   get_system_infos = get_system_infos,
   send = send,
   api_signal = "api"
-}
+}

kong/kong.lua

@@ -30,6 +30,7 @@ local core = require "kong.core.handler"
 local Serf = require "kong.serf"
 local utils = require "kong.tools.utils"
 local Events = require "kong.core.events"
+local constants = require "kong.constants"
 local singletons = require "kong.singletons"
 local DAOFactory = require "kong.dao.factory"
 local plugins_iterator = require "kong.core.plugins_iterator"
@@ -147,6 +148,28 @@ function Kong.init_worker()
   for _, plugin in ipairs(singletons.loaded_plugins) do
     plugin.handler:init_worker()
   end
+
+  local ev = require "resty.worker.events"
+  local handler = function(data, event, source, pid)
+    if source and source == constants.CACHE.CLUSTER then
+      singletons.events:publish(event, data)
+    end
+  end
+
+  ev.register(handler)
+
+  local ok, err = ev.configure {
+      shm = "process_events", -- defined by "lua_shared_dict"
+      timeout = 2,            -- life time of event data in shm
+      interval = 1,           -- poll interval (seconds)
+
+      wait_interval = 0.010,  -- wait before retry fetching event data
+      wait_max = 0.5,         -- max wait time before discarding event
+  }
+  if not ok then
+      ngx.log(ngx.ERR, "failed to start event system: ", err)
+      return
+  end
 end
 
 function Kong.ssl_certificate()
@@ -193,4 +216,4 @@ function Kong.log()
   core.log.after()
 end
 
-return Kong
+return Kong

kong/meta.lua

@@ -19,7 +19,7 @@ return {
   -- to lua-version's `set()` in the form {from, to}
   _DEPENDENCIES = {
     nginx = {"1.9.15.1", "1.11.2.1"},
-    serf = {"0.7.0", "0.7.0"},
+    serf = {"0.7.0", "0.8.0"},
     --resty = {}, -- not version dependent for now
     --dnsmasq = {} -- not version dependent for now
   }

kong/plugins/rate-limiting/policies.lua

@@ -26,11 +26,11 @@ return {
       local periods = timestamp.get_timestamps(current_timestamp)
       for period, period_date in pairs(periods) do
         local cache_key = get_local_key(api_id, identifier, period_date, period)
-        if not cache.rawget(cache_key) then
-          cache.rawset(cache_key, 0, EXPIRATIONS[period])
+        if not cache.sh_get(cache_key) then
+          cache.sh_set(cache_key, 0, EXPIRATIONS[period])
         end
 
-        local _, err = cache.incr(cache_key, value)
+        local _, err = cache.sh_incr(cache_key, value)
         if err then
           ngx_log("[rate-limiting] could not increment counter for period '"..period.."': "..tostring(err))
         end
@@ -39,7 +39,7 @@ return {
     usage = function(conf, api_id, identifier, current_timestamp, name)
       local periods = timestamp.get_timestamps(current_timestamp)
       local cache_key = get_local_key(api_id, identifier, periods[name], name)
-      local current_metric, err = cache.rawget(cache_key)
+      local current_metric, err = cache.sh_get(cache_key)
       if err then
         return nil, err
       end
@@ -138,4 +138,4 @@ return {
       return current_metric and current_metric or 0
     end
   }
-}
+}

kong/plugins/response-ratelimiting/policies.lua

@@ -26,11 +26,11 @@ return {
       local periods = timestamp.get_timestamps(current_timestamp)
       for period, period_date in pairs(periods) do
         local cache_key = get_local_key(api_id, identifier, period_date, name, period)
-        if not cache.rawget(cache_key) then
-          cache.rawset(cache_key, 0, EXPIRATIONS[period])
+        if not cache.sh_get(cache_key) then
+          cache.sh_set(cache_key, 0, EXPIRATIONS[period])
         end
 
-        local _, err = cache.incr(cache_key, value)
+        local _, err = cache.sh_incr(cache_key, value)
         if err then
           ngx_log("[rate-limiting] could not increment counter for period '"..period.."': "..tostring(err))
         end
@@ -39,7 +39,7 @@ return {
     usage = function(conf, api_id, identifier, current_timestamp, period, name)
       local periods = timestamp.get_timestamps(current_timestamp)
       local cache_key = get_local_key(api_id, identifier, periods[period], name, period)
-      local current_metric, err = cache.rawget(cache_key)
+      local current_metric, err = cache.sh_get(cache_key)
       if err then
         return nil, err
       end

kong/templates/nginx_kong.lua

@@ -39,6 +39,7 @@ 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 process_events 1m;
 lua_shared_dict cassandra 1m;
 lua_shared_dict cassandra_prepared 5m;
 lua_socket_log_errors off;