nmon/internal/api/ops_views_test.go

package api_test

import (
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/v10networks/nmon/internal/api"
	"github.com/v10networks/nmon/internal/poller"
	"github.com/v10networks/nmon/internal/store"
)

func TestPollerHealthShapeAndTotals(t *testing.T) {
	e := newEnv(t, 0)
	d := e.device(t, "Core 1", "", "")
	var h api.PollerHealth
	e.getJSON(t, "/api/v1/poller/health", &h)
	if h.Devices == nil || len(h.Devices) != 0 || h.Totals.WriteQueueCapacity != 2_000_000 {
		t.Fatalf("empty health: %+v", h)
	}
	next := e.clk.Now().Add(30 * time.Second)
	e.status = []poller.DeviceStatus{
		{Device: d.Slug, DeviceID: 77, State: store.PollOK, Varbinds: 40, Interval: time.Minute, BaseInterval: time.Minute,
			LastCycleDuration: 1500 * time.Millisecond, NextCycle: next, Profile: "generic"},
		{Device: "ghost", State: store.PollDegraded, Varbinds: 20, Interval: 4 * time.Minute, BaseInterval: time.Minute,
			ConsecutiveFailures: 3, MissedDeadlines: 3, LastError: "snmp: timeout"},
		{Device: "slow", State: store.PollOK, Interval: 2 * time.Minute, BaseInterval: time.Minute},
	}
	rec := e.getJSON(t, "/api/v1/poller/health", &h)
	assertNoLeaks(t, "poller health", rec.Body.Bytes())
	if strings.Contains(rec.Body.String(), "77") {
		t.Errorf("numeric device ID leaked: %s", rec.Body)
	}
	if len(h.Devices) != 3 {
		t.Fatalf("devices: %+v", h.Devices)
	}
	c := h.Devices[0]
	if c.Slug != d.Slug || c.Name != "Core 1" || c.State != "ok" || c.Varbinds != 40 || c.IntervalS != 60 || c.BaseIntervalS != 60 ||
		c.LastCycleS != 1.5 || c.NextCycle == nil || !c.NextCycle.Equal(next) {
		t.Errorf("device row: %+v", c)
	}
	if g := h.Devices[1]; g.State != "degraded" || g.ConsecutiveFailures != 3 || g.MissedDeadlines != 3 || g.LastError != "snmp: timeout" || g.NextCycle != nil {
		t.Errorf("degraded row: %+v", g)
	}
	if h.Devices[2].State != "backoff" || h.Totals.DevicesDegraded != 2 {
		t.Errorf("backoff: %+v degraded=%d", h.Devices[2], h.Totals.DevicesDegraded)
	}
}

func TestPollerHealthPerMinuteRates(t *testing.T) {
	e := newEnv(t, 0)
	pdus := e.reg.Counter("nmon_pdus_total", "path", "poll", "type", "get")
	e.reg.Counter("nmon_pdus_total", "path", "discovery", "type", "get").Add(99999) // not the poll path
	timeouts := e.reg.Counter("nmon_snmp_timeouts_total", "path", "poll")
	missA := e.reg.Counter("nmon_poll_deadline_missed_total", "device", "a")
	missB := e.reg.Counter("nmon_poll_deadline_missed_total", "device", "b")
	e.reg.Gauge("nmon_write_queue_depth", "queue", "hot").Set(1234)
	e.reg.Gauge("nmon_write_queue_depth", "queue", "long").Set(5)

	pdus.Add(600)
	timeouts.Add(6)
	missA.Add(3)
	e.clk.Advance(30 * time.Second)
	var h api.PollerHealth
	e.getJSON(t, "/api/v1/poller/health", &h)
	tot := h.Totals
	if tot.PDUsPerMin != 1200 || tot.TimeoutsPerMin != 12 || tot.DeadlineMissesPerMin != 6 || tot.WriteQueueDepth != 1234 {
		t.Errorf("after 30 s: %+v", tot)
	}
	pdus.Add(600)
	timeouts.Add(1)
	missB.Add(1)
	e.clk.Advance(30 * time.Second)
	e.getJSON(t, "/api/v1/poller/health", &h)
	if tot = h.Totals; tot.PDUsPerMin != 1200 || tot.TimeoutsPerMin != 7 || tot.DeadlineMissesPerMin != 4 {
		t.Errorf("after 60 s: %+v", tot)
	}
	e.clk.Advance(time.Minute) // the window slides: only the last 90 s count
	e.getJSON(t, "/api/v1/poller/health", &h)
	if h.Totals.PDUsPerMin != 400 {
		t.Errorf("sliding window: %+v", h.Totals)
	}
}

