fix: metric loss when a previously expired metric is re-added

.github/workflows/tests.yml

@@ -17,21 +17,18 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v2
-      - run: sudo apt install libpcre2-dev
+      - run: sudo apt install libpcre2-dev libreadline-dev
       - run: pip install hererocks
       # Install latest LuaRocks version plus the Lua version for this build job
       # into 'here' subdirectory.
       - run: hererocks here -r^ --lua ${{ matrix.lua }}
       - run: echo $PWD/here/bin >> $GITHUB_PATH
       - run: eval `luarocks path --bin`
       - run: luarocks install luacheck
-      - run: luarocks install luacov-coveralls
       - run: luarocks install luaunit
       - run: luarocks install lrexlib-pcre2
       - run: luacheck --globals ngx -- prometheus.lua prometheus_keys.lua prometheus_resty_counter.lua
       - run: luacheck --globals luaunit rex_pcre2 ngx TestPrometheus TestKeyIndex -- prometheus_test.lua
-      - run: lua -lluacov prometheus_test.lua
-      - run: luacov-coveralls --include ^prometheus
         env:
           COVERALLS_REPO_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         # Only report test coverage for lua 5.1 to avoid doing it twice.

.gitignore

@@ -1,3 +1,4 @@
 *.src.rock
 *.tar.gz
 nginx-lua-prometheus-0.*/**
+.idea

prometheus.lua

@@ -92,6 +92,9 @@ local DEFAULT_ERROR_METRIC_NAME = "nginx_metric_errors_total"
 -- Default value for per-worker counter sync interval (seconds).
 local DEFAULT_SYNC_INTERVAL = 1
 
+-- Max value for check and remove expired keys interval (seconds).
+local MAX_REMOVE_EXPIRED_KEYS_INTERVAL = 3600
+
 -- Default max size of lookup table
 local DEFAULT_LOOKUP_MAX_SIZE = 1000
 
@@ -412,6 +415,12 @@ local function lookup_or_create(self, label_values)
   local LEAF_KEY = mt -- key used to store full metric names in leaf tables.
   local full_name = t[LEAF_KEY]
   if full_name then
+    if self.exptime and self.exptime > 0 then
+      local err = self._key_index:add(full_name, ERR_MSG_LRU_EVICTION, self.exptime)
+      if err then
+        return nil, err
+      end
+    end
     return full_name
   end
 
@@ -733,15 +742,20 @@ function Prometheus.init(dict_name, options_or_prefix)
       DEFAULT_SYNC_INTERVAL
     self.lookup_max_size = options_or_prefix.lookup_max_size or
       DEFAULT_LOOKUP_MAX_SIZE
+    self.remove_expired_keys_interval = options_or_prefix.remove_expired_keys_interval
+        and options_or_prefix.remove_expired_keys_interval < MAX_REMOVE_EXPIRED_KEYS_INTERVAL
+        and options_or_prefix.remove_expired_keys_interval
+      or MAX_REMOVE_EXPIRED_KEYS_INTERVAL
   else
     self.prefix = options_or_prefix or ''
     self.error_metric_name = DEFAULT_ERROR_METRIC_NAME
     self.sync_interval = DEFAULT_SYNC_INTERVAL
     self.lookup_max_size = DEFAULT_LOOKUP_MAX_SIZE
+    self.remove_expired_keys_interval = MAX_REMOVE_EXPIRED_KEYS_INTERVAL
   end
 
   self.registry = {}
-  self.key_index = key_index_lib.new(self.dict, KEY_INDEX_PREFIX)
+  self.key_index = key_index_lib.new(self.dict, KEY_INDEX_PREFIX, self.remove_expired_keys_interval)
 
   self.initialized = true
 

prometheus_keys.lua

@@ -7,7 +7,14 @@
 local KeyIndex = {}
 KeyIndex.__index = KeyIndex
 
-function KeyIndex.new(shared_dict, prefix)
+
+-- check and remove expired keys
+local function remove_expired_keys(_, self)
+  self:remove_expired_keys()
+end
+
+
+function KeyIndex.new(shared_dict, prefix, remove_expired_keys_interval)
   local self = setmetatable({}, KeyIndex)
   self.dict = shared_dict
   self.key_prefix = prefix .. "key_"
@@ -18,9 +25,27 @@ function KeyIndex.new(shared_dict, prefix)
   self.not_expired_index = 1
   self.keys = {}
   self.index = {}
+  self.expire_keys = {}
+
+  ngx.timer.every(remove_expired_keys_interval or 600, remove_expired_keys, self)
   return self
 end
 
+-- check and remove expired keys
+function KeyIndex:remove_expired_keys()
+  for i, _ in pairs(self.expire_keys) do
+    -- Read i-th key. If it is nil or ttl is < 0, it means it was expired
+    local ttl, err = self.dict:ttl(self.key_prefix .. i)
+    if not (ttl and ttl >= 0 or err and err ~= "not found") then
+      if self.keys[i] then
+        self.index[self.keys[i]] = nil
+        self.keys[i] = nil
+      end
+      self.expire_keys[i] = nil
+    end
+  end
+end
+
 -- Loads new keys that might have been added by other workers since last sync.
 function KeyIndex:sync()
   local delete_count = self.dict:get(self.delete_count) or 0
@@ -33,7 +58,6 @@ function KeyIndex:sync()
     -- Sync only new keys, if there are any.
     self:sync_range(self.last, N)
   end
-  self:sync_expired(N)
   return N
 end
 
@@ -45,41 +69,23 @@ function KeyIndex:sync_range(first, last)
     if key then
       self.keys[i] = key
       self.index[key] = i
+
+      -- if it is nil and ttl not is 0, set expire_keys map
+      if not self.expire_keys[i] then
+        local ttl, _ = self.dict:ttl(self.key_prefix .. i)
+        if ttl and ttl ~= 0 then
+          self.expire_keys[i] = true
+        end
+      end
     elseif self.keys[i] then
       self.index[self.keys[i]] = nil
       self.keys[i] = nil
+      self.expire_keys[i] = nil
     end
   end
   self.last = last
 end
 
-function KeyIndex:sync_expired(N)
-  local first = self.not_expired_index
-  --- the key is sorted by created time, so the key will expire in order
-  for i = first, N do
-    self.not_expired_index = i
-    -- Read i-th key. If it is nil, it means it was expired
-    local ttl, err = self.dict:ttl(self.key_prefix .. i)
-    if ttl then
-      if ttl == 0 then
-        goto CONTINUE
-      else
-        break
-      end
-    else
-      if err ~= "not found" then
-        break
-      end
-      if self.keys[i] then
-        -- we don't need to update self.delete_count and self.key_count
-        self.index[self.keys[i]] = nil
-        self.keys[i] = nil
-      end
-    end
-    ::CONTINUE::
-  end
-end
-
 -- Returns array of all keys.
 function KeyIndex:list()
   self:sync()
@@ -95,10 +101,10 @@ end
 -- Atomically adds one or more keys to the index.
 --
 -- Args:
---   key_or_keys: Single string or a list of strings containing keys to add.
+-- key_or_keys: Single string or a list of strings containing keys to add.
 --
 -- Returns:
---   nil on success, string with error message otherwise
+-- nil on success, string with error message otherwise
 function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
   local keys = key_or_keys
   if type(key_or_keys) == "string" then
@@ -109,18 +115,35 @@ function KeyIndex:add(key_or_keys, err_msg_lru_eviction, exptime)
     while true do
       local N = self:sync()
       if self.index[key] ~= nil then
-        -- key already exists, we can skip it
-        break
+        -- key already exists, if has exptime, set expire
+        local expired = false
+        if exptime then
+          local ok = self.dict:expire(self.key_prefix .. self.index[key], exptime)
+          -- if key has already expired, remove it from the index
+          if not ok then
+            local idx = self.index[key]
+            self.index[key] = nil
+            self.keys[idx] = nil
+            self.expire_keys[idx] = nil
+            expired = true
+          end
+        end
+        if not expired then
+          break
+        end
       end
       N = N+1
       local ok, err, forcible = self.dict:add(self.key_prefix .. N, key, exptime)
       if ok then
         local _, _, forcible2 = self.dict:incr(self.key_count, 1, 0)
         self.keys[N] = key
         self.index[key] = N
+        if exptime and exptime > 0 then
+          self.expire_keys[N] = true
+        end
         if forcible or forcible2 then
           return (err_msg_lru_eviction .. "; key index: add key: idx=" ..
-            self.key_prefix .. N .. ", key=" .. key)
+                  self.key_prefix .. N .. ", key=" .. key)
         end
         break
       elseif err ~= "exists" then
@@ -133,12 +156,13 @@ end
 -- Removes a key based on its value.
 --
 -- Args:
---   key: String value of the key, must exists in this index.
+-- key: String value of the key, must exists in this index.
 function KeyIndex:remove(key, err_msg_lru_eviction)
   local i = self.index[key]
   if i then
     self.index[key] = nil
     self.keys[i] = nil
+    self.expire_keys[i] = nil
     self.dict:set(self.key_prefix .. i, nil)
     self.deleted = self.deleted + 1
 
@@ -152,4 +176,4 @@ function KeyIndex:remove(key, err_msg_lru_eviction)
   end
 end
 
-return KeyIndex
+return KeyIndex

prometheus_test.lua

@@ -145,6 +145,8 @@ function TestPrometheus:setUp()
   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
@@ -177,6 +179,10 @@ function TestPrometheus:testInitOptions()
   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
@@ -738,7 +744,7 @@ TestKeyIndex = {}
 function TestKeyIndex:setUp()
   self.dict = setmetatable({}, SimpleDict)
   ngx.shared.metrics = self.dict
-  self.key_index = require('prometheus_keys').new(self.dict, '_prefix_')
+  self.key_index = require('prometheus_keys').new(self.dict, "_prefix_", 1)
 end
 function TestKeyIndex.tearDown()
   ngx.logs = nil
@@ -915,13 +921,15 @@ function TestPrometheus:testKeyTimeout()
 
   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()
+  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)
@@ -947,4 +955,17 @@ function TestPrometheus:testKeyTimeout()
 
 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
+
 os.exit(luaunit.run())