nmon/internal/api/graph_test.go

package api_test

import (
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net/url"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/v10networks/nmon/internal/api"
	"github.com/v10networks/nmon/internal/store"
	"github.com/v10networks/nmon/internal/vmsup"
)

// graphVM fakes the two stores. hot and long say whether a store has a sample
// at ts; the value is a function of direction and time so tests can check which store answered.
type graphVM struct {
	hot, long func(ts int64) bool
	hotVal    func(dir string, ts int64) float64
	longVal   func(dir string, ts int64) float64
	longErr   error
}

func dirOf(expr string) string {
	if strings.Contains(expr, `dir="in"`) || strings.Contains(expr, "InOctets") {
		return "in"
	}
	return "out"
}

func (g *graphVM) fn(storeName, expr string, from, to time.Time, step time.Duration) ([]byte, error) {
	has, val := g.hot, g.hotVal
	if storeName == "long" {
		if g.longErr != nil {
			return nil, g.longErr
		}
		has, val = g.long, g.longVal
	}
	dir := dirOf(expr)
	vals := map[int64]float64{}
	for ts := from.Unix(); ts <= to.Unix(); ts += int64(step / time.Second) {
		if has == nil || has(ts) {
			vals[ts] = val(dir, ts)
		}
	}
	if len(vals) == 0 {
		return emptyMatrix, nil
	}
	return matrix(mrow{labels: map[string]string{"dir": dir}, vals: vals}), nil
}

func rising(dir string, ts int64) float64 {
	if dir == "in" {
		return 1e6 + float64(ts%100000)
	}
	return 2e6 + float64(ts%100000)
}

func newGraphEnv(t *testing.T, vm *graphVM, mut ...func(*store.Interface)) (*env, store.Interface) {
	e := newEnv(t, 0)
	e.q.fn = vm.fn
	d := e.device(t, "Core", "", "")
	return e, e.iface(t, d, 1, "xe-0/0/1", "uplink", mut...)
}

func iso(ts time.Time) string { return ts.UTC().Format(time.RFC3339) }

func graphURL(id string, from, to time.Time, extra string) string {
	return "/api/v1/graph?iface=" + id + "&from=" + url.QueryEscape(iso(from)) + "&to=" + url.QueryEscape(iso(to)) + extra
}

func TestGraphRawTierWithinHotRetention(t *testing.T) {
	vm := &graphVM{hotVal: rising}
	e, i := newGraphEnv(t, vm)
	var g api.GraphResponse
	rec := e.getJSON(t, graphURL(i.ID, t0.Add(-time.Hour), t0, ""), &g)
	assertNoLeaks(t, "graph", rec.Body.Bytes())
	if g.Tier != "raw" || g.StepSeconds != 60 || g.StitchedAt != nil || len(g.Tiers) != 0 || g.Unreliable {
		t.Fatalf("raw graph: %+v", g)
	}
	if g.Partial == nil || len(g.Partial) != 0 {
		t.Errorf("partial must be [] when complete: %v", g.Partial)
	}
	if g.Totals == nil || g.Totals.InBytes == nil || g.Totals.OutBytes == nil || *g.Totals.InBytes <= 0 ||
		g.Totals.CoveredSeconds != 3600 || g.Totals.RangeSeconds != 3600 {
		t.Errorf("totals: %+v", g.Totals)
	}
	if g.Iface.ID != i.ID || g.Iface.IfName != "xe-0/0/1" || g.Iface.IfAlias != "uplink" || g.Iface.SpeedBPS != 1_000_000_000 || g.Iface.Device.Name != "Core" {
		t.Errorf("iface: %+v", g.Iface)
	}
	if len(g.Series.In) != 61 || len(g.Series.Out) != 61 || g.Series.InMax != nil {
		t.Fatalf("series: %d in, %d out", len(g.Series.In), len(g.Series.Out))
	}
	first, last := g.Series.In[0], g.Series.In[60]
	if first.TS != t0.Add(-time.Hour).Unix() || last.TS != t0.Unix() || first.V == nil || *first.V != rising("in", first.TS) {
		t.Errorf("first %+v last %+v", first, last)
	}
	if *g.Series.Out[10].V != rising("out", g.Series.Out[10].TS) {
		t.Errorf("out series %+v", g.Series.Out[10])
	}
	qs := e.q.queries()
	if len(qs) != 2 || qs[0].store != "hot" || qs[0].step != time.Minute ||
		qs[0].expr != `rate(ifHCInOctets{iface="`+i.ID+`"}[600s])*8` || !strings.Contains(qs[1].expr, "ifHCOutOctets") {
		t.Errorf("queries: %+v", qs)
	}
	var raw map[string]json.RawMessage
	_ = json.Unmarshal(rec.Body.Bytes(), &raw)
	for _, k := range []string{"stitched_at", "tiers", "unreliable_reason", "p95"} {
		if _, ok := raw[k]; ok {
			t.Errorf("%s must be omitted from a plain raw graph", k)
		}
	}
	if !strings.HasPrefix(string(raw["series"]), `{"in":[[`) {
		t.Errorf("points must be [ts,value] pairs: %.60s", raw["series"])
	}
}

