nmon/internal/vmsup/integration_test.go

//go:build integration && linux

package vmsup

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"syscall"
	"testing"
	"time"

	"github.com/v10networks/nmon/internal/metrics"
)

// realStores starts both stores from the real binary in internal/vmsup/bin
// (scripts/fetch-vm.sh), verified against the pinned hash.
func realStores(t *testing.T) *Supervisor {
	t.Helper()
	bin, err := filepath.Abs(filepath.Join("bin", "victoria-metrics-prod-linux-"+runtime.GOARCH))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(bin); err != nil {
		t.Skipf("real store binary not found (%v); run scripts/fetch-vm.sh", err)
	}
	t.Setenv(envBinary, bin)
	t.Setenv(envUnverified, "")
	s, err := New(Options{DataDir: t.TempDir(), Logger: nil})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)
	defer cancel()
	if err := s.Start(ctx); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = s.Stop(context.Background()) })
	return s
}

func queryValue(t *testing.T, qr *Querier, store, q string) (float64, bool) {
	t.Helper()
	b, err := qr.Query(context.Background(), store, q, time.Time{})
	if err != nil {
		return 0, false
	}
	var r struct {
		Data struct {
			Result []struct {
				Value [2]any `json:"value"`
			} `json:"result"`
		} `json:"data"`
	}
	if err := json.Unmarshal(b, &r); err != nil || len(r.Data.Result) != 1 {
		return 0, false
	}
	var v float64
	_, _ = fmt.Sscanf(r.Data.Result[0].Value[1].(string), "%g", &v)
	return v, true
}

func TestIntegrationWriteAndQuery(t *testing.T) {
	s := realStores(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	qr := NewQuerier(s)

	for _, store := range []string{StoreHot, StoreLong} {
		q := NewQueue(store, s, QueueOptions{FlushEvery: 100 * time.Millisecond})
		go q.Run(ctx)
		name := "nmon_it_" + store
		if err := q.Write(ctx, []metrics.Sample{
			metrics.New(name, 42, time.Now(), "device", "d1", "iface", "i1"),
		}); err != nil {
			t.Fatal(err)
		}
		// VM makes new samples searchable after its in-memory flush (about
		// 5-10 s); /internal/force_flush brings that forward for the test.
		deadline := time.Now().Add(15 * time.Second)
		for {
			if q.Depth() == 0 {
				if resp, err := http.Get(s.URL(store) + "/internal/force_flush"); err == nil {
					_ = resp.Body.Close()
				}
			}
			if v, ok := queryValue(t, qr, store, name+`{device="d1"}`); ok {
				t.Logf("%s queryable after %s", store, time.Since(deadline.Add(-15*time.Second)).Round(10*time.Millisecond))
				if v != 42 {
					t.Fatalf("%s value %v", store, v)
				}
				break
			}
			if time.Now().After(deadline) {
				t.Fatalf("%s: sample never became queryable (depth %d)", store, q.Depth())
			}
			time.Sleep(200 * time.Millisecond)
		}
	}
	// the hot sample is not in long and vice versa
	if _, ok := queryValue(t, qr, StoreLong, `nmon_it_hot`); ok {
		t.Fatal("hot sample found in long")
	}
}

func TestIntegrationKillRestartsWithin10s(t *testing.T) {
	s := realStores(t)
	old := status(s, StoreHot).PID
	killed := time.Now()
	if err := syscall.Kill(old, syscall.SIGKILL); err != nil {
		t.Fatal(err)
	}
	eventually(t, 10*time.Second, "hot healthy again", func() bool {
		c := status(s, StoreHot)
		return c.Healthy && c.PID != 0 && c.PID != old
	})
	t.Logf("hot healthy again %s after kill -9", time.Since(killed).Round(10*time.Millisecond))
	if _, err := NewQuerier(s).Query(context.Background(), StoreHot, "1", time.Time{}); err != nil {
		t.Fatal(err)
	}
}

func TestIntegrationLoopbackOnly(t *testing.T) {
	s := realStores(t)
	for _, c := range s.Status() {
		inodes := socketInodes(t, c.PID)
		if len(inodes) == 0 {
			t.Fatalf("%s: no sockets found for pid %d", c.Store, c.PID)
		}
		listening := 0
		for _, f := range []string{"/proc/net/tcp", "/proc/net/tcp6"} {
			for _, l := range listenSockets(t, f) {
				if !inodes[l.inode] {
					continue
				}
				listening++
				ok := l.addr == "0100007F" || l.addr == "00000000000000000000000001000000" ||
					l.addr == "0000000000000000FFFF00000100007F"
				if !ok {
					t.Errorf("%s (pid %d) listens on %s:%d in %s, want loopback only", c.Store, c.PID, l.addr, l.port, f)
				}
			}
		}
		if listening == 0 {
			t.Errorf("%s: no listening socket found", c.Store)
		}
	}
}

func socketInodes(t *testing.T, pid int) map[string]bool {
	t.Helper()
	dir := fmt.Sprintf("/proc/%d/fd", pid)
	ents, err := os.ReadDir(dir)
	if err != nil {
		t.Fatal(err)
	}
	out := map[string]bool{}
	for _, e := range ents {
		l, err := os.Readlink(filepath.Join(dir, e.Name()))
		if err == nil && strings.HasPrefix(l, "socket:[") {
			out[strings.TrimSuffix(strings.TrimPrefix(l, "socket:["), "]")] = true
		}
	}
	return out
}

type listener struct {
	addr  string
	port  int64
	inode string
}

// listenSockets parses /proc/net/tcp{,6} and returns sockets in LISTEN state (0A).
func listenSockets(t *testing.T, path string) []listener {
	t.Helper()
	f, err := os.Open(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		t.Fatal(err)
	}
	defer func() { _ = f.Close() }()
	var out []listener
	sc := bufio.NewScanner(f)
	sc.Scan() // header
	for sc.Scan() {
		fs := strings.Fields(sc.Text())
		if len(fs) < 10 || fs[3] != "0A" {
			continue
		}
		ap := strings.Split(fs[1], ":")
		var port int64
		_, _ = fmt.Sscanf(ap[1], "%X", &port)
		out = append(out, listener{addr: ap[0], port: port, inode: fs[9]})
	}
	return out
}