prometheus_test.lua

-- vim: ts=2:sw=2:sts=2:expandtab
luaunit = require('luaunit')
rex_pcre2 = require('rex_pcre2')

-- Simple implementation of a nginx shared dictionary
local SimpleDict = {}
SimpleDict.__index = SimpleDict
function SimpleDict:set(k, v, exptime)
  local forcible = false
  if k == "willnotfitk" or v == "willnotfitv" then
    forcible = true
  end
  if not self.dict then self.dict = {} end
  if not v then
    self.dict[k] = nil
  else
    local expired = exptime and exptime > 0 and (os.time() + exptime) or nil
    self.dict[k] = {value = v, expired = expired}
  end
  return true, nil, forcible
end
function SimpleDict:add(k, v, exptime)
  local forcible = false
  if k == "willnotfitk" or v == "willnotfitv" then
    forcible = true
  end
  -- ngx.shared.DICT:add only refuses a node that is still live; one that is
  -- past its exptime is reused in place, keeping the same key.
  if self:get(k) ~= nil then
    return false, "exists", false  -- match ngx.shared.DICT:add on present keys
  end
  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
     (self.forcible_keys and self.forcible_keys[k]) then
    forcible = true
  end
  if not self.dict[k] then self.dict[k] = {value = init} end
  self.dict[k]["value"] = self.dict[k]["value"] + (v or 1)
  return self.dict[k]["value"], nil, forcible  -- newval, err, forcible
end
function SimpleDict:get(k)
  -- simulate key not exist
  if k == "gauge2{f2=\"key_not_exist\",f1=\"key_not_exist\"}" then
    return nil, nil
  end
  -- simulate an error
  if k == "gauge2{f2=\"dict_error\",f1=\"dict_error\"}" then
    return nil, "dict error"
  end
  if not self.dict then self.dict = {} end
  -- An entry past its exptime reads as missing but is NOT freed here: like
  -- ngx.shared.DICT, only flush_expired() (or a write reusing the node) frees
  -- it. Pruning here would collapse the "past its ttl" and "node reclaimed"
  -- states, which ttl() has to tell apart.
  local e = self.dict[k]
  if not e or (e["expired"] and e["expired"] < os.time()) then
    return nil, nil
  end
  return e["value"], nil  -- value, err
end
function SimpleDict:get_stale(k)
  local e = self.dict and self.dict[k]
  if not e then return nil end
  return e.value, nil, e.expired ~= nil and e.expired < os.time()
end
function SimpleDict:delete(k)
  self.dict[k] = nil
end
function SimpleDict:expire(k, exptime)
  -- Like ngx.shared.DICT:expire: it looks the node up without checking the
  -- exptime, so a node that is merely past it is resurrected in place, with
  -- its value intact. Only a node that has been freed reports "not found".
  if not self.dict[k] then
    return nil, "not found"
  end
  self.dict[k]["expired"] = exptime ~= 0 and (os.time() + exptime) or nil
  return true  -- match ngx.shared.DICT:expire, which returns true on success
end
function SimpleDict:flush_expired(n)
  -- Mirrors ngx.shared.DICT:flush_expired: scans the whole dict, physically
  -- removes only the entries that are already expired and returns how many
  -- were removed. The optional argument caps the number removed, not scanned.
  if not self.dict then self.dict = {} end
  local flushed = 0
  local now = os.time()
  for k, v in pairs(self.dict) do
    if v["expired"] and v["expired"] < now then
      self.dict[k] = nil
      flushed = flushed + 1
      if n and n > 0 and flushed >= n then break end
    end
  end
  return flushed
end
function SimpleDict:ttl(k)
  -- Like ngx.shared.DICT:ttl, which peeks at the node without checking the
  -- exptime. It reports the three states a slot can be in:
  --   live                         -> a positive number (0 when permanent)
  --   past its exptime, node kept  -> a negative number
  --   node freed                   -> nil, "not found"
  local e = self.dict and self.dict[k]
  if not e then
    return nil, "not found"
  end
  if not e["expired"] then
    return 0
  end
  return e["expired"] - os.time()
end

local function sleep(n)
  local t0 = os.time()
  while os.time() - t0 <= n do end
end

-- Global nginx object
local Nginx = {}
Nginx.__index = Nginx
Nginx.ERR = {}
Nginx.WARN = {}
Nginx.DEBUG = {}
Nginx.header = {}
function Nginx.log(level, ...)
  if level == ngx.DEBUG then return end
  if not ngx.logs then ngx.logs = {} end
  table.insert(ngx.logs, table.concat({...}, " "))
end
function Nginx.print(printed)
  if not ngx.printed then ngx.printed = {} end
  for str in string.gmatch(table.concat(printed, ""), "([^\n]+)") do
    table.insert(ngx.printed, str)
  end
end
Nginx.worker = {}
function Nginx.worker.id()
  return 0
end
function Nginx.worker.count()
  return 1
end
function Nginx.worker.exiting()
  return false
end
function Nginx.sleep() end
function Nginx.now()
  -- the SimpleDict mock keeps expiry in os.time() units
  return os.time()
end
Nginx.timer = {}
function Nginx.timer.every(_, _, _) return true end
function Nginx.timer.at(_, _, _) return true end
function Nginx.get_phase()
  return 'init_worker'
end
Nginx.re = {}
function Nginx.re.match(subject, regexp, _)
  local result = {rex_pcre2.match(subject, regexp)}
  if result[1] == nil or result[1] == false then
    return nil, nil
  end
  return result, nil
end
function Nginx.re.gsub(subject, regexp, replace, _)
  local result, _, substitutions = rex_pcre2.gsub(subject, regexp, replace)
  return result, substitutions, nil
end

ngx = setmetatable({shared={}}, Nginx)

-- Finds index of a given object in a table
local function find_idx(table, element)
  for idx, value in pairs(table) do
    if value == element then
      return idx
    end
  end
end

TestPrometheus = {}
function TestPrometheus:setUp()
  self.dict = setmetatable({}, SimpleDict)
  ngx.shared.metrics = self.dict
  self.p = require('prometheus').init('metrics')
  self.counter1 = self.p:counter("metric1", "Metric 1")
  self.counter2 = self.p:counter("metric2", "Metric 2", {"f2", "f1"})
  self.counter3 = self.p:counter("metric3", "Metric 3", {"f3"})
  self.counter4 = self.p:counter("metric4", "Metric 4", {"f1","f2","f3"})
  self.gauge1 = self.p:gauge("gauge1", "Gauge 1")
  self.gauge2 = self.p:gauge("gauge2", "Gauge 2", {"f2", "f1"})
  self.hist1 = self.p:histogram("l1", "Histogram 1")
  self.hist2 = self.p:histogram("l2", "Histogram 2", {"var", "site"})
  self.counter_exp = self.p:counter("metric_exp", "Metric expire", nil, 1)
  self.gauge_exp = self.p:gauge("gauge_exp", "Gauge expire", nil, 1)
  self.gauge_exp_2 = self.p:gauge("gauge_exp_2", "Gauge expire 2", nil, 1)
  self.hist_exp = self.p:histogram("l_exp", "Histogram expire", nil, nil, 1)
  self.counter_exp_2 = self.p:counter("metric_exp2", "Metric expire 2", nil, 1)

end
function TestPrometheus.tearDown()
  ngx.logs = nil
end
function TestPrometheus:testInit()
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testInitOptions()
  self.dict = setmetatable({}, SimpleDict)
  ngx.shared.metrics = self.dict

  local p1 = require('prometheus').init("metrics")
  assert(p1.prefix == "")
  assert(p1.sync_interval == 1)
  assert(p1.error_metric_name == "nginx_metric_errors_total")

  local p2 = require('prometheus').init("metrics", "test_pref_")
  assert(p2.prefix == "test_pref_")
  assert(p2.sync_interval == 1)
  assert(p2.error_metric_name == "nginx_metric_errors_total")

  local p3 = require('prometheus').init("metrics", {sync_interval=3})
  assert(p3.prefix == "")
  assert(p3.sync_interval == 3)
  assert(p3.error_metric_name == "nginx_metric_errors_total")

  local p4 = require('prometheus').init("metrics", {
    prefix="foo", sync_interval=3, error_metric_name="foobar"})
  assert(p4.prefix == "foo")
  assert(p4.sync_interval == 3)
  assert(p4.error_metric_name == "foobar")
  assert(p4.remove_expired_keys_interval ~= 0 and p4.remove_expired_keys_interval ~= "")

  local p5 = require('prometheus').init("metrics", {remove_expired_keys_interval=3})
  assert(p5.remove_expired_keys_interval == 3)

  luaunit.assertEquals(ngx.logs, nil)
end

