Skip to content

fix(keys): re-claim the expired slot in place instead of allocating a new one #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
AlinsRan wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(keys): re-claim the expired slot in place instead of allocating a new one #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 28 additions & 12 deletions prometheus_keys.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,22 +156,38 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
local ok, err = self.dict:expire(self.key_prefix .. self.index[key], exptime)
if not ok then
if err == "not found" then
-- The slot already expired in the shared dict. Drop the stale
-- local state and bump delete_count so other workers do a full
-- sync and reclaim the slot; without this the old slot lingers in
-- their local self.keys while the metric is re-added at a new slot,
-- desynchronizing the index and causing duplicate metric emission.
-- The dict slot is already gone (expire returned "not found"), so
-- there is no slot to clear here.
-- The slot was reclaimed from the shared dict (expire() only
-- reports "not found" once the node is physically gone; a merely
-- ttl-expired node is resurrected by expire() and keeps its slot).
-- Re-claim the SAME slot number rather than allocating a new one.
-- The key keeps its identity, so no peer's index[key] goes stale,
-- key_count does not grow, and a key can never occupy two slots --
-- which is what produced the duplicate metrics this branch was
-- written for. Nothing has to be broadcast either, so delete_count
-- stays put and no worker is forced into a sync_range(0, key_count)
-- on its request path.
local idx = self.index[key]
local ok2, err2, forcible2 =
self.dict:add(self.key_prefix .. idx, key, exptime)
if ok2 or (err2 == "exists"
and self.dict:get(self.key_prefix .. idx) == key) then
-- either we re-claimed it, or a peer re-claimed it for us
if exptime and exptime > 0 then
self.expire_keys[idx] = true
end
if forcible2 then
return (err_msg_lru_eviction .. "; key index: re-claimed slot: idx="
.. self.key_prefix .. idx .. ", key=" .. key)
end
break
end
-- The slot could not be re-claimed (out of memory, or it now holds
-- a different key because key_count was evicted and restarted).
-- Fall back to allocating a new slot below; list() still reports
-- the key once, from the slot index currently points at.
self.index[key] = nil
self.keys[idx] = nil
self.expire_keys[idx] = nil
self.deleted = self.deleted + 1
local _, incr_err, forcible = self.dict:incr(self.delete_count, 1, 0)
if incr_err or forcible then
return incr_err or err_msg_lru_eviction
end
expired = true
else
-- Unexpected expire error: the slot may still be live, so leave it
Expand Down
83 changes: 69 additions & 14 deletions prometheus_test.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -848,8 +848,8 @@ function TestKeyIndex:testExpiredReAddNoDuplicate()
luaunit.assertEquals(self.key_index.index["expkey"], 1)

-- A second worker sharing the same shared dict syncs the initial state, so it
-- now holds slot 1 in its local self.keys/index. This is the worker that the
-- delete_count bump must later force to re-sync and reclaim the stale slot.
-- now holds slot 1 in its local self.keys/index. It must stay correct across
-- the re-add without having to be told anything.
local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
worker2:sync()
luaunit.assertEquals(worker2.index["expkey"], 1)
Expand All @@ -861,30 +861,85 @@ function TestKeyIndex:testExpiredReAddNoDuplicate()
sleep(2)
luaunit.assertEquals(self.dict:get("_prefix_key_1"), nil)

-- Re-adding the now-expired key takes the expired branch and allocates slot 2.
-- Re-adding the now-expired key re-claims slot 1 in place.
err = self.key_index:add("expkey", "eviction_err", 1)
luaunit.assertEquals(err, nil)
luaunit.assertEquals(self.dict:get("_prefix_key_2"), "expkey")
luaunit.assertEquals(self.dict:get("_prefix_key_1"), "expkey")
luaunit.assertEquals(self.key_index.index["expkey"], 1)

-- delete_count must have been bumped on the expired re-add path so that
-- other workers do a full sync and drop the stale slot.
luaunit.assertEquals(self.dict:get("_prefix_delete_count"), 1)
-- No new slot was allocated, so key_count does not grow ...
luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1)
-- ... and nothing has to be broadcast, so no worker is forced into a full
-- sync_range(0, key_count) on its request path.
luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil)

-- list() must report the key exactly once, not twice.
local keys = self.key_index:list()
luaunit.assertEquals(#keys, 1)
luaunit.assertEquals(keys[1], "expkey")

-- The second worker must converge: its next sync() sees the bumped
-- delete_count, does a full sync, drops the stale slot 1 and picks up slot 2.
-- Without the delete_count bump it would keep slot 1 forever and list() the
-- key twice.
worker2:sync()
luaunit.assertEquals(worker2.keys[1], nil)
luaunit.assertEquals(worker2.index["expkey"], 2)
-- The second worker never went stale: the key kept its slot number, so its
-- local index is still correct and list() reports the key once, with or
-- without an intervening sync().
local keys2 = worker2:list()
luaunit.assertEquals(#keys2, 1)
luaunit.assertEquals(keys2[1], "expkey")
worker2:sync()
luaunit.assertEquals(worker2.index["expkey"], 1)
luaunit.assertEquals(#worker2:list(), 1)
end


-- A key whose slot cannot be re-claimed must still be listed exactly once.
-- The slot is stolen between expire() reporting it gone and add() trying to
-- re-claim it, which is the only way another key can end up on that slot
-- number; add() must then fall back to allocating a new one.
function TestKeyIndex:testExpiredReAddFallbackNoDuplicate()
local err = self.key_index:add("expkey", "eviction_err", 1)
luaunit.assertEquals(err, nil)
self.key_index:sync()
luaunit.assertEquals(self.key_index.index["expkey"], 1)

local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
worker2:sync()
luaunit.assertEquals(worker2.index["expkey"], 1)

sleep(2)
luaunit.assertEquals(self.dict:get("_prefix_key_1"), nil)

-- Simulate a peer writing a different key onto slot 1 in the window between
-- our expire() and our add().
local real_expire = self.dict.expire
self.dict.expire = function(dict_self, k, exptime)
local ok, expire_err = real_expire(dict_self, k, exptime)
if not ok and k == "_prefix_key_1" then
dict_self:set(k, "someone_else")
end
return ok, expire_err
end

err = self.key_index:add("expkey", "eviction_err", 1)
self.dict.expire = real_expire

luaunit.assertEquals(err, nil)
luaunit.assertEquals(self.dict:get("_prefix_key_1"), "someone_else")
luaunit.assertEquals(self.dict:get("_prefix_key_2"), "expkey")
luaunit.assertEquals(self.key_index.index["expkey"], 2)

local seen = 0
for _, k in ipairs(self.key_index:list()) do
if k == "expkey" then seen = seen + 1 end
end
luaunit.assertEquals(seen, 1)

-- worker2 still points at slot 1; list() must not emit the key from a slot
-- the index no longer points at.
worker2:sync()
local seen2 = 0
for _, k in ipairs(worker2:list()) do
if k == "expkey" then seen2 = seen2 + 1 end
end
luaunit.assertEquals(seen2, 1)
end

-- remove_expired_keys() must physically reclaim the shared-dict space of
Expand Down
Loading