From cbff590d49c1ddd0143135430d4c388f0cc63fe2 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Wed, 30 Sep 2026 18:52:56 -0700 Subject: [PATCH 1/2] Dashboard: read gauges per node Every node stores its own gauge readings, and the dashboard took the latest from whichever node wrote last, or added up every reading in a chart slice. Per-node gauges (memory, open connections) showed one node or a pile of ticks, and the cluster-wide counts could show another node's drifted copy: a 20-digit queue (an underflow stored unsigned) or 0 accounts. gaugeReading takes each node's latest reading, adds per-node gauges up, and for queue/user/domain counts takes the largest believable one, which also skips the junk already in history (inbuxa-server stops storing it). Cards, chart slices and the memory tile use it; the tile now reads all nodes from history rather than one node's live value. --- src/features/dashboard/commandCenter.ts | 13 ++-- .../dashboard/components/TrendDeck.tsx | 14 ++--- src/features/dashboard/helpers.test.ts | 61 ++++++++++++++++++- src/features/dashboard/helpers.ts | 46 +++++++++++++- src/i18n/en.json | 2 +- 5 files changed, 119 insertions(+), 17 deletions(-) diff --git a/src/features/dashboard/commandCenter.ts b/src/features/dashboard/commandCenter.ts index 5821195..52e7723 100644 --- a/src/features/dashboard/commandCenter.ts +++ b/src/features/dashboard/commandCenter.ts @@ -10,7 +10,7 @@ * from the components so every figure can be tested on its own. */ import type { Metric } from './types/metrics'; -import { bucketize, metricToScalar } from './helpers'; +import { bucketize, gaugeReading, metricToScalar } from './helpers'; /** Messages the server took in from outside. */ export const RECEIVED = ['queue.message-queued']; @@ -107,9 +107,12 @@ export function mean(samples: Metric[], ids: string[]): number | null { /** Counts per time slice across the period, oldest first. */ export function series(samples: Metric[], ids: string[], from: Date, to: Date, buckets: number): number[] { - return bucketize(samples, from, to, buckets).map((b) => - b.filter((m) => ids.includes(m.metric)).reduce((s, m) => s + metricToScalar(m), 0), - ); + return bucketize(samples, from, to, buckets).map((b) => { + const mine = b.filter((m) => ids.includes(m.metric)); + // A gauge's slice is one reading across the nodes, not the sum of every tick + if (mine.length > 0 && mine.every((m) => m['@type'] === 'Gauge')) return gaugeReading(mine, ids) ?? 0; + return mine.reduce((s, m) => s + metricToScalar(m), 0); + }); } export interface Load { @@ -180,7 +183,7 @@ export function latest(samples: Metric[], ids: string[]): number | null { export function summarize(samples: Metric[], ids: string[], timing: boolean): number | null { if (timing) return mean(samples, ids); const mine = samples.filter((s) => ids.includes(s.metric)); - if (mine.length > 0 && mine.every((s) => s['@type'] === 'Gauge')) return latest(mine, ids); + if (mine.length > 0 && mine.every((s) => s['@type'] === 'Gauge')) return gaugeReading(mine, ids); return total(samples, ids); } diff --git a/src/features/dashboard/components/TrendDeck.tsx b/src/features/dashboard/components/TrendDeck.tsx index 1d6054d..56385ae 100644 --- a/src/features/dashboard/components/TrendDeck.tsx +++ b/src/features/dashboard/components/TrendDeck.tsx @@ -8,7 +8,7 @@ import type { ReactNode } from 'react'; import { useTranslation } from 'react-i18next'; import type { Metric, Period } from '../types/metrics'; import type { ServerFacts } from '../serverFacts'; -import { bucketTimestamps, formatValue, getBucketCount } from '../helpers'; +import { bucketTimestamps, formatValue, gaugeReading, getBucketCount } from '../helpers'; import { DECK_METRICS, AUTH_FAILED, @@ -18,7 +18,6 @@ import { RECEIVED, SENT, SPAM, - latest, load, mean, queueHealth, @@ -149,11 +148,12 @@ export function TrendDeck({ id, samples, window, period, periodName, live, liveV /> ); - // Memory now against its highest in the period: how close to the peak the node is running. + // Memory now against its highest in the period, both summed over the nodes: + // the live feed is one node's, so the reading comes from the stored history. const memoryGauge = () => { - const readings = samples.filter((m) => m.metric === 'server.memory').map((m) => m.count); - const now = liveValues.get('server.memory') ?? latest(samples, ['server.memory']); - const peak = Math.max(0, now ?? 0, ...readings); + const MEMORY = ['server.memory']; + const now = gaugeReading(samples, MEMORY); + const peak = Math.max(0, now ?? 0, ...series(samples, MEMORY, from, to, buckets)); return ( diff --git a/src/features/dashboard/helpers.test.ts b/src/features/dashboard/helpers.test.ts index 0bde2a6..4ffceb5 100644 --- a/src/features/dashboard/helpers.test.ts +++ b/src/features/dashboard/helpers.test.ts @@ -6,7 +6,7 @@ import { describe, expect, it } from 'vitest'; import type { Metric } from './types/metrics'; -import { deltaHistograms } from './helpers'; +import { deltaHistograms, gaugeReading, seriesBucketValue } from './helpers'; const total = (timestamp: string, count: number, sum: number, nodeId?: number): Metric => ({ '@type': 'Histogram', @@ -60,3 +60,62 @@ describe('deltaHistograms', () => { expect(deltaHistograms([counter])).toEqual([counter]); }); }); + +const reading = (metric: string, timestamp: string, count: number, nodeId?: number): Metric => ({ + '@type': 'Gauge', + metric, + count, + timestamp, + nodeId, +}); + +describe('gaugeReading', () => { + it('adds up each node’s latest reading of a per-node gauge', () => { + const v = gaugeReading( + [ + reading('server.memory', '2026-09-30T10:00:00Z', 100, 0), + reading('server.memory', '2026-09-30T11:00:00Z', 180, 0), + reading('server.memory', '2026-09-30T11:00:00Z', 200, 1), + reading('server.memory', '2026-09-30T11:00:01Z', 220, 2), + ], + ['server.memory'], + ); + expect(v).toBe(600); + }); + + it('takes the real cluster-wide count, not a drifted or zero copy', () => { + const samples = [ + reading('queue.count', '2026-09-30T11:00:00Z', 8, 2), + reading('queue.count', '2026-09-30T11:00:01Z', 18_446_744_073_709_551_596, 0), + reading('queue.count', '2026-09-30T11:00:02Z', 18_446_744_073_709_551_613, 1), + reading('user.count', '2026-09-30T11:00:00Z', 7, 2), + reading('user.count', '2026-09-30T11:00:02Z', 0, 1), + ]; + expect(gaugeReading(samples, ['queue.count'])).toBe(8); + expect(gaugeReading(samples, ['user.count'])).toBe(7); + }); + + it('is null with no reading, or only an unbelievable one', () => { + expect(gaugeReading([], ['server.memory'])).toBeNull(); + expect(gaugeReading([reading('queue.count', '2026-09-30T11:00:00Z', 2 ** 64 - 3, 0)], ['queue.count'])).toBeNull(); + }); + + it('reads a single-node server (no nodeId) as before', () => { + const v = gaugeReading( + [reading('server.memory', '2026-09-30T10:00:00Z', 100), reading('server.memory', '2026-09-30T11:00:00Z', 150)], + ['server.memory'], + ); + expect(v).toBe(150); + }); +}); + +describe('seriesBucketValue', () => { + it('reads a gauge slice across the nodes instead of adding up its ticks', () => { + const slice = [ + reading('server.memory', '2026-09-30T10:00:00Z', 100, 0), + reading('server.memory', '2026-09-30T10:01:00Z', 110, 0), + reading('server.memory', '2026-09-30T10:00:30Z', 50, 1), + ]; + expect(seriesBucketValue({ label: 'Memory', metrics: ['server.memory'] }, slice)).toBe(160); + }); +}); diff --git a/src/features/dashboard/helpers.ts b/src/features/dashboard/helpers.ts index a410c9a..6efaf2e 100644 --- a/src/features/dashboard/helpers.ts +++ b/src/features/dashboard/helpers.ts @@ -99,9 +99,49 @@ export function cardValue(card: Card, samples: Metric[]): number { } export function seriesBucketValue(series: Series, bucketSamples: Metric[]): number | null { - const values = bucketSamples.filter((m) => series.metrics.includes(m.metric)).map(metricToScalar); - if (values.length === 0) return null; - return aggregate(values, series.aggregate); + const mine = bucketSamples.filter((m) => series.metrics.includes(m.metric)); + if (mine.length === 0) return null; + // INBUXA: a gauge's slice is one reading across the nodes, not every tick's readings combined + if (mine.every((m) => m['@type'] === 'Gauge')) return gaugeReading(mine, series.metrics); + return aggregate(mine.map(metricToScalar), series.aggregate); +} + +// INBUXA: gauges counting the whole cluster. Only the node that computes +// them has a true reading; history from before inbuxa-server 2026.9.30.2 +// also holds the other nodes' copies, with the queue drifted below zero +// (stored unsigned, so a 20-digit number) and the accounts and domains at 0. +export const CLUSTER_GAUGES = new Set(['queue.count', 'user.count', 'domain.count']); + +/** + * The gauges' reading across the cluster: each node's latest, added up for + * per-node gauges (memory, open connections), and for cluster-wide counts the + * largest believable one. Null when there's no reading. + */ +export function gaugeReading(samples: Metric[], ids: MetricId[]): number | null { + const last = new Map(); + for (const m of samples) { + if (m['@type'] !== 'Gauge' || !ids.includes(m.metric)) continue; + const at = m.timestamp ? Date.parse(m.timestamp) : 0; + const key = `${m.metric}|${m.nodeId ?? ''}`; + const prev = last.get(key); + if (!prev || at >= prev.at) last.set(key, { metric: m.metric, at, value: m.count }); + } + let sum = 0; + let found = false; + const cluster = new Map(); + for (const { metric, value } of last.values()) { + if (!CLUSTER_GAUGES.has(metric)) { + sum += value; + found = true; + } else if (value <= Number.MAX_SAFE_INTEGER) { + cluster.set(metric, Math.max(cluster.get(metric) ?? 0, value)); + } + } + for (const value of cluster.values()) { + sum += value; + found = true; + } + return found ? sum : null; } export function bucketize(samples: Metric[], from: Date, to: Date, bucketCount: number): Metric[][] { diff --git a/src/i18n/en.json b/src/i18n/en.json index 6f9e422..03c89db 100644 --- a/src/i18n/en.json +++ b/src/i18n/en.json @@ -1841,7 +1841,7 @@ "zoneDetail": "fine under {{good}} · slow over {{bad}}", "quiet": "Nothing recorded yet", "memory": "Memory", - "memoryDetail": "this node · peak {{peak}}", + "memoryDetail": "all nodes · peak {{peak}}", "connLoad": "Connection load", "smtpSplit": "SMTP sessions", "smtpIn": "{{n}} in", -- 2.54.0 From 229dd6179ddaaf840e01e96145f065719fae7e58 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Wed, 30 Sep 2026 18:59:39 -0700 Subject: [PATCH 2/2] Dashboard: write the underflowed test readings as expressions no-loss-of-precision rejected the 20-digit literals. --- src/features/dashboard/helpers.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/features/dashboard/helpers.test.ts b/src/features/dashboard/helpers.test.ts index 4ffceb5..4307a6d 100644 --- a/src/features/dashboard/helpers.test.ts +++ b/src/features/dashboard/helpers.test.ts @@ -86,8 +86,8 @@ describe('gaugeReading', () => { it('takes the real cluster-wide count, not a drifted or zero copy', () => { const samples = [ reading('queue.count', '2026-09-30T11:00:00Z', 8, 2), - reading('queue.count', '2026-09-30T11:00:01Z', 18_446_744_073_709_551_596, 0), - reading('queue.count', '2026-09-30T11:00:02Z', 18_446_744_073_709_551_613, 1), + reading('queue.count', '2026-09-30T11:00:01Z', 2 ** 64 - 20, 0), + reading('queue.count', '2026-09-30T11:00:02Z', 2 ** 64 - 3, 1), reading('user.count', '2026-09-30T11:00:00Z', 7, 2), reading('user.count', '2026-09-30T11:00:02Z', 0, 1), ]; -- 2.54.0