function TestPrometheus:testInitWorker()
  self.dict = setmetatable({}, SimpleDict)
  ngx.shared.metrics = self.dict

  local p1 = require('prometheus').init("metrics")
  p1:init_worker(3)

  luaunit.assertEquals(#ngx.logs, 1)
  luaunit.assertStrContains(ngx.logs[1], "do not explicitly call init_worker")
end
function TestPrometheus.testErrorUnitialized()
  local p = require('prometheus')
  p:counter("metric1")
  p:histogram("metric2")
  p:gauge("metric3")
  p:metric_data()

  luaunit.assertEquals(#ngx.logs, 4)
end
function TestPrometheus.testErrorUnknownDict()
  local pok, perr = pcall(require('prometheus').init, "nonexistent")
  luaunit.assertEquals(pok, false)
  luaunit.assertStrContains(perr, "does not seem to exist")
end
function TestPrometheus:testErrorNoMemoryGaugeSet()
  local gauge3 = self.p:gauge("willnotfitk")
  self.counter1:inc(5)
  gauge3:set(111)

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 5)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
  luaunit.assertEquals(self.dict:get("willnotfitk"), 111)
  luaunit.assertEquals(#ngx.logs, 1)
end
function TestPrometheus:testErrorNoMemoryGaugeInc()
  local gauge3 = self.p:gauge("willnotfitk")
  self.counter1:inc(5)
  gauge3:inc(111)

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 5)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
  luaunit.assertEquals(self.dict:get("willnotfitk"), 111)
  luaunit.assertEquals(#ngx.logs, 1)
end
function TestPrometheus:testErrorNoMemoryCounter()
  local counter = self.p:counter("willnotfitk")
  self.counter1:inc(5)
  counter:inc(11)

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 5)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
  luaunit.assertEquals(self.dict:get("willnotfitk"), 11)
  luaunit.assertEquals(#ngx.logs, 1)
end
function TestPrometheus:testErrorInvalidMetricName()
  self.p:histogram("name with a space", "Histogram")
  self.p:gauge("nonprintable\004characters", "Gauge")
  self.p:counter("0startswithadigit", "Counter")
  self.p:counter("__ngx_prom__usesinternalprefix", "Counter no.2")

  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 4)
  luaunit.assertEquals(#ngx.logs, 4)
end
function TestPrometheus:testErrorInvalidLabels()
  self.p:histogram("hist1", "Histogram", {"le"})
  self.p:gauge("count1", "Gauge", {"le"})
  self.p:counter("count1", "Counter", {"foo\002"})

  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 3)
  luaunit.assertEquals(#ngx.logs, 3)
end
function TestPrometheus:testErrorDuplicateMetrics()
  self.p:counter("metric1", "Another metric 1")
  self.p:counter("l1_count", "Conflicts with Histogram 1")
  self.p:counter("l2_sum", "Conflicts with Histogram 2")
  self.p:counter("l2_bucket", "Conflicts with Histogram 2")
  self.p:gauge("metric1", "Conflicts with Metric 1")
  self.p:histogram("l1", "Conflicts with Histogram 1")
  self.p:histogram("metric2", "Conflicts with Metric 2")
  self.p:counter("metric_A_count", "Metric ending with _count")
  self.p:histogram("metric_A", "Conflicts with metric_A_count")
  self.p:counter("metric_B_sum", "Metric ending with _sum")
  self.p:histogram("metric_B", "Conflicts with metric_B_sum")
  self.p:counter("metric_C_bucket", "Metric ending with _bucket")
  self.p:histogram("metric_C", "Conflicts with metric_C_bucket")
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 10)
  luaunit.assertEquals(#ngx.logs, 10)
end
function TestPrometheus:testErrorNegativeValue()
  self.counter1:inc(-5)

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
  luaunit.assertEquals(#ngx.logs, 1)
end
function TestPrometheus:testErrorIncorrectLabels()
  self.counter1:inc(1, {"should-be-no-labels"})
  self.counter2:inc(1, {"too-few-labels"})
  self.counter2:inc(1, {nil, "v"})
  self.counter2:inc(1, {"v", nil})
  self.counter2:inc(1)
  self.counter3:inc(1, {nil})
  self.counter4:inc(1, {"one", nil, "three"})
  self.gauge1:set(1, {"should-be-no-labels"})
  self.gauge2:set(1, {"too-few-labels"})
  self.gauge2:set(1)
  self.hist2:observe(1, {"too", "many", "labels"})
  self.hist2:observe(1, {nil, "label"})
  self.hist2:observe(1, {"label", nil})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), nil)
  luaunit.assertEquals(self.dict:get("l1_count"), nil)
  luaunit.assertEquals(self.dict:get("gauge1"), nil)
  luaunit.assertEquals(self.dict:get("gauge2"), nil)
  luaunit.assertEquals(self.dict:get("l1_count"), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 13)
  luaunit.assertEquals(#ngx.logs, 13)
end
function TestPrometheus:testNumericLabelValues()
  self.counter2:inc(1, {0, 15.5})
  self.gauge2:set(1, {0, 15.5})
  self.hist2:observe(1, {-3, 90000})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('metric2{f2="0",f1="15.5"}'), 1)
  luaunit.assertEquals(self.dict:get('gauge2{f2="0",f1="15.5"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_sum{var="-3",site="90000"}'), 1)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testMultibyteLabelValues()
  self.counter2:inc(1, {"foo", "baz\189\166qux"})
  self.counter2:inc(1, {"bad1\195\195bad", "bad2\224\161\209bad"})
  self.counter2:inc(1, {"bad3\240\144\129\192bad", "bad4\242\129\210bad"})
  self.counter2:inc(1, {"¢€𤭢", "Pay in €. Thanks."})
  self.gauge2:set(1, {"z\001", "\002"})
  self.gauge2:set(1, {"\224\143\175", "\237\129\128"})
  self.hist2:observe(1, {"\166omg", "fooшbar"})
  self.hist2:observe(1, {"\244\143\143\143", "\244"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('metric2{f2="foo",f1="baz"}'), 1)
  luaunit.assertEquals(self.dict:get('metric2{f2="bad1",f1="bad2"}'), 1)
  luaunit.assertEquals(self.dict:get('metric2{f2="bad3",f1="bad4"}'), 1)
  luaunit.assertEquals(self.dict:get('metric2{f2="¢€𤭢",f1="Pay in €. Thanks."}'), 1)
  luaunit.assertEquals(self.dict:get('gauge2{f2="z\001",f1="\002"}'), 1)
  luaunit.assertEquals(self.dict:get('gauge2{f2="",f1="\237\129\128"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_sum{var="",site="fooшbar"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_sum{var="\244\143\143\143",site=""}'), 1)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testNoValues()
  self.counter1:inc()  -- defaults to 1
  self.gauge1:set()  -- should produce an error
  self.hist1:observe()  -- should produce an error

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 2)
  luaunit.assertEquals(#ngx.logs, 2)
end
function TestPrometheus:testCounters()
  self.counter1:inc()
  self.counter1:inc(4)
  self.counter2:inc(1, {"v2", "v1"})
  self.counter2:inc(3, {"v2", "v1"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 5)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v1"}'), 4)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testGaugeIncDec()
  self.gauge1:inc(-1)
  luaunit.assertEquals(self.dict:get("gauge1"), -1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge1:inc(3)
  luaunit.assertEquals(self.dict:get("gauge1"), 2)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge1:inc()
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(1, {"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(5, {"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), 6)
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="othervalue"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(-2, {"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), 4)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(-5, {"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), -1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge1:inc(1, {"should-be-no-labels"})
  self.gauge2:inc(1, {"too-few-labels"})
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), -1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 2)
end
function TestPrometheus:testGaugeDel()
  self.gauge1:inc(1)
  luaunit.assertEquals(self.dict:get("gauge1"), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge1:del()
  luaunit.assertEquals(self.dict:get("gauge1"), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(1, {"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:del({"f2value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)

  self.gauge2:del({"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
end
function TestPrometheus:testCounterDel()
  self.counter1:inc(1)
  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.counter1:del()
  luaunit.assertEquals(self.dict:get("metric1"), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.counter2:inc(1, {"f2value", "f1value"})
  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('metric2{f2="f2value",f1="f1value"}'), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.counter2:del()
  luaunit.assertEquals(self.dict:get('metric2{f2="f2value",f1="f1value"}'), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)

  self.counter2:del({"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('metric2{f2="f2value",f1="f1value"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
end
function TestPrometheus:testReset()
  self.gauge1:inc(1)
  luaunit.assertEquals(self.dict:get("gauge1"), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge1:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("gauge1"), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge1:inc(3)
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(1, {"f2value", "f1value"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), 1)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:inc(4, {"f2value", "f1value2"})
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value2"}'), 4)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.gauge2:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), nil)
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value2"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.counter1:inc()
  self.counter1:inc(4)
  self.counter2:inc(1, {"v2", "v1"})
  self.counter2:inc(3, {"v2", "v2"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 5)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v1"}'), 1)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v2"}'), 3)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.counter1:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("metric1"), nil)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v1"}'), 1)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v2"}'), 3)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.counter1:inc(4)
  self.p._counter:sync()
  self.counter2:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 4)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v1"}'), nil)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="v2"}'), nil)
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value"}'), nil)
  luaunit.assertEquals(self.dict:get('gauge2{f2="f2value",f1="f1value2"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.hist1:observe(0.35)
  self.hist1:observe(0.4)
  self.hist2:observe(0.001, {"ok", "site1"})
  self.hist2:observe(0.15, {"ok", "site1"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 4)
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.300"}'), nil)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.400"}'), 2)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.500"}'), 2)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l1_count'), 2)
  luaunit.assertEquals(self.dict:get('l1_sum'), 0.75)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.005"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.100"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.200"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_count{var="ok",site="site1"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_sum{var="ok",site="site1"}'), 0.151)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.hist1:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 4)
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.400"}'), nil)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.500"}'), nil)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="Inf"}'), nil)
  luaunit.assertEquals(self.dict:get('l1_count'), nil)
  luaunit.assertEquals(self.dict:get('l1_sum'), nil)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.005"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.100"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.200"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_count{var="ok",site="site1"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_sum{var="ok",site="site1"}'), 0.151)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  self.hist1:observe(0.35)
  self.p._counter:sync()
  self.hist2:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("metric1"), 4)
  luaunit.assertEquals(self.dict:get("gauge1"), 3)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.400"}'), 1)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.500"}'), 1)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="Inf"}'), 1)
  luaunit.assertEquals(self.dict:get('l1_count'), 1)
  luaunit.assertEquals(self.dict:get('l1_sum'), 0.35)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.005"}'), nil)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.100"}'), nil)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.200"}'), nil)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="Inf"}'), nil)
  luaunit.assertEquals(self.dict:get('l2_count{var="ok",site="site1"}'), nil)
  luaunit.assertEquals(self.dict:get('l2_sum{var="ok",site="site1"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  -- key not exist
  self.gauge2:inc(4, {"key_not_exist", "key_not_exist"})
  self.gauge2:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get('gauge2{f2="key_not_exist",f1="key_not_exist"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)

  -- error get from dict
  self.gauge2:inc(4, {"dict_error", "dict_error"})
  self.gauge2:reset()
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get('gauge2{f2="dict_error",f1="dict_error"}'), nil)
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 1)
end
function TestPrometheus:testLatencyHistogram()
  self.hist1:observe(0.35)
  self.hist1:observe(0.4)
  self.hist2:observe(0.001, {"ok", "site1"})
  self.hist2:observe(0.15, {"ok", "site1"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.300"}'), nil)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.400"}'), 2)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.500"}'), 2)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l1_count'), 2)
  luaunit.assertEquals(self.dict:get('l1_sum'), 0.75)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.005"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.100"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="00.200"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site1",le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_count{var="ok",site="site1"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_sum{var="ok",site="site1"}'), 0.151)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testLabelEscaping()
  self.counter2:inc(1, {"v2", "\""})
  self.counter2:inc(5, {"v2", "\\"})
  self.gauge2:set(1, {"v2", "\""})
  self.gauge2:set(5, {"v2", "\\"})
  self.gauge2:set(7, {"v3", "foo\nbar"})
  self.hist2:observe(0.001, {"ok", "site\"1"})
  self.hist2:observe(0.15, {"ok", "site\"1"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="\\""}'), 1)
  luaunit.assertEquals(self.dict:get('metric2{f2="v2",f1="\\\\"}'), 5)
  luaunit.assertEquals(self.dict:get('gauge2{f2="v2",f1="\\""}'), 1)
  luaunit.assertEquals(self.dict:get('gauge2{f2="v2",f1="\\\\"}'), 5)
  luaunit.assertEquals(self.dict:get('gauge2{f2="v3",f1="foo\\nbar"}'), 7)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site\\"1",le="00.005"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site\\"1",le="00.100"}'), 1)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site\\"1",le="00.200"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_bucket{var="ok",site="site\\"1",le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_count{var="ok",site="site\\"1"}'), 2)
  luaunit.assertEquals(self.dict:get('l2_sum{var="ok",site="site\\"1"}'), 0.151)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testCustomBucketer1()
  local hist3 = self.p:histogram("l3", "Histogram 3", {"var"}, {1,2,3})
  self.hist1:observe(0.35)
  hist3:observe(2, {"ok"})
  hist3:observe(0.151, {"ok"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.300"}'), nil)
  luaunit.assertEquals(self.dict:get('l1_bucket{le="00.400"}'), 1)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="1.0"}'), 1)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="2.0"}'), 2)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="3.0"}'), 2)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l3_count{var="ok"}'), 2)
  luaunit.assertEquals(self.dict:get('l3_sum{var="ok"}'), 2.151)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testCustomBucketer2()
  local hist3 = self.p:histogram("l3", "Histogram 3", {"var"},
    {0.000005,5,50000})
  hist3:observe(0.000001, {"ok"})
  hist3:observe(3, {"ok"})
  hist3:observe(7, {"ok"})
  hist3:observe(70000, {"ok"})

  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="00000.000005"}'), 1)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="00005.000000"}'), 2)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="50000.000000"}'), 3)
  luaunit.assertEquals(self.dict:get('l3_bucket{var="ok",le="Inf"}'), 4)
  luaunit.assertEquals(self.dict:get('l3_count{var="ok"}'), 4)
  luaunit.assertEquals(self.dict:get('l3_sum{var="ok"}'), 70010.000001)
  luaunit.assertEquals(ngx.logs, nil)
