feat(prometheus) add wasmx metrics

.requirements

@@ -24,9 +24,9 @@ ATC_ROUTER=ffd11db657115769bf94f0c4f915f98300bc26b6 # 1.6.2
 SNAPPY=23b3286820105438c5dbb9bc22f1bb85c5812c8a # 1.2.0
 
 KONG_MANAGER=nightly
-NGX_WASM_MODULE=96b4e27e10c63b07ed40ea88a91c22f23981db35
+NGX_WASM_MODULE=9b1b2c760f73827fc08ade3a936a89fa5473f8fa
 WASMER=3.1.1
-WASMTIME=23.0.2
+WASMTIME=25.0.1
 V8=12.0.267.17
 
 NGX_BROTLI=a71f9312c2deb28875acc7bacfdd5695a111aa53 # master branch of Oct 9, 2023

build/openresty/wasmx/wasmx_repositories.bzl

@@ -42,12 +42,12 @@ wasm_runtimes = {
     },
     "wasmtime": {
         "linux": {
-            "x86_64": "c2fe82f4d707711523e57c2fc8f67d8fc0311fd3cf15050f811f88b30c254980",
-            "aarch64": "4593a131018a99df3aa16b41b1c63838cbbba9a36771c444a39761b25be73469",
+            "x86_64": "5c4c490bbc8ddee6311653dd5c361933202b69e12eaddfe6b3aed371c97b6b4a",
+            "aarch64": "a189e01ef73a5c3c0bfcbc1a26dcc31f5b1904fcbdf344f761cfb19e8ecfd501",
         },
         "macos": {
-            "x86_64": "2939cdf4eca5ce79c7e179c338c46700deb88bc7906da206a272143c3da0ca5b",
-            "aarch64": "cafff668144d15fdee57645918d06330aa05126b6a28b92b836eb69987842cd9",
+            "x86_64": "6d81ab0775ec900285ee1140555ba09a953669324d9317a8bb1fe0572684dbfb",
+            "aarch64": "61b15351c136aad75735eadf42f6101acb42480d6419efef4dbdd81ddb4dd180",
         },
     },
 }

changelog/unreleased/kong/prometheus-wasmx-metrics.yml

@@ -0,0 +1,4 @@
+message: |
+  **Prometheus**: Added support for Proxy-Wasm metrics.
+type: feature
+scope: Plugin

kong-3.9.0-0.rockspec

