nmon/internal/groups/p95.go

package groups

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

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

// Result is a group's monthly p95 with the membership it was computed from.
type Result struct {
	p95.Result
	// Flags are the membership warnings (a group mixing both ends of a link).
	Flags []Flag
}

// GroupP95 returns the group's p95 for month (YYYY-MM in Options.Zone). A
// finished month is answered from the materialised record when one exists for
// the membership that applies to that month today (its members_hash matches);
// otherwise it is computed live from the 5-minute rates.
func (s *Service) GroupP95(ctx context.Context, groupID int64, month string) (Result, error) {
	m, err := p95.ParseMonth(month, s.o.Zone)
	if err != nil {
		return Result{}, err
	}
	p, err := s.newPlan(ctx, groupID, m.Start, m.End)
	if err != nil {
		return Result{}, err
	}
	flags, err := s.flagsOf(ctx, p)
	if err != nil {
		return Result{}, err
	}
	now := s.o.Clock.Now()
	hash := p.hash(m)
	if m.Complete(now) {
		if r, ok, err := s.materialised(ctx, groupID, m, hash); err != nil || ok {
			return Result{Result: r, Flags: flags}, err
		}
	}
	f, err := s.figures(ctx, p, m, now)
	if err != nil {
		return Result{}, err
	}
	return Result{Result: f.result(m, hash), Flags: flags}, nil
}

// figures are the two directions' percentiles of a group month.
type figures struct{ in, out p95.Figure }

func (f figures) result(m p95.Month, hash string) p95.Result {
	return p95.Result{Month: m.Label, InBps: f.in.Bps, OutBps: f.out.Bps, Source: p95.SourceLive,
		Coverage: min(f.in.Coverage(), f.out.Coverage()), MembersHash: hash}
}

// figures computes the percentiles from the summed series. Only windows where
// every contributing member had a sample enter the percentile: a sum missing a
// member is a lower bound, and ranking lower bounds would understate the
// figure. Those windows count against coverage instead.
func (s *Service) figures(ctx context.Context, p plan, m p95.Month, now time.Time) (figures, error) {
	rates, err := s.rd.Rates(ctx, p.interfaceIDs(), m.Start, m.End)
	if err != nil {
		return figures{}, err
	}
	var in, out []p95.Point
	for _, sl := range p.sum(rates, m.Start, m.End) {
		if sl.AnyIn && sl.FullIn {
			in = append(in, p95.Point{TS: sl.TS, V: sl.In})
		}
		if sl.AnyOut && sl.FullOut {
			out = append(out, p95.Point{TS: sl.TS, V: sl.Out})
		}
	}
	fin, okIn := p95.Summarize(in, m, now)
	fout, okOut := p95.Summarize(out, m, now)
	if !okIn && !okOut {
		return figures{}, p95.ErrNoData
	}
	return figures{in: fin, out: fout}, nil
}

// hash identifies the membership used for month m: every contributing row
// clipped to the month, and the LAG relations that dropped members.
func (p plan) hash(m p95.Month) string {
	var lines []string
	for _, r := range p.members {
		from, to := r.EffectiveFrom, m.End
		if r.EffectiveTo != nil && r.EffectiveTo.Before(to) {
			to = *r.EffectiveTo
		}
		if from.Before(m.Start) {
			from = m.Start
		}
		line := fmt.Sprintf("%s|%s|%d|%d", r.InterfaceID, r.Direction, from.Unix(), to.Unix())
		if parent, ok := p.lag[r.InterfaceID]; ok {
			line += "|lag:" + parent
		}
		lines = append(lines, line)
	}
	sort.Strings(lines)
	sum := sha256.Sum256([]byte(strings.Join(lines, "\n")))
	return hex.EncodeToString(sum[:8])
}

func (s *Service) materialised(ctx context.Context, groupID int64, m p95.Month, hash string) (p95.Result, bool, error) {
	q := fmt.Sprintf(`last_over_time({__name__=%q,group=%q,month=%q,members_hash=%q}[1s])`,
		p95.MetricGroupP95, strconv.FormatInt(groupID, 10), m.Label, hash)
	body, err := s.o.Query.Query(ctx, "long", q, m.Start)
	if err != nil {
		return p95.Result{}, false, fmt.Errorf("groups: read materialised: %w", err)
	}
	r, ok, err := p95.DecodeRecords(body, m.Label)
	r.MembersHash = hash
	return r, ok, err
}

// records builds the two directional records of a group.
func records(groupID int64, m p95.Month, hash string, f figures) []metrics.Sample {
	mk := func(dir string, f p95.Figure) metrics.Sample {
		return metrics.New(p95.MetricGroupP95, f.Bps, m.Start, "group", strconv.FormatInt(groupID, 10), metrics.LabelDir, dir,
			"month", m.Label, "members_hash", hash, "coverage", p95.CoverageLabel(f.Coverage()))
	}
	return []metrics.Sample{mk(p95.DirIn, f.in), mk(p95.DirOut, f.out)}
}