end
function TestPrometheus:testCollect()
  local hist3 = self.p:histogram("b1", "Bytes", {"var", "stale"}, {0.1, 100, 2000})
  local hist4 = self.p:histogram("b2", "Labels", {}, {100, 2000})
  self.counter1:inc(5)
  self.counter2:inc(2, {"v2", "v1"})
  self.counter2:inc(2, {"v2", "v1"})
  self.gauge1:set(3)
  self.gauge2:set(2, {"v2", "v1"})
  self.gauge2:set(5, {"v2", "v1"})
  self.hist1:observe(0.000001)
  self.hist2:observe(0.000001, {"ok", "site2"})
  self.hist2:observe(3, {"ok", "site2"})
  self.hist2:observe(7, {"ok", "site2"})
  self.hist2:observe(70000, {"ok","site2"})
  hist3:observe(0.01, {"ok", "true"})
  hist3:observe(50, {"ok", "true"})
  hist3:observe(50, {"ok", "true"})
  hist3:observe(150, {"ok", "true"})
  hist3:observe(5000, {"ok", "true"})
  hist4:observe(50, {})
  hist4:observe(50, {})
  hist4:observe(150, {})
  hist4:observe(5000, {})
  self.p:collect()

  assert(find_idx(ngx.printed, "# HELP metric1 Metric 1") ~= nil)
  assert(find_idx(ngx.printed, "# TYPE metric1 counter") ~= nil)
  assert(find_idx(ngx.printed, "metric1 5") ~= nil)

  assert(find_idx(ngx.printed, "# TYPE metric2 counter") ~= nil)
  assert(find_idx(ngx.printed, 'metric2{f2="v2",f1="v1"} 4') ~= nil)

  assert(find_idx(ngx.printed, "# TYPE gauge1 gauge") ~= nil)
  assert(find_idx(ngx.printed, 'gauge1 3') ~= nil)

  assert(find_idx(ngx.printed, "# TYPE gauge2 gauge") ~= nil)
  assert(find_idx(ngx.printed, 'gauge2{f2="v2",f1="v1"} 5') ~= nil)

  assert(find_idx(ngx.printed, "# TYPE b1 histogram") ~= nil)
  assert(find_idx(ngx.printed, "# HELP b1 Bytes") ~= nil)
  assert(find_idx(ngx.printed, 'b1_bucket{var="ok",stale="true",le="0.1"} 1') ~= nil)
  assert(find_idx(ngx.printed, 'b1_bucket{var="ok",stale="true",le="100"} 3') ~= nil)
  assert(find_idx(ngx.printed, 'b1_sum{var="ok",stale="true"} 5250.01') ~= nil)
  assert(find_idx(ngx.printed, 'b2_bucket{le="100"} 2') ~= nil)
  assert(find_idx(ngx.printed, 'b2_sum{} 5250') ~= nil)

  assert(find_idx(ngx.printed, 'l2_bucket{var="ok",site="site2",le="4"} 2') ~= nil)
  assert(find_idx(ngx.printed, 'l2_bucket{var="ok",site="site2",le="+Inf"} 4') ~= nil)

  -- check that type comment exists and is before any samples for the metric.
  local type_idx = find_idx(ngx.printed, '# TYPE l1 histogram')
  assert (type_idx ~= nil)
  assert (ngx.printed[type_idx-1]:find("^l1") == nil)
  assert (ngx.printed[type_idx+1]:find("^l1") ~= nil)
  luaunit.assertEquals(ngx.logs, nil)
