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 {
	s := &series{tiers: make([]*tierBuf, len(metricTiers))}
	for i, spec := range metricTiers {
		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. One singleton lives below: per-node metrics.
type metricHistory struct {
	mu     sync.Mutex
	series map[string]*series
}

func newMetricHistory() *metricHistory {
	return &metricHistory{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 = newSeries()
		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 is the remaining singleton: 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"}