nmon/internal/api/graph_longtier_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"
)

// TestIntegrationLongTierTakesHighestVersionWithoutLeak writes rewritten and
// unrewritten hourly windows into a real long store and reads them back through
// the graph: the highest v wins even when its value is the lower one, and a
// window holding only an old version does not borrow the newer version of the
// hour before it.
func TestIntegrationLongTierTakesHighestVersionWithoutLeak(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)
	}

	// Four hours 45 days back, so the range lies wholly before hot retention and only the long tier answers.
	h0 := time.Now().Add(-45 * 24 * time.Hour).Truncate(time.Hour)
	at := func(i int) time.Time { return h0.Add(time.Duration(i) * time.Hour) }
	w := func(i int, v string, val float64) metrics.Sample {
		return metrics.New("nmon_if_rate_bps_avg_1h", val, at(i), "device", dev.Slug, "iface", ifc.ID, "dir", "in", "v", v)
	}
	long := []metrics.Sample{
		w(0, "202601001", 900), w(0, "202601002", 100), // rewritten lower: v2 must win
		w(1, "202601001", 50), // only an old version, right after hour 0 whose newest v is higher
		w(3, "202601005", 7000),
		// hour 2 has nothing at all: it must stay empty
	}
	q := vmsup.NewQueue(vmsup.StoreLong, sup, vmsup.QueueOptions{MaxAge: 100 * 24 * time.Hour})
	if err := q.Write(ctx, long); 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})
	query := url.Values{"iface": {ifc.ID}, "dir": {"in"},
		"from": {at(0).UTC().Format(time.RFC3339)}, "to": {at(3).UTC().Format(time.RFC3339)}, "step": {"3600"}}
	var g api.GraphResponse
	deadline := time.Now().Add(30 * time.Second)
	for {
		rec := httptest.NewRecorder()
		h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/graph?"+query.Encode(), nil))
		if rec.Code != http.StatusOK {
			t.Fatalf("graph: %d %s", rec.Code, rec.Body)
		}
		if err := json.Unmarshal(rec.Body.Bytes(), &g); err != nil {
			t.Fatal(err)
		}
		if countValues(g.Series.In) >= 2 || time.Now().After(deadline) {
			break
		}
		time.Sleep(500 * time.Millisecond)
	}
	if g.Tier != "1h" {
		t.Fatalf("tier = %s, want 1h", g.Tier)
	}
	got := map[int64]*float64{}
	for _, p := range g.Series.In {
		got[p.TS] = p.V
	}
	if v := got[at(0).Unix()]; v == nil || *v != 100 {
		t.Errorf("hour 0 = %v, want 100 (highest v wins, not highest value)", deref(v))
	}
	if v := got[at(1).Unix()]; v == nil || *v != 50 {
		t.Errorf("hour 1 = %v, want 50 (no borrowing from the previous window's newer version)", deref(v))
	}
	if v := got[at(2).Unix()]; v != nil {
		t.Errorf("hour 2 = %v, want empty: a lookback leaked an earlier window into the gap", *v)
	}
	if v := got[at(3).Unix()]; v == nil || *v != 7000 {
		t.Errorf("hour 3 = %v, want 7000", deref(v))
	}
}