nmon/internal/vmsup/queue_test.go

package vmsup

import (
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

func mk(n int, ts time.Time) []metrics.Sample {
	out := make([]metrics.Sample, n)
	for i := range out {
		out[i] = metrics.New("m", float64(i), ts, "device", "d1")
	}
	return out
}

// fakeTarget is a store address with a switchable health flag.
type fakeTarget struct {
	url     string
	healthy atomic.Bool
}

func (f *fakeTarget) URL(string) string   { return f.url }
func (f *fakeTarget) Healthy(string) bool { return f.healthy.Load() }

// recorder is an httptest store that records import bodies.
type recorder struct {
	srv    *httptest.Server
	mu     sync.Mutex
	bodies []string
	status atomic.Int32
}

func newRecorder(t *testing.T) *recorder {
	r := &recorder{}
	r.status.Store(http.StatusNoContent)
	r.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		b, _ := io.ReadAll(req.Body)
		if req.URL.Path != "/api/v1/import/prometheus" || req.Method != http.MethodPost {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		if st := int(r.status.Load()); st >= 300 {
			w.WriteHeader(st)
			return
		}
		r.mu.Lock()
		r.bodies = append(r.bodies, string(b))
		r.mu.Unlock()
		w.WriteHeader(http.StatusNoContent)
	}))
	t.Cleanup(r.srv.Close)
	return r
}

func (r *recorder) lines() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	var out []string
	for _, b := range r.bodies {
		out = append(out, strings.Split(strings.TrimSpace(b), "\n")...)
	}
	return out
}

func (r *recorder) posts() int { r.mu.Lock(); defer r.mu.Unlock(); return len(r.bodies) }

func droppedFor(name string) float64 {
	return selfmon.Default.Counter("nmon_dropped_samples_total", "path", name).Value()
}

func TestQueueHoldsWhileDownThenFlushes(t *testing.T) {
	rec := newRecorder(t)
	tg := &fakeTarget{url: rec.srv.URL}
	q := NewQueue(t.Name(), tg, QueueOptions{Clock: clock.NewFake(time.Unix(1000, 0))})
	ctx := context.Background()

	if err := q.Write(ctx, mk(3, time.UnixMilli(1_700_000_000_000))); err != nil {
		t.Fatal(err)
	}
	if err := q.Flush(ctx); !errors.Is(err, ErrStoreDown) {
		t.Fatalf("Flush while down = %v", err)
	}
	if q.Depth() != 3 || rec.posts() != 0 {
		t.Fatalf("depth %d posts %d", q.Depth(), rec.posts())
	}

	tg.healthy.Store(true)
	if err := q.Flush(ctx); err != nil {
		t.Fatal(err)
	}
	if q.Depth() != 0 {
		t.Fatalf("depth %d after flush", q.Depth())
	}
	want := []string{
		`m{device="d1"} 0 1700000000000`,
		`m{device="d1"} 1 1700000000000`,
		`m{device="d1"} 2 1700000000000`,
	}
	got := rec.lines()
	if strings.Join(got, "|") != strings.Join(want, "|") {
		t.Fatalf("got %q", got)
	}
	if hw := selfmon.Default.Gauge("nmon_write_queue_high_water", "queue", t.Name()).Value(); hw != 3 {
		t.Fatalf("high water %v", hw)
	}
	if d := selfmon.Default.Gauge("nmon_write_queue_depth", "queue", t.Name()).Value(); d != 0 {
		t.Fatalf("depth gauge %v", d)
	}
}

