nmon/internal/rollup/integration_test.go

//go:build integration && linux

package rollup

import (
	"context"
	"fmt"
	"math"
	"net/http"
	"os"
	"path/filepath"
	"runtime"
	"testing"
	"time"

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

type itEnv struct {
	sup     *vmsup.Supervisor
	qr      *vmsup.Querier
	long    *LongStore
	st      store.Store
	state   *crashState
	clk     *clock.Fake
	start   time.Time // first sample
	device  string
	ifaceID string
}

func startStores(t *testing.T) (*vmsup.Supervisor, *vmsup.Querier) {
	t.Helper()
	bin, err := filepath.Abs(filepath.Join("..", "vmsup", "bin", "victoria-metrics-prod-linux-"+runtime.GOARCH))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(bin); err != nil {
		t.Skipf("store binary not found (%v); run scripts/fetch-vm.sh", err)
	}
	t.Setenv("NMON_VM_BINARY", bin)
	t.Setenv("NMON_VM_BINARY_UNVERIFIED", "")
	sup, err := vmsup.New(vmsup.Options{DataDir: t.TempDir()})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()
	if err := sup.Start(ctx); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sup.Stop(context.Background()) })
	return sup, vmsup.NewQuerier(sup)
}

func forceFlush(t *testing.T, sup *vmsup.Supervisor, name string) {
	t.Helper()
	resp, err := http.Get(sup.URL(name) + "/internal/force_flush")
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
}

// newITEnv writes three hours of synthetic counters (one-minute samples, no
// jitter, ending a minute and a half ago) for one two-direction interface and
// one gauge into a real hot store. Inbound runs 1..12 Mbit/s over each hour's
// twelve 5-minute windows; outbound is a flat 2 Mbit/s.
func newITEnv(t *testing.T) *itEnv {
	t.Helper()
	sup, qr := startStores(t)
	ctx := context.Background()
	now := time.Now().UTC()
	e := &itEnv{sup: sup, qr: qr, long: NewLongStore(sup, qr), clk: clock.NewFake(now), start: floorTo(now, OneHour).Add(-3 * time.Hour)}
	e.st = openState(t, e.clk)
	e.state = &crashState{Rollup: e.st}

	d := &store.Device{Name: "core1", Kind: store.KindNetwork, Enabled: true}
	if err := e.st.CreateDevice(ctx, d); err != nil {
		t.Fatal(err)
	}
	e.device, e.ifaceID = d.Slug, "01HZZZZZZZZZZZZZZZZZZZZZ01"
	if err := e.st.UpsertInterfaces(ctx, d.ID, []store.Interface{{ID: e.ifaceID, IfIndex: 1, IfName: "eth0", HasHCCounters: true}}); err != nil {
		t.Fatal(err)
	}

	hot := vmsup.NewQueue(vmsup.StoreHot, sup, vmsup.QueueOptions{MaxAge: 72 * time.Hour, MaxSamples: 10_000_000})
	var ss []metrics.Sample
	var cin, cout float64
	end := floorTo(now.Add(-90*time.Second), time.Minute)
	for ts := e.start; !ts.After(end); ts = ts.Add(time.Minute) {
		win := max(0, int((ts.Sub(e.start)-time.Nanosecond)/FiveMinutes)) // the minute ending at ts
		cin += float64(1+win%12) * mbit * 60
		cout += 2 * mbit * 60
		load := 10.0
		if ts.Sub(e.start) >= 90*time.Minute {
			load = 30
		}
		ss = append(ss,
			metrics.New("ifHCInOctets", 3e12+cin, ts, "device", e.device, "iface", e.ifaceID),
			metrics.New("ifHCOutOctets", 4e12+cout, ts, "device", e.device, "iface", e.ifaceID),
			metrics.New("hrProcessorLoad", load, ts, "device", e.device, "component", "1"))
	}
	if err := hot.Write(ctx, ss); err != nil {
		t.Fatal(err)
	}
	if err := hot.Flush(ctx); err != nil {
		t.Fatal(err)
	}
	forceFlush(t, sup, vmsup.StoreHot)
	return e
}

