Fix/policies (Kong#1467)

.travis.yml

@@ -1,5 +1,8 @@
 sudo: false
 
+services:
+  - redis-server
+
 notifications:
   email: false
 

CHANGELOG.md

@@ -8,6 +8,9 @@ The main focus of this release is Kong's new CLI. With a simpler configuration f
 - :warning: Kong uses a new configuration file, with an easier syntax than the previous YAML file.
 - New arguments for the CLI, such as verbose, debug and tracing flags. We also avoid requiring the configuration file as an argument to each command as per the previous CLI.
 - Customization of the Nginx configuration can now be taken care of using two different approaches: with a custom Nginx configuration template and using `kong start --template <file>`, or by using `kong compile` to generate the Kong Nginx sub-configuration, and `include` it in a custom Nginx instance.
+- Plugins:
+  - Rate Limiting: the `continue_on_error` property is now called `cluster_fault_tolerant`.
+  - Response Rate Limiting: the `continue_on_error` property is now called `cluster_fault_tolerant`.
 
 ### Added
 
@@ -23,6 +26,8 @@ The main focus of this release is Kong's new CLI. With a simpler configuration f
   - Bot Detection: Added a new plugin that protects APIs by detecting common bots and crawlers. [#1413](https://github.com/Mashape/kong/pull/1413)
   - correlation-id: new "tracker" generator, identifying requests per worker and connection. [#1288](https://github.com/Mashape/kong/pull/1288)
   - request/response-transformer: ability to add strings including colon characters. [#1353](https://github.com/Mashape/kong/pull/1353)
+  - Rate Limiting: support for new rate-limiting policies (`cluster`, `local` and `redis`), and for a new `limit_by` property to force rate-limiting by `consumer`, `credential` or `ip`.
+  - Response Rate Limiting: support for new rate-limiting policies (`cluster`, `local` and `redis`), and for a new `limit_by` property to force rate-limiting by `consumer`, `credential` or `ip`.
 
 ### Removed
 

kong-0.9.0rc1-0.rockspec

@@ -176,6 +176,7 @@ build = {
     ["kong.plugins.response-ratelimiting.header_filter"] = "kong/plugins/response-ratelimiting/header_filter.lua",
     ["kong.plugins.response-ratelimiting.log"] = "kong/plugins/response-ratelimiting/log.lua",
     ["kong.plugins.response-ratelimiting.schema"] = "kong/plugins/response-ratelimiting/schema.lua",
+    ["kong.plugins.response-ratelimiting.policies"] = "kong/plugins/response-ratelimiting/policies.lua",
     ["kong.plugins.response-ratelimiting.dao.cassandra"] = "kong/plugins/response-ratelimiting/dao/cassandra.lua",
     ["kong.plugins.response-ratelimiting.dao.postgres"] = "kong/plugins/response-ratelimiting/dao/postgres.lua",
 

kong/plugins/rate-limiting/handler.lua

@@ -8,6 +8,7 @@ local BasePlugin = require "kong.plugins.base_plugin"
 local ngx_log = ngx.log
 local pairs = pairs
 local tostring = tostring
+local ngx_timer_at = ngx.timer.at
 
 local RATELIMIT_LIMIT = "X-RateLimit-Limit"
 local RATELIMIT_REMAINING = "X-RateLimit-Remaining"
@@ -105,8 +106,16 @@ function RateLimitingHandler:access(conf)
     end
   end
 
+  local incr = function(premature, conf, api_id, identifier, current_timestamp, value)
+    if premature then return end
+    policies[policy].increment(conf, api_id, identifier, current_timestamp, value)
+  end
+
   -- Increment metrics for all periods if the request goes through
-  policies[policy].increment(conf, api_id, identifier, current_timestamp, 1)
+  local ok, err = ngx_timer_at(0, incr, conf, api_id, identifier, current_timestamp, 1)
+  if not ok then
+    ngx_log(ngx.ERR, "failed to create timer: ", err)
+  end
 end
 
 return RateLimitingHandler

kong/plugins/rate-limiting/policies.lua

@@ -1,14 +1,16 @@
 local singletons = require "kong.singletons"
 local timestamp = require "kong.tools.timestamp"
 local cache = require "kong.tools.database_cache"
+local redis = require "resty.redis"
 local ngx_log = ngx.log
-local ngx_timer_at = ngx.timer.at
 
-local get_local_key = function(api_id, identifier, period_date, period)
-  return string.format("ratelimit:%s:%s:%s:%s", api_id, identifier, period_date, period)
+local pairs = pairs
+local fmt = string.format
+
+local get_local_key = function(api_id, identifier, period_date, name, period)
+  return fmt("ratelimit:%s:%s:%s:%s:%s", api_id, identifier, period_date, name, period)
 end
 
---TODO: Check that it really expires
 local EXPIRATIONS = {
   second = 1,
   minute = 60,
@@ -22,7 +24,6 @@ return {
   ["local"] = {
     increment = function(conf, api_id, identifier, current_timestamp, value)
       local periods = timestamp.get_timestamps(current_timestamp)
-      local ok = true
       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
@@ -31,12 +32,9 @@ return {
 
         local _, err = cache.incr(cache_key, value)
         if err then
-          ok = false
           ngx_log("[rate-limiting] could not increment counter for period '"..period.."': "..tostring(err))
         end
       end
-
-      return ok
     end,
     usage = function(conf, api_id, identifier, current_timestamp, name)
       local periods = timestamp.get_timestamps(current_timestamp)
@@ -50,17 +48,9 @@ return {
   },
   ["cluster"] = {
     increment = function(conf, api_id, identifier, current_timestamp, value)
-      local incr = function(premature, api_id, identifier, current_timestamp, value)
-        if premature then return end
-        local _, stmt_err = singletons.dao.ratelimiting_metrics:increment(api_id, identifier, current_timestamp, value)
-        if stmt_err then
-          ngx_log(ngx.ERR, "failed to increment: ", tostring(stmt_err))
-        end
-      end
-
-      local ok, err = ngx_timer_at(0, incr, api_id, identifier, current_timestamp, 1)
-      if not ok then
-        ngx_log(ngx.ERR, "failed to create timer: ", err)
+      local _, stmt_err = singletons.dao.ratelimiting_metrics:increment(api_id, identifier, current_timestamp, value)
+      if stmt_err then
+        ngx_log(ngx.ERR, "failed to increment: ", tostring(stmt_err))
       end
     end,
     usage = function(conf, api_id, identifier, current_timestamp, name)
@@ -73,20 +63,17 @@ return {
   },
   ["redis"] = {
     increment = function(conf, api_id, identifier, current_timestamp, value)
-      local redis = require "resty.redis"
       local red = redis:new()
-
       red:set_timeout(conf.redis_timeout)
       local ok, err = red:connect(conf.redis_host, conf.redis_port)
       if not ok then
         ngx_log(ngx.ERR, "failed to connect to Redis: ", err)
         return
       end
 
-      if conf.redis_password then
+      if conf.redis_password and conf.redis_password ~= "" then
         local ok, err = red:auth(conf.redis_password)
         if not ok then
-
           ngx_log(ngx.ERR, "failed to connect to Redis: ", err)
           return
         end
@@ -121,24 +108,19 @@ return {
         ngx_log(ngx.ERR, "failed to set Redis keepalive: ", err)
         return
       end
-
-      return true
     end,
     usage = function(conf, api_id, identifier, current_timestamp, name)
-      local redis = require "resty.redis"
       local red = redis:new()
-
       red:set_timeout(conf.redis_timeout)
       local ok, err = red:connect(conf.redis_host, conf.redis_port)
       if not ok then
         ngx_log(ngx.ERR, "failed to connect to Redis: ", err)
         return
       end
 
-      if conf.redis_password then
+      if conf.redis_password and conf.redis_password ~= "" then
         local ok, err = red:auth(conf.redis_password)
         if not ok then
-
           ngx_log(ngx.ERR, "failed to connect to Redis: ", err)
           return
         end

kong/plugins/response-ratelimiting/access.lua

@@ -1,4 +1,4 @@
-local singletons = require "kong.singletons"
+local policies = require "kong.plugins.response-ratelimiting.policies"
 local timestamp = require "kong.tools.timestamp"
 local responses = require "kong.tools.responses"
 local utils = require "kong.tools.utils"
@@ -10,30 +10,34 @@ local _M = {}
 
 local RATELIMIT_REMAINING = "X-RateLimit-Remaining"
 
-local function get_identifier()
+local function get_identifier(conf)
   local identifier
 
   -- Consumer is identified by ip address or authenticated_credential id
-  if ngx.ctx.authenticated_credential then
-    identifier = ngx.ctx.authenticated_credential.id
-  else
-    identifier = ngx.var.remote_addr
+  if conf.limit_by == "consumer" then
+    identifier = ngx.ctx.authenticated_consumer and ngx.ctx.authenticated_consumer.id
+    if not identifier and ngx.ctx.authenticated_credential then -- Fallback on credential
+      identifier = ngx.ctx.authenticated_credential.id
+    end
+  elseif conf.limit_by == "credential" then
+    identifier = ngx.ctx.authenticated_credential and ngx.ctx.authenticated_credential.id
   end
 
+  if not identifier then identifier = ngx.var.remote_addr end
+
   return identifier
 end
 
-local function get_current_usage(api_id, identifier, current_timestamp, limits)
+local function get_usage(conf, api_id, identifier, current_timestamp, limits)
   local usage = {}
 
   for k, v in pairs(limits) do -- Iterate over limit names
     for lk, lv in pairs(v) do -- Iterare over periods
-      local current_metric, err = singletons.dao.response_ratelimiting_metrics:find(api_id, identifier, current_timestamp, lk, k)
+      local current_usage, err = policies[conf.policy].usage(conf, api_id, identifier, current_timestamp, lk, k)
       if err then
-        return false, err
+        return nil, err
       end
 
-      local current_usage = current_metric and current_metric.value or 0
       local remaining = lv - current_usage
 
       if not usage[k] then usage[k] = {} end
@@ -56,13 +60,13 @@ function _M.execute(conf)
   local current_timestamp = timestamp.get_utc()
   ngx.ctx.current_timestamp = current_timestamp -- For later use
   local api_id = ngx.ctx.api.id
-  local identifier = get_identifier()
+  local identifier = get_identifier(conf)
   ngx.ctx.identifier = identifier -- For later use
 
   -- Load current metric for configured period
-  local usage, err = get_current_usage(api_id, identifier, current_timestamp, conf.limits)
+  local usage, err = get_usage(conf, api_id, identifier, current_timestamp, conf.limits)
   if err then
-    if conf.continue_on_error then
+    if conf.cluster_fault_tolerant then
       ngx.log(ngx.ERR, "failed to get usage: ", tostring(err))
       return
     else

kong/plugins/response-ratelimiting/handler.lua

@@ -24,7 +24,7 @@ end
 function ResponseRateLimitingHandler:log(conf)
   if not ngx.ctx.stop_log and ngx.ctx.usage then
     ResponseRateLimitingHandler.super.log(self)
-    log.execute(ngx.ctx.api.id, ngx.ctx.identifier, ngx.ctx.current_timestamp, ngx.ctx.increments, ngx.ctx.usage)
+    log.execute(conf, ngx.ctx.api.id, ngx.ctx.identifier, ngx.ctx.current_timestamp, ngx.ctx.increments, ngx.ctx.usage)
   end
 end
 

kong/plugins/response-ratelimiting/log.lua

@@ -1,28 +1,21 @@
-local singletons = require "kong.singletons"
+local policies = require "kong.plugins.response-ratelimiting.policies"
+local pairs = pairs
 
 local _M = {}
 
-local function increment(api_id, identifier, current_timestamp, value, name)
-  -- Increment metrics for all periods if the request goes through
-  local _, stmt_err = singletons.dao.response_ratelimiting_metrics:increment(api_id, identifier, current_timestamp, value, name)
-  if stmt_err then
-    ngx.log(ngx.ERR, tostring(stmt_err))
-  end
-end
-
-local function log(premature, api_id, identifier, current_timestamp, increments, usage)
+local function log(premature, conf, api_id, identifier, current_timestamp, increments, usage)
   if premature then return end
 
   -- Increment metrics for all periods if the request goes through
   for k, v in pairs(usage) do
     if increments[k] and increments[k] ~= 0 then
-      increment(api_id, identifier, current_timestamp, increments[k], k)
+      policies[conf.policy].increment(conf, api_id, identifier, current_timestamp, increments[k], k)
     end
   end
 end
 
-function _M.execute(api_id, identifier, current_timestamp, increments, usage)
-  local ok, err = ngx.timer.at(0, log, api_id, identifier, current_timestamp, increments, usage)
+function _M.execute(conf, api_id, identifier, current_timestamp, increments, usage)
+  local ok, err = ngx.timer.at(0, log, conf, api_id, identifier, current_timestamp, increments, usage)
   if not ok then
     ngx.log(ngx.ERR, "failed to create timer: ", err)
   end

kong/plugins/response-ratelimiting/migrations/cassandra.lua

@@ -14,5 +14,40 @@ return {
     down = [[
       DROP TABLE response_ratelimiting_metrics;
     ]]
+  },
+  {
+    name = "2016-08-04-321512_response-rate-limiting_policies",
+    up = function(_, _, dao)
+      local rows, err = dao.plugins:find_all {name = "response-ratelimiting"}
+      if err then return err end
+
+      for i = 1, #rows do
+        local response_rate_limiting = rows[i]
+
+        -- Delete the old one to avoid conflicts when inserting the new one
+        local _, err = dao.plugins:delete(response_rate_limiting)
+        if err then return err end
+
+        local _, err = dao.plugins:insert {
+          name = "response-ratelimiting",
+          api_id = response_rate_limiting.api_id,
+          consumer_id = response_rate_limiting.consumer_id,
+          enabled = response_rate_limiting.enabled,
+          config = {
+            second = response_rate_limiting.config.second,
+            minute = response_rate_limiting.config.minute,
+            hour = response_rate_limiting.config.hour,
+            day = response_rate_limiting.config.day,
+            month = response_rate_limiting.config.month,
+            year = response_rate_limiting.config.year,
+            block_on_first_violation = response_rate_limiting.config.block_on_first_violation,
+            limit_by = "consumer",
+            policy = "cluster",
+            cluster_fault_tolerant = response_rate_limiting.config.continue_on_error
+          }
+        }
+        if err then return err end
+      end
+    end
   }
 }

kong/plugins/response-ratelimiting/migrations/postgres.lua

@@ -32,5 +32,40 @@ return {
     down = [[
       DROP TABLE response_ratelimiting_metrics;
     ]]
+  },
+  {
+    name = "2016-08-04-321512_response-rate-limiting_policies",
+    up = function(_, _, dao)
+      local rows, err = dao.plugins:find_all {name = "response-ratelimiting"}
+      if err then return err end
+
+      for i = 1, #rows do
+        local response_rate_limiting = rows[i]
+
+        -- Delete the old one to avoid conflicts when inserting the new one
+        local _, err = dao.plugins:delete(response_rate_limiting)
+        if err then return err end
+
+        local _, err = dao.plugins:insert {
+          name = "response-ratelimiting",
+          api_id = response_rate_limiting.api_id,
+          consumer_id = response_rate_limiting.consumer_id,
+          enabled = response_rate_limiting.enabled,
+          config = {
+            second = response_rate_limiting.config.second,
+            minute = response_rate_limiting.config.minute,
+            hour = response_rate_limiting.config.hour,
+            day = response_rate_limiting.config.day,
+            month = response_rate_limiting.config.month,
+            year = response_rate_limiting.config.year,
+            block_on_first_violation = response_rate_limiting.config.block_on_first_violation,
+            limit_by = "consumer",
+            policy = "cluster",
+            cluster_fault_tolerant = response_rate_limiting.config.continue_on_error
+          }
+        }
+        if err then return err end
+      end
+    end
   }
 }