func TestGraphUses32BitCountersWhenNoHC(t *testing.T) {
	e, i := newGraphEnv(t, &graphVM{hotVal: rising}, func(i *store.Interface) { i.HasHCCounters = false })
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-time.Hour), t0, ""), &g)
	if q := e.q.queries()[0].expr; !strings.Contains(q, "ifInOctets{") || strings.Contains(q, "HC") {
		t.Errorf("32-bit query: %s", q)
	}
}

func TestGraphStepDefaultsAndLimits(t *testing.T) {
	for _, tc := range []struct {
		rng  time.Duration
		step int64
	}{{time.Hour, 60}, {24 * time.Hour, 120}, {7 * 24 * time.Hour, 660}, {30 * 24 * time.Hour, 2640}} {
		e, i := newGraphEnv(t, &graphVM{hotVal: rising})
		var g api.GraphResponse
		e.getJSON(t, graphURL(i.ID, t0.Add(-tc.rng), t0, ""), &g)
		if g.StepSeconds != tc.step || g.Tier != "raw" || g.StitchedAt != nil {
			t.Errorf("range %s: step %d tier %s stitched %v, want step %d raw", tc.rng, g.StepSeconds, g.Tier, g.StitchedAt, tc.step)
		}
		if n := len(g.Series.In); n < 700 && tc.rng >= 24*time.Hour || n > 1001 {
			t.Errorf("range %s: %d points", tc.rng, n)
		}
		for _, p := range g.Series.In {
			if p.TS%g.StepSeconds != 0 {
				t.Fatalf("range %s: sample %d is not aligned to the %d s step", tc.rng, p.TS, g.StepSeconds)
			}
		}
	}
	e, i := newGraphEnv(t, &graphVM{hotVal: rising})
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-time.Hour), t0, "&step=300"), &g)
	if g.StepSeconds != 300 || len(g.Series.In) != 13 {
		t.Errorf("explicit step: %d, %d points", g.StepSeconds, len(g.Series.In))
	}
	isProblem(t, e.do("GET", graphURL(i.ID, t0.Add(-30*24*time.Hour), t0, "&step=1"), ""), 422)
}

