nmon/internal/groups/integration_test.go

//go:build integration && linux

package groups

import (
	"context"
	"os"
	"path/filepath"
	"runtime"
	"sort"
	"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/p95"
	"github.com/v10networks/nmon/internal/rollup"
	"github.com/v10networks/nmon/internal/store"
	"github.com/v10networks/nmon/internal/testutil"
	"github.com/v10networks/nmon/internal/ulid"
	"github.com/v10networks/nmon/internal/vmsup"
)

func startLong(t *testing.T) (*vmsup.Querier, *rollup.LongStore) {
	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()})
	must(t, err)
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()
	must(t, sup.Start(ctx))
	t.Cleanup(func() { _ = sup.Stop(context.Background()) })
	q := vmsup.NewQuerier(sup)
	return q, rollup.NewLongStore(sup, q)
}

// TestRollupWrittenRatesYieldTheSameP95ThroughARealLongStore writes 5-minute
// rates with rollup's acknowledged writer, including a re-run (v2) that
// replaces some windows and a stale v1 write that arrives afterwards, and
// reads the interface and group p95 back, live and materialised.
func TestRollupWrittenRatesYieldTheSameP95ThroughARealLongStore(t *testing.T) {
	q, long := startLong(t)
	now := time.Now().UTC()
	clk := clock.NewFake(now)
	st := testutil.OpenStore(t, clk)
	ctx := context.Background()
	dev := &store.Device{Name: "core1", Kind: store.KindNetwork, Enabled: true}
	must(t, st.CreateDevice(ctx, dev))
	id := ulid.New(now)
	must(t, st.UpsertInterfaces(ctx, dev.ID, []store.Interface{{ID: id, IfIndex: 1, IfName: "eth0", HasHCCounters: true}}))
	g := store.Group{Name: "Acme", Kind: store.GroupCustomer}
	must(t, st.CreateGroup(ctx, &g))

	m := p95.MonthContaining(now, time.UTC).Prev()
	must(t, st.AddGroupMember(ctx, store.GroupMember{GroupID: g.ID, InterfaceID: id, Direction: store.DirAway, EffectiveFrom: m.Start}))
	const windows = 60
	var inVals, outVals []float64
	for i := 0; i < windows; i++ {
		ws := m.Start.Add(time.Duration(i) * p95.Window)
		in, out := float64(1000+i*10), float64(50+i)
		write := func(version int64, in, out float64) {
			v := rollup.VersionLabel(ws, version)
			must(t, long.WriteAcked(ctx, []metrics.Sample{
				metrics.New(longq.MetricRate5m, in, ws, "device", dev.Slug, "iface", id, "dir", "in", "v", v),
				metrics.New(longq.MetricRate5m, out, ws, "device", dev.Slug, "iface", id, "dir", "out", "v", v),
			}))
		}
		write(1, in, out)
		if i%7 == 0 { // a re-run replaces this window; the old value must not leak back
			in, out = in*3, out*3
			write(2, in, out)
			write(1, 1, 1) // a late, stale v1 write
		}
		inVals, outVals = append(inVals, in), append(outVals, out)
	}

	svc := p95.NewService(q, st, p95.Zones{Default: time.UTC}, clk)
	live, err := svc.P95(ctx, id, m.Label)
	must(t, err)
	if live.Source != p95.SourceLive || live.InBps != refP95(inVals) || live.OutBps != refP95(outVals) || !live.Partial() {
		t.Fatalf("live = %+v, want in %v out %v", live, refP95(inVals), refP95(outVals))
	}
	gsvc := NewService(Options{Query: q, Inventory: st, Zone: time.UTC, Clock: clk})
	glive, err := gsvc.GroupP95(ctx, g.ID, m.Label)
	must(t, err)
	if glive.InBps != refP95(outVals) || glive.OutBps != refP95(inVals) { // member direction is "away": swapped
		t.Fatalf("group live = %+v", glive.Result)
	}

	pj, err := p95.NewJob(p95.Options{Query: q, Inventory: st, Writer: long, Zones: p95.Zones{Default: time.UTC}, Clock: clk, Origin: m.Start})
	must(t, err)
	gj, err := NewJob(JobOptions{Service: gsvc, Query: q, Writer: long, Clock: clk, Origin: m.Start})
	must(t, err)
	must(t, pj.RunOnce(ctx))
	must(t, gj.RunOnce(ctx))
	mat, err := svc.P95(ctx, id, m.Label)
	must(t, err)
	gmat, err := gsvc.GroupP95(ctx, g.ID, m.Label)
	must(t, err)
	if mat.Source != p95.SourceMaterialised || mat.InBps != live.InBps || mat.OutBps != live.OutBps || !mat.Partial() {
		t.Fatalf("materialised = %+v, live = %+v", mat, live)
	}
	if gmat.Source != p95.SourceMaterialised || gmat.InBps != glive.InBps || gmat.MembersHash != glive.MembersHash {
		t.Fatalf("group materialised = %+v, live = %+v", gmat.Result, glive.Result)
	}
}

// refP95 is the naive reference: the smallest value whose cumulative share of
// the sorted samples reaches 95%, with the comparison done in integers.
func refP95(vals []float64) float64 {
	s := append([]float64(nil), vals...)
	sort.Float64s(s)
	for i, v := range s {
		if (i+1)*100 >= 95*len(s) {
			return v
		}
	}
	return s[len(s)-1]
}