func TestQueueDropsAfterMaxAge(t *testing.T) {
	rec := newRecorder(t)
	tg := &fakeTarget{url: rec.srv.URL}
	fc := clock.NewFake(time.Unix(1000, 0))
	q := NewQueue(t.Name(), tg, QueueOptions{Clock: fc})
	ctx := context.Background()
	before := droppedFor(t.Name())

	_ = q.Write(ctx, mk(5, fc.Now()))
	fc.Advance(59 * time.Second)
	_ = q.Flush(ctx)
	if q.Depth() != 5 || droppedFor(t.Name()) != before {
		t.Fatalf("dropped early: depth %d", q.Depth())
	}
	fc.Advance(2 * time.Second) // 61 s old
	_ = q.Write(ctx, mk(2, fc.Now()))
	if q.Depth() != 2 {
		t.Fatalf("depth %d, want only the fresh 2", q.Depth())
	}
	if d := droppedFor(t.Name()) - before; d != 5 {
		t.Fatalf("dropped %v, want 5", d)
	}

	// and the survivors are delivered on recovery
	tg.healthy.Store(true)
	if err := q.Flush(ctx); err != nil {
		t.Fatal(err)
	}
	if len(rec.lines()) != 2 {
		t.Fatalf("delivered %d lines", len(rec.lines()))
	}
}

func TestQueueFullDropsOldest(t *testing.T) {
	rec := newRecorder(t)
	tg := &fakeTarget{url: rec.srv.URL}
	fc := clock.NewFake(time.Unix(1000, 0))
	q := NewQueue(t.Name(), tg, QueueOptions{Clock: fc, MaxSamples: 10, BatchSize: 5})
	ctx := context.Background()
	if q.Capacity() != 10 {
		t.Fatal(q.Capacity())
	}
	before := droppedFor(t.Name())
	for i := 0; i < 3; i++ {
		s := mk(5, fc.Now())
		for j := range s {
			s[j].Value = float64(100*i + j)
		}
		_ = q.Write(ctx, s)
	}
	if q.Depth() != 10 {
		t.Fatalf("depth %d", q.Depth())
	}
	if d := droppedFor(t.Name()) - before; d != 5 {
		t.Fatalf("dropped %v", d)
	}
	tg.healthy.Store(true)
	if err := q.Flush(ctx); err != nil {
		t.Fatal(err)
	}
	got := strings.Join(rec.lines(), "\n")
	if strings.Contains(got, " 0 ") && strings.Contains(got, " 3 ") && !strings.Contains(got, " 100 ") {
		t.Fatal("oldest batch should have been dropped")
	}
	if strings.Contains(got, `m{device="d1"} 4 `) || !strings.Contains(got, `m{device="d1"} 100 `) || !strings.Contains(got, `m{device="d1"} 204 `) {
		t.Fatalf("unexpected content:\n%s", got)
	}
}

func TestQueueBatching(t *testing.T) {
	rec := newRecorder(t)
	tg := &fakeTarget{url: rec.srv.URL}
	tg.healthy.Store(true)
	q := NewQueue(t.Name(), tg, QueueOptions{Clock: clock.NewFake(time.Unix(1000, 0)), BatchSize: 2})
	_ = q.Write(context.Background(), mk(5, time.UnixMilli(1)))
	if err := q.Flush(context.Background()); err != nil {
		t.Fatal(err)
	}
	if rec.posts() != 3 || len(rec.lines()) != 5 {
		t.Fatalf("posts %d lines %d", rec.posts(), len(rec.lines()))
	}
}

func TestQueueRetriesServerErrorsAndDropsRejected(t *testing.T) {
	rec := newRecorder(t)
	tg := &fakeTarget{url: rec.srv.URL}
	tg.healthy.Store(true)
	fc := clock.NewFake(time.Unix(1000, 0))
	q := NewQueue(t.Name(), tg, QueueOptions{Clock: fc})
	ctx := context.Background()
	before := droppedFor(t.Name())

	_ = q.Write(ctx, mk(4, fc.Now()))
	rec.status.Store(http.StatusServiceUnavailable)
	if err := q.Flush(ctx); err == nil {
		t.Fatal("want error on 503")
	}
	if q.Depth() != 4 {
		t.Fatalf("503 must keep samples, depth %d", q.Depth())
	}
	rec.status.Store(http.StatusBadRequest)
	if err := q.Flush(ctx); err == nil {
		t.Fatal("want error on 400")
	}
	if q.Depth() != 0 || droppedFor(t.Name())-before != 4 {
		t.Fatalf("400 must drop and count: depth %d dropped %v", q.Depth(), droppedFor(t.Name())-before)
	}
	rec.status.Store(http.StatusNoContent)
	_ = q.Write(ctx, mk(1, fc.Now()))
	if err := q.Flush(ctx); err != nil || q.Depth() != 0 {
		t.Fatalf("recovery: %v depth %d", err, q.Depth())
	}
}

