fix(keys): reclaim expired entries in batches

CHANGELOG.md

@@ -4,6 +4,13 @@ This file only calls out major changes. Please see [the list of Git commits](
 https://github.com/api7/nginx-lua-prometheus/commits/main) for the full list
 of changes.
 
+## Unreleased
+
+- Reclaim expired shared-dict entries in batches, so no single
+  `flush_expired()` call holds the dict lock for a whole backlog, and let
+  callers take the reclamation over with the `auto_flush_expired` option and
+  `prometheus:flush_expired()` (#23).
+
 ## 1.0.0
 
 Starting with this release, this fork (`nginx-lua-prometheus-api7`) uses

README.md

@@ -112,6 +112,13 @@ section of nginx configuration.
     RAM, you might want to increase this to avoid cache getting flushed too
     often. Decreasing this makes sense if you have a very large number of
     metrics or need to minimize memory usage of this library.
+  * `remove_expired_keys_interval` (number): how often, in seconds, each worker
+    drops the metrics that have expired and reclaims the shared dict entries
+    they left behind. Defaults to 3600, which is also the maximum.
+  * `auto_flush_expired` (boolean): whether that reclamation also happens on
+    the interval above, in every worker. Defaults to true. Set it to false to
+    schedule [`prometheus:flush_expired()`](#prometheusflush_expired) yourself,
+    for instance from a single process.
 
 Returns a `prometheus` object that should be used to register metrics.
 
@@ -236,6 +243,28 @@ location /metrics {
 
 Returns metric data as an array of strings.
 
+### prometheus:flush_expired()
+
+**syntax:** prometheus:flush_expired()
+
+Reclaims the shared dict entries left behind by metrics that have expired, and
+returns how many were reclaimed.
+
+Expired entries are only logically gone: every shared dict API reports them as
+missing, but they keep holding their memory until something reclaims them. The
+passive reclamation nginx performs on writes cannot do it, because it stops at
+the first entry that has not expired, and a metric registered without an
+`exptime` inevitably ends up sitting there.
+
+This is done for you on `remove_expired_keys_interval`, in every worker, unless
+`auto_flush_expired` is set to false. Call this instead if you would rather
+reclaim from a single process, or on your own schedule.
+
+The reclamation is issued in batches, because the underlying
+`ngx.shared.DICT:flush_expired()` holds the dict lock until it returns: a
+bounded number of entries is reclaimed per call, and the other workers get the
+lock back in between.
+
 ### counter:inc()
 
 **syntax:** counter:inc(*value*, *label_values*)

prometheus.lua

@@ -746,16 +746,19 @@ function Prometheus.init(dict_name, options_or_prefix)
         and options_or_prefix.remove_expired_keys_interval < MAX_REMOVE_EXPIRED_KEYS_INTERVAL
         and options_or_prefix.remove_expired_keys_interval
       or MAX_REMOVE_EXPIRED_KEYS_INTERVAL
+    self.auto_flush_expired = options_or_prefix.auto_flush_expired ~= false
   else
     self.prefix = options_or_prefix or ''
     self.error_metric_name = DEFAULT_ERROR_METRIC_NAME
     self.sync_interval = DEFAULT_SYNC_INTERVAL
     self.lookup_max_size = DEFAULT_LOOKUP_MAX_SIZE
     self.remove_expired_keys_interval = MAX_REMOVE_EXPIRED_KEYS_INTERVAL
+    self.auto_flush_expired = true
   end
 
   self.registry = {}
-  self.key_index = key_index_lib.new(self.dict, KEY_INDEX_PREFIX, self.remove_expired_keys_interval)
+  self.key_index = key_index_lib.new(self.dict, KEY_INDEX_PREFIX,
+    self.remove_expired_keys_interval, self.auto_flush_expired)
 
   self.initialized = true
 
@@ -930,6 +933,20 @@ end
 -- Returns:
 --   Array of strings with all metrics in a text format compatible with
 --   Prometheus.
+-- Reclaims the expired entries of the metrics shared dict, in batches, and
+-- returns how many were reclaimed.
+--
+-- Only needed by callers that set auto_flush_expired to false and schedule the
+-- reclamation themselves, in a single process instead of in every worker.
+function Prometheus:flush_expired()
+  if not self.initialized then
+    ngx.log(ngx.ERR, "Prometheus module has not been initialized")
+    return 0
+  end
+
+  return self.key_index:flush_expired()
+end
+
 function Prometheus:metric_data()
   if not self.initialized then
     ngx.log(ngx.ERR, "Prometheus module has not been initialized")

prometheus_keys.lua

@@ -14,16 +14,32 @@ KeyIndex.__index = KeyIndex
 -- and the index converges even when far more slots need repairing.
 local MAX_KEY_COUNT_REPAIRS = 1000
 
+-- Entries a single flush_expired() call may reclaim. That call holds the shared
+-- dict lock until it returns, so this is what bounds how long the other workers
+-- can be kept waiting on the lock: measured at ~1ms per 10000 entries reclaimed
+-- on OpenResty 1.29.2.4.
+local FLUSH_EXPIRED_BATCH = 10000
+
+-- Upper bound on the batches one call may run, so a large backlog is drained
+-- over several ticks instead of in a single long stretch. What is left over is
+-- picked up by the next tick.
+local FLUSH_EXPIRED_MAX_BATCHES = 30
+
+-- Pause between batches, so the lock is not taken back to back.
+local FLUSH_EXPIRED_BATCH_DELAY = 1
+
 
 -- check and remove expired keys
 local function remove_expired_keys(_, self)
   self:remove_expired_keys()
 end
 
 
-function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval)
+function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval,
+                      auto_flush_expired)
   local self = setmetatable({}, KeyIndex)
   self.dict = shared_dict
+  self.auto_flush_expired = auto_flush_expired ~= false
   self.key_prefix = prefix .. "key_"
   self.delete_count = prefix .. "delete_count"
   self.key_count = prefix .. "key_count"
@@ -52,19 +68,50 @@ function KeyIndex:remove_expired_keys()
     end
   end
 
-  -- The loop above only drops worker-local references. The expired shared-dict
-  -- entries themselves (both the __ngx_prom__key_N index slots and the metric
-  -- value keys, which live in the same dict) are only *logically* dead: every
-  -- dict API treats them as missing, but their slab pages stay allocated. The
-  -- passive per-write expiry scan cannot reclaim them either, because it stops
-  -- at the first non-expired entry at the LRU tail, and a permanent entry (the
-  -- error metric, or any metric registered without an exptime) inevitably ends
-  -- up sitting there. Without this call the dict grows without bound under
-  -- label churn: index slots are never reused, so free_space steps down on
-  -- every new series and never recovers (apache/apisix#13658). Since expired
-  -- entries are indistinguishable from absent ones through every dict API,
-  -- reclaiming them here cannot change any observable behaviour.
-  self.dict:flush_expired()
+  -- The loop above only drops worker-local references, so the expired entries
+  -- still have to be reclaimed from the dict itself. Callers that schedule
+  -- flush_expired() themselves -- in a single process rather than in every
+  -- worker -- turn this off with auto_flush_expired.
+  if self.auto_flush_expired then
+    self:flush_expired()
+  end
+end
+
+
+-- Reclaims the expired entries of the shared dict, in batches.
+--
+-- The expired entries (both the __ngx_prom__key_N index slots and the metric
+-- value keys, which live in the same dict) are only *logically* dead: every
+-- dict API treats them as missing, but their slab pages stay allocated. The
+-- passive per-write expiry scan cannot reclaim them either, because it stops
+-- at the first non-expired entry at the LRU tail, and a permanent entry (the
+-- error metric, or any metric registered without an exptime) inevitably ends
+-- up sitting there. Without this the dict grows without bound under label
+-- churn: index slots are never reused, so free_space steps down on every new
+-- series and never recovers (apache/apisix#13658). Since expired entries are
+-- indistinguishable from absent ones through every dict API, reclaiming them
+-- cannot change any observable behaviour.
+--
+-- flush_expired() holds the dict lock for its whole scan of the LRU queue, so
+-- it is called with a batch size: a bounded number of entries is reclaimed per
+-- call, and the workers get the lock back in between.
+--
+-- Returns the number of entries reclaimed.
+function KeyIndex:flush_expired()
+  local total = 0
+  for _ = 1, FLUSH_EXPIRED_MAX_BATCHES do
+    local freed = self.dict:flush_expired(FLUSH_EXPIRED_BATCH)
+    total = total + freed
+    -- freeing less than a full batch means this call has already walked the
+    -- whole queue, so there is nothing left to reclaim
+    if freed < FLUSH_EXPIRED_BATCH then
+      break
+    end
+
+    ngx.sleep(FLUSH_EXPIRED_BATCH_DELAY)
+  end
+
+  return total
 end
 
 -- Loads new keys that might have been added by other workers since last sync.

prometheus_test.lua

@@ -920,6 +920,63 @@ function TestKeyIndex:testRemoveExpiredKeysReclaimsSharedDict()
   luaunit.assertNotNil(self.dict.dict["permanent"])
 end
 
+-- flush_expired() holds the dict lock for its whole scan of the LRU queue, so
+-- the reclamation is issued in bounded batches. A backlog larger than one batch
+-- must still be reclaimed in full, by looping.
+function TestKeyIndex:testFlushExpiredRunsInBatches()
+  local expired = 20000
+  for i = 1, expired do
+    self.dict:set("value_" .. i, 1, 1)
+  end
+  -- A permanent entry, like the error metric or any metric without an exptime.
+  self.dict:set("permanent", 1)
+
+  sleep(2)
+
+  -- record what each call asks the dict to reclaim: an unbounded call would
+  -- hold the dict lock for the whole backlog at once
+  local batches = {}
+  local flush_expired = SimpleDict.flush_expired
+  self.dict.flush_expired = function(dict, n)
+    batches[#batches + 1] = n
+    return flush_expired(dict, n)
+  end
+
+  -- a batch reclaims 10000, so reaching all of them takes more than one call
+  luaunit.assertEquals(self.key_index:flush_expired(), expired)
+  luaunit.assertEquals(batches, {10000, 10000, 10000})
+
+  -- nothing left: the loop must stop after the call that comes back short
+  batches = {}
+  luaunit.assertEquals(self.key_index:flush_expired(), 0)
+  luaunit.assertEquals(batches, {10000})
+
+  self.dict.flush_expired = nil
+  luaunit.assertNotNil(self.dict.dict["permanent"])
+end
+
+-- Callers that schedule the reclamation themselves, in a single process rather
+-- than in every worker, turn the automatic flush off. remove_expired_keys()
+-- must then only drop the worker-local references.
+function TestKeyIndex:testAutoFlushExpiredDisabled()
+  local key_index = require('prometheus_keys').new(self.dict, "_noflush_", 1, false)
+  luaunit.assertEquals(key_index:add("expkey", "eviction_err", 1), nil)
+  -- A metric value key: lives in the same dict, and KeyIndex never reads it.
+  self.dict:set("expkey", 1, 1)
+
+  sleep(2)
+  key_index:remove_expired_keys()
+
+  luaunit.assertNil(key_index.index["expkey"])
+  -- both entries are still physically present, unlike with the default
+  luaunit.assertNotNil(self.dict.dict["_noflush_key_1"])
+  luaunit.assertNotNil(self.dict.dict["expkey"])
+
+  luaunit.assertEquals(key_index:flush_expired(), 2)
+  luaunit.assertNil(self.dict.dict["_noflush_key_1"])
+  luaunit.assertNil(self.dict.dict["expkey"])
+end
+
 -- The index slots of a churning metric must not accumulate: each round expires
 -- the previous slot, and the reclaim must keep the dict bounded rather than
 -- letting every new series step free space down for good.