@@ -533,6 +533,7 @@ build = {
     ["kong.plugins.prometheus.prometheus"] = "kong/plugins/prometheus/prometheus.lua",
     ["kong.plugins.prometheus.serve"] = "kong/plugins/prometheus/serve.lua",
     ["kong.plugins.prometheus.schema"] = "kong/plugins/prometheus/schema.lua",
+    ["kong.plugins.prometheus.wasmx"] = "kong/plugins/prometheus/wasmx.lua",
 
     ["kong.plugins.session.handler"] = "kong/plugins/session/handler.lua",
     ["kong.plugins.session.schema"] = "kong/plugins/session/schema.lua",

kong/plugins/prometheus/exporter.lua

@@ -1,11 +1,14 @@
+local balancer = require "kong.runloop.balancer"
+local yield = require("kong.tools.yield").yield
+local wasm = require "kong.plugins.prometheus.wasmx"
+
+
 local kong = kong
 local ngx = ngx
 local get_phase = ngx.get_phase
 local lower = string.lower
 local ngx_timer_pending_count = ngx.timer.pending_count
 local ngx_timer_running_count = ngx.timer.running_count
-local balancer = require("kong.runloop.balancer")
-local yield = require("kong.tools.yield").yield
 local get_all_upstreams = balancer.get_all_upstreams
 if not balancer.get_all_upstreams then -- API changed since after Kong 2.5
   get_all_upstreams = require("kong.runloop.balancer.upstreams").get_all_upstreams
@@ -517,6 +520,7 @@ local function metric_data(write_fn)
   -- notify the function if prometheus plugin is enabled,
   -- so that it can avoid exporting unnecessary metrics if not
   prometheus:metric_data(write_fn, not IS_PROMETHEUS_ENABLED)
+  wasm.metric_data()
 end
 
 local function collect()

kong/plugins/prometheus/wasmx.lua

@@ -0,0 +1,206 @@
+local buffer = require "string.buffer"
+local wasm = require "kong.runloop.wasm"
+local wasmx_shm
+
+
+local fmt = string.format
+local str_find = string.find
+local str_match = string.match
+local str_sub = string.sub
+local table_insert = table.insert
+local table_sort = table.sort
+local buf_new = buffer.new
+local ngx_say = ngx.say
+
+
+local _M = {}
+
+
+local FLUSH_EVERY = 100
+
+
+local function sorted_iter(ctx)
+  local v = ctx.t[ctx.sorted_keys[ctx.i]]
+  ctx.i = ctx.i + 1
+
+  return v
+end
+
+
+local function sorted_pairs(t)
+  local sorted_keys = {}
+
+  for k, _ in pairs(t) do
+    table_insert(sorted_keys, k)
+  end
+
+  table_sort(sorted_keys)
+
+  return sorted_iter, { t = t, sorted_keys = sorted_keys, i = 1 }
+end
+
+
+local function parse_pw_key(key)
+  local name = key
+  local labels = {}
+  local header_size = 3  -- pw.
+  local first_label = #key
+
+  local second_dot_pos, _ = str_find(key, "%.", header_size + 1)
+  local filter_name = str_sub(key, header_size + 1, second_dot_pos - 1)
+
+  local filter_config = wasm.filters_by_name[filter_name].config or {}
+  local patterns = filter_config.pw_metrics
+                   and filter_config.pw_metrics.label_patterns or {}
+
+  for _, pair in ipairs(patterns) do
+    local label_kv, label_v = str_match(key, pair.pattern)
+    if label_kv then
+      local label_k = str_sub(label_kv, 0, str_find(label_kv, "="))
+      local label_k_start, _ = str_find(key, label_k)
+
+      first_label = (label_k_start < first_label) and label_k_start or first_label
+
+      table_insert(labels, { pair.label, label_v })
+    end
+  end
+
+  if first_label ~= #key then
+    name = str_sub(key, 0, first_label - 1)
+  end
+
+  return name, labels
+end
+
+
+local function parse_key(key)
+  -- TODO: parse wa. (WasmX metrics) and lua. (metrics defined in Lua land)
+  local header = { pw = "pw." }
+
+  local name = key
+  local labels = {}
+
+  local is_pw = #key > #header.pw and key:sub(0, #header.pw) == header.pw
+
+  if is_pw then
+    name, labels = parse_pw_key(key)
+  end
+
+  name = name:gsub("%.", "_")
+
+  return name, labels
+end
+
+
+local function serialize_labels(labels)
+  local buf = buf_new()
+
+  for _, pair in ipairs(labels) do
+    buf:put(fmt(',%s="%s"', pair[1], pair[2]))
+  end
+
+  buf:get(1)  -- discard trailing comma
+
+  return "{" .. buf:get() .. "}"
+end
+
+
+local function serialize_metric(m, buf)
+  buf:put(fmt("# HELP %s\n# TYPE %s %s", m.name, m.name, m.type))
+
+  if m.type == "histogram" then
+    local sum_lines_buf = buf_new()
+    local count_lines_buf = buf_new()
+
+    for _, pair in ipairs(m.labels) do
+      local count, sum = 0, 0
+      local labels, labeled_m = pair[1], pair[2]
+      local slabels = (#labels > 0) and serialize_labels(labels) or ""
+
+      local blabels = (#labels > 0) and (slabels:sub(1, #slabels - 1) .. ",") or "{"
+
+      for _, bin in ipairs(labeled_m.value) do
+        local ub = (bin.ub ~= 4294967295) and bin.ub or "+Inf"
+        local ubl = fmt('le="%s"', ub)
+
+        count = count + bin.count
+
+        buf:put(fmt("\n%s%s %s", m.name, blabels .. ubl .. "}", count))
+      end
+
+      sum = sum + labeled_m.sum
+
+      sum_lines_buf:put(fmt("\n%s_sum%s %s", m.name, slabels, sum))
+      count_lines_buf:put(fmt("\n%s_count%s %s", m.name, slabels, count))
+    end
+
+    buf:put(sum_lines_buf:get())
+    buf:put(count_lines_buf:get())
+
+  else
+    for _, pair in ipairs(m.labels) do
+      local labels, labeled_m = pair[1], pair[2]
+      local slabels = (#labels > 0) and serialize_labels(labels) or ""
+
+      buf:put(fmt("\n%s%s %s", m.name, slabels, labeled_m.value))
+    end
+  end
+
+  buf:put("\n")
+end
+
+
+_M.metric_data = function()
+  local i = 0
+  local metrics = {}
+  local parsed = {}
+  local buf = buf_new()
+
+  -- delayed require of the WasmX module, to ensure it is loaded
+  -- after ngx_wasm_module.so is loaded.
+  if not wasmx_shm then
+    local ok, _wasmx_shm = pcall(require, "resty.wasmx.shm")
+    if ok then
+      wasmx_shm = _wasmx_shm
+    end
+  end
+
+  if not wasmx_shm then
+    return
+  end
+
+  wasmx_shm.metrics:lock()
+
+  for key in wasmx_shm.metrics:iterate_keys() do
+    table_insert(metrics, { key, wasmx_shm.metrics:get_by_name(key, { prefix = false })})
+  end
+
+  wasmx_shm.metrics:unlock()
+
+  -- in WasmX the different labels of a metric are stored as separate metrics;
+  -- aggregate those separate metrics into a single one.
+  for _, pair in ipairs(metrics) do
+    local key = pair[1]
+    local m = pair[2]
+    local name, labels = parse_key(key)
+
+    parsed[name] = parsed[name] or { name = name, type = m.type, labels = {} }
+
+    table_insert(parsed[name].labels, { labels, m })
+  end
+
+  for metric_by_label in sorted_pairs(parsed) do
+    buf:put(serialize_metric(metric_by_label, buf))
+
+    i = i + 1
+
+    if i % FLUSH_EVERY == 0 then
+      ngx_say(buf:get())
+    end
+  end
+
+  ngx_say(buf:get())
+end
+
+
+return _M

kong/runloop/wasm.lua

@@ -402,6 +402,8 @@ local function rebuild_state(db, version, old_state)
 
       for _, filter in ipairs(chain.filters) do
         if filter.enabled then
+          _M.filters_by_name[filter.name].config = cjson_decode(filter.config) or filter.config
+
           -- Serialize all JSON configurations up front
           --
           -- NOTE: there is a subtle difference between a raw, non-JSON filter
@@ -778,6 +780,13 @@ local function register_property_handlers()
     return ok, value, const
   end)
 
+  properties.add_getter("kong.route_name", function(_, _, ctx)
+    local value = ctx.route and ctx.route.name
+    local ok = value ~= nil
+    local const = ok
+    return ok, value, const
+  end)
+
   properties.add_getter("kong.service.response.status", function(kong)
     return true, kong.service.response.get_status(), false
   end)
@@ -789,6 +798,13 @@ local function register_property_handlers()
     return ok, value, const
   end)
 
+  properties.add_getter("kong.service_name", function(_, _, ctx)
+    local value = ctx.service and ctx.service.name
+    local ok = value ~= nil
+    local const = ok
+    return ok, value, const
+  end)
+
   properties.add_getter("kong.version", function(kong)
     return true, kong.version, true
   end)

spec/02-integration/20-wasm/04-proxy-wasm_spec.lua

@@ -38,6 +38,7 @@ describe("proxy-wasm filters (#wasm) (#" .. strategy .. ")", function()
     mock_service = assert(bp.services:insert {
       host = helpers.mock_upstream_host,
       port = helpers.mock_upstream_port,
+      name = "mock_service",
     })
 
     local mock_upstream = assert(bp.upstreams:insert {
@@ -50,12 +51,14 @@ describe("proxy-wasm filters (#wasm) (#" .. strategy .. ")", function()
     })
 
     r_single = assert(bp.routes:insert {
+      name = "r_single",
       paths = { "/single" },
       strip_path = true,
       service = mock_service,
     })
 
     local r_double = assert(bp.routes:insert {
+      name = "r_double",
       paths = { "/double" },
       strip_path = true,
       service = mock_service,
@@ -687,6 +690,26 @@ describe("proxy-wasm filters (#wasm) (#" .. strategy .. ")", function()
       assert.logfile().has.no.line("[crit]",  true, 0)
     end)
 
+    it("read kong.route_name", function()
+      local client = helpers.proxy_client()
+      finally(function() client:close() end)
+
+      local res = assert(client:send {
+        method = "GET",
+        path = "/single/status/201",
+        headers = {
+          [HEADER_NAME_TEST] = "get_kong_property",
+          [HEADER_NAME_INPUT] = "route_name",
+          [HEADER_NAME_DISPATCH_ECHO] = "on",
+        }
+      })
+
+      local body = assert.res_status(200, res)
+      assert.equal(r_single.name, body)
+      assert.logfile().has.no.line("[error]", true, 0)
+      assert.logfile().has.no.line("[crit]",  true, 0)
+    end)
+
     it("read kong.service_id", function()
       local client = helpers.proxy_client()
       finally(function() client:close() end)
@@ -707,6 +730,26 @@ describe("proxy-wasm filters (#wasm) (#" .. strategy .. ")", function()
       assert.logfile().has.no.line("[crit]",  true, 0)
     end)
 
+    it("read kong.service_name", function()
+      local client = helpers.proxy_client()
+      finally(function() client:close() end)
+
+      local res = assert(client:send {
+        method = "GET",
+        path = "/single/status/201",
+        headers = {
+          [HEADER_NAME_TEST] = "get_kong_property",
+          [HEADER_NAME_INPUT] = "service_name",
+          [HEADER_NAME_DISPATCH_ECHO] = "on",
+        }
+      })
+
+      local body = assert.res_status(200, res)
+      assert.equal(mock_service.name, body)
+      assert.logfile().has.no.line("[error]", true, 0)
+      assert.logfile().has.no.line("[crit]",  true, 0)
+    end)
+
     it("read kong.ctx.shared[<attr>]", function()
       local client = helpers.proxy_client()
       finally(function() client:close() end)