nmon/internal/api/device_metrics_graph.go

package api

import (
	"context"
	"time"

	"github.com/danielgtaylor/huma/v2"

	"github.com/v10networks/nmon/internal/profile"
)

type metricGraphIn struct {
	Slug      string    `path:"slug" doc:"Device slug."`
	Metric    string    `query:"metric" required:"true" minLength:"1" doc:"Profile metric name, as /metrics lists it."`
	Component string    `query:"component" doc:"Row label; empty for a device-wide scalar."`
	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."`
	Aggregate string    `query:"aggregate" enum:"avg,sum,max" doc:"Without component: one series over all components."`
}

// allComponents, as the component of metricSel, selects every component of the metric.
const allComponents = "\x00all"

// metricSel is the selector of one component of a metric.
func metricSel(name, slug, comp string) string {
	if comp == allComponents {
		return name + `{device=` + quote(slug) + `}`
	}
	return name + `{device=` + quote(slug) + `,component=` + quote(comp) + `}`
}

// valueExpr is the metric's value at one component with scale_by applied. A metric with sentinels
// drops its negative values, which are words rather than numbers.
func valueExpr(m profile.Metric, slug, comp string) string {
	e := metricSel(m.Name, slug, comp)
	if len(m.Sentinels) > 0 {
		e = `(` + e + ` >= 0)`
	}
	if m.ScaleBy != "" {
		e = `(` + e + ` * on(device, component) ` + metricSel(m.ScaleBy, slug, comp) + `)`
	}
	return e
}

// graphExprs returns the series expression and, for a metric with a capacity that is not drawn as
// percent, the capacity expression.
func graphExprs(res *profile.Resolved, m profile.Metric, slug, comp string, step, every time.Duration) (expr, maxExpr string) {
	display := m.EffectiveDisplay()
	val := valueExpr(m, slug, comp)
	counter := m.Type == profile.Counter32 || m.Type == profile.Counter64
	if counter && display == "rate" {
		val = `rate(` + metricSel(m.Name, slug, comp) + `[` + secs(max(step, rateWindowFor(every))) + `])`
	}
	om, hasOf := res.Metric(m.Of)
	if !hasOf {
		return val, ""
	}
	den := valueExpr(om, slug, comp)
	if display == "percent" {
		return `100 * ` + val + ` / on(device, component) (` + den + ` > 0)`, ""
	}
	return val, den
}

func (s *server) metricGraph(ctx context.Context, in *metricGraphIn) (*struct{ Body MetricGraphResponse }, error) {
	d, err := s.Store.GetDeviceBySlug(ctx, in.Slug)
	if err != nil {
		return nil, s.storeErr(err)
	}
	if d.PurgedAt != nil || s.Profiles == nil {
		return nil, huma.Error404NotFound("not found")
	}
	res := s.Profiles.Get(d.Profile)
	m, ok := metricOf(res, in.Metric)
	if !ok {
		return nil, huma.Error404NotFound("no such metric in the device's profile")
	}
	plan, err := s.metricPlan(in)
	if err != nil {
		return nil, err
	}
	step := plan.newest().step
	comp := in.Component
	if in.Aggregate != "" && comp == "" {
		comp = allComponents
	}
	expr, maxExpr := graphExprs(res, m, d.Slug, comp, step, d.PollInterval(s.pollEvery()))
	if comp == allComponents {
		expr, maxExpr = in.Aggregate+`(`+expr+`)`, ""
	}
	resp := MetricGraphResponse{
		Unit: unitOfMetric(res, m), Display: m.EffectiveDisplay(), StepSeconds: int64(step / time.Second), Series: []Point{},
	}
	grid := gridOf(plan)
	series, err := s.fetchMetricSeries(ctx, expr, plan)
	if err != nil {
		return nil, err
	}
	resp.Series = pointsOf(grid, series)
	if maxExpr != "" {
		mx, err := s.fetchMetricSeries(ctx, maxExpr, plan)
		if err != nil {
			return nil, err
		}
		resp.MaxSeries = pointsOf(grid, mx)
	}
	return &struct{ Body MetricGraphResponse }{resp}, nil
}

func metricOf(res *profile.Resolved, name string) (profile.Metric, bool) {
	if res == nil {
		return profile.Metric{}, false
	}
	m, ok := res.Metric(name)
	return m, ok && m.NeedsCategory()
}

func unitOfMetric(res *profile.Resolved, m profile.Metric) string {
	return (&metricsBuilder{res: res}).unitOf(m)
}

// metricPlan applies the /graph range defaults and step rules, limited to the hot store.
func (s *server) metricPlan(in *metricGraphIn) (*graphPlan, error) {
	p, _, err := s.hotPlan(in.From, in.To, in.Step)
	return p, err
}

// hotPlan applies the /graph range defaults (to = now, from = an hour earlier) and step rules to a range that
// only the hot store answers: a from before hot retention is moved up to its start, and clamped says so.
func (s *server) hotPlan(from, to time.Time, step int64) (p *graphPlan, clamped bool, err error) {
	if to.IsZero() {
		to = s.Clock.Now()
	}
	if from.IsZero() {
		from = to.Add(-time.Hour)
	}
	hotStart := s.Clock.Now().Add(-s.HotRetention)
	from = from.UTC()
	if from.Before(hotStart) {
		from, clamped = hotStart, true
	}
	if !from.Before(to) {
		return nil, clamped, huma.Error422UnprocessableEntity("from must be before to, within the hot retention")
	}
	p, err = planGraph(from, to.UTC(), hotStart, time.Duration(step)*time.Second)
	if err != nil {
		return nil, clamped, huma.Error422UnprocessableEntity(err.Error())
	}
	return p, clamped, nil
}

// fetchMetricSeries runs the expression over the plan's hot windows and returns ts -> value.
func (s *server) fetchMetricSeries(ctx context.Context, expr string, p *graphPlan) (map[int64]float64, error) {
	release, err := s.admit(ctx, false)
	if err != nil {
		return nil, err
	}
	defer release()
	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
	defer cancel()
	out := map[int64]float64{}
	for _, w := range p.tiers {
		if w.tier != tierRaw {
			continue // only the hot store is searched
		}
		if err := s.fetchOne(qctx, "hot", expr, w, func(ts int64, v float64) { out[ts] = v }); err != nil {
			return nil, err
		}
	}
	return out, nil
}