func TestQueueRunFlushesPeriodically(t *testing.T) {
	rec := newRecorder(t)
	tg := &fakeTarget{url: rec.srv.URL}
	q := NewQueue(t.Name(), tg, QueueOptions{FlushEvery: 10 * time.Millisecond})
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { q.Run(ctx); close(done) }()
	defer func() { cancel(); <-done }()

	_ = q.Write(ctx, mk(3, time.Now()))
	time.Sleep(50 * time.Millisecond)
	if rec.posts() != 0 {
		t.Fatal("posted while store was down")
	}
	tg.healthy.Store(true)
	eventually(t, 2*time.Second, "Run to flush", func() bool { return len(rec.lines()) == 3 && q.Depth() == 0 })
}

func TestQueueWriteDoesNotBlockOnSlowStore(t *testing.T) {
	block := make(chan struct{})
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-block }))
	defer srv.Close()
	defer close(block)
	tg := &fakeTarget{url: srv.URL}
	tg.healthy.Store(true)
	q := NewQueue(t.Name(), tg, QueueOptions{})
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_ = q.Write(ctx, mk(1, time.Now()))
	go func() { _ = q.Flush(ctx) }() // stuck in the POST
	time.Sleep(50 * time.Millisecond)
	start := time.Now()
	for i := 0; i < 100; i++ {
		_ = q.Write(ctx, mk(10, time.Now()))
	}
	if time.Since(start) > time.Second {
		t.Fatal("Write blocked behind a slow flush")
	}
}

func TestQuerier(t *testing.T) {
	var gotForm atomic.Value
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_ = r.ParseForm()
		gotForm.Store(r.URL.Path + "?" + r.PostForm.Encode())
		if r.PostForm.Get("query") == "bad(" {
			w.WriteHeader(http.StatusUnprocessableEntity)
			_, _ = w.Write([]byte("parse error"))
			return
		}
		_, _ = w.Write([]byte(`{"status":"success"}`))
	}))
	defer srv.Close()
	tg := &fakeTarget{url: srv.URL}
	qr := NewQuerier(tg)
	ctx := context.Background()

	start := time.Now()
	if _, err := qr.Query(ctx, "hot", "up", time.Time{}); !errors.Is(err, ErrStoreDown) {
		t.Fatalf("err = %v", err)
	}
	if time.Since(start) > 100*time.Millisecond {
		t.Fatal("did not fail fast")
	}
	tg.healthy.Store(true)
	b, err := qr.Query(ctx, "hot", "up", time.Unix(1700000000, 0))
	if err != nil || string(b) != `{"status":"success"}` {
		t.Fatalf("%s %v", b, err)
	}
	if got := gotForm.Load().(string); got != "/api/v1/query?query=up&time=1700000000.000" {
		t.Fatalf("form %q", got)
	}
	if _, err := qr.QueryRange(ctx, "hot", "up", time.Unix(100, 0), time.Unix(200, 0), 30*time.Second); err != nil {
		t.Fatal(err)
	}
	if got := gotForm.Load().(string); got != "/api/v1/query_range?end=200.000&query=up&start=100.000&step=30" {
		t.Fatalf("form %q", got)
	}
	if _, err := qr.Query(ctx, "hot", "bad(", time.Time{}); err == nil || !strings.Contains(err.Error(), "parse error") {
		t.Fatalf("err = %v", err)
	}
	if _, err := qr.QueryRange(ctx, "hot", "up", time.Unix(1, 0), time.Unix(2, 0), 0); err == nil {
		t.Fatal("zero step accepted")
	}
}