func (e *itEnv) job(t *testing.T, mut func(*Options)) *Job {
	t.Helper()
	o := Options{
		Query: e.qr, Inventory: e.st, State: e.state, Writer: e.long, Deleter: e.long, Clock: e.clk, Origin: e.start,
		Gauges: func() []string { return []string{"hrProcessorLoad"} },
	}
	if mut != nil {
		mut(&o)
	}
	j, err := New(o)
	if err != nil {
		t.Fatal(err)
	}
	return j
}

// latest reads a long-store selector with the reader-side highest-version rule.
func (e *itEnv) latest(t *testing.T, selector string, ts time.Time) (val float64, n int) {
	t.Helper()
	body, err := e.qr.Query(context.Background(), vmsup.StoreLong, longq.LatestVersion(selector), ts)
	if err != nil {
		t.Fatal(err)
	}
	rows, err := parseResult(body)
	if err != nil {
		t.Fatal(err)
	}
	for _, r := range rows {
		if len(r.pts) > 0 {
			val = r.pts[0].v
			n++
		}
	}
	return val, n
}

func (e *itEnv) count(t *testing.T, q string, ts time.Time) float64 {
	t.Helper()
	body, err := e.qr.Query(context.Background(), vmsup.StoreLong, q, ts)
	if err != nil {
		t.Fatal(err)
	}
	v, err := firstValue(body)
	if err != nil {
		t.Fatal(err)
	}
	return v
}

func TestIntegrationThreeHoursOfCountersRollUpInRealStores(t *testing.T) {
	e := newITEnv(t)
	if err := e.job(t, nil).RunOnce(context.Background()); err != nil {
		t.Fatal(err)
	}
	sel := func(m, dir string) string {
		return fmt.Sprintf(`%s{device=%q,iface=%q,dir=%q}`, m, e.device, e.ifaceID, dir)
	}
	for k := 0; k < 3*12; k++ {
		ws := e.start.Add(time.Duration(k) * FiveMinutes)
		in, n := e.latest(t, sel(longq.MetricRate5m, "in"), ws)
		want := float64(1+k%12) * 1e6
		if n != 1 || math.Abs(in-want) > 0.005*want {
			t.Fatalf("window %d in = %v (n=%d), want %v", k, in, n, want)
		}
		if out, _ := e.latest(t, sel(longq.MetricRate5m, "out"), ws); math.Abs(out-2e6) > 0.005*2e6 {
			t.Fatalf("window %d out = %v, want 2e6", k, out)
		}
	}
	for h := 0; h < 2; h++ { // the third hour may still be open
		hs := e.start.Add(time.Duration(h) * OneHour)
		if mx, n := e.latest(t, sel(longq.MetricRateMax1h, "in"), hs); n != 1 || math.Abs(mx-12e6) > 0.005*12e6 {
			t.Errorf("hour %d max = %v (n=%d), want the largest of the 12 rates, 12e6", h, mx, n)
		}
		if av, _ := e.latest(t, sel(longq.MetricRateAvg1h, "in"), hs); math.Abs(av-6.5e6) > 0.005*6.5e6 {
			t.Errorf("hour %d avg = %v, want 6.5e6", h, av)
		}
	}
	g := fmt.Sprintf(`hrProcessorLoad_avg_5m{device=%q,component="1"}`, e.device)
	if v, n := e.latest(t, g, e.start.Add(time.Hour)); n != 1 || v != 10 {
		t.Errorf("gauge 5m avg = %v n=%d", v, n)
	}
	gm := fmt.Sprintf(`hrProcessorLoad_max_1h{device=%q,component="1"}`, e.device)
	if v, _ := e.latest(t, gm, e.start.Add(time.Hour)); v != 30 {
		t.Errorf("gauge hourly max = %v, want 30", v)
	}
}

