fix(rate-limiting): counters accuracy with redis policy & sync_rate (#11859)

* fix(rate-limiting): redis async updates

When the periodic sync to redis feature is turned on, using the
`sync_rate` configuration option, keys are incremented by steps of 2
instead of 1 for requests that arrive after the `sync_rate` interval
has expired.

This happens because after each sync, the key is loaded again from redis
and also incremented atomically (see: #10559)
however the next call to `increment` also adds 1 to its value, so
the key is incremented by 2 every time it's loaded from redis.

This fix sets a negative delta for the key when
`conf.sync_rate ~= SYNC_RATE_REALTIME` and the key was loaded from redis
in order to invalidate the next call to `increment`.

Includes a small code refactor

Author: samugi

changelog/unreleased/kong/rate-limiting-fix-redis-sync-rate.yml

@@ -0,0 +1,3 @@
+message: "**Rate Limiting**: fix to provide better accuracy in counters when sync_rate is used with the redis policy."
+type: bugfix
+scope: Plugin

kong/plugins/rate-limiting/policies/init.lua

@@ -206,14 +206,9 @@ local function update_local_counters(conf, periods, limits, identifier, value)
     if limits[period] then
       local cache_key = get_local_key(conf, identifier, period, period_date)
 
-      if cur_delta[cache_key] then
-        cur_delta[cache_key] = cur_delta[cache_key] + value
-      else
-        cur_delta[cache_key] = value
-      end
+      cur_delta[cache_key] = (cur_delta[cache_key] or 0) + value
     end
   end
-  
 end
 
 return {
@@ -346,7 +341,9 @@ return {
       if conf.sync_rate ~= SYNC_RATE_REALTIME then
         cur_usage[cache_key] = current_metric or 0
         cur_usage_expire_at[cache_key] = periods[period] + EXPIRATION[period]
-        cur_delta[cache_key] = 0
+        -- The key was just read from Redis using `incr`, which incremented it
+        -- by 1. Adjust the value to account for the prior increment.
+        cur_delta[cache_key] = -1
       end
 
       return current_metric or 0

spec/03-plugins/23-rate-limiting/02-policies_spec.lua

@@ -176,53 +176,63 @@ describe("Plugin: rate-limiting (policies)", function()
     end)
   end
 
-  for _, sync_rate in ipairs{1, SYNC_RATE_REALTIME} do
-    describe("redis with sync rate: " .. sync_rate, function()
-      local identifier = uuid()
-      local conf       = {
-        route_id = uuid(),
-        service_id = uuid(),
-        redis_host = helpers.redis_host,
-        redis_port = helpers.redis_port,
-        redis_database = 0,
-        sync_rate = sync_rate,
-      }
-
-      before_each(function()
-        local red = require "resty.redis"
-        local redis = assert(red:new())
-        redis:set_timeout(1000)
-        assert(redis:connect(conf.redis_host, conf.redis_port))
-        redis:flushall()
-        redis:close()
-      end)
-
-      it("increase & usage", function()
-        --[[
-          Just a simple test:
-          - increase 1
-          - check usage == 1
-          - increase 1
-          - check usage == 2
-          - increase 1 (beyond the limit)
-          - check usage == 3
-        --]]
-
-        local current_timestamp = 1424217600
-        local periods = timestamp.get_timestamps(current_timestamp)
+  for _, sync_rate in ipairs{0.5, SYNC_RATE_REALTIME} do
+    local current_timestamp = 1424217600
+    local periods = timestamp.get_timestamps(current_timestamp)
+
+    for period in pairs(periods) do
+      describe("redis with sync rate: " .. sync_rate .. " period: " .. period, function()
+        local identifier = uuid()
+        local conf       = {
+          route_id = uuid(),
+          service_id = uuid(),
+          redis_host = helpers.redis_host,
+          redis_port = helpers.redis_port,
+          redis_database = 0,
+          sync_rate = sync_rate,
+        }
 
-        for period in pairs(periods) do
+        before_each(function()
+          local red = require "resty.redis"
+          local redis = assert(red:new())
+          redis:set_timeout(1000)
+          assert(redis:connect(conf.redis_host, conf.redis_port))
+          redis:flushall()
+          redis:close()
+        end)
+
+        it("increase & usage", function()
+          --[[
+            Just a simple test:
+            - increase 1
+            - check usage == 1
+            - increase 1
+            - check usage == 2
+            - increase 1 (beyond the limit)
+            - check usage == 3
+          --]]
 
           local metric = assert(policies.redis.usage(conf, identifier, period, current_timestamp))
           assert.equal(0, metric)
 
           for i = 1, 3 do
-            assert(policies.redis.increment(conf, { [period] = 2 }, identifier, current_timestamp, 1))
-            metric = assert(policies.redis.usage(conf, identifier, period, current_timestamp))
-            assert.equal(i, metric)
+            -- "second" keys expire too soon to check the async increment.
+            -- Let's verify all the other scenarios:
+            if not (period == "second" and sync_rate ~= SYNC_RATE_REALTIME) then
+              assert(policies.redis.increment(conf, { [period] = 2 }, identifier, current_timestamp, 1))
+
+              -- give time to the async increment to happen
+              if sync_rate ~= SYNC_RATE_REALTIME then
+                local sleep_time = 1 + (sync_rate > 0 and sync_rate or 0)
+                ngx.sleep(sleep_time)
+              end
+
+              metric = assert(policies.redis.usage(conf, identifier, period, current_timestamp))
+              assert.equal(i, metric)
+            end
           end
-        end
+        end)
       end)
-    end)
+    end
   end
 end)