nmon/internal/vmsup/supervisor_test.go

//go:build linux

package vmsup

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"
)

// fakeEnv points the supervisor at this test binary, which TestMain turns into
// a fake store, skipping the pin check.
func fakeEnv(t *testing.T) {
	t.Helper()
	exe, err := os.Executable()
	if err != nil {
		t.Fatal(err)
	}
	t.Setenv(envBinary, exe)
	t.Setenv(envUnverified, "1")
}

type events struct {
	mu sync.Mutex
	ev []Event
}

func (e *events) add(ev Event) { e.mu.Lock(); e.ev = append(e.ev, ev); e.mu.Unlock() }
func (e *events) count(typ, store string) int {
	e.mu.Lock()
	defer e.mu.Unlock()
	n := 0
	for _, ev := range e.ev {
		if ev.Type == typ && ev.Store == store {
			n++
		}
	}
	return n
}

func newFake(t *testing.T, ev *events, mod func(*Options)) *Supervisor {
	t.Helper()
	fakeEnv(t)
	o := Options{DataDir: t.TempDir(), OnEvent: ev.add, backoff: 20 * time.Millisecond, healthEvery: 200 * time.Millisecond}
	if mod != nil {
		mod(&o)
	}
	s, err := New(o)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	if err := s.Start(ctx); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = s.Stop(context.Background()) })
	return s
}

func status(s *Supervisor, store string) ChildStatus {
	for _, c := range s.Status() {
		if c.Store == store {
			return c
		}
	}
	return ChildStatus{}
}

func eventually(t *testing.T, d time.Duration, what string, f func() bool) {
	t.Helper()
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		if f() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("timed out after %s waiting for %s", d, what)
}

func TestStartStop(t *testing.T) {
	s := newFake(t, &events{}, nil)
	for _, st := range []string{StoreHot, StoreLong} {
		c := status(s, st)
		if !c.Healthy || c.PID == 0 || c.External {
			t.Fatalf("%s: %+v", st, c)
		}
		if !strings.HasPrefix(c.URL, "http://127.0.0.1:") {
			t.Fatalf("%s url %q not on loopback", st, c.URL)
		}
		// own process group, so signals aimed at nmon's group don't reach it
		if pg, err := syscall.Getpgid(c.PID); err != nil || pg != c.PID {
			t.Fatalf("%s pgid %d err %v, want own group %d", st, pg, err, c.PID)
		}
	}
	if s.HotURL() == s.LongURL() {
		t.Fatal("hot and long share a URL")
	}
	hot, long := status(s, StoreHot).PID, status(s, StoreLong).PID
	if err := s.Stop(context.Background()); err != nil {
		t.Fatal(err)
	}
	for _, pid := range []int{hot, long} {
		if err := syscall.Kill(pid, 0); err == nil {
			t.Fatalf("child %d still alive after Stop", pid)
		}
	}
	if s.Healthy(StoreHot) {
		t.Fatal("hot healthy after Stop")
	}
}

func TestRestartAfterKillWithin10s(t *testing.T) {
	ev := &events{}
	// Production backoff (1 s) and a 1 s health interval.
	s := newFake(t, ev, func(o *Options) { o.backoff = backoffBase; o.healthEvery = time.Second })
	old := status(s, StoreHot).PID
	other := status(s, StoreLong).PID

	killed := time.Now()
	if err := syscall.Kill(old, syscall.SIGKILL); err != nil {
		t.Fatal(err)
	}
	eventually(t, 10*time.Second, "hot restarted and healthy", func() bool {
		c := status(s, StoreHot)
		return c.Healthy && c.PID != 0 && c.PID != old
	})
	if d := time.Since(killed); d > 10*time.Second {
		t.Fatalf("restart took %s", d)
	}
	if status(s, StoreLong).PID != other {
		t.Fatal("long store was disturbed")
	}
	eventually(t, 2*time.Second, "events", func() bool {
		return ev.count(EventStoreRestarted, StoreHot) == 1 && ev.count(EventStoreRecovered, StoreHot) == 1
	})
	if ev.count(EventStoreFailed, StoreHot) != 0 {
		t.Fatal("unexpected store_failed")
	}
}

func TestStoreFailedAfterThreeRestarts(t *testing.T) {
	ev := &events{}
	s := newFake(t, ev, nil)

	killAndWait := func(wantNewChild bool) {
		old := status(s, StoreHot).PID
		if err := syscall.Kill(old, syscall.SIGKILL); err != nil {
			t.Fatal(err)
		}
		if !wantNewChild {
			return
		}
		eventually(t, 5*time.Second, "restart", func() bool {
			c := status(s, StoreHot)
			return c.Healthy && c.PID != 0 && c.PID != old
		})
	}
	for i := 0; i < 3; i++ { // three restarts are allowed
		killAndWait(true)
	}
	if ev.count(EventStoreFailed, StoreHot) != 0 {
		t.Fatal("store_failed raised too early")
	}
	killAndWait(false) // the fourth failure in the window is one too many
	eventually(t, 5*time.Second, "store_failed", func() bool { return ev.count(EventStoreFailed, StoreHot) == 1 })

	c := status(s, StoreHot)
	if !c.Failed || c.Healthy || c.PID != 0 {
		t.Fatalf("status after failure: %+v", c)
	}
	time.Sleep(500 * time.Millisecond) // 25x the backoff: a restart would have happened by now
	if c := status(s, StoreHot); c.PID != 0 || !c.Failed {
		t.Fatalf("restarted despite store_failed: %+v", c)
	}
	if !s.Healthy(StoreLong) || status(s, StoreLong).PID == 0 {
		t.Fatal("long store should keep running")
	}
	if err := s.Restart(StoreLong); err == nil {
		t.Fatal("Restart of a healthy store should be an error")
	}

	if err := s.Restart(StoreHot); err != nil {
		t.Fatal(err)
	}
	eventually(t, 5*time.Second, "hot resumed", func() bool {
		c := status(s, StoreHot)
		return c.Healthy && !c.Failed && c.PID != 0
	})
	if err := s.Restart(StoreHot); err == nil {
		t.Fatal("second Restart should be an error")
	}
	// the history was cleared: a new kill restarts rather than failing at once
	killAndWait(true)
	if ev.count(EventStoreFailed, StoreHot) != 1 {
		t.Fatal("history not cleared by Restart")
	}
}

