3x-ui-agent/internal/web/service/metric_history.go

package service

import (
	"slices"
	"sync"
	"time"
)

// MetricSample is one point of any time-series we keep in memory.
// The frontend deserializes both keys, so they must stay short.
type MetricSample struct {
	T int64   `json:"t"`
	V float64 `json:"v"`
}

// tierSpec defines one resolution layer of the rollup ladder: a fixed bucket
// size in seconds and how many buckets to retain. window = resolution*capacity.
type tierSpec struct {
	resolution int
	capacity   int
}

// metricTiers is the rollup ladder applied to every series. High resolution is
// kept only for the recent past; older samples roll up into progressively
// coarser, cheaper layers (RRDtool-style). Per series this totals ~5700 samples
// (~90 KiB) yet spans a live 2s view through ~7 days of history.
var metricTiers = []tierSpec{
	{resolution: 2, capacity: 1800},   // 1h at 2s
	{resolution: 60, capacity: 2880},  // 48h at 1m
	{resolution: 600, capacity: 1008}, // 7d at 10m
}

// tierBuf is one fixed-resolution ring of a series. Samples land in an open
// bucket and are averaged into the ring only when the next bucket begins, so a
// coarse tier carries one mean per bucket instead of every raw point.
type tierBuf struct {
	resolution int
	capacity   int
	samples    []MetricSample
	open       bool
	openStart  int64
	openSum    float64
	openCount  int
}

func (tb *tierBuf) add(unixSec int64, v float64) {
	res := int64(tb.resolution)
	b := (unixSec / res) * res
	if tb.open && b != tb.openStart {
		tb.flush()
	}
	tb.open = true
	tb.openStart = b
	tb.openSum += v
	tb.openCount++
}

func (tb *tierBuf) flush() {
	if tb.openCount == 0 {
		tb.open = false
		return
	}
	tb.samples = append(tb.samples, MetricSample{T: tb.openStart, V: tb.openSum / float64(tb.openCount)})
	if len(tb.samples) > tb.capacity {
		tb.samples = tb.samples[len(tb.samples)-tb.capacity:]
	}
	tb.open = false
	tb.openStart = 0
	tb.openSum = 0
	tb.openCount = 0
}

// readSamples returns a copy of the closed buckets plus the still-open one, so
// the most recent point is visible before its bucket boundary closes.
func (tb *tierBuf) readSamples() []MetricSample {
	out := make([]MetricSample, len(tb.samples), len(tb.samples)+1)
	copy(out, tb.samples)
	if tb.openCount > 0 {
		out = append(out, MetricSample{T: tb.openStart, V: tb.openSum / float64(tb.openCount)})
	}
	return out
}

// series is the rollup ladder for one metric: a sample is fed to every tier.
type series struct {
	tiers []*tierBuf
}

func newSeries() *series {
	return newSeriesWithTiers(metricTiers)
}

func newSeriesWithTiers(specs []tierSpec) *series {
	s := &series{tiers: make([]*tierBuf, len(specs))}
	for i, spec := range specs {
		s.tiers[i] = &tierBuf{resolution: spec.resolution, capacity: spec.capacity}
	}
	return s
}

func (s *series) add(unixSec int64, v float64) {
	for _, tb := range s.tiers {
		tb.add(unixSec, v)
	}
}

// pickTier returns the finest tier whose window covers spanSeconds, falling back
// to the coarsest (longest-window) tier when nothing covers it.
func (s *series) pickTier(spanSeconds int64) *tierBuf {
	for _, tb := range s.tiers {
		if int64(tb.resolution)*int64(tb.capacity) >= spanSeconds {
			return tb
		}
	}
	return s.tiers[len(s.tiers)-1]
}

// metricHistory is a thread-safe, in-memory store of tiered series keyed by
// arbitrary strings. Two singletons live below: per-node metrics (full
// rollup ladder) and the homepage overview series (one small fixed ring).
type metricHistory struct {
	mu     sync.Mutex
	tiers  []tierSpec
	series map[string]*series
}

func newMetricHistory() *metricHistory {
	return newMetricHistoryWithTiers(metricTiers)
}

