nmon/internal/api/api.go

// Package api is nmon's HTTP API: huma v2 on net/http, errors as RFC 9457
// problem details, OpenAPI generated from the handlers at /openapi.json.
package api

import (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/danielgtaylor/huma/v2"
	"github.com/danielgtaylor/huma/v2/adapters/humago"

	"github.com/v10networks/nmon/internal/clock"
	"github.com/v10networks/nmon/internal/p95"
	"github.com/v10networks/nmon/internal/poller"
	"github.com/v10networks/nmon/internal/profile"
	"github.com/v10networks/nmon/internal/selfmon"
	"github.com/v10networks/nmon/internal/store"
)

// Limits.
const (
	// QueryTimeout bounds one /query call.
	QueryTimeout = 30 * time.Second
	// MaxSeries is the most series a /query call may return; more is a 422.
	// It is enforced on the result rather than with a store flag because the
	// rollup job needs the store's own limit far higher (-search.maxUniqueTimeseries).
	MaxSeries = 100_000
	// DiscoveryTimeout bounds the seed discovery run by POST /devices.
	DiscoveryTimeout = 30 * time.Second
	// Version is the API's OpenAPI info.version; it changes only with the route contract, never with a build.
	Version = "1.0.0"
)

// Querier runs read-only MetricsQL against the hot store. *vmsup.Querier implements it.
type Querier interface {
	Query(ctx context.Context, store, q string, ts time.Time) ([]byte, error)
	QueryRange(ctx context.Context, store, q string, from, to time.Time, step time.Duration) ([]byte, error)
}

// Refresher is told when the device list changed. *poller.Poller implements it.
type Refresher interface{ Refresh() }

// Deps are what the handlers need. Store, Querier and Poller are required.
type Deps struct {
	Store    store.Store
	Querier  Querier
	Poller   Refresher
	Clock    clock.Clock       // default clock.Real()
	Registry *selfmon.Registry // served at /metrics; default selfmon.Default
	Logger   *slog.Logger      // default slog.Default()
	SNMPPort int               // dev/test override of UDP 161; 0 = 161
	// Profiles select a device's profile at seed discovery. Nil: no profile matching.
	Profiles *profile.Set

	// PollerStatus feeds GET /poller/health. *poller.Poller's Status fits.
	PollerStatus func() []poller.DeviceStatus
	// AfterDiscover is called after POST /devices/{slug}/discover succeeded, e.g. to make the poller reload that device.
	AfterDiscover func(deviceID int64)

	// DiscoveryTimeout overrides the 30 s seed discovery limit (tests).
	DiscoveryTimeout time.Duration

	// Reachability decides up/down/warning per device. Default: what is persisted now
	// (poll state, last_seen, the ICMP and SNMP up series). M4 swaps in the alerter's state.
	Reachability ReachabilitySource
	// DeviceWarnings adds live warnings (for example icmp_unsupported) to a device's warnings list.
	DeviceWarnings func(slug string) []string
	// Dependencies checks parent edits for cycles and is refreshed after a write. Default: read the
	// edges from the store on each check.
	Dependencies DependencyGraph
	// OnMaintenanceChange is called after a maintenance window is created or deleted, so the
	// evaluators see it at once instead of at their next refresh.
	OnMaintenanceChange func(ctx context.Context)

	// PollInterval is the normal poll cadence; a device unheard of for three of them is down. Default 60 s.
	PollInterval time.Duration

	// HotRetention, FiveMinuteRetention and LongRetention are the stores' retention periods. They decide where a
	// graph stitches and what GET /meta reports. Defaults: 30 d, 120 d, 2 y.
	HotRetention, FiveMinuteRetention, LongRetention time.Duration
	// WriteQueueCapacity is the hot write queue's size in samples, for GET /poller/health. Default 2,000,000.
	WriteQueueCapacity int

	// MaxClientQueries and MaxGlobalQueries bound concurrent free-form queries per client and metric-store
	// queries overall (defaults 4 and 16). AdmissionWait is how long a request waits for a global slot (default 5 s).
	MaxClientQueries, MaxGlobalQueries int
	AdmissionWait                      time.Duration

	// M5 collaborators. A route whose collaborator is nil answers 503 "not configured", so tests and
	// `nmon openapi` need none of them.
	Tokens  PushTokenAdmin
	Secrets SecretWriter
	Rollup  RollupRepairer
	P95     InterfaceP95
	Groups  GroupTraffic
	Zones   p95.Zones
	Checks  CheckResults
	Proxmox ProxmoxStatus
}

type server struct {
	Deps
	adm  *admission
	rate *counterRates
}

// withDefaults fills every unset optional dependency.
func (d Deps) withDefaults() Deps {
	if d.Clock == nil {
		d.Clock = clock.Real()
	}
	if d.Registry == nil {
		d.Registry = selfmon.Default
	}
	if d.Logger == nil {
		d.Logger = slog.Default()
	}
	d.DiscoveryTimeout = cmp.Or(d.DiscoveryTimeout, DiscoveryTimeout)
	d.PollInterval = cmp.Or(d.PollInterval, time.Minute)
	d.HotRetention = cmp.Or(d.HotRetention, 30*day)
	d.FiveMinuteRetention = cmp.Or(d.FiveMinuteRetention, 120*day)
	d.LongRetention = cmp.Or(d.LongRetention, 2*365*day)
	d.WriteQueueCapacity = cmp.Or(d.WriteQueueCapacity, 2_000_000)
	d.MaxClientQueries = cmp.Or(d.MaxClientQueries, 4)
	d.MaxGlobalQueries = cmp.Or(d.MaxGlobalQueries, 16)
	d.AdmissionWait = cmp.Or(d.AdmissionWait, 5*time.Second)
	if d.Dependencies == nil {
		d.Dependencies = storeGraph{src: d.Store}
	}
	if d.Reachability == nil {
		d.Reachability = &persistedReachability{q: d.Querier, clk: d.Clock, interval: d.PollInterval}
	}
	return d
}