func TestMeta(t *testing.T) {
	e := newEnvDeps(t, 0, func(d *api.Deps) {
		d.HotRetention, d.FiveMinuteRetention, d.LongRetention = 720*time.Hour, 2880*time.Hour, 17520*time.Hour
	})
	var m api.Meta
	rec := e.getJSON(t, "/api/v1/meta", &m)
	if m.Retention.Hot != "720h0m0s" || m.Retention.FiveMinute != "2880h0m0s" || m.Retention.Long != "17520h0m0s" {
		t.Errorf("retention: %+v", m.Retention)
	}
	if !m.Now.Equal(e.clk.Now()) || m.Version == "" || m.Build == "" || !strings.Contains(rec.Body.String(), `"now":"2026-10-22T12:00:00Z"`) {
		t.Errorf("meta: %+v %s", m, rec.Body)
	}
	for _, s := range []string{m.Retention.Hot, m.Retention.FiveMinute, m.Retention.Long} {
		if _, err := time.ParseDuration(s); err != nil {
			t.Errorf("retention %q is not a Go duration", s)
		}
	}
	var def api.Meta
	newEnv(t, 0).getJSON(t, "/api/v1/meta", &def)
	if def.Retention.Hot != "720h0m0s" || def.Retention.FiveMinute != "2880h0m0s" || def.Retention.Long != "17520h0m0s" {
		t.Errorf("default retention (30 d, 120 d, 2 y): %+v", def.Retention)
	}
}

func TestQueryAdmissionPerClientAndGlobal(t *testing.T) {
	var running, peak atomic.Int32
	release := make(chan struct{})
	started := make(chan struct{}, 32)
	e := newEnvDeps(t, 0, func(d *api.Deps) { d.MaxClientQueries, d.MaxGlobalQueries, d.AdmissionWait = 4, 6, 50*time.Millisecond })
	e.q.fn = func(string, string, time.Time, time.Time, time.Duration) ([]byte, error) {
		n := running.Add(1)
		for {
			p := peak.Load()
			if n <= p || peak.CompareAndSwap(p, n) {
				break
			}
		}
		started <- struct{}{}
		<-release
		running.Add(-1)
		return seriesBody(1), nil
	}
	var wg sync.WaitGroup
	do := func(addr string) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if rec := e.doFrom("GET", "/api/v1/query?q=up", addr); rec.Code != 200 {
				t.Errorf("admitted query got %d", rec.Code)
			}
		}()
	}
	for range 4 {
		do("192.0.2.1:1000")
	}
	for range 4 {
		<-started
	}
	// the fifth query from the same client is refused at once; another client is admitted
	rec := e.doFrom("GET", "/api/v1/query?q=up", "192.0.2.1:2000")
	isProblem(t, rec, http.StatusTooManyRequests)
	do("192.0.2.2:1000")
	do("192.0.2.3:1000")
	for range 2 {
		<-started
	}
	// six are running: the global limit; a seventh from a fresh client waits, then is refused
	isProblem(t, e.doFrom("GET", "/api/v1/query?q=up", "192.0.2.4:1000"), http.StatusTooManyRequests)
	close(release)
	wg.Wait()
	if peak.Load() > 6 {
		t.Errorf("%d queries ran at once, the global cap is 6", peak.Load())
	}
	// slots are returned: the client can query again
	if rec := e.doFrom("GET", "/api/v1/query?q=up", "192.0.2.1:3000"); rec.Code != 200 {
		t.Errorf("after release: %d %s", rec.Code, rec.Body)
	}
}