func TestGraphStitchesHourlyLongDataToRawHot(t *testing.T) {
	join := t0.Add(-30 * 24 * time.Hour)
	vm := &graphVM{
		hot: func(ts int64) bool { return ts >= join.Unix() }, hotVal: rising,
		longVal: func(dir string, ts int64) float64 { return -1 }, // long values are recognisable
	}
	e, i := newGraphEnv(t, vm)
	from := t0.Add(-40 * 24 * time.Hour)
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, from, t0, "&max=true"), &g)
	if g.StitchedAt == nil || g.Tier != "raw" || len(g.Tiers) != 2 {
		t.Fatalf("stitched graph: stitched_at=%v tier=%s tiers=%+v", g.StitchedAt, g.Tier, g.Tiers)
	}
	if g.Tiers[0].Tier != "1h" || g.Tiers[0].StepSeconds != 3600 || g.Tiers[1].Tier != "raw" || g.Tiers[1].StepSeconds != g.StepSeconds {
		t.Errorf("tiers: %+v step=%d", g.Tiers, g.StepSeconds)
	}
	if !g.Tiers[0].To.Equal(*g.StitchedAt) || !g.Tiers[1].From.Equal(*g.StitchedAt) || g.StitchedAt.Before(join) ||
		g.StitchedAt.Sub(join) >= time.Duration(g.StepSeconds)*time.Second {
		t.Errorf("stitched_at %v should be the hot retention edge %v rounded up by less than a step", g.StitchedAt, join)
	}
	var long, rawN int
	prev := int64(0)
	for _, p := range g.Series.In {
		if p.TS <= prev {
			t.Fatalf("series not strictly increasing at %d", p.TS)
		}
		prev = p.TS
		if p.V == nil {
			t.Fatalf("hole at %d: both stores have data here", p.TS)
		}
		if *p.V == -1 {
			long++
			if p.TS >= g.StitchedAt.Unix() || p.TS%3600 != 0 {
				t.Errorf("long sample %d misplaced", p.TS)
			}
		} else {
			rawN++
			if p.TS < g.StitchedAt.Unix() {
				t.Errorf("raw sample %d before the join", p.TS)
			}
		}
	}
	if long < 200 || rawN < 400 {
		t.Errorf("%d long and %d raw samples", long, rawN)
	}
	if len(g.Partial) != 0 {
		t.Errorf("no gaps expected: %v", g.Partial)
	}
	if len(g.Series.InMax) != len(g.Series.In) || len(g.Series.OutMax) != len(g.Series.Out) {
		t.Errorf("max series lengths %d/%d", len(g.Series.InMax), len(g.Series.OutMax))
	}
	var sawLong, sawHot bool
	for _, c := range e.q.queries() {
		switch c.store {
		case "long":
			sawLong = true
			if !strings.Contains(c.expr, "sum without(v)") || !strings.Contains(c.expr, `iface="`+i.ID+`"`) ||
				(!strings.Contains(c.expr, "nmon_if_rate_bps_avg_1h") && !strings.Contains(c.expr, "nmon_if_rate_bps_max_1h")) {
				t.Errorf("long query: %s", c.expr)
			}
		case "hot":
			sawHot = true
			if c.from.Before(*g.StitchedAt) {
				t.Errorf("hot queried before the join: %v", c.from)
			}
		}
	}
	if !sawLong || !sawHot {
		t.Error("a stitched graph must query both stores")
	}
}

func TestGraphLongRangeUsesCoarserStepAndHourlyAverages(t *testing.T) {
	e, i := newGraphEnv(t, &graphVM{hotVal: rising, longVal: rising})
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-365*24*time.Hour), t0, ""), &g)
	if g.StitchedAt == nil || g.Tiers[0].StepSeconds%3600 != 0 || g.Tiers[0].StepSeconds <= 3600 {
		t.Fatalf("1y tiers: %+v", g.Tiers)
	}
	var longExpr string
	for _, c := range e.q.queries() {
		if c.store == "long" {
			longExpr = c.expr
		}
	}
	if !strings.Contains(longExpr, "avg_over_time((sum without(v)") || !strings.Contains(longExpr, ":3600s])") {
		t.Errorf("coarse long query should average the hourly points: %s", longExpr)
	}
}

