nmon/internal/rollup/fakes_test.go

package rollup

import (
	"context"
	"encoding/json"
	"fmt"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/v10networks/nmon/internal/clock"
	"github.com/v10networks/nmon/internal/metrics"
	"github.com/v10networks/nmon/internal/store"
)

// t0 is the start of the synthetic data: aligned to the hour so windows line up.
var t0 = time.Date(2026, 10, 20, 8, 0, 0, 0, time.UTC)

// fakeVM answers the queries the job issues from in-memory series, and keeps
// everything written to "long" so tests can read it back.
type fakeVM struct {
	mu     sync.Mutex
	hot    map[string][]point // key device|metric|iface
	gauges map[string][]gaugeSeries
	long   []metrics.Sample
	writes int
}

type gaugeSeries struct {
	labels map[string]string
	pts    []point
}

func newFakeVM() *fakeVM {
	return &fakeVM{hot: map[string][]point{}, gauges: map[string][]gaugeSeries{}}
}

func hotKey(dev, metric, iface string) string { return dev + "|" + metric + "|" + iface }

func (f *fakeVM) addPoint(dev, metric, iface string, ts time.Time, v float64) {
	f.mu.Lock()
	defer f.mu.Unlock()
	k := hotKey(dev, metric, iface)
	f.hot[k] = append(f.hot[k], point{ts: ts.UnixMilli(), v: v})
	sort.Slice(f.hot[k], func(a, b int) bool { return f.hot[k][a].ts < f.hot[k][b].ts })
}

var (
	reDevice = regexp.MustCompile(`device="([^"]+)"`)
	reRange  = regexp.MustCompile(`\[(\d+)s\]`)
	reNames  = regexp.MustCompile(`__name__=~"([^"]+)"`)
	reAvgOT  = regexp.MustCompile(`^avg_over_time\(([A-Za-z0-9_]+)\[(\d+)s\]\)$`)
)

type resultRow struct {
	Metric map[string]string `json:"metric"`
	Values [][]any           `json:"values,omitempty"`
	Value  []any             `json:"value,omitempty"`
}

func respond(rows []resultRow) []byte {
	b, _ := json.Marshal(map[string]any{"status": "success", "data": map[string]any{"resultType": "matrix", "result": rows}})
	return b
}

func pt(p point) []any { return []any{float64(p.ts) / 1000, strconv.FormatFloat(p.v, 'g', -1, 64)} }

func (f *fakeVM) Query(_ context.Context, storeName, q string, ts time.Time) ([]byte, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if m := reAvgOT.FindStringSubmatch(q); m != nil {
		return f.gaugeAvg(m[1], ts), nil
	}
	if strings.HasPrefix(q, "count(") {
		return nil, fmt.Errorf("fake: unexpected %s", q)
	}
	dev := reDevice.FindStringSubmatch(q)
	rng := reRange.FindStringSubmatch(q)
	if dev == nil || rng == nil {
		return nil, fmt.Errorf("fake: cannot parse query %q", q)
	}
	secs, _ := strconv.Atoi(rng[1])
	lo, hi := ts.Add(-time.Duration(secs)*time.Second).UnixMilli(), ts.UnixMilli()
	if storeName == "hot" {
		return f.hotMatrix(dev[1], lo, hi), nil
	}
	names := map[string]bool{}
	for _, n := range strings.Split(reNames.FindStringSubmatch(q)[1], "|") {
		names[n] = true
	}
	return f.longMatrix(dev[1], names, lo, hi), nil
}

func (f *fakeVM) hotMatrix(dev string, lo, hi int64) []byte {
	var rows []resultRow
	keys := make([]string, 0, len(f.hot))
	for k := range f.hot {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		parts := strings.Split(k, "|")
		if parts[0] != dev {
			continue
		}
		r := resultRow{Metric: map[string]string{"__name__": parts[1], "device": dev, "iface": parts[2]}}
		for _, p := range f.hot[k] {
			if p.ts > lo && p.ts <= hi {
				r.Values = append(r.Values, pt(p))
			}
		}
		if len(r.Values) > 0 {
			rows = append(rows, r)
		}
	}
	return respond(rows)
}

func (f *fakeVM) gaugeAvg(name string, ts time.Time) []byte {
	var rows []resultRow
	hi := ts.UnixMilli()
	lo := ts.Add(-FiveMinutes).UnixMilli()
	for _, g := range f.gauges[name] {
		var sum float64
		n := 0
		for _, p := range g.pts {
			if p.ts > lo && p.ts <= hi {
				sum += p.v
				n++
			}
		}
		if n > 0 {
			rows = append(rows, resultRow{Metric: g.labels, Value: pt(point{ts: hi, v: sum / float64(n)})})
		}
	}
	return respond(rows)
}

func (f *fakeVM) longMatrix(dev string, names map[string]bool, lo, hi int64) []byte {
	byKey := map[string]*resultRow{}
	var order []string
	for _, s := range f.long {
		if !names[s.Name] || s.TS.UnixMilli() <= lo || s.TS.UnixMilli() > hi {
			continue
		}
		lbl := map[string]string{"__name__": s.Name}
		for _, l := range s.Labels {
			lbl[l.Name] = l.Value
		}
		if lbl["device"] != dev {
			continue
		}
		k := s.SeriesKey()
		r := byKey[k]
		if r == nil {
			r = &resultRow{Metric: lbl}
			byKey[k] = r
			order = append(order, k)
		}
		r.Values = append(r.Values, pt(point{ts: s.TS.UnixMilli(), v: s.Value}))
	}
	sort.Strings(order)
	rows := make([]resultRow, 0, len(order))
	for _, k := range order {
		rows = append(rows, *byKey[k])
	}
	return respond(rows)
}

