mirror of
https://github.com/OptimalBits/bull.git
synced 2026-07-02 08:27:43 +08:00
perf(metrics): save zeros as much as max data points (#2767)
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
--[[
|
||||
Functions to collect metrics based on a current and previous count of jobs.
|
||||
Granualarity is fixed at 1 minute.
|
||||
]]
|
||||
|
||||
-- Includes
|
||||
--- @include "batches"
|
||||
|
||||
local function collectMetrics(metaKey, dataPointsList, maxDataPoints, timestamp)
|
||||
-- Increment current count
|
||||
local count = rcall("HINCRBY", metaKey, "count", 1) - 1
|
||||
|
||||
-- Compute how many data points we need to add to the list, N.
|
||||
local prevTS = rcall("HGET", metaKey, "prevTS")
|
||||
|
||||
if not prevTS then
|
||||
-- If prevTS is nil, set it to the current timestamp
|
||||
rcall("HSET", metaKey, "prevTS", timestamp, "prevCount", 0)
|
||||
return
|
||||
end
|
||||
|
||||
local N = math.min(math.floor((timestamp - prevTS) / 60000), tonumber(maxDataPoints))
|
||||
|
||||
if N > 0 then
|
||||
local delta = count - rcall("HGET", metaKey, "prevCount")
|
||||
-- If N > 1, add N-1 zeros to the list
|
||||
if N > 1 then
|
||||
local points = {}
|
||||
points[1] = delta
|
||||
for i = 2, N do points[i] = 0 end
|
||||
|
||||
for from, to in batches(#points, 7000) do
|
||||
rcall("LPUSH", dataPointsList, unpack(points, from, to))
|
||||
end
|
||||
else
|
||||
-- LPUSH delta to the list
|
||||
rcall("LPUSH", dataPointsList, delta)
|
||||
end
|
||||
|
||||
-- LTRIM to keep list to its max size
|
||||
rcall("LTRIM", dataPointsList, 0, maxDataPoints - 1)
|
||||
|
||||
-- update prev count with current count
|
||||
rcall("HSET", metaKey, "prevCount", count, "prevTS", timestamp)
|
||||
end
|
||||
end
|
||||
@@ -43,52 +43,9 @@
|
||||
local rcall = redis.call
|
||||
|
||||
-- Includes
|
||||
--- @include "includes/collectMetrics"
|
||||
--- @include "includes/removeLock"
|
||||
--- @include "includes/removeDebounceKeyIfNeeded"
|
||||
--- @include "includes/batches"
|
||||
|
||||
--[[
|
||||
Functions to collect metrics based on a current and previous count of jobs.
|
||||
Granualarity is fixed at 1 minute.
|
||||
]]
|
||||
local function collectMetrics(metaKey, dataPointsList, maxDataPoints, timestamp)
|
||||
-- Increment current count
|
||||
local count = rcall("HINCRBY", metaKey, "count", 1) - 1
|
||||
|
||||
-- Compute how many data points we need to add to the list, N.
|
||||
local prevTS = rcall("HGET", metaKey, "prevTS")
|
||||
|
||||
if not prevTS then
|
||||
-- If prevTS is nil, set it to the current timestamp
|
||||
rcall("HSET", metaKey, "prevTS", timestamp, "prevCount", 0)
|
||||
return
|
||||
end
|
||||
|
||||
local N = math.floor((timestamp - prevTS) / 60000)
|
||||
|
||||
if N > 0 then
|
||||
local delta = count - rcall("HGET", metaKey, "prevCount")
|
||||
-- If N > 1, add N-1 zeros to the list
|
||||
if N > 1 then
|
||||
local points = {}
|
||||
points[1] = delta
|
||||
for i = 2, N do points[i] = 0 end
|
||||
|
||||
for from, to in batches(#points, 7000) do
|
||||
rcall("LPUSH", dataPointsList, unpack(points, from, to))
|
||||
end
|
||||
else
|
||||
-- LPUSH delta to the list
|
||||
rcall("LPUSH", dataPointsList, delta)
|
||||
end
|
||||
|
||||
-- LTRIM to keep list to its max size
|
||||
rcall("LTRIM", dataPointsList, 0, maxDataPoints - 1)
|
||||
|
||||
-- update prev count with current count
|
||||
rcall("HSET", metaKey, "prevCount", count, "prevTS", timestamp)
|
||||
end
|
||||
end
|
||||
|
||||
if rcall("EXISTS", KEYS[3]) == 1 then -- // Make sure job exists
|
||||
local errorCode = removeLock(KEYS[3], KEYS[8], ARGV[5], ARGV[1])
|
||||
|
||||
Reference in New Issue
Block a user