nmon/internal/api/graph_p95.go

package api

import (
	"context"
	"errors"
	"time"

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

const (
	p95Quantile = 0.95
	p95Slot     = 5 * time.Minute
	// maxP95Span bounds the live computation to what one range query can return (about 25,000 slots).
	maxP95Span = 85 * 24 * time.Hour
)

// graphP95 answers p95=true. A range that is one whole calendar month of the interface's site is the monthly
// figure and comes from p95.Service (materialised when the month is done and recorded); anything else is
// computed live over the drawn range.
func (s *server) graphP95(ctx context.Context, q *graphQuery, p *graphPlan, gs graphSite) (*GraphP95, error) {
	if m, ok := s.wholeMonth(gs.site, p); ok && s.P95 != nil {
		r, err := s.P95.P95(ctx, q.iface.ID, m.Label)
		switch {
		case err == nil:
			g := p95Graph(r)
			g.AggregateBPS = s.liveAggregate(ctx, q, p, gs.hotStart)
			return g, nil
		case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
			return nil, err
		case !errors.Is(err, p95.ErrNoData):
			s.Logger.Warn("monthly p95 unavailable; computing live", "iface", q.iface.ID, "month", m.Label, "err", err)
		}
	}
	return s.livep95(ctx, q, p, gs.hotStart)
}

// wholeMonth reports whether the planned range is one calendar month of the site's timezone, give or take a step.
func (s *server) wholeMonth(site string, p *graphPlan) (p95.Month, bool) {
	tol := max(p.newest().step, p95Slot)
	m := p95.MonthContaining(p.from.Add(tol), s.Zones.For(site))
	near := func(a, b time.Time) bool { return a.Sub(b).Abs() <= tol }
	return m, near(p.from, m.Start) && near(p.to, m.End)
}

// livep95 computes the 95th percentile of 5-minute average rates over the
// raw (hot) part of the drawn range, by nearest rank. M5 adds the materialised
// monthly value; until then the source is "live" and coverage says how much of
// the range's 5-minute slots had data, so a short history never passes for a full month.
func (s *server) livep95(ctx context.Context, q *graphQuery, p *graphPlan, hotStart time.Time) (*GraphP95, error) {
	out := &GraphP95{Source: string(p95.SourceLive)}
	slots, ok, err := s.p95Slots(ctx, q, p, hotStart)
	if err != nil {
		return nil, err
	}
	if !ok {
		out.Partial = p.newest().tier != tierRaw
		return out, nil
	}
	expected := max(int64(p.to.Sub(p.from)/p95Slot), 1)
	coverage := 1.0
	for _, dir := range q.dirs {
		coverage = min(coverage, float64(len(slots[dir]))/float64(expected))
		if len(slots[dir]) == 0 {
			continue
		}
		v := nearestRank(mapValues(slots[dir]), p95Quantile)
		if dir == dirIn {
			out.InBPS = &v
		} else {
			out.OutBPS = &v
		}
	}
	out.AggregateBPS = aggregateP95(slots)
	out.Coverage = min(coverage, 1)
	out.Partial = out.Coverage < p95.LowCoverage
	return out, nil
}

// p95Slots fetches the 5-minute average rates of every requested direction over the raw part of the range, by
// direction and slot time. ok is false when the range has no raw slots to compute from.
func (s *server) p95Slots(
	ctx context.Context, q *graphQuery, p *graphPlan, hotStart time.Time,
) (slots map[string]map[int64]float64, ok bool, err error) {
	raw := p.newest()
	if raw.tier != tierRaw {
		return nil, false, nil
	}
	start := ceilTo(latest(raw.from, hotStart, raw.bound.Add(-maxP95Span)), p95Slot)
	end := floorTo(raw.bound, p95Slot)
	if end.Before(start) {
		return nil, false, nil
	}
	slots = map[string]map[int64]float64{}
	w := tierWindow{tier: tierRaw, from: start, last: end, bound: end, step: p95Slot}
	for _, dir := range q.dirs {
		m := map[int64]float64{}
		expr := `rate(` + q.octets(dir) + `{iface=` + quote(q.iface.ID) + `}[` + secs(max(p95Slot, rateWindowFor(q.every))) + `])*8`
		if err := s.fetchOne(ctx, "hot", expr, w, func(ts int64, v float64) { m[ts] = v }); err != nil {
			return nil, false, err
		}
		slots[dir] = m
	}
	return slots, true, nil
}

// aggregateP95 is the 95th percentile of in+out per 5-minute slot, the figure LibreNMS calls the aggregate. A slot
// counts only when both directions have a sample, so a one-sided gap never lowers the sum. Nil when either
// direction was not requested or no slot has both.
func aggregateP95(slots map[string]map[int64]float64) *float64 {
	in, out := slots[dirIn], slots[dirOut]
	if in == nil || out == nil {
		return nil
	}
	sums := make([]float64, 0, len(in))
	for ts, a := range in {
		if b, ok := out[ts]; ok {
			sums = append(sums, a+b)
		}
	}
	if len(sums) == 0 {
		return nil
	}
	v := nearestRank(sums, p95Quantile)
	return &v
}

func mapValues(m map[int64]float64) []float64 {
	out := make([]float64, 0, len(m))
	for _, v := range m {
		out = append(out, v)
	}
	return out
}

// latest returns the latest of ts.
func latest(ts ...time.Time) time.Time {
	out := ts[0]
	for _, t := range ts[1:] {
		if t.After(out) {
			out = t
		}
	}
	return out
}

// liveAggregate is the aggregate 95th for a figure whose in and out come from the materialised month: the
// month's own job stores no aggregate, so it is computed from the hot store when the month lies in it (nil
// otherwise, and when the store fails, since the figure is a bonus next to a good monthly answer).
func (s *server) liveAggregate(ctx context.Context, q *graphQuery, p *graphPlan, hotStart time.Time) *float64 {
	slots, ok, err := s.p95Slots(ctx, q, p, hotStart)
	if err != nil || !ok {
		return nil
	}
	return aggregateP95(slots)
}