func TestGraphOlderThanHotIsLongTierOnly(t *testing.T) {
	vm := &graphVM{hot: func(int64) bool { return false }, hotVal: rising, longVal: rising}
	e, i := newGraphEnv(t, vm)
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-90*24*time.Hour), t0.Add(-60*24*time.Hour), ""), &g)
	if g.Tier != "1h" || g.StepSeconds != 3600 || g.StitchedAt != nil || len(g.Tiers) != 0 || len(g.Series.In) < 700 {
		t.Errorf("long-only graph: tier=%s step=%d stitched=%v %d points", g.Tier, g.StepSeconds, g.StitchedAt, len(g.Series.In))
	}
	for _, c := range e.q.queries() {
		if c.store != "long" {
			t.Errorf("hot queried for a range wholly beyond hot retention: %+v", c)
		}
	}
}

func TestGraphReportsPartialWhereNeitherStoreHasData(t *testing.T) {
	// Until M5's rollup exists the long store is empty: the stitched span is partial, honestly.
	join := t0.Add(-30 * 24 * time.Hour)
	vm := &graphVM{hot: func(ts int64) bool { return ts >= join.Unix() }, hotVal: rising, long: func(int64) bool { return false }}
	e, i := newGraphEnv(t, vm)
	from := t0.Add(-40 * 24 * time.Hour)
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, from, t0, ""), &g)
	if g.StitchedAt == nil || len(g.Partial) != 1 {
		t.Fatalf("partial: %+v stitched %v", g.Partial, g.StitchedAt)
	}
	p := g.Partial[0]
	if p.From.After(from.Add(time.Hour)) || p.To.Before(join) || p.To.After(join.Add(2*time.Hour)) {
		t.Errorf("gap %v..%v, want about %v..%v", p.From, p.To, from, join)
	}
	for _, pt := range g.Series.In {
		if pt.TS < g.StitchedAt.Unix() && pt.V != nil {
			t.Fatalf("a null expected before the join, got %v at %d", *pt.V, pt.TS)
		}
	}

	// a short hole in raw data is a partial span; a single missed sample is not
	e2, i2 := newGraphEnv(t, &graphVM{hotVal: rising, hot: func(ts int64) bool {
		m := (ts - t0.Add(-time.Hour).Unix()) / 60
		return m < 20 || m > 50 && m != 55
	}})
	e2.getJSON(t, graphURL(i2.ID, t0.Add(-time.Hour), t0, ""), &g)
	if len(g.Partial) != 1 || !g.Partial[0].From.Equal(t0.Add(-41*time.Minute)) || !g.Partial[0].To.Equal(t0.Add(-9*time.Minute)) {
		t.Errorf("hole: %+v", g.Partial)
	}

	// no data at all: one partial span covering the range
	e3, i3 := newGraphEnv(t, &graphVM{hotVal: rising, hot: func(int64) bool { return false }})
	e3.getJSON(t, graphURL(i3.ID, t0.Add(-time.Hour), t0, ""), &g)
	if len(g.Partial) != 1 || !g.Partial[0].From.Equal(t0.Add(-time.Hour)) || !g.Partial[0].To.Equal(t0) {
		t.Errorf("empty: %+v", g.Partial)
	}
	// data that starts late: a leading gap
	e4, i4 := newGraphEnv(t, &graphVM{hotVal: rising, hot: func(ts int64) bool { return ts >= t0.Add(-20*time.Minute).Unix() }})
	e4.getJSON(t, graphURL(i4.ID, t0.Add(-time.Hour), t0, ""), &g)
	if len(g.Partial) != 1 || !g.Partial[0].From.Equal(t0.Add(-time.Hour)) || !g.Partial[0].To.Equal(t0.Add(-20*time.Minute)) {
		t.Errorf("leading gap: %+v", g.Partial)
	}
}

func TestGraphLongStoreDownIsPartialNotAnError(t *testing.T) {
	join := t0.Add(-30 * 24 * time.Hour)
	vm := &graphVM{hot: func(ts int64) bool { return ts >= join.Unix() }, hotVal: rising, longErr: fmt.Errorf("%w: long", vmsup.ErrStoreDown)}
	e, i := newGraphEnv(t, vm)
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-40*24*time.Hour), t0, ""), &g)
	if g.StitchedAt == nil || len(g.Partial) != 1 {
		t.Errorf("long store down: %+v", g.Partial)
	}
}

