nixos-router/pkgs/router-webui/metrics.go

package main

import (
	"bytes"
	"encoding/gob"
	"math"
	"os"
	"slices"
	"sync"
	"time"
)

// Metric history: every series keeps a ring of averages per resolution. A
// slot's index follows from its time, so writing needs no head pointer and
// gaps (the UI was not running) simply stay NaN.
//
//	5 s for 1 h, 1 min for 24 h, 15 min for 7 days, 1 h for 30 days.
var tiers = []struct {
	Step  int64 // seconds
	Slots int
}{
	{5, 720},
	{60, 1440},
	{900, 672},
	{3600, 720},
}

// Ranges the dashboard offers, and the tier each one reads.
var ranges = []struct {
	Name string
	Tier int
}{
	{"1h", 0},
	{"24h", 1},
	{"7d", 2},
	{"30d", 3},
}

func rangeTier(name string) (int, string) {
	for _, r := range ranges {
		if r.Name == name {
			return r.Tier, r.Name
		}
	}
	return 0, ranges[0].Name
}

type Tier struct {
	Step   int64
	Values []float64 // NaN: no data
	Last   int64     // start of the newest written bucket (unix seconds)
	// The bucket being filled.
	Bucket int64
	Sum    float64
	Count  int
}

func newTier(step int64, slots int) *Tier {
	t := &Tier{Step: step, Values: make([]float64, slots)}
	for i := range t.Values {
		t.Values[i] = math.NaN()
	}
	return t
}

func (t *Tier) slot(b int64) int { return int((b / t.Step) % int64(len(t.Values))) }

func (t *Tier) add(at int64, v float64) {
	b := at - at%t.Step
	if b != t.Bucket {
		t.flush()
		t.Bucket, t.Sum, t.Count = b, 0, 0
	}
	t.Sum += v
	t.Count++
}

// flush writes the bucket being filled into the ring.
func (t *Tier) flush() {
	if t.Count == 0 || t.Bucket < t.Last {
		return
	}
	n := int64(len(t.Values))
	// Clear the slots skipped since the last write (at most one full turn).
	for b, i := t.Last+t.Step, int64(0); b < t.Bucket && i < n; b, i = b+t.Step, i+1 {
		t.Values[t.slot(b)] = math.NaN()
	}
	t.Values[t.slot(t.Bucket)] = t.Sum / float64(t.Count)
	t.Last = t.Bucket
	t.Count = 0
}

// read returns the buckets of the last full turn of the ring up to now, the
// one being filled included, oldest first, and the first bucket's start.
func (t *Tier) read(now int64) (int64, []float64) {
	n := int64(len(t.Values))
	end := now - now%t.Step
	start := end - (n-1)*t.Step
	out := make([]float64, n)
	for i := range out {
		b := start + int64(i)*t.Step
		switch {
		case b == t.Bucket && t.Count > 0:
			out[i] = t.Sum / float64(t.Count)
		case b > t.Last || b <= t.Last-n*t.Step:
			out[i] = math.NaN()
		default:
			out[i] = t.Values[t.slot(b)]
		}
	}
	return start, out
}

type Series struct {
	Tiers []*Tier
}

func newSeries() *Series {
	s := &Series{}
	for _, t := range tiers {
		s.Tiers = append(s.Tiers, newTier(t.Step, t.Slots))
	}
	return s
}

// Event is a change of the uplink's state.
type Event struct {
	Time      time.Time
	Connected bool
	Detail    string
}

// Store is the history that survives restarts.
type Store struct {
	mu     sync.Mutex
	Series map[string]*Series
	// The uplink's state as last seen, and its recent changes.
	Connected bool
	Since     time.Time
	Events    []Event
}

const maxEvents = 50

func newStore() *Store { return &Store{Series: map[string]*Series{}} }

func (s *Store) Add(name string, at time.Time, v float64) {
	if math.IsNaN(v) || math.IsInf(v, 0) {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	ser := s.Series[name]
	if ser == nil {
		ser = newSeries()
		s.Series[name] = ser
	}
	for _, t := range ser.Tiers {
		t.add(at.Unix(), v)
	}
}

// Read returns a series at the resolution of tier: the start, step and values.
func (s *Store) Read(name string, tier int, now time.Time) (time.Time, time.Duration, []float64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	t := tiers[tier]
	ser := s.Series[name]
	if ser == nil {
		ser = newSeries() // all NaN
	}
	start, vals := ser.Tiers[tier].read(now.Unix())
	return time.Unix(start, 0), time.Duration(t.Step) * time.Second, vals
}

func (s *Store) Names() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	names := make([]string, 0, len(s.Series))
	for n := range s.Series {
		names = append(names, n)
	}
	slices.Sort(names)
	return names
}

// SetConnected records the uplink's state; a change becomes an event.
func (s *Store) SetConnected(connected bool, at time.Time, detail string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if connected == s.Connected && !s.Since.IsZero() {
		return
	}
	s.Connected, s.Since = connected, at
	s.Events = append(s.Events, Event{Time: at, Connected: connected, Detail: detail})
	if len(s.Events) > maxEvents {
		s.Events = s.Events[len(s.Events)-maxEvents:]
	}
}

func (s *Store) Uplink() (bool, time.Time, []Event) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.Connected, s.Since, slices.Clone(s.Events)
}

// The on-disk format; bump storeVersion when it changes incompatibly (an old
// file is then ignored).
const storeVersion = 1

type storeFile struct {
	Version   int
	Series    map[string]*Series
	Connected bool
	Since     time.Time
	Events    []Event
}

func (s *Store) Save(path string) error {
	s.mu.Lock()
	f := storeFile{storeVersion, s.Series, s.Connected, s.Since, s.Events}
	var buf bytes.Buffer
	err := gob.NewEncoder(&buf).Encode(f)
	s.mu.Unlock()
	if err != nil {
		return err
	}
	return writeFileAtomic(path, buf.Bytes(), 0o600)
}

func loadStore(path string) (*Store, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return newStore(), err
	}
	var f storeFile
	if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&f); err != nil || f.Version != storeVersion {
		return newStore(), err
	}
	s := &Store{Series: map[string]*Series{}, Connected: f.Connected, Since: f.Since, Events: f.Events}
	for name, ser := range f.Series {
		// Keep only series whose shape matches the current tiers.
		ok := len(ser.Tiers) == len(tiers)
		for i := 0; ok && i < len(tiers); i++ {
			ok = ser.Tiers[i].Step == tiers[i].Step && len(ser.Tiers[i].Values) == tiers[i].Slots
		}
		if ok {
			s.Series[name] = ser
		}
	}
	return s, nil
}