end

function TestPrometheus:testCollectWithPrefix()
  self.dict = setmetatable({}, SimpleDict)
  ngx.shared.metrics = self.dict
  local p = require('prometheus').init("metrics", "test_pref_")

  local counter1 = p:counter("metric1", "Metric 1")
  local gauge1 = p:gauge("gauge1", "Gauge 1")
  local hist1 = p:histogram("b1", "Bytes", {"var"}, {100, 2000})
  counter1:inc(5)
  gauge1:set(3)
  hist1:observe(50, {"ok"})
  hist1:observe(50, {"ok"})
  hist1:observe(150, {"ok"})
  hist1:observe(5000, {"ok"})
  p:collect()

  assert(find_idx(ngx.printed, "# HELP test_pref_metric1 Metric 1") ~= nil)
  assert(find_idx(ngx.printed, "# TYPE test_pref_metric1 counter") ~= nil)
  assert(find_idx(ngx.printed, "test_pref_metric1 5") ~= nil)

  assert(find_idx(ngx.printed, "# HELP test_pref_gauge1 Gauge 1") ~= nil)
  assert(find_idx(ngx.printed, "# TYPE test_pref_gauge1 gauge") ~= nil)
  assert(find_idx(ngx.printed, "test_pref_gauge1 3") ~= nil)

  assert(find_idx(ngx.printed, "# TYPE test_pref_b1 histogram") ~= nil)
  assert(find_idx(ngx.printed, "# HELP test_pref_b1 Bytes") ~= nil)
  assert(find_idx(ngx.printed, 'test_pref_b1_bucket{var="ok",le="100"} 2') ~= nil)
  assert(find_idx(ngx.printed, 'test_pref_b1_sum{var="ok"} 5250') ~= nil)
end

TestKeyIndex = {}
function TestKeyIndex:setUp()
  self.dict = setmetatable({}, SimpleDict)
  ngx.shared.metrics = self.dict
  self.key_index = require('prometheus_keys').new(self.dict, "_prefix_", 1)
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)
end
function TestKeyIndex:testAdd()
  self.key_index:add("single", "eviction_err")
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1)
  luaunit.assertEquals(self.dict:get("_prefix_key_1"), "single")

  self.key_index:add({"multiple", "keys"}, "eviction_err")
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 3)
  luaunit.assertEquals(self.dict:get("_prefix_key_2"), "multiple")
  luaunit.assertEquals(self.dict:get("_prefix_key_3"), "keys")

  -- adding already existing key should do nothing
  self.key_index:add("single", "eviction_err")
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 3)

  -- error should be returned when memory is full
  local err = self.key_index:add("willnotfitv", "eviction_err")
  luaunit.assertEquals(err,
    "eviction_err; key index: add key: idx=_prefix_key_4, key=willnotfitv")
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 4)
end
function TestKeyIndex:testRemove()
  self.key_index:add({"key1", "key2", "key3"}, "eviction_err")

  self.key_index:remove("key2")
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 3)
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), 1)
  local keys = self.key_index:list()
  luaunit.assertEquals(#keys, 2)
  luaunit.assertEquals(keys[1], "key1")
  luaunit.assertEquals(keys[2], "key3")

  self.key_index:remove("key4")
  luaunit.assertEquals(#ngx.logs, 1)
  keys = self.key_index:list()
  luaunit.assertEquals(#keys, 2)
  luaunit.assertEquals(keys[1], "key1")
  luaunit.assertEquals(keys[2], "key3")
end
function TestKeyIndex:testList()
  self.key_index:add({"key1", "key2", "key3"}, "eviction_err")
  local keys = self.key_index:list()
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(#keys, 3)
  luaunit.assertEquals(keys[1], "key1")
  luaunit.assertEquals(keys[2], "key2")
  luaunit.assertEquals(keys[3], "key3")
end

-- A metric that expires and comes back while its node is still in the dict
-- keeps its slot: expire() resurrects the node with its value intact, so no
-- number is consumed and nothing has to be told to the other workers.
function TestKeyIndex:testExpiredReAddKeepsItsSlot()
  local err = self.key_index:add("expkey", "eviction_err", 1)
  luaunit.assertEquals(err, nil)
  self.key_index:sync()
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1)
  luaunit.assertEquals(self.dict:get("_prefix_key_1"), "expkey")

  -- a second worker holds the same slot in its local state
  local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  worker2:sync()
  luaunit.assertEquals(worker2.index["expkey"], 1)

  sleep(2)
  -- past its ttl, but the node is still there
  luaunit.assertEquals(self.dict:get("_prefix_key_1"), nil)
  luaunit.assertTrue(self.dict:ttl("_prefix_key_1") < 0)
  -- a reclaim round must leave the number alone in this state; reclaiming is
  -- left out here, so the node stays in the dict
  self.dict.flush_expired = function() return 0 end
  self.key_index:remove_expired_keys()
  self.dict.flush_expired = nil
  luaunit.assertEquals(self.key_index.index["expkey"], 1)

  err = self.key_index:add("expkey", "eviction_err", 1)
  luaunit.assertEquals(err, nil)

  -- same slot, no new number, and nothing broadcast
  luaunit.assertEquals(self.dict:get("_prefix_key_1"), "expkey")
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1)
  luaunit.assertEquals(self.dict:get("_prefix_key_2"), nil)
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil)

  -- both workers list it exactly once; worker2 never had to re-sync
  luaunit.assertEquals(self.key_index:list(), {"expkey"})
  luaunit.assertEquals(worker2:list(), {"expkey"})
end


function TestKeyIndex:testFreshReaderFindsRevivedSlot()
  local writer = self.key_index
  writer:add("target", "eviction_err", 60)
  writer:add("tail", "eviction_err")
  self.dict.dict._prefix_key_1.expired = os.time() - 1
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  luaunit.assertEquals(reader.last, 2)
  luaunit.assertEquals(writer:add("target", "eviction_err", 60), nil)
  local keys = reader:list()
  table.sort(keys)
  luaunit.assertEquals(keys, {"tail", "target"})
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 2)
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil)
end

function TestKeyIndex:testReaderFindsRenewalAcrossExpirationBoundary()
  self.key_index:add("target", "eviction_err", 60)
  self.key_index:add("tail", "eviction_err")
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  local injected = false
  self.dict.get = function(dict, key)
    local value = SimpleDict.get(dict, key)
    if not injected and key == "_prefix_key_1" then
      injected = true
      dict.dict[key].expired = os.time() - 1
      reader:sync()
    end
    return value
  end
  luaunit.assertEquals(self.key_index:add("target", "eviction_err", 60), nil)
  luaunit.assertTrue(injected)
  luaunit.assertEquals(reader.index.target, 1)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 2)
end

function TestKeyIndex:testReclaimedCanonicalFallsBackToAlias()
  self.dict:set("_prefix_key_1", "target", 60)
  self.dict:set("_prefix_key_2", "target", 60)
  self.dict:set("_prefix_key_count", 2)
  self.key_index:sync()
  luaunit.assertEquals(self.key_index:list(), {"target"})
  self.dict:delete("_prefix_key_2")
  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)
  luaunit.assertEquals(self.key_index.aliases, {})
  luaunit.assertEquals(self.key_index:add("target", "eviction_err", 60), nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 2)
end

function TestKeyIndex:testStaleAliasDoesNotHideLiveSlotAfterReclamation()
  self.dict:set("_prefix_key_1", "target", 60)
  self.dict:set("_prefix_key_2", "target", 60)
  self.dict.dict._prefix_key_2.expired = os.time() - 1
  self.dict:set("_prefix_key_count", 2)
  self.key_index:sync()
  -- a round keeps the expired alias and then flushes it; the next one sees
  -- it gone
  self.key_index:remove_expired_keys()
  self.key_index:remove_expired_keys()
  luaunit.assertEquals(self.key_index:list(), {"target"})
  luaunit.assertEquals(self.key_index.index.target, 1)
  luaunit.assertEquals(self.key_index.aliases, {})
end

