fix(keys): keep a returning metric on its slot, and stop broadcasting

CHANGELOG.md

@@ -4,6 +4,12 @@ This file only calls out major changes. Please see [the list of Git commits](
 https://github.com/api7/nginx-lua-prometheus/commits/main) for the full list
 of changes.
 
+## Unreleased
+
+- Reuse expired index slots while their dictionary nodes still exist, avoid
+  expiry-driven broadcasts, renew known slots without synchronizing the full
+  index, and detect a re-created `key_count` (#25).
+
 ## 1.0.0
 
 Starting with this release, this fork (`nginx-lua-prometheus-api7`) uses

integration/README.md

@@ -6,3 +6,8 @@ requests to nginx in a loop for a predefined amount of time. After all requests
 are sent, the test collects metrics and compares request counters with the
 total number of requests sent by clients. A few other metric checks are
 performed as well.
+
+Before generating traffic, the test runs shared-dictionary regressions for
+stale-slot discovery, renewal races, alias reclamation, and the public gauge
+API. The renewal test inserts a reader between the writer's dictionary read and
+renewal to reproduce a process interleaving.

integration/index.lua

@@ -0,0 +1,90 @@
+local KeyIndex = require('prometheus_keys')
+local d = ngx.shared.index_metrics
+
+local function new(prefix)
+  return KeyIndex.new(d, prefix, 3600)
+end
+
+-- A fresh reader must remember a stale node before a writer revives it.
+local writer = new("reader_")
+assert(not writer:add("target", "eviction", 0.01))
+assert(not writer:add("tail", "eviction"))
+ngx.sleep(0.02)
+local reader = new("reader_")
+reader:sync()
+assert(not writer:add("target", "eviction", 60))
+assert(reader.index.target == writer.index.target, "revived slot was not discovered")
+assert(d:get("reader_key_count") == 2, "renewal allocated an unnecessary slot")
+
+-- Expiration can occur between the fast-path read and the renewal.
+writer = new("boundary_")
+assert(not writer:add("target", "eviction", 60))
+assert(not writer:add("tail", "eviction"))
+reader = new("boundary_")
+local injected = false
+writer.dict = setmetatable({get = function(_, slot)
+  local value = d:get(slot)
+  if not injected and slot == "boundary_key_1" then
+    injected = true
+    assert(d:expire(slot, 0.001))
+    ngx.sleep(0.01)
+    reader:sync()
+    assert(d:get(slot) == nil, "slot did not expire")
+  end
+  return value
+end}, {__index = function(_, method)
+  return function(_, ...) return d[method](d, ...) end
+end})
+assert(not writer:add("target", "eviction", 60))
+assert(injected and reader.index.target == 1, "renewal race hid the metric")
+
+-- Concurrent registrations can leave a live low slot and a stale high alias.
+assert(d:set("alias_key_1", "target", 60))
+assert(d:set("alias_key_2", "target", 0.01))
+assert(d:set("alias_key_count", 2))
+ngx.sleep(0.02)
+reader = new("alias_")
+reader:sync()
+d:flush_expired()
+reader:remove_expired_keys()
+assert(reader.index.target == 1, "reclaimed alias hid the live slot")
+assert(#reader:list() == 1, "duplicate or missing metric")
+assert(not reader:add("target", "eviction", 60))
+assert(d:get("alias_key_count") == 2, "alias recovery allocated another slot")
+reader:remove("target", "eviction")
+assert(#reader:list() == 0, "deleted metric retained an alias")
+
+-- A reused alias belongs to its new writer, not to the deleting reader.
+assert(d:set("owner_key_1", "target", 60))
+assert(d:set("owner_key_2", "target", 60))
+assert(d:set("owner_key_count", 2))
+reader = new("owner_")
+reader:sync()
+d:delete("owner_key_1")
+d:delete("owner_key_count")
+local replacement = new("owner_")
+assert(not replacement:add("replacement", "eviction", 60))
+assert(replacement.index.replacement == 1)
+reader:remove("target", "eviction")
+assert(d:get("owner_key_1") == "replacement", "deleted another slot owner")
+assert(d:get_stale("owner_key_2") == nil, "target alias remained")
+
+-- Exercise the public metric API and the explicit-deletion broadcast.
+local p = index_writer
+local peer = index_reader
+local target = p:gauge("target", "Returning metric", nil, 0.01)
+local removable = p:gauge("removable", "Deleted metric")
+local tail = p:gauge("tail", "Permanent metric")
+target:set(1)
+removable:set(1)
+tail:set(1)
+ngx.sleep(0.02)
+peer:gauge("target", "Returning metric", nil, 0.01)
+local peer_removable = peer:gauge("removable", "Deleted metric")
+peer:gauge("tail", "Permanent metric")
+assert(not table.concat(peer:metric_data()):find("\ntarget 1\n", 1, true))
+peer_removable:del()
+target:set(42)
+assert(table.concat(peer:metric_data()):find("\ntarget 42\n", 1, true),
+       "public API omitted the returning metric")
+ngx.say("ok")

integration/nginx.conf

@@ -10,13 +10,16 @@ http {
     tcp_nodelay on;
 
     lua_shared_dict prometheus_metrics 10M;
+    lua_shared_dict index_metrics 1M;
     lua_package_path "/nginx-lua-prometheus/?.lua;;";
 
     error_log stderr;
 
     init_worker_by_lua_block {
         prometheus = require("prometheus").init("prometheus_metrics",
 	  {sync_interval=0.4})
+        index_writer = require("prometheus").init("index_metrics")
+        index_reader = require("prometheus").init("index_metrics")
         metric_requests = prometheus:counter("requests_total",
           "Number of HTTP requests", {"host", "status"})
         metric_latency = prometheus:histogram("request_duration_seconds",
@@ -37,6 +40,10 @@ http {
         root /nginx-lua-prometheus/integration/www;
         index index.txt;
 
+        location = /test_index {
+            log_by_lua_block { return }
+            content_by_lua_file /nginx-lua-prometheus/integration/index.lua;
+        }
         location /error {
             return 500;
         }

integration/test.go

@@ -112,6 +112,18 @@ func checkBucketBoundaries(mfs map[string]*dto.MetricFamily, metric string) erro
 func main() {
 	flag.Parse()
 
+	indexClient := &http.Client{Timeout: 10 * time.Second}
+	indexResp, err := indexClient.Get("http://localhost:18001/test_index")
+	if err != nil {
+		log.Fatalf("Key index regression test failed: %v", err)
+	}
+	if err := indexResp.Body.Close(); err != nil {
+		log.Fatalf("Could not close the key index regression response: %v", err)
+	}
+	if indexResp.StatusCode != http.StatusOK {
+		log.Fatalf("Key index regression test returned HTTP %d", indexResp.StatusCode)
+	}
+
 	// Use a custom http client with a lower idle connection timeout and a request timeout.
 	client := &http.Client{
 		Timeout:   500 * time.Millisecond,