diff --git a/CHANGELOG.md b/CHANGELOG.md index ddd13ba..5037359 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index ff28be6..2b22e62 100644 --- a/README.md +++ b/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. diff --git a/integration/cleanup.lua b/integration/cleanup.lua new file mode 100644 index 0000000..085fe0a --- /dev/null +++ b/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") diff --git a/integration/nginx.conf b/integration/nginx.conf index f8bc3ae..f0d1f4e 100644 --- a/integration/nginx.conf +++ b/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; } diff --git a/integration/test.go b/integration/test.go index 07c8c76..1bce01d 100644 --- a/integration/test.go +++ b/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, diff --git a/prometheus.lua b/prometheus.lua index 8556b8d..51a130f 100644 --- a/prometheus.lua +++ b/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 diff --git a/prometheus_keys.lua b/prometheus_keys.lua index 90bffe0..43c5536 100644 --- a/prometheus_keys.lua +++ b/prometheus_keys.lua @@ -14,20 +14,66 @@ KeyIndex.__index = KeyIndex -- and the index converges even when far more slots need repairing. local MAX_KEY_COUNT_REPAIRS = 1000 +local DEFAULT_REMOVE_EXPIRED_KEYS_INTERVAL = 600 + +-- Yield between groups of shared-dictionary TTL checks. +local REMOVE_EXPIRED_KEYS_YIELD_EVERY = 500 + +KeyIndex.REMOVE_EXPIRED_KEYS_YIELD_EVERY = REMOVE_EXPIRED_KEYS_YIELD_EVERY + -- check and remove expired keys -local function remove_expired_keys(_, self) - self:remove_expired_keys() +local function remove_expired_keys(premature, self) + if not premature then + self:remove_expired_keys() + end end -function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval) +local function run_remove_expired_keys_every(self, interval) + local ok, err = ngx.timer.every(interval, remove_expired_keys, self) + if not ok then + ngx.log(ngx.ERR, "failed to start the remove expired keys timer: ", err) + end +end + + +-- Stagger worker cleanup timers to avoid starting every scan together. +local function start_remove_expired_keys_timer(self, interval) + local id = ngx.worker.id() or 0 + local count = ngx.worker.count() + + local offset = interval * id / count + if offset <= 0 then + run_remove_expired_keys_every(self, interval) + return + end + + local ok, err = ngx.timer.at(offset, function(premature, s) + if premature then + return + end + run_remove_expired_keys_every(s, interval) + s:remove_expired_keys() + end, self) + if not ok then + ngx.log(ngx.ERR, "failed to start the remove expired keys timer: ", err) + end +end + + +-- remove_expired_keys_interval: seconds between cleanup rounds; 0 or less +-- starts no cleanup timer. auto_flush_expired: false leaves +-- dict:flush_expired() to the caller. +function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval, + auto_flush_expired) local self = setmetatable({}, KeyIndex) self.dict = shared_dict self.key_prefix = prefix .. "key_" self.delete_count = prefix .. "delete_count" self.key_count = prefix .. "key_count" self.generation = prefix .. "generation" + self.flush_marker = prefix .. "flush_expired" self.last = 0 self.deleted = 0 self.keys = {} @@ -35,7 +81,16 @@ function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval) self.aliases = {} self.expire_keys = {} - ngx.timer.every(remove_expired_keys_interval or 600, remove_expired_keys, self) + local interval = remove_expired_keys_interval or + DEFAULT_REMOVE_EXPIRED_KEYS_INTERVAL + self.remove_expired_keys_interval = interval + -- The dict truncates exptime to milliseconds, and 0 would never expire. + self.flush_marker_ttl = math.max(interval, 0.001) + -- the flush marker expires after the interval, so it needs a positive one + self.auto_flush_expired = auto_flush_expired ~= false and interval > 0 + if interval > 0 then + start_remove_expired_keys_timer(self, interval) + end return self end @@ -46,14 +101,61 @@ local function slot_gone(self, i, err) return err == "not found" and self.dict:get_stale(self.key_prefix .. i) == nil end +-- Drops the worker-local references to the slots that are gone. +local function remove_expired_slots(self) + -- Snapshot due slots before yielding: add() may mutate expire_keys while + -- this coroutine is suspended. Future deadlines need no dictionary lookup. + local now = ngx.now() + local due = {} + for i, deadline in pairs(self.expire_keys) do + if deadline <= now then + due[#due + 1] = i + end + end + + for j, i in ipairs(due) do + if j > 1 and (j - 1) % REMOVE_EXPIRED_KEYS_YIELD_EVERY == 0 then + ngx.sleep(0) + -- do not hold up a reload: what is left is gone with the worker anyway + if ngx.worker.exiting() then + return + end + now = ngx.now() + end + + -- re-read: while this round yielded the slot may have been renewed, or + -- dropped by add() + local deadline = self.expire_keys[i] + if deadline and deadline <= now then + -- Keep expired nodes until reclamation: expire() can renew them in place. + local ttl, err = self.dict:ttl(self.key_prefix .. i) + if ttl and ttl > 0 then + self.expire_keys[i] = now + ttl + elseif slot_gone(self, i, err) then + self:forget_slot(i) + end + end + end +end + -- check and remove expired keys function KeyIndex:remove_expired_keys() - for i, _ in pairs(self.expire_keys) do - -- Keep expired nodes until reclamation: expire() can renew them in place. - local _, err = self.dict:ttl(self.key_prefix .. i) - if slot_gone(self, i, err) then - self:forget_slot(i) - end + -- A round yields, so the timer can fire again before the previous run + -- finished (ngx.timer.every does not wait for it). Two rounds would only ask + -- the dict about the same slots twice, so the second one steps aside. + if self.removing then + return + end + + self.removing = true + -- pcall so that a throwing ngx.sleep -- a caller reaching this from a phase + -- that does not allow yielding -- cannot leave the guard set and stop the + -- local scan for the life of the worker. + local ok, err = pcall(remove_expired_slots, self) + self.removing = false + + if not ok then + ngx.log(ngx.ERR, "failed to remove expired keys: ", tostring(err)) end -- The loop above only drops worker-local references. The expired shared-dict @@ -66,7 +168,28 @@ function KeyIndex:remove_expired_keys() -- 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). - self.dict:flush_expired() + -- + -- flush_expired() walks the whole dict under its lock, and one call reclaims + -- for every worker, so two reclamations are at least an interval apart: the + -- first process to finish its local cleanup after the marker has expired + -- runs it. The marker is never deleted. A round that finds it still set + -- skips the reclamation, which delays it until a later round. + -- + -- A worker that has never seen an entry with an exptime has nothing to + -- reclaim, and must not take the interval from one that has. The flag is + -- never cleared: forgetting the last expiring slot does not mean its metric + -- values are gone, as counters renew them up to sync_interval later. + if not self.auto_flush_expired or not self.has_expiring_entries + or ngx.worker.exiting() then + return + end + -- safe_add: add() would force-evict a live entry to store the marker. A dict + -- too full to hold it is the one that most needs reclaiming, so flush anyway. + local added, add_err = self.dict:safe_add(self.flush_marker, true, + self.flush_marker_ttl) + if added or add_err == "no memory" then + self.dict:flush_expired() + end end -- Tokens for the generation key (see advance_key_count()), unique per @@ -128,13 +251,15 @@ function KeyIndex:sync_range(first, last) local ttl, err = self.dict:ttl(self.key_prefix .. i) if ttl then if ttl ~= 0 then - self.expire_keys[i] = true + self.expire_keys[i] = ngx.now() + ttl + self.has_expiring_entries = true end elseif slot_gone(self, i, err) then self:forget_slot(i) else -- Kept at the ttl() sentinel; check it again on the next round. - self.expire_keys[i] = true + self.expire_keys[i] = ngx.now() + self.has_expiring_entries = true end end elseif self.keys[i] then @@ -242,6 +367,13 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime) keys = { key_or_keys } end + -- when the slot's expiry is renewed or set below, remember when it lands, so + -- that remove_expired_keys() does not have to ask the dict about it + local now = exptime and exptime > 0 and ngx.now() + if now then + self.has_expiring_entries = true + end + for _, key in pairs(keys) do local retried = false local repairs = 0 @@ -255,6 +387,9 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime) if mine and exptime and self.dict:get(self.key_prefix .. mine) == key then local ok, err = self.dict:expire(self.key_prefix .. mine, exptime) if ok then + if now then + self.expire_keys[mine] = now + exptime + end goto renewed end if err ~= "not found" then @@ -271,6 +406,9 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime) local expired = false if exptime then local ok, err = self.dict:expire(self.key_prefix .. self.index[key], exptime) + if ok and now then + self.expire_keys[self.index[key]] = now + exptime + end if not ok then if err == "not found" then -- The node is gone, so this number is no longer ours to renew. @@ -309,8 +447,8 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime) if ok then local _, _, forcible2 = advance_key_count(self) self:remember_slot(N, key) - if exptime and exptime > 0 then - self.expire_keys[N] = true + if now then + self.expire_keys[N] = now + exptime end if forcible or forcible2 or repair_forcible then return (err_msg_lru_eviction .. "; key index: add key: idx=" .. diff --git a/prometheus_test.lua b/prometheus_test.lua index 22d2f91..5b63bd7 100644 --- a/prometheus_test.lua +++ b/prometheus_test.lua @@ -32,6 +32,10 @@ function SimpleDict:add(k, v, exptime) self:set(k, v, exptime) return true, nil, forcible -- ok, err, forcible end +function SimpleDict:safe_add(k, v, exptime) + local ok, err = self:add(k, v, exptime) + return ok, err +end function SimpleDict:incr(k, v, init) local forcible = false if k == "willnotfitk" or v == "willnotfitv" or @@ -137,7 +141,13 @@ function Nginx.print(printed) end Nginx.worker = {} function Nginx.worker.id() - return 'testworker' + return 0 +end +function Nginx.worker.count() + return 1 +end +function Nginx.worker.exiting() + return false end function Nginx.sleep() end function Nginx.now() @@ -145,7 +155,8 @@ function Nginx.now() return os.time() end Nginx.timer = {} -function Nginx.timer.every(_, _, _) end +function Nginx.timer.every(_, _, _) return true end +function Nginx.timer.at(_, _, _) return true end function Nginx.get_phase() return 'init_worker' end @@ -231,6 +242,7 @@ function TestPrometheus:testInitOptions() luaunit.assertEquals(ngx.logs, nil) end + function TestPrometheus:testInitWorker() self.dict = setmetatable({}, SimpleDict) ngx.shared.metrics = self.dict @@ -793,6 +805,9 @@ function TestKeyIndex:setUp() end function TestKeyIndex.tearDown() ngx.logs = nil + -- the cleanup tests stub ngx.sleep; drop the stub here too, so a failing + -- assertion before their own restore cannot leak it into the next test + ngx.sleep = nil end function TestKeyIndex.testInit() luaunit.assertEquals(ngx.logs, nil) @@ -933,7 +948,7 @@ function TestKeyIndex:testReclaimedCanonicalFallsBackToAlias() self.key_index:sync() luaunit.assertEquals(self.key_index:list(), {"target"}) self.dict:delete("_prefix_key_2") - self.key_index.expire_keys[2] = true + self.key_index.expire_keys[2] = 0 self.key_index:remove_expired_keys() luaunit.assertEquals(self.key_index:list(), {"target"}) luaunit.assertEquals(self.key_index.index.target, 1) @@ -1168,6 +1183,482 @@ function TestKeyIndex:testSyncTracksSlotsImportedAtTheTtlSentinel() luaunit.assertNil(reader.expire_keys[i]) end +function TestKeyIndex:testCleanupTimersAreStaggeredByWorker() + local id, count = ngx.worker.id, ngx.worker.count + local at, every = ngx.timer.at, ngx.timer.every + local scheduled + ngx.worker.count = function() return 4 end + ngx.timer.at = function(delay, callback, context) + scheduled = {kind="at", delay=delay, callback=callback, context=context} + return true + end + ngx.timer.every = function(delay, callback, context) + scheduled = {kind="every", delay=delay, callback=callback, context=context} + return true + end + + local cases = {{id=0}, {id=2}, {}} + local results = {} + for i, case in ipairs(cases) do + ngx.worker.id = function() return case.id end + local key_index = require('prometheus_keys').new(self.dict, "_timer_", 60) + local calls = 0 + key_index.remove_expired_keys = function() calls = calls + 1 end + local first = scheduled + if first.kind == "at" then + first.callback(true, first.context) + local after_shutdown = calls + first.callback(false, first.context) + results[i] = {first.kind, first.delay, scheduled.kind, scheduled.delay, + after_shutdown, calls} + else + results[i] = {first.kind, first.delay} + end + end + ngx.worker.id, ngx.worker.count = id, count + ngx.timer.at, ngx.timer.every = at, every + + luaunit.assertEquals(results, {{"every", 60}, + {"at", 30, "every", 60, 0, 1}, {"every", 60}}) +end + +function TestKeyIndex:testCleanupTimerSkipsPrematureCallbacks() + local every = ngx.timer.every + local callback, context + ngx.timer.every = function(_, fn, key_index) + callback, context = fn, key_index + return true + end + local key_index = require('prometheus_keys').new(self.dict, "_timer_", 60) + ngx.timer.every = every + + local calls = 0 + key_index.remove_expired_keys = function() calls = calls + 1 end + callback(true, context) + luaunit.assertEquals(calls, 0) + callback(false, context) + luaunit.assertEquals(calls, 1) +end + +function TestKeyIndex:testRemoveExpiredKeysKeepsPermanentSlots() + luaunit.assertNil(self.key_index:add("permanent", "eviction_err", 600)) + local i = self.key_index.index["permanent"] + local slot = self.key_index.key_prefix .. i + self.dict:set(slot, "permanent") + self.key_index.expire_keys[i] = 0 + + self.key_index:remove_expired_keys() + self.key_index:remove_expired_keys() + + luaunit.assertEquals(self.key_index.index["permanent"], i) + luaunit.assertEquals(self.dict:get(slot), "permanent") +end + +-- ttl() also reads 0 at a slot's exact expiry millisecond, so a zero keeps the +-- slot tracked until its node is reclaimed. +function TestKeyIndex:testRemoveExpiredKeysRechecksZeroTTLSlots() + luaunit.assertNil(self.key_index:add("boundary", "eviction_err", 600)) + local i = self.key_index.index["boundary"] + local slot = self.key_index.key_prefix .. i + self.key_index.expire_keys[i] = 0 + + self.dict.ttl = function(_, key) + luaunit.assertEquals(key, slot) + return 0 + end + self.key_index:remove_expired_keys() + self.dict.ttl = nil + luaunit.assertNotNil(self.key_index.expire_keys[i]) + + self.dict:delete(slot) + self.key_index:remove_expired_keys() + luaunit.assertNil(self.key_index.keys[i]) + luaunit.assertNil(self.key_index.index["boundary"]) +end + +-- Metrics registered without an exptime are permanent; the workers that import +-- their slots must not check them on every round. +function TestKeyIndex:testSyncDoesNotTrackPermanentSlots() + luaunit.assertNil(self.key_index:add("permanent", "eviction_err")) + local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1) + reader:sync() + local i = reader.index["permanent"] + luaunit.assertNotNil(i) + luaunit.assertNil(reader.expire_keys[i]) +end + +-- Future deadlines require no shared-dictionary TTL checks. +function TestKeyIndex:testRemoveExpiredKeysSkipsSlotsThatCannotHaveLapsed() + local keys = {} + for i = 1, 100 do + keys[i] = "live_" .. i + end + luaunit.assertEquals(self.key_index:add(keys, "eviction_err", 600), nil) + + local checked = 0 + local ttl = SimpleDict.ttl + self.dict.ttl = function(dict, k) + checked = checked + 1 + return ttl(dict, k) + end + + self.key_index:remove_expired_keys() + luaunit.assertEquals(checked, 0) + luaunit.assertEquals(self.key_index.index["live_1"], 1) + + self.dict.ttl = nil + luaunit.assertEquals(ngx.logs, nil) +end + +-- A renewal on the fast path moves the deadline, and tracks a slot that was +-- not tracked before. +function TestKeyIndex:testRenewalRecordsTheNewDeadline() + luaunit.assertNil(self.key_index:add("renewed", "eviction_err", 1)) + local i = self.key_index.index["renewed"] + self.key_index.expire_keys[i] = nil + + local sync = self.key_index.sync + self.key_index.sync = function() error("renewal must not sync") end + luaunit.assertNil(self.key_index:add("renewed", "eviction_err", 600)) + self.key_index.sync = sync + + luaunit.assertTrue(self.key_index.expire_keys[i] >= ngx.now() + 600) +end + +-- The slots whose deadline has passed do have to be asked about, and there can +-- be as many of them as the index is large. One round asks about all of them, +-- so the local index keeps up with the expiry rate, but yields every +-- REMOVE_EXPIRED_KEYS_YIELD_EVERY of them so the worker keeps serving. +local function lapse_all_slots(self, prefix, count) + local keys = {} + for i = 1, count do + keys[i] = prefix .. i + end + luaunit.assertEquals(self.key_index:add(keys, "eviction_err", 600), nil) + + -- what a round running after the exptime sees, without waiting for it: every + -- recorded deadline has passed, so every slot has to be asked about + for i in pairs(self.key_index.expire_keys) do + self.key_index.expire_keys[i] = 0 + end + + local counter = {checked = 0} + local ttl = SimpleDict.ttl + self.dict.ttl = function(dict, k) + counter.checked = counter.checked + 1 + return ttl(dict, k) + end + return counter +end + +function TestKeyIndex:testRemoveExpiredKeysYieldsWhileScanning() + local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY + local counter = lapse_all_slots(self, "yield_", 2 * every + 10) + + local yields, checked_at_yield = 0, {} + ngx.sleep = function() + yields = yields + 1 + checked_at_yield[yields] = counter.checked + end + self.key_index:remove_expired_keys() + ngx.sleep = nil + + -- every lapsed slot is asked about in this one round, in steps of `every` + luaunit.assertEquals(counter.checked, 2 * every + 10) + luaunit.assertEquals(checked_at_yield, {every, 2 * every}) + + -- the slots are all still live, so they are kept, and their deadlines are + -- refreshed rather than re-checked on every round from now on + luaunit.assertEquals(self.key_index.index["yield_1"], 1) + luaunit.assertEquals(self.key_index.index["yield_" .. (2 * every + 10)], + 2 * every + 10) + counter.checked = 0 + self.key_index:remove_expired_keys() + luaunit.assertEquals(counter.checked, 0) + + self.dict.ttl = nil + luaunit.assertEquals(ngx.logs, nil) +end + +-- A round yields, so the timer can fire again before it finishes. The second +-- round must step aside instead of asking the dict about the same slots again. +function TestKeyIndex:testRemoveExpiredKeysDoesNotOverlap() + local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY + local counter = lapse_all_slots(self, "overlap_", every + 10) + + local checked_by_reentry + ngx.sleep = function() + if checked_by_reentry == nil then + local before = counter.checked + self.key_index:remove_expired_keys() + checked_by_reentry = counter.checked - before + end + end + self.key_index:remove_expired_keys() + ngx.sleep = nil + + luaunit.assertEquals(checked_by_reentry, 0) + luaunit.assertEquals(counter.checked, every + 10) + + self.dict.ttl = nil + luaunit.assertEquals(ngx.logs, nil) +end + +-- A worker that is shutting down stops scanning at the next yield, so it does +-- not hold up a reload. +function TestKeyIndex:testRemoveExpiredKeysStopsWhenExiting() + local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY + local counter = lapse_all_slots(self, "exiting_", every + 10) + + ngx.worker.exiting = function() return true end + self.key_index:remove_expired_keys() + ngx.worker.exiting = function() return false end + + luaunit.assertEquals(counter.checked, every) + + self.dict.ttl = nil + luaunit.assertEquals(ngx.logs, nil) +end + +-- A throwing ngx.sleep -- a caller reaching remove_expired_keys() from a phase +-- that does not allow yielding -- must not leave the guard set, or the worker +-- would stop scanning for good. +function TestKeyIndex:testRemoveExpiredKeysReleasesTheGuardAfterAnError() + local every = self.key_index.REMOVE_EXPIRED_KEYS_YIELD_EVERY + local counter = lapse_all_slots(self, "error_", every + 10) + + ngx.sleep = function() error("API disabled in the current context") end + self.key_index:remove_expired_keys() + ngx.sleep = nil + + luaunit.assertEquals(#ngx.logs, 1) + luaunit.assertStrContains(ngx.logs[1], "failed to remove expired keys") + ngx.logs = nil + luaunit.assertEquals(counter.checked, every) + + -- the guard is released, so the next round asks about the rest + counter.checked = 0 + self.key_index:remove_expired_keys() + luaunit.assertEquals(counter.checked, 10) + + self.dict.ttl = nil +end + +-- A timer that cannot be created leaves the worker without cleanup, so say so. +function TestKeyIndex:testCleanupTimerFailureIsLogged() + local id, count = ngx.worker.id, ngx.worker.count + local at, every = ngx.timer.at, ngx.timer.every + local function fail() return nil, "too many pending timers" end + local first + ngx.worker.count = function() return 2 end + ngx.timer.every = fail + + -- worker 0 starts the periodic timer directly + require('prometheus_keys').new(self.dict, "_timer_", 60) + luaunit.assertEquals(#ngx.logs, 1) + + -- worker 1 cannot start its staggered first round + ngx.worker.id = function() return 1 end + ngx.timer.at = fail + require('prometheus_keys').new(self.dict, "_timer_", 60) + luaunit.assertEquals(#ngx.logs, 2) + + -- or starts it, but not the periodic timer from there + ngx.timer.at = function(_, callback, context) + first = {callback=callback, context=context} + return true + end + local key_index = require('prometheus_keys').new(self.dict, "_timer_", 60) + key_index.remove_expired_keys = function() end + first.callback(false, first.context) + ngx.worker.id, ngx.worker.count = id, count + ngx.timer.at, ngx.timer.every = at, every + + luaunit.assertEquals(#ngx.logs, 3) + for _, log in ipairs(ngx.logs) do + luaunit.assertStrContains(log, "failed to start the remove expired keys timer") + luaunit.assertStrContains(log, "too many pending timers") + end +end + +-- A non-positive interval turns the cleanup off: no timer, so neither the +-- local cleanup nor the reclamation runs. +function TestKeyIndex:testNonPositiveIntervalDisablesCleanup() + local at, every = ngx.timer.at, ngx.timer.every + -- init() also starts the counter sync timer; count only cleanup timers. + local intervals = {} + local function schedule(delay, _, context) + if type(context) == "table" and context.expire_keys then + intervals[#intervals + 1] = delay + end + return true + end + ngx.timer.at, ngx.timer.every = schedule, schedule + local zero = require('prometheus_keys').new(self.dict, "_zero_", 0) + require('prometheus_keys').new(self.dict, "_negative_", -1) + require('prometheus_keys').new(self.dict, "_default_") + local p = require('prometheus').init("metrics", {remove_expired_keys_interval=0}) + local capped = require('prometheus').init("metrics", + {remove_expired_keys_interval=7200}) + ngx.timer.at, ngx.timer.every = at, every + luaunit.assertEquals(intervals, {600, 3600}) + luaunit.assertEquals(p.remove_expired_keys_interval, 0) + luaunit.assertEquals(capped.remove_expired_keys_interval, 3600) + luaunit.assertFalse(zero.auto_flush_expired) +end + +local function count_flushes(dict) + local flushes = {n = 0} + dict.flush_expired = function(d) + flushes.n = flushes.n + 1 + return SimpleDict.flush_expired(d) + end + return flushes +end + +-- flush_expired() reclaims for every worker, so one call per interval is +-- enough: the first worker to finish its local cleanup runs it, the others +-- skip it until the interval has passed. +function TestKeyIndex:testOneFlushPerInterval() + luaunit.assertEquals(self.key_index:add("expiring", "eviction_err", 3600), nil) + local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1) + worker2:sync() + local flushes = count_flushes(self.dict) + + self.key_index:remove_expired_keys() + worker2:remove_expired_keys() + self.key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 1) + + sleep(2) + worker2:remove_expired_keys() + self.key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 2) + self.dict.flush_expired = nil +end + +-- The marker keeps two reclamations an interval apart, whatever the number of +-- workers, and is never stored without an expiry: the dict truncates exptime +-- to milliseconds. +function TestKeyIndex:testFlushMarkerLastsTheInterval() + local count = ngx.worker.count + local ttls = {} + for _, case in ipairs({{1, 60}, {4, 60}, {1, 0.0005}}) do + local workers, interval = case[1], case[2] + ngx.worker.count = function() return workers end + local key_index = require('prometheus_keys').new(self.dict, "_ttl_", + interval) + luaunit.assertEquals(key_index:add("expiring", "eviction_err", 3600), nil) + self.dict.safe_add = function(_, _, _, exptime) + ttls[#ttls + 1] = exptime + return true + end + key_index:remove_expired_keys() + self.dict.safe_add = nil + end + ngx.worker.count = count + luaunit.assertEquals(ttls, {60, 60, 0.001}) +end + +-- Storing the marker must not evict a live entry, and a dict too full to hold +-- it is the one that most needs reclaiming. +function TestKeyIndex:testFlushWhenTheMarkerDoesNotFit() + luaunit.assertEquals(self.key_index:add("expiring", "eviction_err", 3600), nil) + local flushes = count_flushes(self.dict) + self.dict.safe_add = function() return nil, "no memory" end + + self.key_index:remove_expired_keys() + self.key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 2) + self.dict.safe_add = nil + self.dict.flush_expired = nil +end + +-- A worker on its way out must not take the interval: it could exit before +-- reclaiming, and no other worker would reclaim until the interval passed. +function TestKeyIndex:testExitingWorkerLeavesTheFlushToOthers() + luaunit.assertEquals(self.key_index:add("expiring", "eviction_err", 3600), nil) + local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1) + worker2:sync() + local flushes = count_flushes(self.dict) + local exiting = ngx.worker.exiting + ngx.worker.exiting = function() return true end + self.key_index:remove_expired_keys() + ngx.worker.exiting = exiting + luaunit.assertEquals(flushes.n, 0) + luaunit.assertNil(self.dict:get("_prefix_flush_expired")) + + worker2:remove_expired_keys() + luaunit.assertEquals(flushes.n, 1) + self.dict.flush_expired = nil +end + +-- auto_flush_expired = false leaves the reclamation to the caller, but each +-- worker still drops its references to the slots that are gone. +function TestKeyIndex:testFlushExpiredDisabledKeepsLocalCleanup() + local at, every = ngx.timer.at, ngx.timer.every + local intervals = {} + local function schedule(delay, _, context) + if type(context) == "table" and context.expire_keys then + intervals[#intervals + 1] = delay + end + return true + end + ngx.timer.at, ngx.timer.every = schedule, schedule + local p = require('prometheus').init("metrics", {auto_flush_expired=false}) + local key_index = require('prometheus_keys').new(self.dict, "_noflush_", 1, + false) + ngx.timer.at, ngx.timer.every = at, every + luaunit.assertEquals(intervals, {3600, 1}) + luaunit.assertFalse(p.key_index.auto_flush_expired) + + local flushes = count_flushes(self.dict) + luaunit.assertEquals(key_index:add("expiring", "eviction_err", 1), nil) + local i = key_index.index["expiring"] + sleep(2) + key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 0) + luaunit.assertNil(self.dict:get("_noflush_flush_expired")) + luaunit.assertEquals(key_index.keys[i], "expiring") + + -- the caller reclaims; the next round forgets the slot + SimpleDict.flush_expired(self.dict) + key_index:remove_expired_keys() + luaunit.assertNil(key_index.keys[i]) + luaunit.assertNil(key_index.index["expiring"]) + self.dict.flush_expired = nil +end + +-- Without an exptime nothing in the dict expires, so the walk under the dict +-- lock is skipped, and the interval is left to a worker that has something to +-- reclaim. Once a worker has seen an expiring entry it keeps reclaiming, even +-- after the last expiring slot is forgotten: a counter renews its values up to +-- sync_interval after its slot, so they can outlive it. +function TestKeyIndex:testFlushOnlyOnceEntriesCanExpire() + local flushes = count_flushes(self.dict) + luaunit.assertEquals(self.key_index:add("permanent", "eviction_err"), nil) + self.key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 0) + luaunit.assertNil(self.dict:get("_prefix_flush_expired")) + + -- another worker registers a series that expires; this one learns of it by + -- syncing the index + local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1) + luaunit.assertEquals(worker2:add("expiring", "eviction_err", 1), nil) + self.key_index:sync() + self.key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 1) + + sleep(2) + self.key_index:remove_expired_keys() + self.key_index:remove_expired_keys() + luaunit.assertEquals(next(self.key_index.expire_keys), nil) + sleep(2) + self.key_index:remove_expired_keys() + luaunit.assertEquals(flushes.n, 3) + self.dict.flush_expired = nil +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. @@ -1271,7 +1762,7 @@ end function TestKeyIndex:testReaderFindsSlotReissuedAfterKeyCountEviction() local reader = evict_key_count_and_slot(self, 5) -- the reader's own round finds slot 5 gone and forgets it - reader.expire_keys[5] = true + reader.expire_keys[5] = 0 reader:remove_expired_keys() luaunit.assertNil(reader.index["key5"]) @@ -1282,7 +1773,7 @@ end function TestKeyIndex:testReaderFindsSlotReissuedAfterKeyCountClimbsBack() local reader = evict_key_count_and_slot(self, 5) - reader.expire_keys[5] = true + reader.expire_keys[5] = 0 reader:remove_expired_keys() luaunit.assertEquals(self.key_index:add("key5", "eviction_err", 60), nil) @@ -1311,7 +1802,7 @@ end function TestKeyIndex:testReaderFindsLowestSlotReissuedWithoutRepair() local reader = evict_key_count_and_slot(self, 1) - reader.expire_keys[1] = true + reader.expire_keys[1] = 0 reader:remove_expired_keys() -- slot 1 is free, so this takes it without any key_count repair, and @@ -1323,7 +1814,7 @@ end function TestKeyIndex:testReaderFindsSlotReissuedByFreshWorker() local reader = evict_key_count_and_slot(self, 1) - reader.expire_keys[1] = true + reader.expire_keys[1] = 0 reader:remove_expired_keys() -- a respawned worker: its self.last is 0, so it cannot see key_count going @@ -1350,7 +1841,7 @@ local function evict_key_count_and_all_slots(self) self.dict:delete("_prefix_key_count") for i = 1, 100 do self.dict:delete("_prefix_key_" .. i) - reader.expire_keys[i] = true + reader.expire_keys[i] = 0 end reader:remove_expired_keys() luaunit.assertNil(reader.index["key5"]) @@ -1742,6 +2233,8 @@ function TestPrometheus:testKeyTimeout() sleep(1) self.p.key_index:sync() + -- the round above reclaimed for this interval; let the interval pass + self.dict:delete("__ngx_prom__flush_expired") -- the first round keeps the expired node and flushes it, the second one -- finds it reclaimed and drops the reference self.p.key_index:remove_expired_keys()