func newMetricHistoryWithTiers(tiers []tierSpec) *metricHistory {
	return &metricHistory{tiers: tiers, series: map[string]*series{}}
}

// append stores a single sample for the given metric across all tiers.
func (h *metricHistory) append(metric string, t time.Time, v float64) {
	h.mu.Lock()
	defer h.mu.Unlock()
	s := h.series[metric]
	if s == nil {
		s = newSeriesWithTiers(h.tiers)
		h.series[metric] = s
	}
	s.add(t.Unix(), v)
}

// drop removes the entire history for one metric. Used when a node is deleted so
// its old samples don't linger forever in the singleton.
func (h *metricHistory) drop(metric string) {
	h.mu.Lock()
	delete(h.series, metric)
	h.mu.Unlock()
}

// aggregate returns up to maxPoints buckets of size bucketSeconds, each carrying
// the arithmetic mean of the underlying samples from the finest tier that covers
// the requested span. Bucket alignment is to absolute Unix-second boundaries so
// two concurrent calls see identical x-axes.
func (h *metricHistory) aggregate(metric string, bucketSeconds int, maxPoints int) []map[string]any {
	empty := []map[string]any{}
	if bucketSeconds <= 0 || maxPoints <= 0 {
		return empty
	}
	span := int64(bucketSeconds) * int64(maxPoints)
	cutoff := time.Now().Unix() - span

	h.mu.Lock()
	s := h.series[metric]
	if s == nil {
		h.mu.Unlock()
		return empty
	}
	raw := s.pickTier(span).readSamples()
	h.mu.Unlock()

	startIdx := len(raw)
	for i, r := range slices.Backward(raw) {
		if r.T < cutoff {
			break
		}
		startIdx = i
	}
	tmp := raw[startIdx:]
	if len(tmp) == 0 {
		return empty
	}

	bSize := int64(bucketSeconds)
	curBucket := (tmp[0].T / bSize) * bSize
	var out []map[string]any
	var acc []float64
	flush := func(ts int64) {
		if len(acc) == 0 {
			return
		}
		sum := 0.0
		for _, v := range acc {
			sum += v
		}
		out = append(out, map[string]any{"t": ts, "v": sum / float64(len(acc))})
		acc = acc[:0]
	}
	for _, p := range tmp {
		b := (p.T / bSize) * bSize
		if b != curBucket {
			flush(curBucket)
			curBucket = b
		}
		acc = append(acc, p.V)
	}
	flush(curBucket)
	if len(out) > maxPoints {
		out = out[len(out)-maxPoints:]
	}
	if out == nil {
		return empty
	}
	return out
}

// nodeMetrics holds per-node CPU/Mem/net series fed by NodeHeartbeatJob
// (via UpdateHeartbeat) and read by the node history panel. The whole-host
// system series and the xray expvar series that used to live here were
// removed with the system-history and xray-metrics features, together with
// the gob persistence of the host series.
var nodeMetrics = newMetricHistory()

// NodeMetricKeys lists the per-node metric names NodeHeartbeatJob writes.
var NodeMetricKeys = []string{"cpu", "mem", "netUp", "netDown"}

// overviewMetrics is the homepage overview store (P8): a deliberately small,
// memory-only ring feeding the index-page sparkline seed — NOT a revival of
// the removed system-history feature. One tier at the status cadence (2s)
// holding 360 samples = a 12-minute window; eight series of 360 samples are
// ~46 KB of payload, bounded by construction. Nothing is persisted: the
// ring starts empty on every panel start and the overview curves simply
// accumulate from there.
var overviewMetrics = newMetricHistoryWithTiers([]tierSpec{{resolution: 2, capacity: 360}})

// OverviewMetricKeys lists the host metric names ServerService writes on
// every status sample for the homepage overview. It matches the frontend
// useOverviewHistory SERIES_KEYS exactly; the controller validates incoming
// history requests against this allow-list.
var OverviewMetricKeys = []string{"cpu", "mem", "swap", "diskUsage", "netUp", "netDown", "tcpCount", "udpCount"}