function TestKeyIndex:testRemoveAllKnownAliases()
  for i = 1, 3 do self.dict:set("_prefix_key_" .. i, "target", 60) end
  self.dict:set("_prefix_key_count", 3)
  self.key_index:sync()
  local peer = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  peer:sync()
  self.key_index:remove("target", "eviction_err")
  luaunit.assertEquals(self.key_index:list(), {})
  luaunit.assertEquals(peer:list(), {})
  luaunit.assertEquals(self.key_index.aliases, {})
  luaunit.assertEquals(peer.aliases, {})
  for i = 1, 3 do luaunit.assertNil(self.dict:get_stale("_prefix_key_" .. i)) end
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), 1)
end

function TestKeyIndex:testRemovePreservesReassignedAlias()
  self.dict:set("_prefix_key_1", "target", 60)
  self.dict:set("_prefix_key_2", "target", 60)
  self.dict:set("_prefix_key_count", 2)
  self.key_index:sync()
  -- Eviction and a counter reset let a fresh writer reuse the lower slot.
  self.dict:delete("_prefix_key_1")
  self.dict:delete("_prefix_key_count")
  local peer = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  luaunit.assertNil(peer:add("replacement", "eviction_err", 60))
  luaunit.assertEquals(peer.index.replacement, 1)
  self.key_index:remove("target", "eviction_err")
  luaunit.assertEquals(self.dict:get("_prefix_key_1"), "replacement")
  luaunit.assertNil(self.dict:get_stale("_prefix_key_2"))
  luaunit.assertNil(self.key_index.index.target)
  luaunit.assertEquals(self.key_index.aliases, {})
end

function TestKeyIndex:testSlotReassignmentDropsOldOwner()
  self.dict:set("_prefix_key_1", "old", 60)
  self.dict:set("_prefix_key_2", "old", 60)
  self.dict:set("_prefix_key_count", 2)
  self.key_index:sync()
  self.dict:set("_prefix_key_2", "new")
  self.key_index:sync_range(0, 2)
  luaunit.assertEquals(self.key_index.index.old, 1)
  luaunit.assertEquals(self.key_index.index.new, 2)
  luaunit.assertEquals(self.key_index.aliases, {})
  self.dict:delete("_prefix_key_1")
  self.key_index:sync_range(0, 2)
  luaunit.assertNil(self.key_index.index.old)
  luaunit.assertEquals(self.key_index:list(), {"new"})
end

function TestKeyIndex:testSlotReclaimedBetweenReadAndTTL()
  self.dict:set("_prefix_key_1", "target", 60)
  self.dict:set("_prefix_key_count", 1)
  self.dict.ttl = function(dict, key)
    if key == "_prefix_key_1" then dict:delete(key) end
    return SimpleDict.ttl(dict, key)
  end
  luaunit.assertEquals(self.key_index:list(), {})
  luaunit.assertNil(self.key_index.index.target)
  luaunit.assertEquals(self.key_index.expire_keys, {})
end

-- Regression test for apache/apisix#11934 (duplicate metrics) and for removing
-- the delete_count broadcast that came with it.
--
-- Once the node itself is gone the number is not the key's any more, so the key
-- takes a fresh slot. That slot is above every worker's self.last, so their
-- incremental sync picks it up: nothing has to be broadcast. The peer that
-- still holds the old, dead slot must neither list the key twice nor -- once
-- its own reclaim round runs -- lose it, which is what the broadcast used to
-- paper over.
function TestKeyIndex:testReclaimedSlotTakesANewOneWithoutBroadcasting()
  luaunit.assertEquals(self.key_index:add("expkey", "eviction_err", 1), nil)
  -- a handful of other metrics, so the dead slot sits below the peer's
  -- incremental range, as an old slot on a long-running gateway does
  for i = 1, 5 do
    luaunit.assertEquals(self.key_index:add("other" .. i, "eviction_err"), nil)
  end
  local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  worker2:sync()
  luaunit.assertEquals(worker2.index["expkey"], 1)

  sleep(2)
  self.dict:flush_expired()
  luaunit.assertNil(self.dict.dict["_prefix_key_1"])

  luaunit.assertEquals(self.key_index:add("expkey", "eviction_err", 60), nil)
  local moved = self.key_index.index["expkey"]
  luaunit.assertNotEquals(moved, 1)
  luaunit.assertEquals(self.dict:get("_prefix_key_" .. moved), "expkey")
  -- no broadcast: nothing forces the other workers into a full sync
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil)

  -- the peer finds the new slot through its incremental sync, and lists the
  -- key once although its own self.keys still holds the dead slot
  local listed = {}
  for _, k in ipairs(worker2:list()) do
    listed[k] = (listed[k] or 0) + 1
  end
  luaunit.assertEquals(listed["expkey"], 1)

  -- and its own reclaim round must not take the key down with the dead slot
  worker2:remove_expired_keys()
  luaunit.assertEquals(worker2.index["expkey"], moved)
  listed = {}
  for _, k in ipairs(worker2:list()) do
    listed[k] = (listed[k] or 0) + 1
  end
  luaunit.assertEquals(listed["expkey"], 1)
end


-- A live slot must survive a reclaim round untouched.
function TestKeyIndex:testRemoveExpiredKeysKeepsLiveSlots()
  luaunit.assertEquals(self.key_index:add("livekey", "eviction_err", 60), nil)
  luaunit.assertEquals(self.key_index:add("permkey", "eviction_err"), nil)

  self.key_index:remove_expired_keys()

  local listed = {}
  for _, k in ipairs(self.key_index:list()) do
    listed[k] = true
  end
  luaunit.assertTrue(listed["livekey"])
  luaunit.assertTrue(listed["permkey"])
  luaunit.assertEquals(self.key_index.index["livekey"], 1)
end


