nmon/internal/api/integration_test.go

//go:build integration && linux

package api_test

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"testing"
	"time"

	"github.com/v10networks/nmon/internal/api"
	"github.com/v10networks/nmon/internal/clock"
	"github.com/v10networks/nmon/internal/metrics"
	"github.com/v10networks/nmon/internal/store"
	"github.com/v10networks/nmon/internal/testutil"
	"github.com/v10networks/nmon/internal/ulid"
	"github.com/v10networks/nmon/internal/vmsup"
)

// TestIntegrationGraphFromRealStores writes synthetic octet counters whose rate
// rises, and hourly rollups for the weeks before, into real hot and long
// stores, then asks the API for a one-hour graph and a stitched 40-day graph.
func TestIntegrationGraphFromRealStores(t *testing.T) {
	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()) })

	clk := clock.Real()
	st := testutil.OpenStore(t, clk)
	dev := &store.Device{Name: "Core", Kind: store.KindNetwork, Enabled: true, Profile: "generic"}
	if err := st.CreateDevice(ctx, dev); err != nil {
		t.Fatal(err)
	}
	ifc := store.Interface{ID: ulid.New(clk.Now()), DeviceID: dev.ID, IfIndex: 1, IfName: "xe-0/0/1", IfSpeed: 1e9, IfHighSpeed: 1000,
		AdminStatus: 1, OperStatus: 1, HasHCCounters: true}
	if err := st.UpsertInterfaces(ctx, dev.ID, []store.Interface{ifc}); err != nil {
		t.Fatal(err)
	}

	// 3 h of one-minute samples ending 2 min ago. The in rate climbs from 1 to 4 Mbit/s; out is a flat 500 kbit/s.
	end := time.Now().Add(-2 * time.Minute).Truncate(time.Minute)
	const minutes = 3 * 60
	var hot []metrics.Sample
	var in, out float64
	for i := 0; i <= minutes; i++ {
		ts := end.Add(-time.Duration(minutes-i) * time.Minute)
		inBps := 1e6 + 3e6*float64(i)/minutes
		in += inBps / 8 * 60
		out += 500e3 / 8 * 60
		hot = append(hot,
			metrics.New("ifHCInOctets", in, ts, "device", dev.Slug, "iface", ifc.ID),
			metrics.New("ifHCOutOctets", out, ts, "device", dev.Slug, "iface", ifc.ID))
	}
	// Hourly rollups for 40 days back to 31 days ago, 800 kbit/s in and 200 out.
	var long []metrics.Sample
	for h := 40 * 24; h >= 31*24; h-- {
		ts := time.Now().Add(-time.Duration(h) * time.Hour).Truncate(time.Hour)
		long = append(long,
			metrics.New("nmon_if_rate_bps_avg_1h", 800e3, ts, "device", dev.Slug, "iface", ifc.ID, "dir", "in", "v", "1"),
			metrics.New("nmon_if_rate_bps_avg_1h", 200e3, ts, "device", dev.Slug, "iface", ifc.ID, "dir", "out", "v", "1"))
	}
	for name, ss := range map[string][]metrics.Sample{vmsup.StoreHot: hot, vmsup.StoreLong: long} {
		q := vmsup.NewQueue(name, sup, vmsup.QueueOptions{MaxAge: 100 * 24 * time.Hour})
		if err := q.Write(ctx, ss); err != nil {
			t.Fatal(err)
		}
		if err := q.Flush(ctx); err != nil {
			t.Fatal(err)
		}
	}

	h := api.New(api.Deps{Store: st, Querier: vmsup.NewQuerier(sup), Poller: noRefresh{}, Clock: clk})
	get := func(q url.Values) api.GraphResponse {
		rec := httptest.NewRecorder()
		h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/graph?"+q.Encode(), nil))
		if rec.Code != http.StatusOK {
			t.Fatalf("graph: %d %s", rec.Code, rec.Body)
		}
		var g api.GraphResponse
		if err := json.Unmarshal(rec.Body.Bytes(), &g); err != nil {
			t.Fatal(err)
		}
		return g
	}
	// The store needs a moment to make flushed samples searchable.
	var g api.GraphResponse
	deadline := time.Now().Add(30 * time.Second)
	for {
		now := time.Now()
		g = get(url.Values{"iface": {ifc.ID}, "from": {now.Add(-time.Hour).UTC().Format(time.RFC3339)}, "to": {now.UTC().Format(time.RFC3339)},
			"max": {"true"}, "p95": {"true"}})
		if countValues(g.Series.In) > 30 || time.Now().After(deadline) {
			break
		}
		time.Sleep(500 * time.Millisecond)
	}
	if g.Tier != "raw" || g.StitchedAt != nil || g.StepSeconds != 60 {
		t.Fatalf("1h graph: tier=%s step=%d stitched=%v", g.Tier, g.StepSeconds, g.StitchedAt)
	}
	first, last := firstLast(g.Series.In)
	if first == nil || last == nil || *last <= *first*1.1 || *first < 2.5e6 || *last > 4.2e6 {
		t.Errorf("in series should rise from about 2.5 to 4 Mbit/s: first %v last %v", deref(first), deref(last))
	}
	if f, l := firstLast(g.Series.Out); f == nil || *f < 450e3 || *f > 550e3 || *l < 450e3 || *l > 550e3 {
		t.Errorf("out series should be a flat 500 kbit/s: %v %v", deref(f), deref(l))
	}
	if countValues(g.Series.InMax) < 30 {
		t.Errorf("max series: %d values", countValues(g.Series.InMax))
	}
	if g.P95 == nil || g.P95.InBPS == nil || *g.P95.InBPS < 2.5e6 || g.P95.Source != "live" || g.P95.Coverage <= 0 {
		t.Errorf("p95: %+v", g.P95)
	}

	now := time.Now()
	g = get(url.Values{"iface": {ifc.ID}, "from": {now.Add(-40 * 24 * time.Hour).UTC().Format(time.RFC3339)}, "to": {now.UTC().Format(time.RFC3339)}})
	if g.StitchedAt == nil || len(g.Tiers) != 2 || g.Tiers[0].Tier != "1h" {
		t.Fatalf("40d graph should be stitched: %+v", g.Tiers)
	}
	var fromLong int
	for _, p := range g.Series.In {
		if p.TS < g.StitchedAt.Unix() && p.V != nil {
			fromLong++
			if *p.V != 800e3 {
				t.Fatalf("long sample %d = %v, want 800000", p.TS, *p.V)
			}
		}
	}
	if fromLong < 150 {
		t.Errorf("only %d hourly samples came from the long store", fromLong)
	}
	if len(g.Partial) == 0 {
		t.Error("hot has 3 h of data in a 30-day window: the rest is partial")
	}

	// A year needs a coarser step: the store must accept the averaging subquery and average equal hours to the same value.
	g = get(url.Values{"iface": {ifc.ID}, "from": {now.Add(-365 * 24 * time.Hour).UTC().Format(time.RFC3339)}, "to": {now.UTC().Format(time.RFC3339)}})
	if g.StitchedAt == nil || g.Tiers[0].StepSeconds <= 3600 {
		t.Fatalf("1y tiers: %+v", g.Tiers)
	}
	var coarse int
	for _, p := range g.Series.In {
		if p.TS < g.StitchedAt.Unix() && p.V != nil {
			coarse++
			if *p.V < 799e3 || *p.V > 801e3 {
				t.Errorf("coarse long sample %d = %v, want 800000", p.TS, *p.V)
			}
		}
	}
	if coarse == 0 {
		t.Error("the 1y graph has no long-store samples")
	}
}

type noRefresh struct{}

func (noRefresh) Refresh() {}

func countValues(ps []api.Point) int {
	n := 0
	for _, p := range ps {
		if p.V != nil {
			n++
		}
	}
	return n
}

func firstLast(ps []api.Point) (first, last *float64) {
	for _, p := range ps {
		if p.V == nil {
			continue
		}
		if first == nil {
			first = p.V
		}
		last = p.V
	}
	return first, last
}

func deref(p *float64) float64 {
	if p == nil {
		return -1
	}
	return *p
}