348 lines
11 KiB
JavaScript
348 lines
11 KiB
JavaScript
/**
|
|
* In-memory tiered metric ring buffers.
|
|
*
|
|
* Tier 0: high-res (1s) short retention (hot path)
|
|
* Tier 1: downsampled averages — also flushed to HyperDB warm storage
|
|
*
|
|
* See docs/STORAGE-HYPERDB.md
|
|
*/
|
|
import { EventEmitter } from 'events'
|
|
import { SAMPLE_INTERVAL_MS, CHART_BY_ID, chartSummary } from '../../shared/metrics.js'
|
|
import { getDb } from '../db/index.js'
|
|
import { queryRemoteMetricPoints, listRemoteDbs } from '../db/remote.js'
|
|
|
|
function envInt(name, fallback) {
|
|
const n = Number(process.env[name])
|
|
return Number.isFinite(n) && n > 0 ? n : fallback
|
|
}
|
|
|
|
export class MetricStore extends EventEmitter {
|
|
constructor() {
|
|
super()
|
|
this.tier0Max = envInt('PEARDATA_TIER0_POINTS', 3600) // 1h @ 1s
|
|
this.tier1Max = envInt('PEARDATA_TIER1_POINTS', 1440) // 24h @ 1m
|
|
this.tier1Every = envInt('PEARDATA_TIER1_EVERY', 60) // downsample every N samples
|
|
/** @type {Map<string, { points: Array<{ts:number, values: Record<string, number|null>}>, tier1: Array<{ts:number, values: Record<string, number|null>}>, acc: object|null, accCount: number }>} */
|
|
this.series = new Map()
|
|
}
|
|
|
|
/**
|
|
* Live-update retention knobs (from Data Manager) and trim rings.
|
|
* @param {{ tier0Max?: number, tier1Max?: number, tier1Every?: number }} opts
|
|
*/
|
|
setRetention(opts = {}) {
|
|
if (opts.tier0Max != null && Number.isFinite(Number(opts.tier0Max)) && opts.tier0Max > 0) {
|
|
this.tier0Max = Math.floor(Number(opts.tier0Max))
|
|
}
|
|
if (opts.tier1Max != null && Number.isFinite(Number(opts.tier1Max)) && opts.tier1Max > 0) {
|
|
this.tier1Max = Math.floor(Number(opts.tier1Max))
|
|
}
|
|
if (opts.tier1Every != null && Number.isFinite(Number(opts.tier1Every)) && opts.tier1Every > 0) {
|
|
this.tier1Every = Math.floor(Number(opts.tier1Every))
|
|
}
|
|
this.trimToRetention()
|
|
return {
|
|
tier0Max: this.tier0Max,
|
|
tier1Max: this.tier1Max,
|
|
tier1Every: this.tier1Every,
|
|
}
|
|
}
|
|
|
|
/** Trim all series to current tier maxima. */
|
|
trimToRetention() {
|
|
for (const entry of this.series.values()) {
|
|
if (entry.points.length > this.tier0Max) {
|
|
entry.points.splice(0, entry.points.length - this.tier0Max)
|
|
}
|
|
if (entry.tier1.length > this.tier1Max) {
|
|
entry.tier1.splice(0, entry.tier1.length - this.tier1Max)
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Approximate in-memory footprint for Data Manager. */
|
|
memoryStats() {
|
|
let tier0Points = 0
|
|
let tier1Points = 0
|
|
let charts = 0
|
|
let oldestTs = null
|
|
let newestTs = null
|
|
for (const entry of this.series.values()) {
|
|
charts++
|
|
tier0Points += entry.points.length
|
|
tier1Points += entry.tier1.length
|
|
const first = entry.points[0]?.ts
|
|
const last = entry.points[entry.points.length - 1]?.ts
|
|
if (first != null && (oldestTs == null || first < oldestTs)) oldestTs = first
|
|
if (last != null && (newestTs == null || last > newestTs)) newestTs = last
|
|
}
|
|
// Rough: ~48 bytes overhead + ~8 per numeric dim (assume ~6 dims) + JSON-ish
|
|
const approxBytes = charts * 256 + (tier0Points + tier1Points) * 96
|
|
return {
|
|
charts,
|
|
tier0Points,
|
|
tier1Points,
|
|
tier0Max: this.tier0Max,
|
|
tier1Max: this.tier1Max,
|
|
tier1Every: this.tier1Every,
|
|
approxBytes,
|
|
oldestTs,
|
|
newestTs,
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {Array<{ chart: string, context: string, ts: number, values: Record<string, number|null> }>} batch
|
|
*/
|
|
ingest(batch) {
|
|
for (const s of batch) {
|
|
let entry = this.series.get(s.chart)
|
|
if (!entry) {
|
|
entry = { points: [], tier1: [], acc: null, accCount: 0 }
|
|
this.series.set(s.chart, entry)
|
|
}
|
|
entry.points.push({ ts: s.ts, values: s.values })
|
|
if (entry.points.length > this.tier0Max + 60) {
|
|
entry.points.splice(0, entry.points.length - this.tier0Max)
|
|
}
|
|
|
|
// accumulate for tier1
|
|
if (!entry.acc) {
|
|
entry.acc = { ...s.values }
|
|
entry.accCount = 1
|
|
} else {
|
|
for (const [k, v] of Object.entries(s.values)) {
|
|
if (v == null || Number.isNaN(v)) continue
|
|
entry.acc[k] = (entry.acc[k] || 0) + v
|
|
}
|
|
entry.accCount++
|
|
}
|
|
if (entry.accCount >= this.tier1Every) {
|
|
/** @type {Record<string, number|null>} */
|
|
const avg = {}
|
|
for (const [k, v] of Object.entries(entry.acc)) {
|
|
avg[k] = entry.accCount ? v / entry.accCount : null
|
|
}
|
|
const warm = { ts: s.ts, values: avg }
|
|
entry.tier1.push(warm)
|
|
if (entry.tier1.length > this.tier1Max + 30) {
|
|
entry.tier1.splice(0, entry.tier1.length - this.tier1Max)
|
|
}
|
|
entry.acc = null
|
|
entry.accCount = 0
|
|
this.emit('warm', {
|
|
chart: s.chart,
|
|
context: s.context,
|
|
ts: warm.ts,
|
|
values: warm.values,
|
|
tier: 1,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {string} chart
|
|
*/
|
|
getMeta(chart) {
|
|
const def = CHART_BY_ID.get(chart)
|
|
const entry = this.series.get(chart)
|
|
const first = entry?.points[0]?.ts
|
|
const last = entry?.points[entry.points.length - 1]?.ts
|
|
if (!def) return null
|
|
return chartSummary(def, {
|
|
firstEntry: first ? Math.floor(first / 1000) : 0,
|
|
lastEntry: last ? Math.floor(last / 1000) : 0,
|
|
updateEvery: SAMPLE_INTERVAL_MS / 1000,
|
|
})
|
|
}
|
|
|
|
listChartSummaries() {
|
|
/** @type {Record<string, any>} */
|
|
const charts = {}
|
|
for (const id of CHART_BY_ID.keys()) {
|
|
const meta = this.getMeta(id)
|
|
if (meta) charts[id] = meta
|
|
}
|
|
return charts
|
|
}
|
|
|
|
/**
|
|
* Query points for a chart (after/before/points windowing).
|
|
* Memory first; optional HyperDB warm fallback when window exceeds hot buffer.
|
|
*
|
|
* @param {{ chart: string, after?: number, before?: number, points?: number, group?: string, tier?: number }} opts
|
|
*/
|
|
async query(opts) {
|
|
const chart = opts.chart
|
|
const entry = this.series.get(chart)
|
|
const def = CHART_BY_ID.get(chart)
|
|
if (!def) {
|
|
return { error: 'unknown chart', chart }
|
|
}
|
|
|
|
const useTier1 = opts.tier === 1
|
|
const src = entry ? (useTier1 ? entry.tier1 : entry.points) : []
|
|
const nowSec = Math.floor(Date.now() / 1000)
|
|
let before = opts.before == null || opts.before === 0 ? nowSec : Number(opts.before)
|
|
let after = opts.after == null ? -Math.min(opts.points || 60, src.length || 60) : Number(opts.after)
|
|
|
|
if (before <= 0) before = nowSec + before
|
|
if (after <= 0) after = before + after // relative seconds
|
|
|
|
const afterMs = after * 1000
|
|
const beforeMs = before * 1000
|
|
|
|
let windowed = src.filter((p) => p.ts >= afterMs && p.ts <= beforeMs)
|
|
let source = useTier1 ? 'memory-tier1' : 'memory-tier0'
|
|
|
|
const oldestMem = src.length ? src[0].ts : null
|
|
const windowExceedsMemory =
|
|
oldestMem != null && afterMs < oldestMem - 1000
|
|
const sparse =
|
|
!windowed.length ||
|
|
(opts.tier >= 1 && windowed.length < (opts.points || 60) / 2) ||
|
|
windowExceedsMemory
|
|
|
|
// HyperDB warm fallback when memory misses, is sparse, or cannot cover the after window
|
|
if (sparse) {
|
|
const db = getDb()
|
|
if (db) {
|
|
try {
|
|
const warm = await db.queryMetricPoints({
|
|
chart,
|
|
afterMs,
|
|
beforeMs,
|
|
limit: Math.max(opts.points || 60, 10_000),
|
|
tier: 1,
|
|
})
|
|
if (warm.length) {
|
|
if (!windowed.length || warm.length >= windowed.length || windowExceedsMemory) {
|
|
windowed = warm
|
|
source = 'hyperdb-warm'
|
|
}
|
|
}
|
|
} catch {
|
|
// keep memory result
|
|
}
|
|
}
|
|
// Linked peer warm pull (replicated Corestore / remote bee)
|
|
if ((!windowed.length || source !== 'hyperdb-warm') && listRemoteDbs().length) {
|
|
try {
|
|
const remote = await queryRemoteMetricPoints({
|
|
chart,
|
|
afterMs,
|
|
beforeMs,
|
|
limit: Math.max(opts.points || 60, 10_000),
|
|
tier: 1,
|
|
})
|
|
if (remote.length) {
|
|
if (!windowed.length || remote.length >= windowed.length || windowExceedsMemory) {
|
|
windowed = remote
|
|
source = 'hyperdb-remote'
|
|
}
|
|
}
|
|
} catch {
|
|
// keep prior result
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!windowed.length && src.length) {
|
|
const n = Math.min(opts.points || 60, src.length)
|
|
windowed = src.slice(-n)
|
|
source = useTier1 ? 'memory-tier1' : 'memory-tier0'
|
|
}
|
|
|
|
const want = Math.min(opts.points || windowed.length || 60, 10_000)
|
|
const sampled = downsample(windowed, want, opts.group || 'average', def.dimensions.map((d) => d.id))
|
|
|
|
const labels = ['time', ...def.dimensions.map((d) => d.id)]
|
|
const data = sampled.map((p) => {
|
|
const row = [Math.floor(p.ts / 1000)]
|
|
for (const dim of def.dimensions) {
|
|
const v = p.values[dim.id]
|
|
row.push(v == null || Number.isNaN(v) ? null : round4(v))
|
|
}
|
|
return row
|
|
})
|
|
|
|
return {
|
|
chart,
|
|
context: def.context,
|
|
labels,
|
|
data,
|
|
view_update_every: SAMPLE_INTERVAL_MS / 1000,
|
|
after: after,
|
|
before: before,
|
|
points: data.length,
|
|
format: 'json',
|
|
source,
|
|
}
|
|
}
|
|
|
|
latestValues() {
|
|
/** @type {Record<string, { ts: number, values: Record<string, number|null> }>} */
|
|
const out = {}
|
|
for (const [chart, entry] of this.series) {
|
|
const last = entry.points[entry.points.length - 1]
|
|
if (last) out[chart] = last
|
|
}
|
|
return out
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param {Array<{ts:number, values: object}>} points
|
|
* @param {number} want
|
|
* @param {string} group
|
|
* @param {string[]} dims
|
|
*/
|
|
function downsample(points, want, group, dims) {
|
|
if (points.length <= want) return points
|
|
const bucketSize = points.length / want
|
|
/** @type {typeof points} */
|
|
const out = []
|
|
for (let i = 0; i < want; i++) {
|
|
const start = Math.floor(i * bucketSize)
|
|
const end = Math.floor((i + 1) * bucketSize)
|
|
const slice = points.slice(start, Math.max(start + 1, end))
|
|
const acc = {}
|
|
for (const d of dims) acc[d] = []
|
|
for (const p of slice) {
|
|
for (const d of dims) {
|
|
const v = p.values[d]
|
|
if (v != null && !Number.isNaN(v)) acc[d].push(v)
|
|
}
|
|
}
|
|
/** @type {Record<string, number|null>} */
|
|
const values = {}
|
|
for (const d of dims) {
|
|
values[d] = aggregate(acc[d], group)
|
|
}
|
|
out.push({ ts: slice[slice.length - 1].ts, values })
|
|
}
|
|
return out
|
|
}
|
|
|
|
function aggregate(arr, group) {
|
|
if (!arr.length) return null
|
|
if (group === 'min') return Math.min(...arr)
|
|
if (group === 'max') return Math.max(...arr)
|
|
if (group === 'sum') return arr.reduce((a, b) => a + b, 0)
|
|
// average default
|
|
return arr.reduce((a, b) => a + b, 0) / arr.length
|
|
}
|
|
|
|
function round4(n) {
|
|
return Math.round(n * 10000) / 10000
|
|
}
|
|
|
|
/** @type {MetricStore|null} */
|
|
let singleton = null
|
|
|
|
export function getStore() {
|
|
if (!singleton) singleton = new MetricStore()
|
|
return singleton
|
|
}
|