-- With no exptime -- the default in APISIX -- nothing ever expires, so none of
-- this comes into play and no shared counter is written at all.
function TestKeyIndex:testPermanentMetricsTouchNoSharedCounters()
  for i = 1, 3 do
    luaunit.assertEquals(self.key_index:add("perm" .. i, "eviction_err"), nil)
  end
  self.key_index:remove_expired_keys()
  for i = 1, 3 do
    luaunit.assertEquals(self.key_index:add("perm" .. i, "eviction_err"), nil)
  end

  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 3)
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil)
  luaunit.assertEquals(#self.key_index:list(), 3)
end


-- remove_expired_keys() must physically reclaim the shared-dict space of
-- expired entries, not just drop the worker-local references to them.
-- Logically expired entries keep occupying slab pages, and the passive
-- per-write expiry scan cannot reach them once a permanent entry sits at the
-- LRU tail, so without an explicit flush the dict grows without bound under
-- label churn (apache/apisix#13658).
function TestKeyIndex:testRemoveExpiredKeysReclaimsSharedDict()
  local err = self.key_index:add("expkey", "eviction_err", 1)
  luaunit.assertEquals(err, nil)
  -- A metric value key: lives in the same dict, and KeyIndex never reads it.
  self.dict:set("expkey", 1, 1)
  -- A permanent entry, like the error metric or any metric without an exptime.
  self.dict:set("permanent", 1)

  sleep(2)

  -- Both entries are logically gone but still physically present: get()
  -- reports them as missing while they still hold their slab pages, and ttl()
  -- is what tells that state apart from a node that is really gone.
  luaunit.assertEquals(self.dict:get("_prefix_key_1"), nil)
  luaunit.assertTrue(self.dict:ttl("_prefix_key_1") < 0)
  luaunit.assertNotNil(self.dict.dict["_prefix_key_1"])
  luaunit.assertNotNil(self.dict.dict["expkey"])

  self.key_index:remove_expired_keys()

  -- Both the index slot and the metric value must be reclaimed in the same
  -- round. Before the fix remove_expired_keys() issued no dict writes at all,
  -- so both entries survived and the dict only ever grew.
  luaunit.assertNil(self.dict.dict["_prefix_key_1"])
  luaunit.assertNil(self.dict.dict["expkey"])
  -- Entries that have not expired must be left alone.
  luaunit.assertNotNil(self.dict.dict["permanent"])

  -- Only now, with the node gone, is the number given up: a second round sees
  -- "not found" and drops the local reference to it.
  local _, err2 = self.dict:ttl("_prefix_key_1")
  luaunit.assertEquals(err2, "not found")
  self.key_index:remove_expired_keys()
  luaunit.assertNil(self.key_index.index["expkey"])
end

-- A slot first imported at the ttl() sentinel (expired exactly 5ms ago) is
-- kept, so it still needs a reclamation check.
function TestKeyIndex:testSyncTracksSlotsImportedAtTheTtlSentinel()
  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
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)

  self.dict.ttl = function(_, key)
    luaunit.assertEquals(key, slot)
    return nil, "not found"
  end
  reader:sync()
  self.dict.ttl = nil
  luaunit.assertEquals(reader.index["boundary"], i)
  luaunit.assertNotNil(reader.expire_keys[i])

  self.dict:delete(slot)
  reader:remove_expired_keys()
  luaunit.assertNil(reader.keys[i])
  luaunit.assertNil(reader.index["boundary"])
  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.
function TestKeyIndex:testExpiredSlotsDoNotAccumulate()
  for _ = 1, 3 do
    luaunit.assertEquals(self.key_index:add("churn", "eviction_err", 1), nil)
    sleep(2)
    self.key_index:remove_expired_keys()
  end

  local live = 0
  for k in pairs(self.dict.dict) do
    if k:find("_prefix_key_", 1, true) == 1 then live = live + 1 end
  end
  -- Slot numbers keep advancing (key_count is monotonic), but at most the most
  -- recent slot may still be physically present; the earlier ones are gone.
  luaunit.assertTrue(live <= 1)
end

-- list() must de-duplicate when self.keys transiently holds the same key in two
-- slots (the window before a stale slot is reclaimed): only the slot the index
-- points at is canonical. This isolates the list() guard from the add() path.
function TestKeyIndex:testListDeduplicatesStaleSlot()
  self.key_index.keys = {[1] = "dup", [2] = "dup"}
  self.key_index.index = {dup = 2}
  self.key_index.last = 2

  local keys = self.key_index:list()
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(#keys, 1)
  luaunit.assertEquals(keys[1], "dup")
end

-- Regression test for apache/apisix#12275 (workers pinned at 100% CPU).
-- key_count is an ordinary shared-dict node: it is only refreshed when a new
-- key is registered, so under steady traffic it goes cold and, once the dict
-- is full, gets LRU-evicted while higher-numbered slots survive. sync() then
-- reads it as 0 and add() keeps calling dict:add() on the same occupied slot:
-- "exists" -> retry -> "exists", forever, without yielding. Before the fix
-- this call never returned and the worker spun at 100% CPU; the fix advances
-- key_count past occupied slots so add() terminates and repairs the counter.
function TestKeyIndex:testAddAfterKeyCountEvicted()
  for i = 1, 50 do
    self.key_index:add("key" .. i, "eviction_err")
  end
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 50)

  -- the shared dict force-evicts its LRU-tail node, which is key_count
  self.dict:delete("_prefix_key_count")

  local err = self.key_index:add("newkey", "eviction_err")
  luaunit.assertEquals(err, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_51"), "newkey")
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 51)
  luaunit.assertEquals(self.key_index.index["newkey"], 51)
end

-- Same eviction, seen from a freshly started worker (e.g. respawned after a
-- crash, or after nginx reload) whose local index is empty: it must adopt the
-- surviving slots while repairing the counter instead of spinning.
function TestKeyIndex:testAddAfterKeyCountEvictedFreshWorker()
  for i = 1, 50 do
    self.key_index:add("key" .. i, "eviction_err")
  end
  self.dict:delete("_prefix_key_count")

  local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  local err = worker2:add("newkey", "eviction_err")
  luaunit.assertEquals(err, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_51"), "newkey")
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 51)
  -- surviving slots were adopted into the local index along the way
  luaunit.assertEquals(worker2.index["key7"], 7)
  luaunit.assertEquals(worker2.index["newkey"], 51)
end

-- key_count and an old slot are evicted together, so the slot's number is
-- handed out again below what a reader has already synced. The reader has to
-- look at it again, whether it syncs before or after key_count has climbed
-- back past its own self.last.
local function list_counts(key_index)
  local listed = {}
  for _, k in ipairs(key_index:list()) do
    listed[k] = (listed[k] or 0) + 1
  end
  return listed
end

local function evict_key_count_and_slot(self, slot)
  for i = 1, 100 do
    luaunit.assertEquals(self.key_index:add("key" .. i, "eviction_err", 60), nil)
  end
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  luaunit.assertEquals(reader.last, 100)
  self.dict:delete("_prefix_key_count")
  self.dict:delete("_prefix_key_" .. slot)
  return reader
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] = 0
  reader:remove_expired_keys()
  luaunit.assertNil(reader.index["key5"])

  luaunit.assertEquals(self.key_index:add("key5", "eviction_err", 60), nil)
  luaunit.assertEquals(self.key_index.index["key5"], 5)
  luaunit.assertEquals(list_counts(reader)["key5"], 1)
end

function TestKeyIndex:testReaderFindsSlotReissuedAfterKeyCountClimbsBack()
  local reader = evict_key_count_and_slot(self, 5)
  reader.expire_keys[5] = 0
  reader:remove_expired_keys()

  luaunit.assertEquals(self.key_index:add("key5", "eviction_err", 60), nil)
  -- new series keep coming before the reader syncs again, so by then
  -- key_count is past its self.last and a plain N < self.last check misses it
  for i = 1, 50 do
    luaunit.assertEquals(self.key_index:add("new" .. i, "eviction_err", 60), nil)
  end
  luaunit.assertTrue(self.dict:get("_prefix_key_count") > reader.last)
  luaunit.assertEquals(list_counts(reader)["key5"], 1)
end

