nmon/internal/api/graph.go

package api

import (
	"context"
	"net/http"
	"time"

	"github.com/danielgtaylor/huma/v2"

	"github.com/v10networks/nmon/internal/p95"
	"github.com/v10networks/nmon/internal/store"
)

func (s *server) registerGraph(api huma.API) {
	huma.Register(api, huma.Operation{
		OperationID: "get-graph", Method: http.MethodGet, Path: "/api/v1/graph", Tags: []string{"graph"},
		Summary: "Bits per second in and out of one interface",
		Description: "From the hot store when the range lies within hot retention; older data comes from the long store's " +
			"hourly rollups, and a range that needs both is stitched at stitched_at. Stretches with no data in either store " +
			"are listed in partial. An unreliable interface returns no series. max=true adds the per-step maximum; " +
			"p95=true adds the 95th percentile of 5-minute averages: the monthly figure for a whole calendar month, otherwise " +
			"computed live over the range; source says which. totals integrates the average series into bytes in and out " +
			"over the range; covered_seconds says how much of it had data.",
		Errors: []int{http.StatusNotFound, http.StatusUnprocessableEntity, http.StatusTooManyRequests, http.StatusServiceUnavailable},
	}, s.graph)
}

type graphIn struct {
	Iface string    `query:"iface" required:"true" minLength:"1" doc:"Interface ULID."`
	From  time.Time `query:"from" doc:"Start, RFC 3339. Default one hour before to."`
	To    time.Time `query:"to" doc:"End, RFC 3339. Default now."`
	Step  int64     `query:"step" minimum:"1" doc:"Step in seconds. Default max(60, range/1000) rounded up to whole minutes."`
	Dir   string    `query:"dir" enum:"in,out" doc:"Only this direction; the other series is empty."`
	Max   bool      `query:"max" doc:"Also return in_max and out_max, the highest poll-to-poll rate within each step."`
	P95   bool      `query:"p95" doc:"Also return the 95th percentile: monthly for a whole calendar month, else live over the range."`
	Month string    `query:"month" pattern:"^\\d{4}-(0[1-9]|1[0-2])$" doc:"YYYY-MM in the site timezone. Replaces from and to with that month's bounds."` //nolint:lll // struct tag text is the OpenAPI doc; cannot wrap
}

func (s *server) graph(ctx context.Context, in *graphIn) (*struct{ Body GraphResponse }, error) {
	i, err := s.Store.GetInterface(ctx, in.Iface)
	if err != nil {
		return nil, s.storeErr(err)
	}
	dev, err := s.Store.GetDeviceByID(ctx, i.DeviceID)
	if err != nil {
		return nil, s.storeErr(err)
	}
	from, to, err := s.graphRange(in, dev.Site)
	if err != nil {
		return nil, err
	}
	resp := GraphResponse{
		Iface: GraphIface{
			ID: i.ID, IfName: i.IfName, IfAlias: i.IfAlias, SpeedBPS: speedBPS(i), Device: DeviceRef{Slug: dev.Slug, Name: dev.Name},
		},
		Partial: []TimeSpan{}, Series: GraphSeries{In: []Point{}, Out: []Point{}},
	}
	hotStart := s.Clock.Now().Add(-s.HotRetention)
	plan, err := planGraph(from, to, hotStart, time.Duration(in.Step)*time.Second)
	if err != nil {
		return nil, huma.Error422UnprocessableEntity(err.Error())
	}
	resp.StepSeconds, resp.Tier = int64(plan.newest().step/time.Second), plan.newest().tier
	if i.Unreliable {
		resp.Unreliable, resp.UnreliableReason = true, unreliableReason(i)
		return &struct{ Body GraphResponse }{resp}, nil
	}
	if err := s.fillGraph(ctx, &resp, i, plan, in, graphSite{hotStart: hotStart, site: dev.Site,
		every: dev.PollInterval(s.pollEvery()), from: from, to: to}); err != nil {
		return nil, err
	}
	return &struct{ Body GraphResponse }{resp}, nil
}

// graphRange applies the defaults and validates the requested range.
func (s *server) graphRange(in *graphIn, site string) (from, to time.Time, err error) {
	if in.Month != "" {
		m, err := p95.ParseMonth(in.Month, s.Zones.For(site))
		if err != nil {
			return from, to, huma.Error422UnprocessableEntity(err.Error())
		}
		return m.Start.UTC(), m.End.UTC(), nil
	}
	to = in.To
	if to.IsZero() {
		to = s.Clock.Now()
	}
	from = in.From
	if from.IsZero() {
		from = to.Add(-time.Hour)
	}
	if !from.Before(to) {
		return from, to, huma.Error422UnprocessableEntity("from must be before to")
	}
	return from.UTC(), to.UTC(), nil
}

// graphSite is what p95 needs to know about the interface's place: where hot retention begins and its site.
type graphSite struct {
	hotStart time.Time
	site     string
	every    time.Duration // the device's effective poll interval
	from, to time.Time     // the requested range, unaligned
}

// fillGraph queries the stores the plan names and lays the samples out on the grid.
func (s *server) fillGraph(
	ctx context.Context, resp *GraphResponse, i *store.Interface, p *graphPlan, in *graphIn, gs graphSite,
) error {
	release, err := s.admit(ctx, false)
	if err != nil {
		return err
	}
	defer release()
	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
	defer cancel()
	q := &graphQuery{iface: i, dirs: []string{dirIn, dirOut}, withMax: in.Max, every: gs.every}
	if in.Dir != "" {
		q.dirs = []string{in.Dir}
	}
	data := newGraphData()
	for _, w := range p.tiers {
		if err := s.fetchTier(qctx, q, w, data); err != nil {
			return err
		}
	}
	assemble(resp, p, q, data)
	resp.Totals = graphTotals(p, q.dirs, data.avg, gs.from, gs.to)
	if in.P95 {
		p95, err := s.graphP95(qctx, q, p, gs)
		if err != nil {
			return err
		}
		resp.P95 = p95
	}
	return nil
}

// assemble turns the samples into series, tiers, stitched_at and partial spans.
func assemble(resp *GraphResponse, p *graphPlan, q *graphQuery, d *graphData) {
	grid := gridOf(p)
	for _, dir := range q.dirs {
		resp.Series.set(dir, false, pointsOf(grid, d.avg[dir]))
		if q.withMax {
			resp.Series.set(dir, true, pointsOf(grid, d.max[dir]))
		}
	}
	has := func(ts int64) bool {
		for _, dir := range q.dirs {
			if _, ok := d.avg[dir][ts]; ok {
				return true
			}
		}
		return false
	}
	resp.Partial = append(resp.Partial, findGaps(grid, has, p.from.Unix(), p.to.Unix())...)
	if p.stitchedAt != nil {
		resp.StitchedAt = p.stitchedAt
		for _, w := range p.tiers {
			resp.Tiers = append(resp.Tiers, GraphTier{From: w.from.UTC(), To: w.bound.UTC(), Tier: w.tier, StepSeconds: int64(w.step / time.Second)})
		}
	}
}