Policy: rate_limit_headers fix cache issues.

When testing rate_limit_headers the number of APICast workers are always set to
1. When the number of workers increased, the lru_cache is not longer effective
due to multiple threads.

Because we only have information about usage statistics on Authrep, and this is
using ngx.shared info, this policy also need to use shared information to be
able to reply without doing the authrep in all sessions.

Fix THREESCALE-3795

Signed-off-by: Eloy Coto <eloy.coto@acalustra.com>

CHANGELOG.md

@@ -18,7 +18,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Added
 
 - Added upstream Mutual TLS policy [THREESCALE-672](https://issues.jboss.org/browse/THREESCALE-672) [PR #1182](https://github.com/3scale/APIcast/pull/1182)
-- Added Rate-limit headers policy [THREESCALE-3795](https://issues.jboss.org/browse/THREESCALE-3795) [PR #1166](https://github.com/3scale/APIcast/pull/1166)
+- Added Rate-limit headers policy [THREESCALE-3795](https://issues.jboss.org/browse/THREESCALE-3795) [PR #1166](https://github.com/3scale/APIcast/pull/1166) [PR #1197](https://github.com/3scale/APIcast/pull/1197)
 - Added Content-caching policy [THREESCALE-2894](https://issues.jboss.org/browse/THREESCALE-2894) [PR #1182](https://github.com/3scale/APIcast/pull/1182)
 - Added Nginx request_id variable to context [PR #1184](https://github.com/3scale/APIcast/pull/1184)
 - Added HTTP verb on url_rewriten [PR #1187](https://github.com/3scale/APIcast/pull/1187)  [THREESCALE-5259](https://issues.jboss.org/browse/THREESCALE-5259)

gateway/http.d/shdict.conf

@@ -1,3 +1,4 @@
 lua_shared_dict api_keys 10m;
+lua_shared_dict rate_limit_headers 20m;
 lua_shared_dict configuration 10m;
 lua_shared_dict locks 1m;

gateway/src/apicast/policy/rate_limit_headers/cache.lua

@@ -7,10 +7,10 @@ local mt = { __index = _M }
 local default_namespace = "rate_limit_namespace"
 
 
-function _M.new(size, namespace)
-  local cache_size = tonumber(size) or 1000
+function _M.new(namespace)
   local self = setmetatable({}, mt)
-  local cache, err = lrucache.new(cache_size)
+  -- LRU cache used only for unittesting, where ngx.shared is not enabled.
+  local cache, err = ngx.shared.rate_limit_headers or lrucache.new(1)
   if err then
       ngx.log(ngx.ERR, "Cannot start cache for usage metrics, err=", err)
       return err
@@ -30,25 +30,23 @@ function _M:decrement_usage_metric(usage)
     end
 
     local key = self:get_key(usage)
-    local data = self.cache:get(key)
-
-    if not data then
+    local raw_data = self.cache:get(key)
+    if not raw_data then
         -- If it's here should not, because this should be called in the
         -- post_action, so return an empty one0
         return cache_entry.Init_empty(usage)
     end
-    -- data:decrement(delta)
-    -- Take care here, delta can be from usage.delta
+    local data = cache_entry.import(usage, raw_data)
 
     data:decrement(1)
-    self.cache:set(key, data)
+    self.cache:set(key, data:export())
     return data
 end
 
 function _M:reset_or_create_usage_metric(usage, max, remaining, reset)
     local key = self:get_key(usage)
     local data = cache_entry.new(usage, max, remaining, reset)
-    self.cache:set(key, data)
+    self.cache:set(key, data:export())
     return data
 end
 

gateway/src/apicast/policy/rate_limit_headers/cache_entry.lua

@@ -2,6 +2,7 @@ local now = ngx.now
 
 local counter = require "resty.counter"
 local countdown_counter =  require("apicast.policy.rate_limit_headers.countdown_counter")
+local stringx = require 'pl.stringx'
 
 local _M = {}
 local mt = { __index = _M }
@@ -41,4 +42,25 @@ function _M:reset(max, remaining, reset)
   self.reset = countdown_counter.new(reset, now())
 end
 
+function _M:export()
+  return string.format("%s#%s#%s",
+    self.limit:__tostring(),
+    self.remaining:__tostring(),
+    self.reset:remaining_secs_positive(now()))
+end
+
+function _M.import(usage, exported_data)
+  if not usage then
+    return nil
+  end
+
+  if not exported_data then
+    return _M.Init_empty(usage)
+  end
+
+  local data = stringx.split(exported_data, "#")
+  return _M.new(usage, data[1] or 0, data[2] or 0, data[3] or 0)
+end
+
+
 return _M

gateway/src/apicast/policy/rate_limit_headers/rate_limit_headers.lua

@@ -20,7 +20,7 @@ local reset_header = "RateLimit-Reset"
 
 function _M.new(config)
   local self = new(config)
-  self.cache = usage_cache.new(1000, "rate_limit_headers")
+  self.cache = usage_cache.new("rate_limit_headers")
   return self
 end
 

spec/policy/custom_metrics/custom_metrics_spec.lua

@@ -0,0 +1,79 @@
+local custom_metrics =  require("apicast.policy.custom_metrics")
+local ngx_variable = require 'apicast.policy.ngx_variable'
+local backend_client = require('apicast.backend_client')
+local Usage = require('apicast.usage')
+
+
+describe('custom metrics policy', function()
+  local context = {}
+  local usage = {}
+
+  before_each(function()
+    ngx.var = {}
+    ngx.header = {}
+
+    stub(ngx_variable, 'available_context', function(context) return context end)
+    stub(ngx.req, 'get_headers', function() return {} end)
+    stub(ngx.resp, 'get_headers', function() return {} end)
+
+    context = { usage = {} }
+    stub(context.usage, 'merge', function() return end)
+
+    usage = Usage.new()
+    usage:add("foo", 1)
+  end)
+
+  it('if Auth cache is disabled, report the usage', function()
+      local config = {
+        rules = {
+          {
+            metric = "foo",
+            increment = "1",
+            condition = {
+              combine_op = "and",
+              operations = {
+                { left = "foo", op = "==", right = "foo" }
+              }
+            }
+          }
+        }
+      }
+
+      local policy = custom_metrics.new(config)
+
+      stub(policy, "report", function(context, usage) return end)
+
+      policy:post_action(context)
+      assert.spy(policy.report).was_called()
+      assert.spy(policy.report).was.called_with(context, usage)
+  end)
+
+  it('if Auth cache is enabled, usage is incremented', function()
+    ngx.var.cached_key = true
+
+    local config = {
+      rules = {
+        {
+          metric = "foo",
+          increment = "1",
+          condition = {
+            combine_op = "and",
+            operations = {
+              { left = "foo", op = "==", right = "foo" }
+            }
+          }
+        }
+      }
+    }
+
+    local policy = custom_metrics.new(config)
+
+    stub(policy, "report", function(context, usage) return end)
+    policy:post_action(context)
+    assert.spy(policy.report).was_not_called()
+    assert.spy(context.usage.merge).was_called()
+
+    assert.spy(context.usage.merge).was_called_with(context.usage, usage)
+
+  end)
+end)

spec/policy/rate_limit_headers/cache_entry_spec.lua

@@ -26,5 +26,35 @@ describe("Cache key", function()
     assert.same(key.remaining:__tostring(), "-1")
   end)
 
+  describe("Import/export methods",  function()
+
+    it("Works as expected", function()
+
+      local entry = cache_entry.new(usage, 1, 10, 3)
+      assert.same(entry:export(), "1#10#3")
+
+      local data = cache_entry.import(usage, entry:export())
+      assert.same(data.limit:__tostring(), "1")
+      assert.same(data.remaining:__tostring(), "10")
+    end)
+
+
+    it("invalid usage returns nil", function()
+      assert.falsy(cache_entry.import())
+    end)
+
+    it("invalid import raw data", function()
+      local data = cache_entry.import(usage, "1#asd#123")
+      assert.same(data.limit:__tostring(), "1")
+      assert.same(data.remaining:__tostring(), "0")
+    end)
+
+    it("no raw_data return empty data", function()
+      local data = cache_entry.import(usage, "")
+      assert.same(data.limit:__tostring(), "0")
+      assert.same(data.remaining:__tostring(), "0")
+    end)
+
+  end)
 
 end)

spec/policy/rate_limit_headers/cache_spec.lua

@@ -8,15 +8,15 @@ describe("Cache key", function()
   end)
 
   it("New works as expected", function()
-    local c = cache.new(100, "namespace")
+    local c = cache.new("namespace")
     assert.Same(c.namespace, "namespace")
     assert.is_not(c.cache, nil)
 
     assert.Same(c:get_key(usage), "namespace::usage%5Ba%5D=3")
   end)
 
   it("Decrement when no usage was before in there", function()
-    local c = cache.new(100, "namespace")
+    local c = cache.new("namespace")
     local entry = c:decrement_usage_metric(nil):dump_data()
     assert.Same(entry.limit, "0")
     assert.Same(entry.remaining, "0")
@@ -27,7 +27,7 @@ describe("Cache key", function()
   end)
 
   it("Decrement works as expected", function()
-    local c = cache.new(100, "namespace")
+    local c = cache.new("namespace")
     c:reset_or_create_usage_metric(usage, 10, 10, 10)
 
     local entry = c:decrement_usage_metric(usage):dump_data()
@@ -43,7 +43,7 @@ describe("Cache key", function()
     usage:add("j", 5)
     usage:add("b", 2)
 
-    local c = cache.new(100, "namespace")
+    local c = cache.new("namespace")
     c:reset_or_create_usage_metric(usage, 10, 10, 10)
 
     local entry = c:decrement_usage_metric(usage):dump_data()