function TestKeyIndex:testReaderFindsNewKeyInReissuedSlot()
  local reader = evict_key_count_and_slot(self, 5)
  -- a new series falls into the hole while the reader still maps slot 5 to
  -- its old key, and the node is live again, so its own round keeps that
  luaunit.assertEquals(self.key_index:add("newkey", "eviction_err", 60), nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_5"), "newkey")
  for i = 1, 50 do
    luaunit.assertEquals(self.key_index:add("new" .. i, "eviction_err", 60), nil)
  end
  local listed = list_counts(reader)
  luaunit.assertEquals(listed["newkey"], 1)
  luaunit.assertNil(listed["key5"])
end

function TestKeyIndex:testReaderFindsLowestSlotReissuedWithoutRepair()
  local reader = evict_key_count_and_slot(self, 1)
  reader.expire_keys[1] = 0
  reader:remove_expired_keys()

  -- slot 1 is free, so this takes it without any key_count repair, and
  -- nothing else is registered: only the reader can tell key_count went back
  luaunit.assertEquals(self.key_index:add("key1", "eviction_err", 60), nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1)
  luaunit.assertEquals(list_counts(reader)["key1"], 1)
end

function TestKeyIndex:testReaderFindsSlotReissuedByFreshWorker()
  local reader = evict_key_count_and_slot(self, 1)
  reader.expire_keys[1] = 0
  reader:remove_expired_keys()

  -- a respawned worker: its self.last is 0, so it cannot see key_count going
  -- backwards; only its key_count repair shows it
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  luaunit.assertEquals(fresh:add("key1", "eviction_err", 60), nil)
  luaunit.assertEquals(fresh.index["key1"], 1)
  for i = 1, 150 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  luaunit.assertTrue(self.dict:get("_prefix_key_count") > reader.last)
  luaunit.assertEquals(list_counts(reader)["key1"], 1)
end

-- Under churn the reclamation frees most old slots, so after key_count is lost
-- the next registrations refill 1, 2, 3, ... without a single collision, and
-- key_count climbs past every reader's self.last before any of them syncs.
local function evict_key_count_and_all_slots(self)
  for i = 1, 100 do
    luaunit.assertEquals(self.key_index:add("key" .. i, "eviction_err", 60), nil)
  end
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  self.dict:delete("_prefix_key_count")
  for i = 1, 100 do
    self.dict:delete("_prefix_key_" .. i)
    reader.expire_keys[i] = 0
  end
  reader:remove_expired_keys()
  luaunit.assertNil(reader.index["key5"])
  return reader
end

function TestKeyIndex:testReaderFindsContiguousRefillByFreshWorker()
  local reader = evict_key_count_and_all_slots(self)
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  luaunit.assertEquals(fresh:add("key5", "eviction_err", 60), nil)
  luaunit.assertEquals(fresh.index["key5"], 1)
  for i = 1, 120 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  luaunit.assertTrue(self.dict:get("_prefix_key_count") > reader.last)
  local listed = list_counts(reader)
  luaunit.assertEquals(listed["key5"], 1)
  luaunit.assertEquals(listed["new120"], 1)
end

function TestKeyIndex:testRenewalDoesNotTakeOverAReissuedSlot()
  local reader = evict_key_count_and_all_slots(self)
  -- the original writer still maps key5 to slot 5 and only renews, so it
  -- never syncs while key_count is low; a fresh worker refills the holes
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  for i = 1, 120 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  luaunit.assertEquals(self.dict:get("_prefix_key_5"), "new5")

  luaunit.assertEquals(self.key_index:add("key5", "eviction_err", 60), nil)
  luaunit.assertNotEquals(self.key_index.index["key5"], 5)
  local listed = list_counts(reader)
  luaunit.assertEquals(listed["key5"], 1)
  luaunit.assertEquals(listed["new5"], 1)
end

-- Losing the signal itself only costs a full sync, never a missed one.
function TestKeyIndex:testEvictedGenerationForcesAFullSync()
  local reader = evict_key_count_and_slot(self, 5)
  luaunit.assertEquals(self.key_index:add("newkey", "eviction_err", 60), nil)
  reader:sync()
  self.dict:delete("_prefix_generation")
  -- a local entry only a full sync can restore (key_count is 5 by now)
  reader.keys[3] = nil
  reader:sync()
  luaunit.assertEquals(reader.keys[3], "key3")
end

-- A re-creation that lands between sync()'s reads of key_count and of the
-- token leaves the token looking seen, with key_count below self.last.
function TestKeyIndex:testKeyCountBelowLastForcesAFullSync()
  local reader = evict_key_count_and_all_slots(self)
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  for i = 1, 3 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  reader.synced_generation = self.dict:get("_prefix_generation")
  local listed = list_counts(reader)
  for i = 1, 3 do
    luaunit.assertEquals(listed["new" .. i], 1)
  end
end

-- The writer that re-creates key_count cannot publish the token: readers must
-- still notice, even once key_count has climbed past their self.last.
function TestKeyIndex:testFailedTokenPublishForcesAFullSync()
  local reader = evict_key_count_and_all_slots(self)
  local set = SimpleDict.set
  self.dict.set = function(d, k, ...)
    if k == "_prefix_generation" then return false, "no memory" end
    return set(d, k, ...)
  end
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  for i = 1, 120 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  self.dict.set = nil
  luaunit.assertTrue(self.dict:get("_prefix_key_count") > reader.last)
  luaunit.assertEquals(list_counts(reader)["new5"], 1)
end

-- The token is evicted after a reader already synced its absence, then a
-- second re-creation's token is evicted too.
function TestKeyIndex:testTokenLostTwiceForcesAFullSync()
  for i = 1, 100 do
    luaunit.assertEquals(self.key_index:add("key" .. i, "eviction_err", 60), nil)
  end
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  self.dict:delete("_prefix_generation")
  reader:sync()
  luaunit.assertEquals(reader.last, 100)

  self.dict:delete("_prefix_key_count")
  for i = 1, 100 do
    self.dict:delete("_prefix_key_" .. i)
  end
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  for i = 1, 120 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  self.dict:delete("_prefix_generation")
  luaunit.assertEquals(list_counts(reader)["new5"], 1)
end

-- key_count is re-created between sync()'s reads of key_count and of the
-- token: the reader pairs the old count with the new token, and key_count is
-- past its self.last again by its next sync.
function TestKeyIndex:testOldKeyCountWithNewTokenForcesAFullSync()
  for i = 1, 100 do
    luaunit.assertEquals(self.key_index:add("key" .. i, "eviction_err", 60), nil)
  end
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  local get = SimpleDict.get
  self.dict.get = function(d, k)
    local v = get(d, k)
    if k == "_prefix_key_count" and d.get ~= get then
      d.get = nil
      d:delete("_prefix_key_count")
      for i = 1, 100 do
        d:delete("_prefix_key_" .. i)
      end
      luaunit.assertEquals(fresh:add("new1", "eviction_err", 60), nil)
    end
    return v
  end
  reader:sync()
  self.dict.get = nil
  for i = 2, 120 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  luaunit.assertTrue(self.dict:get("_prefix_key_count") > reader.last)
  local listed = list_counts(reader)
  luaunit.assertEquals(listed["new1"], 1)
  luaunit.assertEquals(listed["new5"], 1)
end

-- The token is missing and key_count is re-created after the reader read it
-- but before the reader publishes its own token, whose re-creation token is
-- lost as well: the reader's token must not vouch for the old key_count.
function TestKeyIndex:testPublishedTokenDoesNotCoverAnOldKeyCount()
  for i = 1, 100 do
    luaunit.assertEquals(self.key_index:add("key" .. i, "eviction_err", 60), nil)
  end
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  self.dict:delete("_prefix_generation")
  local fresh = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  self.dict.add = function(d, k, ...)
    if k == "_prefix_generation" then
      d.add = nil
      d:delete("_prefix_key_count")
      for i = 1, 100 do
        d:delete("_prefix_key_" .. i)
      end
      luaunit.assertEquals(fresh:add("new1", "eviction_err", 60), nil)
      d:delete("_prefix_generation")
    end
    return SimpleDict.add(d, k, ...)
  end
  reader:sync()
  self.dict.add = nil
  for i = 2, 120 do
    luaunit.assertEquals(fresh:add("new" .. i, "eviction_err", 60), nil)
  end
  luaunit.assertTrue(self.dict:get("_prefix_key_count") > reader.last)
  for _ = 1, 2 do
    local listed = list_counts(reader)
    luaunit.assertEquals(listed["new1"], 1)
    luaunit.assertEquals(listed["new5"], 1)
  end
end

-- Publishing the token can evict another entry; that is reported like any
-- other forcible write.
function TestKeyIndex:testForcibleTokenPublishIsReported()
  self.dict.set = function(d, k, ...)
    local ok, err = SimpleDict.set(d, k, ...)
    return ok, err, k == "_prefix_generation"
  end
  local err = self.key_index:add("key1", "eviction_err")
  self.dict.set = nil
  luaunit.assertStrContains(err, "eviction_err")
  luaunit.assertEquals(self.key_index.index["key1"], 1)
end

-- ttl() reads "not found" for a node that expired exactly 5ms ago
-- (NGX_DECLINED is -5); the node is still there for expire() to revive.
function TestKeyIndex:testTtlSentinelKeepsAPresentNode()
  luaunit.assertEquals(self.key_index:add("key1", "eviction_err", 60), nil)
  luaunit.assertEquals(self.key_index:add("key2", "eviction_err"), nil)
  local reader = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  reader:sync()
  self.dict.dict["_prefix_key_1"].expired = os.time() - 1
  local ttl = SimpleDict.ttl
  self.dict.ttl = function(d, k)
    if k == "_prefix_key_1" then return nil, "not found" end
    return ttl(d, k)
  end
  -- the node is past its ttl but not reclaimed yet
  self.dict.flush_expired = function() return 0 end
  reader:remove_expired_keys()
  self.dict.ttl = nil
  self.dict.flush_expired = nil
  luaunit.assertEquals(reader.keys[1], "key1")

  luaunit.assertEquals(self.key_index:add("key1", "eviction_err", 60), nil)
  luaunit.assertEquals(self.key_index.index["key1"], 1)
  luaunit.assertEquals(list_counts(reader)["key1"], 1)
end

-- One add() call bounds its repair work (MAX_KEY_COUNT_REPAIRS = 1000) and
-- degrades by dropping the sample, like the existing dict-full path. The
-- advanced counter is shared, so a later call resumes and converges.
function TestKeyIndex:testKeyCountRepairIsBounded()
  local keys = {}
  for i = 1, 1500 do
    keys[i] = "key" .. i
  end
  self.key_index:add(keys, "eviction_err")
  self.dict:delete("_prefix_key_count")

  local err = self.key_index:add("newkey", "eviction_err")
  luaunit.assertStrContains(err, "key_count fell behind occupied slots")
  -- bounded progress was made and persisted for the next call
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1000)

  err = self.key_index:add("newkey", "eviction_err")
  luaunit.assertEquals(err, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_1501"), "newkey")
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 1501)
end

-- If the repair increments forcibly displaced other entries and the requested
-- key then turns out to already exist (it gets adopted from an occupied slot
-- during the repair crawl), the eviction must still be reported through the
-- LRU-eviction error, exactly like the new-slot success path reports
-- forcible/forcible2.
function TestKeyIndex:testKeyCountRepairReportsForcibleEviction()
  for i = 1, 50 do
    self.key_index:add("key" .. i, "eviction_err")
  end
  self.dict:delete("_prefix_key_count")
  -- re-creating/incrementing key_count now displaces other entries
  self.dict.forcible_keys = {["_prefix_key_count"] = true}

  -- a worker with an empty local index registers a key that already exists in
  -- the dict at a slot ahead of the (rewound) counter: the repair crawl adopts
  -- it and exits through the existing-key break.
  local worker2 = require('prometheus_keys').new(self.dict, "_prefix_", 1)
  local err = worker2:add("key30", "eviction_err")
  luaunit.assertStrContains(err,
    "eviction_err; key index: adopted key after key_count repair: " ..
    "idx=_prefix_key_30, key=key30")
  luaunit.assertEquals(worker2.index["key30"], 30)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), 30)
