fix(keys): reduce cleanup scans and coordinate expired-entry reclamation

CHANGELOG.md

@@ -9,6 +9,15 @@ of changes.
 - Reuse expired index slots while their dictionary nodes still exist, avoid
   expiry-driven broadcasts, renew known slots without synchronizing the full
   index, and detect a re-created `key_count` (#25).
+- Stagger worker cleanup timers, skip slots whose deadlines have not passed,
+  yield between groups of TTL checks, and log a cleanup timer that fails to
+  start (#23).
+- Reclaim expired entries once per `remove_expired_keys_interval` for the
+  whole dictionary instead of once per worker, and not at all while no metric
+  has an `exptime` (#23).
+- Add the `auto_flush_expired` option to leave reclamation to the caller, and
+  turn the cleanup off with a `remove_expired_keys_interval` of 0 or less
+  (#23).
 
 ## 1.0.0
 

README.md

@@ -112,6 +112,36 @@ 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): cleanup interval in seconds.
+    Defaults to 3600, which is also the maximum: larger values are capped to
+    it. Cleanup checks slots whose recorded deadlines have passed and yields
+    between groups of TTL checks. Choose an interval based on dictionary
+    capacity and label churn: longer intervals retain more expired entries
+    until reclamation. A value of 0 or less turns the cleanup off: no worker
+    drops its references to expired slots and nothing is reclaimed.
+    Worker cleanup timers are staggered across workers. The interval is also
+    the minimum time between two reclamations of expired entries for the
+    whole dictionary; each one is run by the first worker to finish its
+    cleanup after the previous one is an interval old, so the time between
+    them can exceed the interval. The reclamation holds the
+    dictionary lock while it scans all entries, so its latency grows with the
+    number of entries in the dictionary. A worker that has not seen a metric
+    with an `exptime` does not reclaim, so without expiring metrics the
+    dictionary is never scanned. The minimum time between reclamations is
+    kept in the shared dictionary, so it applies to every instance using the
+    dictionary: instances that share one should use the same cleanup
+    settings, as with different intervals the longest one can keep the
+    others from reclaiming at their own. It also survives a reload: after
+    the interval is shortened, the next reclamation waits for what is left
+    of the previous interval. A delayed reclamation keeps expired entries in
+    the dictionary longer, which can lead to evictions when it is full.
+  * `auto_flush_expired` (boolean): whether the cleanup reclaims expired
+    entries with `flush_expired()` on the shared dictionary. Defaults to
+    `true`. When `false`, expired entries stay in the dictionary until the
+    caller calls `ngx.shared[dict_name]:flush_expired()`, and workers drop
+    their references to expired slots only once the caller has reclaimed
+    them; with metrics that have an `exptime`, the dictionary otherwise fills
+    up.
 
 Returns a `prometheus` object that should be used to register metrics.
 

integration/cleanup.lua

@@ -0,0 +1,75 @@
+local p = cleanup_prometheus
+local d = ngx.shared.cleanup_metrics
+local marker = "__ngx_prom__flush_expired"
+
+local function set_expired()
+  for i = 1, 10 do
+    assert(d:set("expired_" .. i, i, 0.01))
+  end
+  ngx.sleep(0.02)
+end
+
+-- no metric has an exptime yet, so the dict lock is not taken to reclaim
+d:delete(marker)
+d:set("permanent", 42)
+set_expired()
+p.key_index:remove_expired_keys()
+assert(d:get_stale("expired_1") ~= nil, "reclaimed without an expiring metric")
+assert(d:get(marker) == nil, "took the interval without an expiring metric")
+
+-- start from an interval nobody has reclaimed in yet
+p:gauge("expiring", "Gauge with an exptime", nil, 3600):set(1)
+p.key_index:remove_expired_keys()
+for i = 1, 10 do
+  assert(d:get_stale("expired_" .. i) == nil, "did not reclaim the expired entries")
+end
+assert(d:get("permanent") == 42, "permanent entry was reclaimed")
+assert(d:get(marker), "did not take the interval")
+
+-- the interval is taken: another round leaves the reclamation alone
+set_expired()
+p.key_index:remove_expired_keys()
+assert(d:get_stale("expired_1") ~= nil, "reclaimed twice in one interval")
+
+-- once it has passed, the next round reclaims again
+d:delete(marker)
+p.key_index:remove_expired_keys()
+assert(d:get_stale("expired_1") == nil, "did not reclaim in the next interval")
+
+-- auto_flush_expired = false leaves the reclamation to the caller
+local nd = ngx.shared.cleanup_noflush_metrics
+cleanup_noflush:gauge("expiring", "Gauge with an exptime", nil, 3600):set(1)
+assert(nd:set("expired", 1, 0.01))
+ngx.sleep(0.02)
+cleanup_noflush.key_index:remove_expired_keys()
+assert(nd:get_stale("expired") ~= nil, "reclaimed with auto_flush_expired off")
+assert(nd:get(marker) == nil, "took the interval with auto_flush_expired off")
+
+-- the cleanup timer reclaims on its own, unless the interval turns it off
+local td = ngx.shared.cleanup_timer_metrics
+local od = ngx.shared.cleanup_off_metrics
+for _, p in ipairs({cleanup_timer, cleanup_off}) do
+  p:gauge("expiring", "Gauge with an exptime", nil, 3600):set(1)
+end
+assert(td:set("expired", 1, 0.01))
+assert(od:set("expired", 1, 0.01))
+ngx.sleep(2.5)
+assert(td:get_stale("expired") == nil, "the cleanup timer did not reclaim")
+assert(od:get_stale("expired") ~= nil, "reclaimed with the cleanup turned off")
+
+-- the dict truncates exptime to milliseconds; a marker TTL below 1ms would
+-- never expire, and stop reclamation after the first round. The rounds are
+-- run by hand, so the key index starts no timer.
+local sd = ngx.shared.cleanup_tiny_metrics
+local at, every = ngx.timer.at, ngx.timer.every
+ngx.timer.at = function() return true end
+ngx.timer.every = ngx.timer.at
+local tiny = require("prometheus_keys").new(sd, "_tiny_", 0.0005)
+ngx.timer.at, ngx.timer.every = at, every
+assert(tiny:add("expiring", "eviction", 3600) == nil)
+tiny:remove_expired_keys()
+assert(sd:set("expired", 1, 0.001))
+ngx.sleep(0.01)
+tiny:remove_expired_keys()
+assert(sd:get_stale("expired") == nil, "the flush marker did not expire")
+ngx.say("ok")

integration/nginx.conf

@@ -11,6 +11,11 @@ http {
 
     lua_shared_dict prometheus_metrics 10M;
     lua_shared_dict index_metrics 1M;
+    lua_shared_dict cleanup_metrics 16M;
+    lua_shared_dict cleanup_noflush_metrics 1M;
+    lua_shared_dict cleanup_timer_metrics 1M;
+    lua_shared_dict cleanup_off_metrics 1M;
+    lua_shared_dict cleanup_tiny_metrics 1M;
     lua_package_path "/nginx-lua-prometheus/?.lua;;";
 
     error_log stderr;
@@ -20,6 +25,13 @@ http {
 	  {sync_interval=0.4})
         index_writer = require("prometheus").init("index_metrics")
         index_reader = require("prometheus").init("index_metrics")
+        cleanup_prometheus = require("prometheus").init("cleanup_metrics")
+        cleanup_noflush = require("prometheus").init("cleanup_noflush_metrics",
+          {auto_flush_expired=false})
+        cleanup_timer = require("prometheus").init("cleanup_timer_metrics",
+          {remove_expired_keys_interval=1})
+        cleanup_off = require("prometheus").init("cleanup_off_metrics",
+          {remove_expired_keys_interval=0})
         metric_requests = prometheus:counter("requests_total",
           "Number of HTTP requests", {"host", "status"})
         metric_latency = prometheus:histogram("request_duration_seconds",
@@ -44,6 +56,10 @@ http {
             log_by_lua_block { return }
             content_by_lua_file /nginx-lua-prometheus/integration/index.lua;
         }
+        location = /test_cleanup {
+            log_by_lua_block { return }
+            content_by_lua_file /nginx-lua-prometheus/integration/cleanup.lua;
+        }
         location /error {
             return 500;
         }

integration/test.go

@@ -124,6 +124,17 @@ func main() {
 		log.Fatalf("Key index regression test returned HTTP %d", indexResp.StatusCode)
 	}
 
+	cleanupResp, err := indexClient.Get("http://localhost:18001/test_cleanup")
+	if err != nil {
+		log.Fatalf("Cleanup regression test failed: %v", err)
+	}
+	if err := cleanupResp.Body.Close(); err != nil {
+		log.Fatalf("Could not close the cleanup regression response: %v", err)
+	}
+	if cleanupResp.StatusCode != http.StatusOK {
+		log.Fatalf("Cleanup regression test returned HTTP %d", cleanupResp.StatusCode)
+	}
+
 	// Use a custom http client with a lower idle connection timeout and a request timeout.
 	client := &http.Client{
 		Timeout:   500 * time.Millisecond,

prometheus.lua

@@ -742,20 +742,23 @@ function Prometheus.init(dict_name, options_or_prefix)
       DEFAULT_SYNC_INTERVAL
     self.lookup_max_size = options_or_prefix.lookup_max_size or
       DEFAULT_LOOKUP_MAX_SIZE
-    self.remove_expired_keys_interval = options_or_prefix.remove_expired_keys_interval
-        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
+    local interval = options_or_prefix.remove_expired_keys_interval or
+      MAX_REMOVE_EXPIRED_KEYS_INTERVAL
+    self.remove_expired_keys_interval =
+      math.min(interval, 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