const day = 24 * time.Hour

// New returns the API handler with every route registered.
func New(d Deps) http.Handler {
	d = d.withDefaults()
	s := &server{Deps: d, adm: newAdmission(d.MaxClientQueries, d.MaxGlobalQueries, d.AdmissionWait), rate: &counterRates{}}
	s.rate.perMinute(s.readCounters()) // baseline, so the first poller/health call has a window to measure
	mux := http.NewServeMux()
	api := s.newHuma(mux)
	registerDocs(mux)
	s.register(api)
	return problemFallback(mux)
}

// Spec returns the OpenAPI 3.1 document of the API as indented JSON. It is the
// same document served at /openapi.json, generated without any dependencies:
// `nmon openapi > openapi.json` checks it in.
func Spec() ([]byte, error) {
	s := &server{Deps: Deps{}.withDefaults()}
	api := s.newHuma(http.NewServeMux())
	s.register(api)
	return json.MarshalIndent(api.OpenAPI(), "", "  ")
}

// register adds every operation to api.
func (s *server) register(api huma.API) {
	s.registerDevices(api)
	s.registerDeviceList(api)
	s.registerDeviceViews(api)
	s.registerInterfaces(api)
	s.registerGraph(api)
	s.registerOps(api)
	s.registerHealth(api)
	s.registerReviews(api)
	s.registerQuery(api)
	s.registerMeta(api)
	s.registerAlerts(api)
	s.registerMaintenance(api)
	s.registerMetrics(api)
	s.registerTokens(api)
	s.registerSecrets(api)
	s.registerChecks(api)
	s.registerGroups(api)
	s.registerGroupGraph(api)
	s.registerP95(api)
	s.registerRollup(api)
	s.registerProxmox(api)
}

// newHuma builds the huma API on mux. /docs is served by registerDocs from embedded assets, not huma's CDN page.
func (s *server) newHuma(mux *http.ServeMux) huma.API {
	configureHuma()
	cfg := huma.DefaultConfig("nmon", Version)
	cfg.Info.Description = "nmon network monitor API."
	cfg.OpenAPIPath = "/openapi"
	cfg.DocsPath = ""
	cfg.CreateHooks = nil // no $schema fields or Link headers in responses
	cfg.OnAddOperation = append(cfg.OnAddOperation, addDefaultProblem)
	api := humago.New(mux, cfg)
	api.UseMiddleware(clientKeyMiddleware)
	return api
}

// storeErr maps store sentinels to problem responses; anything else is a 500
// whose detail does not leak internals.
func (s *server) storeErr(err error) error {
	switch {
	case errors.Is(err, store.ErrNotFound):
		return huma.Error404NotFound("not found")
	case errors.Is(err, store.ErrConflict):
		return huma.Error409Conflict(trimSentinel(err, store.ErrConflict))
	case errors.Is(err, store.ErrInvalid):
		return huma.Error422UnprocessableEntity(trimSentinel(err, store.ErrInvalid))
	}
	s.Logger.Error("store error", "err", err)
	return huma.Error500InternalServerError("internal error")
}

// trimSentinel returns err's text without the leading "store: x: " sentinel.
func trimSentinel(err, sentinel error) string {
	msg := err.Error()
	if rest, ok := strings.CutPrefix(msg, sentinel.Error()+": "); ok {
		return rest
	}
	return strings.TrimPrefix(msg, "store: ")
}

// problemFallback answers requests that match no route (404, 405) with
// problem+json instead of the mux's plain text.
func problemFallback(mux *http.ServeMux) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		h, pattern := mux.Handler(r)
		if pattern != "" {
			mux.ServeHTTP(w, r) // sets the path values
			return
		}
		cw := &capture{h: http.Header{}}
		h.ServeHTTP(cw, r)
		status := cw.status
		if status == 0 {
			status = http.StatusNotFound
		}
		if allow := cw.h.Get("Allow"); allow != "" {
			w.Header().Set("Allow", allow)
		}
		w.Header().Set("Content-Type", "application/problem+json")
		w.WriteHeader(status)
		_ = json.NewEncoder(w).Encode(map[string]any{
			"title": http.StatusText(status), "status": status,
			"detail": r.Method + " " + r.URL.Path + ": " + strings.ToLower(http.StatusText(status)),
		})
	})
}

// capture records only the status and headers the mux would have written.
type capture struct {
	h      http.Header
	status int
}

func (c *capture) Header() http.Header         { return c.h }
func (c *capture) Write(b []byte) (int, error) { return len(b), nil }
func (c *capture) WriteHeader(s int)           { c.status = s }

func itoa(n int) string { return strconv.Itoa(n) }