nmon/internal/backup/integration_test.go

//go:build integration && vmembed && linux

package backup

import (
	"bytes"
	"context"
	"fmt"
	"net/http"
	"path/filepath"
	"testing"
	"time"

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

// TestBackupRestoreRoundTripWithRealStores is the M6 backup gate: write series into both real stores, back
// up online, restore into a clean directory, and verify row counts and the sampled series against the
// restored stores, which are started for the check.
func TestBackupRestoreRoundTripWithRealStores(t *testing.T) {
	ctx := context.Background()
	dir := t.TempDir()
	keyFile := filepath.Join(t.TempDir(), "secret.key")
	if err := secrets.GenerateKeyFile(keyFile); err != nil {
		t.Fatal(err)
	}
	kr, err := secrets.LoadKeyring(keyFile)
	if err != nil {
		t.Fatal(err)
	}
	st, err := store.Open(ctx, store.DriverSQLite, filepath.Join(dir, "inventory.db"), kr, clock.Real())
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = st.Close() }()
	if err := st.Migrate(ctx); err != nil {
		t.Fatal(err)
	}
	for _, n := range []string{"sw1", "sw2"} {
		if err := st.CreateDevice(ctx, &store.Device{Name: n, Kind: store.KindNetwork, Enabled: true}); err != nil {
			t.Fatal(err)
		}
	}

	sup, err := vmsup.New(vmsup.Options{DataDir: dir})
	if err != nil {
		t.Fatal(err)
	}
	startCtx, cancel := context.WithTimeout(ctx, time.Minute)
	defer cancel()
	if err := sup.Start(startCtx); err != nil {
		t.Fatal(err)
	}
	defer func() { _ = sup.Stop(context.Background()) }()
	writeSeries(t, sup)

	out := filepath.Join(t.TempDir(), "nmon.tar")
	var log bytes.Buffer
	res, err := Backup(ctx, Options{
		DataDir: dir, OutPath: out, Store: st, Stores: StoresFromStatus(dir, sup.Status()), Log: &log, Settle: time.Minute,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Log(log.String())
	for _, sm := range res.Manifest.Stores {
		if len(sm.Series) < 2 {
			t.Fatalf("%s: only %d series fingerprinted", sm.Name, len(sm.Series))
		}
	}

	target := filepath.Join(t.TempDir(), "restored")
	rr, err := Restore(ctx, RestoreOptions{File: out, DataDir: target, Log: &log})
	if err != nil {
		t.Fatalf("restore: %v\n%s", err, log.String())
	}
	if rr.RowsVerified < 2 || rr.SeriesVerified < 4 {
		t.Fatalf("verified too little: %+v", rr)
	}
}

func writeSeries(t *testing.T, sup *vmsup.Supervisor) {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	base := time.Now().Add(-10 * time.Minute)
	for _, name := range []string{vmsup.StoreHot, vmsup.StoreLong} {
		q := vmsup.NewQueue(name, sup, vmsup.QueueOptions{FlushEvery: 100 * time.Millisecond})
		go q.Run(ctx)
		var batch []metrics.Sample
		for dev := 0; dev < 3; dev++ {
			for i := 0; i < 10; i++ {
				batch = append(batch, metrics.New("nmon_it_backup", float64(dev*100+i), base.Add(time.Duration(i)*30*time.Second),
					"device", fmt.Sprintf("d%d", dev), "iface", "i1"))
			}
		}
		if err := q.Write(ctx, batch); err != nil {
			t.Fatal(err)
		}
		deadline := time.Now().Add(20 * time.Second)
		for q.Depth() > 0 && time.Now().Before(deadline) {
			time.Sleep(100 * time.Millisecond)
		}
		if resp, err := http.Get(sup.URL(name) + "/internal/force_flush"); err == nil {
			_ = resp.Body.Close()
		}
	}
}