func TestGraphUnreliableInterfaceHasNoSeries(t *testing.T) {
	for name, tc := range map[string]struct {
		mut    func(*store.Interface)
		reason string
	}{
		"32-bit fast link":  {func(i *store.Interface) { i.Unreliable, i.HasHCCounters = true, false }, "counter_32bit_fast_link"},
		"discontinuities":   {func(i *store.Interface) { i.Unreliable, i.CounterDiscontinuity = true, 3 }, "counter_discontinuity"},
		"unreliable for HC": {func(i *store.Interface) { i.Unreliable = true }, "counter_unreliable"},
	} {
		t.Run(name, func(t *testing.T) {
			e, i := newGraphEnv(t, &graphVM{hotVal: rising}, tc.mut)
			var g api.GraphResponse
			rec := e.getJSON(t, graphURL(i.ID, t0.Add(-time.Hour), t0, "&max=true&p95=true"), &g)
			if !g.Unreliable || g.UnreliableReason != tc.reason || len(g.Series.In) != 0 || len(g.Series.Out) != 0 ||
				g.P95 != nil || g.Totals != nil || g.Partial == nil || g.Tier != "raw" || g.StepSeconds != 60 {
				t.Errorf("unreliable graph: %+v", g)
			}
			if n := len(e.q.queries()); n != 0 {
				t.Errorf("an unreliable interface must not be queried, %d queries", n)
			}
			if !strings.Contains(rec.Body.String(), `"in":[],"out":[]`) {
				t.Errorf("series must be empty arrays: %s", rec.Body)
			}
		})
	}
}

func TestGraphDirectionFilterAndMax(t *testing.T) {
	e, i := newGraphEnv(t, &graphVM{hotVal: rising})
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-time.Hour), t0, "&dir=out&max=true"), &g)
	if len(g.Series.In) != 0 || len(g.Series.Out) != 61 || len(g.Series.OutMax) != 61 || g.Series.InMax != nil {
		t.Errorf("dir=out: in=%d out=%d out_max=%d", len(g.Series.In), len(g.Series.Out), len(g.Series.OutMax))
	}
	if g.Totals == nil || g.Totals.InBytes != nil || g.Totals.OutBytes == nil {
		t.Errorf("dir=out totals: %+v", g.Totals)
	}
	qs := e.q.queries()
	if len(qs) != 2 {
		t.Fatalf("queries %d", len(qs))
	}
	for _, c := range qs {
		if strings.Contains(c.expr, "In") {
			t.Errorf("dir=out queried the in counter: %s", c.expr)
		}
	}
	if !strings.Contains(qs[1].expr, "max_over_time((irate(ifHCOutOctets{iface=") || !strings.Contains(qs[1].expr, "[60s:60s])") {
		t.Errorf("max query: %s", qs[1].expr)
	}
	isProblem(t, e.do("GET", graphURL(i.ID, t0.Add(-time.Hour), t0, "&dir=sideways"), ""), 422)
}

// referenceP95 is an independent nearest-rank implementation: the smallest
// value such that at least 95% of the values are at or below it.
func referenceP95(vals []float64) float64 {
	s := append([]float64(nil), vals...)
	sort.Float64s(s)
	for _, v := range s {
		n := 0
		for _, x := range s {
			if x <= v {
				n++
			}
		}
		if float64(n)*100 >= 95*float64(len(s)) {
			return v
		}
	}
	return math.NaN()
}