// WriteAcked implements Writer.
func (f *fakeVM) WriteAcked(_ context.Context, s []metrics.Sample) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.long = append(f.long, s...)
	f.writes++
	return nil
}

// latest returns, per series key without v, the value of the highest version at ts.
func (f *fakeVM) latest(name string, ts time.Time) map[string]float64 {
	f.mu.Lock()
	defer f.mu.Unlock()
	best := map[string]int64{}
	out := map[string]float64{}
	for _, s := range f.long {
		if s.Name != name || !s.TS.Equal(ts) {
			continue
		}
		var kv []string
		var v int64
		for _, l := range s.Labels {
			if l.Name == "v" {
				v, _ = strconv.ParseInt(l.Value, 10, 64)
				continue
			}
			kv = append(kv, l.Name+"="+l.Value)
		}
		k := strings.Join(kv, ",")
		if v > best[k] {
			best[k], out[k] = v, s.Value
		}
	}
	return out
}

func (f *fakeVM) samples(name string) []metrics.Sample {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []metrics.Sample
	for _, s := range f.long {
		if s.Name == name {
			out = append(out, s)
		}
	}
	return out
}

func label(s metrics.Sample, name string) string {
	for _, l := range s.Labels {
		if l.Name == name {
			return l.Value
		}
	}
	return ""
}

// fakeInv is an in-memory Inventory.
type fakeInv struct {
	devs   []store.Device
	ifaces map[int64][]store.Interface
}

func (f *fakeInv) ListDevices(_ context.Context, _ store.DeviceFilter, p store.Page) ([]store.Device, int64, error) {
	lo := min(int(p.Offset), len(f.devs))
	hi := min(lo+int(p.Limit), len(f.devs))
	return f.devs[lo:hi], int64(len(f.devs)), nil
}

func (f *fakeInv) ListInterfaces(_ context.Context, id int64, _ bool) ([]store.Interface, error) {
	return f.ifaces[id], nil
}

func (f *fakeInv) add(slug string, ifs ...store.Interface) {
	id := int64(len(f.devs) + 1)
	f.devs = append(f.devs, store.Device{ID: id, Slug: slug})
	if f.ifaces == nil {
		f.ifaces = map[int64][]store.Interface{}
	}
	f.ifaces[id] = ifs
}

type fakeQueue struct {
	mu    sync.Mutex
	depth int
}

func (q *fakeQueue) Depth() int    { q.mu.Lock(); defer q.mu.Unlock(); return q.depth }
func (q *fakeQueue) Capacity() int { return 100 }
func (q *fakeQueue) set(d int)     { q.mu.Lock(); q.depth = d; q.mu.Unlock() }

// crashState fails the final commit of a window, as a crash after the long write would.
type crashState struct {
	store.Rollup
	crash bool
}

func (c *crashState) PutRollupWindow(ctx context.Context, w store.RollupWindow) error {
	if c.crash && w.Status != store.RollupPending {
		return fmt.Errorf("simulated crash before commit")
	}
	return c.Rollup.PutRollupWindow(ctx, w)
}

func openState(t *testing.T, clk clock.Clock) store.Store {
	t.Helper()
	ctx := context.Background()
	st, err := store.Open(ctx, store.DriverSQLite, filepath.Join(t.TempDir(), "inv.db"), nil, clk)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = st.Close() })
	if err := st.Migrate(ctx); err != nil {
		t.Fatal(err)
	}
	return st
}

func hcIface(id string) store.Interface { return store.Interface{ID: id, HasHCCounters: true} }

// rig wires a Job to the fakes.
type rig struct {
	vm    *fakeVM
	inv   *fakeInv
	st    store.Store
	clk   *clock.Fake
	state *crashState
	q     *fakeQueue
	job   *Job
}

func newRig(t *testing.T, now time.Time, mut func(*Options)) *rig {
	t.Helper()
	r := &rig{vm: newFakeVM(), inv: &fakeInv{}, clk: clock.NewFake(now), q: &fakeQueue{}}
	r.st = openState(t, r.clk)
	r.state = &crashState{Rollup: r.st}
	o := Options{Query: r.vm, Inventory: r.inv, State: r.state, Writer: r.vm, Clock: r.clk, Queue: r.q, Origin: t0, PollInterval: time.Minute}
	if mut != nil {
		mut(&o)
	}
	var err error
	if r.job, err = New(o); err != nil {
		t.Fatal(err)
	}
	return r
}

// addLinear adds a counter growing at bytesPerSec, sampled every step with a per-sample timing jitter.
// base is the counter value at start; wrapAt, if non-zero, is the modulus applied to the stored value.
func (f *fakeVM) addLinear(dev, metric, iface string, start time.Time, n int, step time.Duration, bytesPerSec, base, wrapAt float64) {
	f.mu.Lock()
	defer f.mu.Unlock()
	k := hotKey(dev, metric, iface)
	for i := 0; i < n; i++ {
		off := time.Duration(i)*step + time.Duration((i*7)%5)*300*time.Millisecond // up to 1.2 s of jitter
		v := base + bytesPerSec*off.Seconds()
		if wrapAt > 0 {
			v = float64(uint64(v) % uint64(wrapAt))
		}
		f.hot[k] = append(f.hot[k], point{ts: start.Add(off).UnixMilli(), v: v})
	}
}