end

function TestKeyIndex:testSync()
  self.key_index:sync()
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(self.dict:get("_prefix_key_count"), nil)
  luaunit.assertEquals(self.dict:get("_prefix_delete_count"), nil)

  -- key added by another worker
  self.dict:set("_prefix_key_count", 1)
  self.dict:set("_prefix_key_1", "key1")
  self.key_index:sync()
  local keys = self.key_index:list()
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(#keys, 1)
  luaunit.assertEquals(keys[1], "key1")

  -- multiple keys added by another worker
  self.dict:set("_prefix_key_count", 3)
  self.dict:set("_prefix_key_2", "key2")
  self.dict:set("_prefix_key_3", "key3")
  self.key_index:sync()
  keys = self.key_index:list()
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(#keys, 3)
  luaunit.assertEquals(keys[1], "key1")
  luaunit.assertEquals(keys[2], "key2")
  luaunit.assertEquals(keys[3], "key3")

  -- key deleted by another worker
  self.dict:set("_prefix_delete_count", 1)
  self.dict:delete("_prefix_key_2")
  self.dict:set("_prefix_key_3", "key3")
  self.key_index:sync()
  keys = self.key_index:list()
  luaunit.assertEquals(ngx.logs, nil)
  luaunit.assertEquals(#keys, 2)
  luaunit.assertEquals(keys[1], "key1")
  luaunit.assertEquals(keys[2], "key3")
end

function TestPrometheus:testLookupMaxSize()
  local c = self.p:counter("testc", "Counter", {"tag"})
  local g = self.p:gauge("testg", "Gauge", {"tag"})
  local h = self.p:histogram("testh", "Histogram", {"tag"})

  luaunit.assertEquals(c.lookup_size, 0)
  luaunit.assertEquals(g.lookup_size, 0)
  luaunit.assertEquals(h.lookup_size, 0)
  luaunit.assertEquals(c.lookup_max_size, 1000)
  luaunit.assertEquals(g.lookup_max_size, 1000)
  luaunit.assertEquals(h.lookup_max_size, 1000)

  for _, max_size in ipairs({10, 17, 101}) do
    local p1 = require('prometheus').init("metrics",
      {lookup_max_size=max_size})
    local c1 = p1:counter("testc1", "Counter", {"tag"})
    local g1 = p1:gauge("testg1", "Gauge", {"tag"})
    local h1 = p1:histogram("testh1", "Histogram", {"tag"})

    luaunit.assertEquals(p1.lookup_max_size, max_size)
    luaunit.assertEquals(c1.lookup_max_size, max_size)
    luaunit.assertEquals(g1.lookup_max_size, max_size)
    luaunit.assertEquals(h1.lookup_max_size, max_size)

    for i = 0, max_size * 2.1, 1 do
      c1:inc(1, {tostring(i)})
      g1:set(3, {tostring(i)})
      h1:observe(50, {tostring(i)})

      luaunit.assertEquals(c1.lookup_size, (i % max_size)+1)
      luaunit.assertEquals(g1.lookup_size, (i % max_size)+1)
      luaunit.assertEquals(h1.lookup_size, (i % max_size)+1)
    end
  end

  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)
end

function TestPrometheus.testPrintfTable()
  local p = require('prometheus')
  luaunit.assertEquals(p._table_to_string(nil), "nil")
  luaunit.assertEquals(p._table_to_string({}), "<>")
  luaunit.assertEquals(p._table_to_string({"foo"}), "<foo>")
  luaunit.assertEquals(p._table_to_string({"foo",2,"bar"}), "<foo,2,bar>")
  -- table ends at the first value before nil.
  luaunit.assertEquals(p._table_to_string({nil,2,nil,"foo",nil}), "<nil,2>")
end

function TestPrometheus:testKeyTimeout()
  self.counter_exp:inc(1)
  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get("metric_exp"), 1)
  local i = self.p.key_index.index["metric_exp"]
  luaunit.assertEquals(self.dict:get("__ngx_prom__key_" .. i), "metric_exp")
  luaunit.assertEquals(self.p.key_index.keys[i], "metric_exp")

  sleep(1)
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("metric_exp"), nil)
  luaunit.assertEquals(self.dict:get("__ngx_prom__key_" .. i), nil)
  -- the node is only past its ttl, so the slot is still this key's; the metric
  -- is absent from the output because metric_data() skips a missing value
  luaunit.assertEquals(self.p.key_index.index["metric_exp"], i)
  luaunit.assertEquals(self.dict:get("metric_exp"), nil)

  self.gauge_exp:inc(1)
  luaunit.assertEquals(self.dict:get("gauge_exp"), 1)
  i = self.p.key_index.index["gauge_exp"]
  luaunit.assertEquals(self.dict:get("__ngx_prom__key_" .. i), "gauge_exp")
  luaunit.assertEquals(self.p.key_index.keys[i], "gauge_exp")

  sleep(1)
  self.p.key_index:sync()
  luaunit.assertEquals(self.dict:get("gauge_exp"), nil)
  luaunit.assertEquals(self.dict:get("__ngx_prom__key_" .. i), nil)
  luaunit.assertEquals(self.p.key_index.index["gauge_exp"], i)
  -- the reference survives as long as the node does; the metric is out of the
  -- output because its value is gone, which is what metric_data() checks
  luaunit.assertEquals(self.p.key_index.keys[i], "gauge_exp")
  luaunit.assertEquals(self.dict:get("gauge_exp"), nil)

  self.gauge_exp_2:set(1)
  self.p.key_index:sync()
  self.p.key_index:remove_expired_keys()
  luaunit.assertEquals(self.dict:get("gauge_exp_2"), 1)
  i = self.p.key_index.index["gauge_exp_2"]
  luaunit.assertEquals(self.dict:get("__ngx_prom__key_" .. i), "gauge_exp_2")
  luaunit.assertEquals(self.p.key_index.keys[i], "gauge_exp_2")

  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()
  self.p.key_index:remove_expired_keys()
  luaunit.assertEquals(self.dict:get("gauge_exp_2"), nil)
  luaunit.assertEquals(self.dict:get("__ngx_prom__key_" .. i), nil)
  luaunit.assertEquals(self.p.key_index.index["gauge_exp_2"], nil)
  luaunit.assertEquals(self.p.key_index.keys[i], nil)

  self.hist_exp:observe(0.35)
  self.hist_exp:observe(0.4)
  self.p._counter:sync()
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="00.300"}'), nil)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="00.400"}'), 2)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="00.500"}'), 2)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="Inf"}'), 2)
  luaunit.assertEquals(self.dict:get('l_exp_count'), 2)
  luaunit.assertEquals(self.dict:get('l_exp_sum'), 0.75)

  sleep(1)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="00.300"}'), nil)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="00.400"}'), nil)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="00.500"}'), nil)
  luaunit.assertEquals(self.dict:get('l_exp_bucket{le="Inf"}'), nil)
  luaunit.assertEquals(self.dict:get('l_exp_count'), nil)
  luaunit.assertEquals(self.dict:get('l_exp_sum'), nil)

end

function TestPrometheus:testKeyTimeout2()
  self.counter_exp_2:inc(1)
  self.p._counter:sync()
  sleep(2)
  self.p.key_index:sync()
  self.p.key_index:remove_expired_keys()
  self.counter_exp_2:inc(1)
  self.p._counter:sync()
  local i = self.p.key_index.index["metric_exp2"]
  luaunit.assertEquals(self.p.key_index.keys[i], "metric_exp2")

end

-- Same as TestKeyIndex:testRemoveExpiredKeysReclaimsSharedDict, but driven
-- through the public metric API: once a counter with an exptime goes stale,
-- neither its value key nor its index slot may keep occupying the shared dict.
function TestPrometheus:testExpiredMetricIsReclaimedFromDict()
  self.counter_exp:inc(1)
  self.p._counter:sync()
  local i = self.p.key_index.index["metric_exp"]
  luaunit.assertNotNil(i)
  luaunit.assertEquals(self.dict:get("metric_exp"), 1)

  sleep(2)
  -- Expired, so invisible through the dict API, yet still holding its space.
  luaunit.assertNotNil(self.dict.dict["metric_exp"])
  luaunit.assertNotNil(self.dict.dict["__ngx_prom__key_" .. i])

  self.p.key_index:remove_expired_keys()

  luaunit.assertNil(self.dict.dict["metric_exp"])
  luaunit.assertNil(self.dict.dict["__ngx_prom__key_" .. i])
  -- Metrics registered without an exptime are untouched.
  luaunit.assertEquals(self.dict:get("nginx_metric_errors_total"), 0)
end

os.exit(luaunit.run())