func TestGraphP95IsNearestRankOverFiveMinuteAverages(t *testing.T) {
	// 24 h of 5-minute averages; one hour of the day has no data.
	from := t0.Add(-24 * time.Hour)
	vals := func(dir string, ts int64) float64 {
		k := float64((ts / 300 * 7919) % 1000) // scrambled but deterministic
		if dir == "out" {
			k *= 3
		}
		return k * 1000
	}
	hole := func(ts int64) bool { return ts < t0.Add(-5*time.Hour).Unix() || ts >= t0.Add(-4*time.Hour).Unix() }
	e, i := newGraphEnv(t, &graphVM{hotVal: vals, hot: hole})
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, from, t0, "&p95=true"), &g)
	if g.P95 == nil || g.P95.Source != "live" || g.P95.InBPS == nil || g.P95.OutBPS == nil {
		t.Fatalf("p95: %+v", g.P95)
	}
	var in, out []float64
	for ts := from.Unix(); ts <= t0.Unix(); ts += 300 {
		if hole(ts) {
			in, out = append(in, vals("in", ts)), append(out, vals("out", ts))
		}
	}
	if *g.P95.InBPS != referenceP95(in) || *g.P95.OutBPS != referenceP95(out) {
		t.Errorf("p95 in %v out %v, reference %v / %v", *g.P95.InBPS, *g.P95.OutBPS, referenceP95(in), referenceP95(out))
	}
	agg := make([]float64, len(in))
	for k := range in {
		agg[k] = in[k] + out[k]
	}
	if g.P95.AggregateBPS == nil || *g.P95.AggregateBPS != referenceP95(agg) {
		t.Errorf("aggregate p95 %v, reference %v", g.P95.AggregateBPS, referenceP95(agg))
	}
	wantCov := float64(len(in)) / (24 * 12)
	if math.Abs(g.P95.Coverage-wantCov) > 0.01 || g.P95.Coverage >= 1 {
		t.Errorf("coverage %v, want about %v", g.P95.Coverage, wantCov)
	}
	var withoutP95 api.GraphResponse
	e.getJSON(t, graphURL(i.ID, from, t0, ""), &withoutP95)
	if withoutP95.P95 != nil {
		t.Error("p95 only when asked")
	}
}

func TestGraphErrors(t *testing.T) {
	e, i := newGraphEnv(t, &graphVM{hotVal: rising})
	isProblem(t, e.do("GET", "/api/v1/graph", ""), 422)
	isProblem(t, e.do("GET", "/api/v1/graph?iface=01ARZ3NDEKTSV4RRFFQ69G5FAV", ""), 404)
	isProblem(t, e.do("GET", graphURL(i.ID, t0, t0.Add(-time.Hour), ""), ""), 422)
	isProblem(t, e.do("GET", "/api/v1/graph?iface="+i.ID+"&from=yesterday", ""), 422)
	isProblem(t, e.do("GET", "/api/v1/graph?iface="+i.ID+"&step=0", ""), 422)
	e.q.fn = func(string, string, time.Time, time.Time, time.Duration) ([]byte, error) {
		return nil, fmt.Errorf("%w: hot", vmsup.ErrStoreDown)
	}
	isProblem(t, e.do("GET", graphURL(i.ID, t0.Add(-time.Hour), t0, ""), ""), 503)
	e.q.fn = func(string, string, time.Time, time.Time, time.Duration) ([]byte, error) {
		return nil, errors.New("boom")
	}
	isProblem(t, e.do("GET", graphURL(i.ID, t0.Add(-time.Hour), t0, ""), ""), 502)
	// defaults: to=now, from=1h earlier
	e.q.fn = (&graphVM{hotVal: rising}).fn
	var g api.GraphResponse
	e.getJSON(t, "/api/v1/graph?iface="+i.ID, &g)
	if len(g.Series.In) != 61 || g.Series.In[60].TS != t0.Unix() {
		t.Errorf("default range: %d points", len(g.Series.In))
	}
}

func TestGraphAggregateP95NeedsBothDirections(t *testing.T) {
	e, i := newGraphEnv(t, &graphVM{hotVal: rising})
	var g api.GraphResponse
	e.getJSON(t, graphURL(i.ID, t0.Add(-2*time.Hour), t0, "&p95=true&dir=in"), &g)
	if g.P95 == nil || g.P95.InBPS == nil || g.P95.AggregateBPS != nil {
		t.Fatalf("one direction asked for: p95 %+v", g.P95)
	}
}