nmon/internal/api/groups_graph.go

package api

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

	"github.com/danielgtaylor/huma/v2"

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

// maxGroupSpan is how far back a group graph may reach: the 5-minute rates it sums live for retention.fiveminute.
const maxGroupSpan = 125 * day

// GroupGraphResponse is the oriented sum of a group's members.
type GroupGraphResponse struct {
	Group       GroupView   `json:"group"`
	From        time.Time   `json:"from" doc:"First sample time of the grid."`
	To          time.Time   `json:"to"`
	StepSeconds int64       `json:"step_seconds"`
	Series      GraphSeries `json:"series" doc:"Bits per second in and out, oriented by each member's direction. Points are null where no member had data."` //nolint:lll // struct tag text is the OpenAPI doc; cannot wrap
	Partial     []TimeSpan  `json:"partial" doc:"Stretches where at least one member had no sample, so the sum is a lower bound."`
	Flags       []GroupFlag `json:"flags"`
	P95         *GraphP95   `json:"p95,omitempty" doc:"The group's monthly p95 of the summed series (never a sum of per-port figures)."`
}

type groupGraphIn struct {
	ID    string    `path:"id" doc:"Group id."`
	From  time.Time `query:"from" doc:"Start, RFC 3339. Default 24 hours before to."`
	To    time.Time `query:"to" doc:"End, RFC 3339. Default now."`
	Step  int64     `query:"step" minimum:"1" doc:"Step in seconds, rounded up to a multiple of 300. Default max(300, range/1000)."`
	P95   bool      `query:"p95" doc:"Also return the group's p95 for month."`
	Month string    `query:"month" pattern:"^\\d{4}-\\d{2}$" doc:"Month for p95, YYYY-MM in the server timezone. Default: the month containing to."` //nolint:lll // struct tag text is the OpenAPI doc; cannot wrap
}

func (s *server) registerGroupGraph(api huma.API) {
	huma.Register(api, huma.Operation{
		OperationID: "get-group-graph", Method: http.MethodGet, Path: "/api/v1/groups/{id}/graph", Tags: []string{"groups"},
		Summary: "Oriented in/out sum of a group's members",
		Description: "Membership is resolved per timestamp, so a port added in February never changes January. Interfaces with " +
			"unreliable counters are left out, a LAG member is dropped when its LAG is in the group, and a member with direction " +
			"away has in and out swapped. Built from 5-minute rates, so it reaches back retention.fiveminute only.",
		Errors: []int{http.StatusNotFound, http.StatusUnprocessableEntity, http.StatusServiceUnavailable},
	}, s.groupGraph)
}

func (s *server) groupGraph(ctx context.Context, in *groupGraphIn) (*struct{ Body GroupGraphResponse }, error) {
	if s.Groups == nil {
		return nil, errNotConfigured("group traffic")
	}
	id, err := idParam(in.ID)
	if err != nil {
		return nil, err
	}
	ref, err := s.Store.GetGroup(ctx, id)
	if err != nil {
		return nil, s.storeErr(err)
	}
	from, to, step, err := s.groupRange(in)
	if err != nil {
		return nil, err
	}
	release, err := s.admit(ctx, false)
	if err != nil {
		return nil, err
	}
	defer release()
	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
	defer cancel()
	res, err := s.Groups.GroupSeries(qctx, id, from, to, step)
	if err != nil {
		return nil, s.vmErr(qctx, err)
	}
	out := GroupGraphResponse{
		Group: GroupView{ID: strconv.FormatInt(ref.ID, 10), Name: ref.Name, Kind: ref.Kind}, From: from.UTC(), To: to.UTC(),
		StepSeconds: int64(step / time.Second), Partial: []TimeSpan{}, Flags: []GroupFlag{},
	}
	out.Series.In, out.Series.Out = gridPoints(res.In, from, to, step), gridPoints(res.Out, from, to, step)
	for _, sp := range res.Partial {
		out.Partial = append(out.Partial, TimeSpan{From: sp.From, To: sp.To})
	}
	for _, f := range res.Flags {
		out.Flags = append(out.Flags, GroupFlag{Kind: f.Kind, A: f.A, B: f.B})
	}
	if in.P95 {
		if out.P95, err = s.groupMonthP95(qctx, id, in.Month, to); err != nil {
			return nil, err
		}
	}
	return &struct{ Body GroupGraphResponse }{out}, nil
}

// groupRange applies the defaults and the 5-minute grid.
func (s *server) groupRange(in *groupGraphIn) (from, to time.Time, step time.Duration, err error) {
	to = in.To
	if to.IsZero() {
		to = s.Clock.Now()
	}
	from = in.From
	if from.IsZero() {
		from = to.Add(-24 * time.Hour)
	}
	to, from = to.UTC(), from.UTC()
	switch {
	case !from.Before(to):
		return from, to, 0, huma.Error422UnprocessableEntity("from must be before to")
	case to.Sub(from) > maxGroupSpan:
		return from, to, 0, huma.Error422UnprocessableEntity("a group graph reaches back at most 125 days: it is built from 5-minute rates")
	}
	step = time.Duration(in.Step) * time.Second
	if step <= 0 {
		step = to.Sub(from) / targetPoints
	}
	step = max(p95.Window, ceilDur(step, p95.Window))
	if n := to.Sub(from) / step; n > maxGraphPoints {
		return from, to, 0, huma.Error422UnprocessableEntity("too many points; use a larger step")
	}
	return from.Truncate(p95.Window), to, step, nil
}

// gridPoints lays sparse [ts, value] points on the bucket grid, null where a bucket had none.
func gridPoints(pts [][2]float64, from, to time.Time, step time.Duration) []Point {
	byTS := make(map[int64]float64, len(pts))
	for _, p := range pts {
		byTS[int64(p[0])] = p[1]
	}
	out := []Point{}
	for t := from; t.Before(to); t = t.Add(step) {
		p := Point{TS: t.Unix()}
		if v, ok := byTS[t.Unix()]; ok {
			p.V = &v
		}
		out = append(out, p)
	}
	return out
}

// groupMonthP95 is the group's p95 for month (default: the month containing to, in the server timezone).
func (s *server) groupMonthP95(ctx context.Context, id int64, month string, to time.Time) (*GraphP95, error) {
	if month == "" {
		month = p95.MonthContaining(to.Add(-time.Nanosecond), s.Zones.For("")).Label
	}
	r, err := s.Groups.GroupP95(ctx, id, month)
	switch {
	case errors.Is(err, p95.ErrNoData):
		return &GraphP95{Source: string(p95.SourceLive), Month: month, Partial: true}, nil
	case err != nil:
		return nil, s.p95Err(ctx, err)
	}
	return p95Graph(r.Result), nil
}

// p95Graph is the GraphP95 form of a figure.
func p95Graph(r p95.Result) *GraphP95 {
	in, out := r.InBps, r.OutBps
	return &GraphP95{InBPS: &in, OutBPS: &out, Source: string(r.Source), Coverage: r.Coverage, Month: r.Month, Partial: r.Partial()}
}

// p95Err maps p95 and group failures to problems.
func (s *server) p95Err(ctx context.Context, err error) error {
	switch {
	case errors.Is(err, store.ErrNotFound):
		return huma.Error404NotFound("not found")
	case errors.Is(err, p95.ErrUnreliable):
		return huma.Error422UnprocessableEntity("the interface's counters are unreliable, so no p95 is reported")
	case errors.Is(err, p95.ErrNoData):
		return huma.Error422UnprocessableEntity("no 5-minute samples in that month")
	}
	return s.vmErr(ctx, err)
}