Metric Correlations

server/services/weights.js

@@ -0,0 +1,579 @@
+/**
+ * Metric Correlations / weights scoring engine.
+ *
+ * Highlight-vs-baseline methods (ks2, volume) plus single-window modes
+ * (anomaly-rate, value). When no highlight window is provided, falls back to
+ * anomaly-alert weights (method=alerts).
+ */
+import { CHART_BY_ID } from '../../shared/metrics.js'
+import { getStore } from './store.js'
+import { getAnomalyEngine } from './anomaly.js'
+
+const MIN_POINTS = 15
+const DEFAULT_POINTS = 500
+const MAX_POINTS = 10_000
+const DEFAULT_TIMEOUT_MS = 30_000
+const DEFAULT_BASELINE_MULT = 4
+
+/** @typedef {'ks2'|'volume'|'anomaly-rate'|'value'|'alerts'} WeightsMethod */
+
+/**
+ * @param {string|undefined} raw
+ * @returns {WeightsMethod}
+ */
+export function normalizeWeightsMethod(raw) {
+  const m = String(raw || '')
+    .toLowerCase()
+    .replace(/_/g, '-')
+  if (m === 'ks2' || m === 'ks') return 'ks2'
+  if (m === 'volume' || m === 'vol') return 'volume'
+  if (m === 'anomaly-rate' || m === 'anomaly' || m === 'ar') return 'anomaly-rate'
+  if (m === 'value' || m === 'cv') return 'value'
+  if (m === 'alerts' || m === 'alert') return 'alerts'
+  return ''
+}
+
+/**
+ * Main entry — RPC / REST.
+ * @param {object} [opts]
+ */
+export async function computeWeights(opts = {}) {
+  const started = Date.now()
+  const timeoutMs = Math.max(1000, Number(opts.timeout) || DEFAULT_TIMEOUT_MS)
+  const hasWindow =
+    opts.after != null ||
+    opts.before != null ||
+    opts.highlight_after != null ||
+    opts.highlight_before != null
+
+  let method = normalizeWeightsMethod(opts.method)
+  if (!method) method = hasWindow ? 'volume' : 'alerts'
+  if (!hasWindow && (method === 'ks2' || method === 'volume' || method === 'value')) {
+    method = 'alerts'
+  }
+  if (!hasWindow && method === 'anomaly-rate') {
+    // AR can use a default recent window
+  }
+
+  if (method === 'alerts') {
+    const legacy = getAnomalyEngine().getWeights({
+      chart: opts.chart || opts.context,
+      limit: opts.limit,
+    })
+    return {
+      ...legacy,
+      method: 'alerts',
+      view: null,
+      stats: { durationMs: Date.now() - started, chartsScored: legacy.results?.length || 0 },
+    }
+  }
+
+  const windows = resolveWindows(opts, method)
+  if (windows.error) {
+    return {
+      error: windows.error,
+      method,
+      results: [],
+      ts: Date.now(),
+      stats: { durationMs: Date.now() - started },
+    }
+  }
+
+  const store = getStore()
+  const chartIds = listCandidateCharts(store, opts)
+  const limit = Math.max(1, Math.min(500, Number(opts.limit) || 100))
+  const timeGroup = String(opts.time_group || opts.group || (method === 'value' ? 'cv' : 'average'))
+  const points = windows.points
+  const shifts = windows.shifts
+
+  /** @type {Array<{ id: string, weight: number, context: string, family: string, info: string, dimensions?: Record<string, number> }>} */
+  const scored = []
+  let examined = 0
+  let skipped = 0
+
+  const concurrency = 8
+  let idx = 0
+
+  async function worker() {
+    while (idx < chartIds.length) {
+      if (Date.now() - started > timeoutMs) return
+      const i = idx++
+      const chartId = chartIds[i]
+      examined++
+      try {
+        const row = await scoreChart(store, chartId, {
+          method,
+          windows,
+          points,
+          shifts,
+          timeGroup,
+          dimensions: opts.dimensions,
+        })
+        if (row) scored.push(row)
+        else skipped++
+      } catch {
+        skipped++
+      }
+    }
+  }
+
+  await Promise.all(Array.from({ length: concurrency }, () => worker()))
+
+  const timedOut = Date.now() - started > timeoutMs
+  spreadEvenly(scored)
+  scored.sort((a, b) => b.weight - a.weight || a.id.localeCompare(b.id))
+
+  return {
+    method,
+    view: {
+      highlight: {
+        after: windows.highlightAfter,
+        before: windows.highlightBefore,
+        duration: windows.highlightBefore - windows.highlightAfter,
+      },
+      baseline:
+        method === 'ks2' || method === 'volume'
+          ? {
+              after: windows.baselineAfter,
+              before: windows.baselineBefore,
+              duration: windows.baselineBefore - windows.baselineAfter,
+              shifts,
+            }
+          : null,
+      points,
+      time_group: timeGroup,
+    },
+    results: scored.slice(0, limit).map((r) => ({
+      id: r.id,
+      weight: r.weight,
+      score: r.weight,
+      context: r.context,
+      family: r.family,
+      info: r.info,
+      dimensions: r.dimensions,
+    })),
+    ts: Date.now(),
+    stats: {
+      durationMs: Date.now() - started,
+      chartsExamined: examined,
+      chartsScored: scored.length,
+      chartsSkipped: skipped,
+      timedOut,
+    },
+  }
+}
+
+/**
+ * @param {object} opts
+ * @param {WeightsMethod} method
+ */
+export function resolveWindows(opts, method) {
+  const nowSec = Math.floor(Date.now() / 1000)
+  let after = pickNum(opts.after, opts.highlight_after)
+  let before = pickNum(opts.before, opts.highlight_before)
+  let baselineAfter = pickNum(opts.baseline_after, opts.baselineAfter)
+  let baselineBefore = pickNum(opts.baseline_before, opts.baselineBefore)
+
+  // Defaults: last 60s highlight when method needs a window but none given
+  if (after == null && before == null) {
+    before = 0
+    after = -60
+  }
+
+  before = resolveRelative(before, nowSec, nowSec)
+  after = resolveRelative(after, nowSec, before)
+
+  if (!(before > after)) {
+    return { error: 'Invalid selected time-range.' }
+  }
+
+  const highDelta = before - after
+  if (highDelta < 15 && (method === 'ks2' || method === 'volume')) {
+    return { error: 'Highlight window must be at least 15 seconds.' }
+  }
+
+  let points = Math.max(0, Number(opts.points) || 0)
+  if (!points) points = method === 'ks2' || method === 'volume' ? DEFAULT_POINTS : 120
+  points = Math.min(MAX_POINTS, Math.max(MIN_POINTS, points))
+
+  let shifts = 0
+  if (method === 'ks2' || method === 'volume') {
+    if (baselineBefore == null) baselineBefore = after
+    else baselineBefore = resolveRelative(baselineBefore, nowSec, after)
+
+    if (baselineAfter == null) {
+      baselineAfter = baselineBefore - highDelta * DEFAULT_BASELINE_MULT
+    } else {
+      baselineAfter = resolveRelative(baselineAfter, nowSec, baselineBefore)
+    }
+
+    if (!(baselineBefore > baselineAfter)) {
+      return { error: 'Invalid baseline time-range.' }
+    }
+
+    let baseDelta = baselineBefore - baselineAfter
+    let multiplier = Math.max(1, Math.round(baseDelta / highDelta))
+    // Snap to power of two
+    if ((multiplier & (multiplier - 1)) !== 0) {
+      multiplier = nextPowerOfTwo(multiplier)
+    }
+    while (multiplier > 1) {
+      shifts++
+      multiplier >>= 1
+    }
+    while (shifts && points << shifts > MAX_POINTS) shifts--
+    while (points << shifts > MAX_POINTS) points >>= 1
+    if (points < MIN_POINTS) {
+      return { error: 'Too few points available, at least 15 are needed.' }
+    }
+    baselineAfter = baselineBefore - (highDelta << shifts)
+  }
+
+  return {
+    highlightAfter: after,
+    highlightBefore: before,
+    baselineAfter: baselineAfter ?? null,
+    baselineBefore: baselineBefore ?? null,
+    points,
+    shifts,
+  }
+}
+
+function pickNum(...vals) {
+  for (const v of vals) {
+    if (v == null || v === '') continue
+    const n = Number(v)
+    if (Number.isFinite(n)) return n
+  }
+  return null
+}
+
+/** @param {number|null} v @param {number} nowSec @param {number} anchor */
+function resolveRelative(v, nowSec, anchor) {
+  if (v == null) return nowSec
+  if (v <= 0) {
+    // relative to anchor (Netdata-style): before=0 → now; after=-N → before-N
+    return anchor + v
+  }
+  return v
+}
+
+function nextPowerOfTwo(n) {
+  let m = Math.max(1, n) - 1
+  m |= m >> 1
+  m |= m >> 2
+  m |= m >> 4
+  m |= m >> 8
+  m |= m >> 16
+  return m + 1
+}
+
+function listCandidateCharts(store, opts) {
+  const filterChart = opts.chart || opts.context
+  const filterCharts = opts.charts
+    ? String(opts.charts)
+        .split(',')
+        .map((s) => s.trim())
+        .filter(Boolean)
+    : null
+  const filterContexts = opts.contexts
+    ? String(opts.contexts)
+        .split('|')
+        .map((s) => s.trim())
+        .filter(Boolean)
+    : null
+
+  /** @type {string[]} */
+  const ids = []
+  const seen = new Set()
+  for (const id of store.series.keys()) {
+    if (!CHART_BY_ID.has(id)) continue
+    seen.add(id)
+    ids.push(id)
+  }
+  for (const id of CHART_BY_ID.keys()) {
+    if (seen.has(id)) continue
+    // Include catalog charts that may have warm data only
+    ids.push(id)
+  }
+
+  return ids.filter((id) => {
+    if (filterChart && id !== filterChart && !id.startsWith(String(filterChart))) return false
+    if (filterCharts && !filterCharts.includes(id)) return false
+    if (filterContexts) {
+      const ctx = CHART_BY_ID.get(id)?.context || ''
+      if (!filterContexts.some((c) => ctx === c || ctx.startsWith(c))) return false
+    }
+    return true
+  })
+}
+
+/**
+ * @param {import('./store.js').MetricStore} store
+ * @param {string} chartId
+ * @param {object} cfg
+ */
+async function scoreChart(store, chartId, cfg) {
+  const def = CHART_BY_ID.get(chartId)
+  if (!def) return null
+  const { method, windows, points, shifts, timeGroup } = cfg
+
+  if (method === 'anomaly-rate') {
+    return scoreAnomalyRate(chartId, def, windows)
+  }
+
+  const highQ = await store.query({
+    chart: chartId,
+    after: windows.highlightAfter,
+    before: windows.highlightBefore,
+    points,
+    group: timeGroup === 'cv' ? 'average' : timeGroup,
+  })
+  if (highQ.error || !highQ.data?.length) return null
+
+  const dimNames = (highQ.labels || []).slice(1)
+  if (!dimNames.length) return null
+  const wantedDims = filterDims(dimNames, cfg.dimensions)
+
+  if (method === 'value') {
+    let best = 0
+    /** @type {Record<string, number>} */
+    const dims = {}
+    for (const dim of wantedDims) {
+      const series = extractDim(highQ.data, highQ.labels, dim)
+      const v = aggregateSeries(series, timeGroup)
+      if (!Number.isFinite(v) || v === 0) continue
+      dims[dim] = Math.abs(v)
+      best = Math.max(best, Math.abs(v))
+    }
+    if (best <= 0) return null
+    return {
+      id: chartId,
+      weight: best,
+      context: def.context,
+      family: def.family || '',
+      info: def.title || chartId,
+      dimensions: dims,
+    }
+  }
+
+  // ks2 / volume need baseline
+  const basePoints = method === 'ks2' ? points << shifts : points
+  const baseQ = await store.query({
+    chart: chartId,
+    after: windows.baselineAfter,
+    before: windows.baselineBefore,
+    points: Math.min(MAX_POINTS, Math.max(MIN_POINTS, basePoints)),
+    group: 'average',
+  })
+  if (baseQ.error || !baseQ.data?.length) return null
+
+  let best = 0
+  /** @type {Record<string, number>} */
+  const dims = {}
+  for (const dim of wantedDims) {
+    const high = extractDim(highQ.data, highQ.labels, dim)
+    const base = extractDim(baseQ.data, baseQ.labels, dim)
+    if (high.length < MIN_POINTS || base.length < 2) continue
+    let w = 0
+    if (method === 'volume') w = volumeScore(base, high)
+    else w = ks2Score(base, high, shifts)
+    if (!Number.isFinite(w) || w <= 0) continue
+    dims[dim] = w
+    best = Math.max(best, w)
+  }
+  if (best <= 0) return null
+  return {
+    id: chartId,
+    weight: best,
+    context: def.context,
+    family: def.family || '',
+    info: def.title || chartId,
+    dimensions: dims,
+  }
+}
+
+function scoreAnomalyRate(chartId, def, windows) {
+  const eng = getAnomalyEngine()
+  const afterMs = windows.highlightAfter * 1000
+  const beforeMs = windows.highlightBefore * 1000
+  const dur = Math.max(1, beforeMs - afterMs)
+  let hits = 0
+  let maxScore = 0
+  for (const ev of eng.recent || []) {
+    if (ev.chart !== chartId || ev.cleared) continue
+    const ts = Number(ev.ts) || 0
+    if (ts < afterMs || ts > beforeMs) continue
+    hits++
+    maxScore = Math.max(maxScore, Number(ev.score) || (ev.severity === 'critical' ? 1 : 0.5))
+  }
+  const st = [...eng.status.entries()].find(([cfgId]) => eng.configs.get(cfgId)?.chart === chartId)
+  if (st?.[1] === 'CRITICAL') maxScore = Math.max(maxScore, 1)
+  if (st?.[1] === 'WARNING') maxScore = Math.max(maxScore, 0.65)
+  if (hits === 0 && maxScore === 0) return null
+  // Rate of events per minute of highlight + severity
+  const rate = hits / (dur / 60_000)
+  const weight = Math.min(1, maxScore * 0.6 + Math.min(1, rate) * 0.4)
+  if (weight <= 0) return null
+  return {
+    id: chartId,
+    weight,
+    context: def.context,
+    family: def.family || '',
+    info: def.title || chartId,
+  }
+}
+
+/** Volume heuristic (Netdata-inspired). Higher = more changed. */
+export function volumeScore(baseline, highlight) {
+  const baseAvg = mean(baseline)
+  const highAvg = mean(highlight)
+  if (!Number.isFinite(highAvg)) return 0
+  if (baseAvg === highAvg) return 0
+  const threshold = baseAvg
+  const above = highAvg >= threshold
+  let count = 0
+  for (const v of highlight) {
+    if (!Number.isFinite(v)) continue
+    if (above ? v > threshold : v < threshold) count++
+  }
+  const frac = count / Math.max(1, highlight.length)
+  if (Number.isFinite(baseAvg) && baseAvg !== 0) {
+    return Math.abs((highAvg - baseAvg) / baseAvg) * frac
+  }
+  return frac
+}
+
+/** KS2 on pairwise diffs. Higher = more different. */
+export function ks2Score(baseline, highlight, shifts = 2) {
+  if (baseline.length < 2 || highlight.length < 2) return 0
+  const baseDiffs = pairwiseDiffs(baseline)
+  const highDiffs = pairwiseDiffs(highlight)
+  if (!baseDiffs.length || !highDiffs.length) return 0
+  const d = ksStatistic(baseDiffs, highDiffs)
+  if (!Number.isFinite(d) || d <= 0) return 0
+  // Approximate p-value; flip so higher weight = more different
+  const n = (baseDiffs.length * highDiffs.length) / (baseDiffs.length + highDiffs.length)
+  const p = ksPValue(Math.round(n), d)
+  const weight = 1 - p
+  // Mild boost when baseline is much longer (shifts)
+  return weight * (1 + Math.min(3, shifts) * 0.02)
+}
+
+function pairwiseDiffs(arr) {
+  /** @type {number[]} */
+  const out = []
+  for (let i = 1; i < arr.length; i++) {
+    const a = arr[i - 1]
+    const b = arr[i]
+    if (!Number.isFinite(a) || !Number.isFinite(b)) continue
+    out.push(b - a)
+  }
+  return out
+}
+
+/** Two-sample KS statistic on sorted samples. */
+export function ksStatistic(a, b) {
+  const A = [...a].sort((x, y) => x - y)
+  const B = [...b].sort((x, y) => x - y)
+  const n1 = A.length
+  const n2 = B.length
+  let i = 0
+  let j = 0
+  let d = 0
+  while (i < n1 || j < n2) {
+    const va = i < n1 ? A[i] : Infinity
+    const vb = j < n2 ? B[j] : Infinity
+    if (va <= vb) i++
+    if (vb <= va) j++
+    d = Math.max(d, Math.abs(i / n1 - j / n2))
+  }
+  return d
+}
+
+/** Kolmogorov–Smirnov survival function approximation (KSfbar-ish). */
+export function ksPValue(en, d) {
+  if (!Number.isFinite(en) || en <= 0 || !Number.isFinite(d) || d <= 0) return 1
+  // Marsaglia/Tsang-style approximation via series of exp terms
+  const lambda = (Math.sqrt(en) + 0.12 + 0.11 / Math.sqrt(en)) * d
+  if (lambda <= 0) return 1
+  let sum = 0
+  let sign = 1
+  for (let k = 1; k <= 100; k++) {
+    const term = sign * Math.exp(-2 * lambda * lambda * k * k)
+    sum += term
+    if (Math.abs(term) < 1e-12) break
+    sign = -sign
+  }
+  const p = Math.max(0, Math.min(1, 2 * sum))
+  return p
+}
+
+function aggregateSeries(series, timeGroup) {
+  const vals = series.filter((v) => Number.isFinite(v))
+  if (!vals.length) return NaN
+  const g = String(timeGroup || 'average').toLowerCase()
+  if (g === 'min') return Math.min(...vals)
+  if (g === 'max') return Math.max(...vals)
+  if (g === 'sum') return vals.reduce((a, b) => a + b, 0)
+  if (g === 'cv' || g === 'stddev') {
+    const m = mean(vals)
+    const sd = stddev(vals, m)
+    if (g === 'stddev') return sd
+    if (!m) return sd > 0 ? 1 : 0
+    return Math.abs(sd / m)
+  }
+  return mean(vals)
+}
+
+function mean(arr) {
+  const vals = arr.filter((v) => Number.isFinite(v))
+  if (!vals.length) return NaN
+  return vals.reduce((a, b) => a + b, 0) / vals.length
+}
+
+function stddev(arr, m) {
+  const vals = arr.filter((v) => Number.isFinite(v))
+  if (vals.length < 2) return 0
+  const mu = m ?? mean(vals)
+  let s = 0
+  for (const v of vals) s += (v - mu) ** 2
+  return Math.sqrt(s / (vals.length - 1))
+}
+
+function extractDim(data, labels, dim) {
+  const idx = labels.indexOf(dim)
+  if (idx < 0) return []
+  /** @type {number[]} */
+  const out = []
+  for (const row of data) {
+    const v = row[idx]
+    if (v == null || Number.isNaN(v)) continue
+    out.push(Number(v))
+  }
+  return out
+}
+
+function filterDims(dimNames, dimensions) {
+  if (!dimensions) return dimNames
+  const want = String(dimensions)
+    .split(/[,|]/)
+    .map((s) => s.trim())
+    .filter(Boolean)
+  if (!want.length) return dimNames
+  return dimNames.filter((d) => want.includes(d))
+}
+
+/** Spread raw weights evenly across (0,1] — higher remains more interesting. */
+export function spreadEvenly(rows) {
+  if (!rows.length) return
+  const sorted = [...rows].sort((a, b) => a.weight - b.weight)
+  const n = sorted.length
+  if (n === 1) {
+    sorted[0].weight = 1
+    return
+  }
+  for (let i = 0; i < n; i++) {
+    sorted[i].weight = (i + 1) / n
+  }
+}