func TestHungStoreIsKilledAndRestarted(t *testing.T) {
	// SIGSTOP leaves the process alive but not answering /health.
	ev := &events{}
	s := newFake(t, ev, nil)
	old := status(s, StoreHot).PID
	if err := syscall.Kill(old, syscall.SIGSTOP); err != nil {
		t.Fatal(err)
	}
	eventually(t, 10*time.Second, "hung store replaced", func() bool {
		c := status(s, StoreHot)
		return c.Healthy && c.PID != 0 && c.PID != old
	})
}

func TestChecksumMismatchRefusesStart(t *testing.T) {
	fakeEnv(t)
	t.Setenv(envUnverified, "") // verify against the pin: the test binary is not victoria-metrics
	s, err := New(Options{DataDir: t.TempDir()})
	if err != nil {
		t.Fatal(err)
	}
	err = s.Start(context.Background())
	if !errors.Is(err, ErrChecksum) {
		t.Fatalf("Start = %v, want ErrChecksum", err)
	}
	for _, c := range s.Status() {
		if c.PID != 0 {
			t.Fatalf("a child was started: %+v", c)
		}
	}
}

func TestBinaryTamperedBetweenStarts(t *testing.T) {
	// The hash is re-checked before every spawn, not only at Start.
	ev := &events{}
	exe, _ := os.Executable()
	data, err := os.ReadFile(exe)
	if err != nil {
		t.Fatal(err)
	}
	cp := t.TempDir() + "/vm"
	if err := os.WriteFile(cp, data, 0o755); err != nil {
		t.Fatal(err)
	}
	want, _ := fileSHA256(cp)
	s := newFake(t, ev, nil)
	s.spec = binarySpec{path: cp, want: want}
	hot := status(s, StoreHot).PID
	// swap in a modified binary, then kill the child: the restart must be refused
	f, err := os.OpenFile(cp, os.O_APPEND|os.O_WRONLY, 0)
	if err != nil {
		t.Fatal(err)
	}
	_, _ = f.WriteString("tampered")
	_ = f.Close()
	_ = syscall.Kill(hot, syscall.SIGKILL)
	eventually(t, 5*time.Second, "store_failed on checksum", func() bool { return ev.count(EventStoreFailed, StoreHot) == 1 })
	ev.mu.Lock()
	defer ev.mu.Unlock()
	for _, e := range ev.ev {
		if e.Type == EventStoreFailed && e.Store == StoreHot && !strings.Contains(e.Detail, "checksum") {
			t.Fatalf("detail %q does not mention the checksum", e.Detail)
		}
	}
}

func TestNoEmbeddedBinary(t *testing.T) {
	if len(embeddedGz) > 0 {
		t.Skip("built with vmembed")
	}
	t.Setenv(envBinary, "")
	s, err := New(Options{DataDir: t.TempDir()})
	if err != nil {
		t.Fatal(err)
	}
	if err := s.Start(context.Background()); !errors.Is(err, ErrNoBinary) {
		t.Fatalf("Start = %v, want ErrNoBinary", err)
	}
}

func TestExternalStores(t *testing.T) {
	var mu sync.Mutex
	var got []string
	mkSrv := func() *httptest.Server {
		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if r.URL.Path == "/health" {
				_, _ = w.Write([]byte("OK"))
				return
			}
			buf := make([]byte, 4096)
			n, _ := r.Body.Read(buf)
			mu.Lock()
			got = append(got, r.URL.Path+" "+string(buf[:n]))
			mu.Unlock()
		}))
	}
	hot, long := mkSrv(), mkSrv()
	defer hot.Close()
	defer long.Close()
	t.Setenv(envBinary, "") // nothing to extract: no child is needed
	s, err := New(Options{HotURL: hot.URL + "/", LongURL: long.URL})
	if err != nil {
		t.Fatal(err)
	}
	if err := s.Start(context.Background()); err != nil {
		t.Fatal(err)
	}
	if s.HotURL() != hot.URL || s.LongURL() != long.URL {
		t.Fatalf("urls %s %s", s.HotURL(), s.LongURL())
	}
	for _, c := range s.Status() {
		if !c.External || c.PID != 0 || !c.Healthy {
			t.Fatalf("%+v", c)
		}
	}
	if err := s.Restart(StoreHot); err == nil {
		t.Fatal("Restart of an external store should fail")
	}
	// Stop flushes registered queues while the store is still reachable.
	q := NewQueue(StoreHot, s, QueueOptions{})
	_ = q.Write(context.Background(), mk(1, time.Now()))
	if err := s.Stop(context.Background()); err != nil {
		t.Fatal(err)
	}
	mu.Lock()
	defer mu.Unlock()
	if len(got) != 1 || !strings.HasPrefix(got[0], "/api/v1/import/prometheus m{") {
		t.Fatalf("Stop did not flush the queue: %q", got)
	}
}

func TestNewValidation(t *testing.T) {
	if _, err := New(Options{}); err == nil {
		t.Fatal("missing DataDir should fail")
	}
	if _, err := New(Options{DataDir: "x", HotURL: "ftp://h"}); err == nil {
		t.Fatal("bad URL should fail")
	}
}