func TestIntegrationCrashAfterLongWriteGivesOneValuePerTimestamp(t *testing.T) {
	e := newITEnv(t)
	ctx := context.Background()
	e.state.crash = true
	if err := e.job(t, nil).RunOnce(ctx); err == nil {
		t.Fatal("expected the simulated crash")
	}
	e.state.crash = false
	if err := e.job(t, nil).RunOnce(ctx); err != nil { // a fresh Job: the restarted process
		t.Fatal(err)
	}
	// The window that crashed is the oldest in the redo span.
	ws := floorTo(e.clk.Now().Add(-time.Minute), FiveMinutes).Add(-2 * time.Hour)
	w, err := e.st.GetRollupWindow(ctx, ws, store.ResolutionFiveMinute)
	if err != nil || w.Version != 2 || w.Status != store.RollupDone {
		t.Fatalf("crashed window record = %+v, %v; want done v2", w, err)
	}
	sel := fmt.Sprintf(`%s{device=%q,iface=%q,dir="in"}`, longq.MetricRate5m, e.device, e.ifaceID)
	if raw := e.count(t, `count(last_over_time(`+sel+`[1s]))`, ws); raw != 2 {
		t.Fatalf("store holds %v versions at the timestamp, want 2 (v1 from before the crash, v2 after)", raw)
	}
	if _, n := e.latest(t, sel, ws); n != 1 {
		t.Fatalf("latest-version read returned %d series, want exactly 1", n)
	}
}

func TestIntegrationLatestVersionPrefersHighestVNotHighestValueAndDoesNotLeak(t *testing.T) {
	e := newITEnv(t)
	ctx := context.Background()
	if err := e.job(t, nil).RunOnce(ctx); err != nil {
		t.Fatal(err)
	}
	ws := e.start.Add(time.Hour) // window 12: 1 Mbit/s
	lower := metrics.New(longq.MetricRate5m, 123, ws, "device", e.device, "iface", e.ifaceID, "dir", "in", "v", VersionLabel(ws, 7))
	if err := e.long.WriteAcked(ctx, []metrics.Sample{lower}); err != nil {
		t.Fatal(err)
	}
	sel := fmt.Sprintf(`%s{device=%q,iface=%q,dir="in"}`, longq.MetricRate5m, e.device, e.ifaceID)
	if v, n := e.latest(t, sel, ws); n != 1 || v != 123 {
		t.Errorf("window with v7 = %v (n=%d), want the v7 value 123 even though v1 is larger", v, n)
	}
	if v, _ := e.latest(t, sel, ws.Add(FiveMinutes)); math.Abs(v-2e6) > 0.005*2e6 {
		t.Errorf("next window = %v: v7 of the previous window leaked forward", v)
	}
}

func TestIntegrationPruneDropsOnlyWholeExpiredMonths(t *testing.T) {
	e := newITEnv(t)
	ctx := context.Background()
	now := e.clk.Now()
	ts := floorTo(now.Add(-2*time.Hour), FiveMinutes)
	prev := ts.AddDate(0, -1, 0) // only its v label's month matters to the matcher
	old := metrics.New(longq.MetricRate5m, 1, ts, "device", e.device, "iface", e.ifaceID, "dir", "in", "v", VersionLabel(prev, 1))
	cur := metrics.New(longq.MetricRate5m, 2, ts, "device", e.device, "iface", e.ifaceID, "dir", "out", "v", VersionLabel(ts, 1))
	gauge := metrics.New("hrProcessorLoad_avg_5m", 3, ts, "component", "1", "device", e.device, "v", VersionLabel(prev, 1))
	hourly := metrics.New(longq.MetricRateMax1h, 4, ts, "device", e.device, "iface", e.ifaceID, "dir", "in", "v", VersionLabel(prev, 1))
	for _, s := range []metrics.Sample{old, cur, gauge, hourly} {
		if err := e.long.WriteAcked(ctx, []metrics.Sample{s}); err != nil {
			t.Fatal(err)
		}
	}
	j := e.job(t, func(o *Options) { o.FiveMinuteRetention = time.Hour; o.LongRetention = 90 * 24 * time.Hour })
	if err := j.Prune(ctx); err != nil {
		t.Fatal(err)
	}
	has := func(name string) bool {
		return e.count(t, fmt.Sprintf(`count(last_over_time({__name__=%q,device=%q}[1s]))`, name, e.device), ts) > 0
	}
	if has("hrProcessorLoad_avg_5m") {
		t.Error("expired gauge 5m series survived")
	}
	if !has(longq.MetricRateMax1h) {
		t.Error("hourly series must never be pruned")
	}
	body, _ := e.qr.Query(ctx, vmsup.StoreLong, fmt.Sprintf(`last_over_time(%s{device=%q}[1s])`, longq.MetricRate5m, e.device), ts)
	rows, _ := parseResult(body)
	if len(rows) != 1 || rows[0].labels["dir"] != "out" {
		t.Errorf("after prune the 5m series are %v, want only the current month's", rows)
	}
}