nmon/internal/api/graph_fetch.go

package api

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/v10networks/nmon/internal/longq"
	"github.com/v10networks/nmon/internal/store"
	"github.com/v10networks/nmon/internal/vmsup"
)

const (
	dirIn  = "in"
	dirOut = "out"
	// minRateWindow keeps the rate window above two poll intervals so a step of
	// one minute still sees two samples.
	minRateWindow = 2 * time.Minute
	// maxInnerStep is the resolution of the max-over-step subquery.
	maxInnerStep = time.Minute
)

// graphData holds the samples found, by direction and Unix second.
type graphData struct {
	avg map[string]map[int64]float64
	max map[string]map[int64]float64
}

func newGraphData() *graphData {
	return &graphData{avg: map[string]map[int64]float64{}, max: map[string]map[int64]float64{}}
}

func (g *graphData) put(isMax bool, dir string, ts int64, v float64) {
	m := g.avg
	if isMax {
		m = g.max
	}
	if m[dir] == nil {
		m[dir] = map[int64]float64{}
	}
	if old, ok := m[dir][ts]; !ok || v > old {
		m[dir][ts] = v // stores agree on duplicates; the highest wins
	}
}

// graphQuery is what one fetch needs to know.
type graphQuery struct {
	iface   *store.Interface
	dirs    []string
	withMax bool
}

// octets is the counter that carries direction dir for the interface.
func (q *graphQuery) octets(dir string) string {
	switch {
	case q.iface.HasHCCounters && dir == dirIn:
		return "ifHCInOctets"
	case q.iface.HasHCCounters:
		return "ifHCOutOctets"
	case dir == dirIn:
		return "ifInOctets"
	}
	return "ifOutOctets"
}

func secs(d time.Duration) string { return strconv.FormatInt(int64(d/time.Second), 10) + "s" }

// rawExpr is the average bits per second of dir over the rate window.
func (q *graphQuery) rawExpr(dir string, step time.Duration) string {
	sel := q.octets(dir) + `{iface=` + quote(q.iface.ID) + `}`
	return `rate(` + sel + `[` + secs(max(step, minRateWindow)) + `])*8`
}

// rawMaxExpr is the highest one-minute rate seen within each step.
func (q *graphQuery) rawMaxExpr(dir string, step time.Duration) string {
	sel := q.octets(dir) + `{iface=` + quote(q.iface.ID) + `}`
	return `max_over_time((rate(` + sel + `[` + secs(maxInnerStep) + `])*8)[` + secs(step) + `:` + secs(maxInnerStep) + `])`
}

// longExpr reads an hourly rollup metric. A re-run of a window writes the next
// version, so longq.LatestVersion keeps the highest v per hour (a plain max
// would let a lower rewrite lose to a stale higher value). A coarser step
// averages (or takes the max of) the hours it covers.
func (q *graphQuery) longExpr(metric, over string, dir string, step time.Duration) string {
	inner := longq.LatestVersion(metric + `{iface=` + quote(q.iface.ID) + `,dir=` + quote(dir) + `}`)
	if step <= longStepMin {
		return inner
	}
	return over + `((` + inner + `)[` + secs(step) + `:` + secs(longStepMin) + `])`
}

// fetchTier fills d from the store that owns the tier. Hot errors fail the
// graph; a long store that is down only leaves its stretch empty, which the
// response reports as partial.
func (s *server) fetchTier(ctx context.Context, q *graphQuery, w tierWindow, d *graphData) error {
	storeName := "hot"
	if w.tier == tier1h {
		storeName = "long"
	}
	for _, dir := range q.dirs {
		for _, wantMax := range []bool{false, true} {
			if wantMax && !q.withMax {
				continue
			}
			err := s.fetchOne(ctx, storeName, q.exprFor(w, dir, wantMax), w, func(ts int64, v float64) { d.put(wantMax, dir, ts, v) })
			if err == nil {
				continue
			}
			if storeName == "long" && !errors.Is(err, context.Canceled) {
				s.Logger.Warn("long store query failed; its stretch is reported partial", "err", err)
				continue
			}
			return err
		}
	}
	return nil
}

func (q *graphQuery) exprFor(w tierWindow, dir string, wantMax bool) string {
	switch {
	case w.tier == tierRaw && wantMax:
		return q.rawMaxExpr(dir, w.step)
	case w.tier == tierRaw:
		return q.rawExpr(dir, w.step)
	case wantMax:
		return q.longExpr(longq.MetricRateMax1h, "max_over_time", dir, w.step)
	}
	return q.longExpr(longq.MetricRateAvg1h, "avg_over_time", dir, w.step)
}

// fetchOne runs one range query over the tier's window and hands every finite sample to add.
func (s *server) fetchOne(ctx context.Context, storeName, expr string, w tierWindow, add func(ts int64, v float64)) error {
	body, err := s.Querier.QueryRange(ctx, storeName, expr, w.from, w.last, w.step)
	if err != nil {
		return s.vmErr(ctx, err)
	}
	rows, err := parseVM(body)
	if err != nil {
		return fmt.Errorf("graph: %w", err)
	}
	for _, r := range rows {
		for _, p := range r.Values {
			if p.OK {
				add(p.TS, p.V)
			}
		}
	}
	return nil
}

// vmErr maps a metric-store failure to a problem. It is shared by /query and /graph.
func (s *server) vmErr(ctx context.Context, err error) error {
	switch {
	case errors.Is(err, vmsup.ErrStoreDown):
		return errStoreDown()
	case errors.Is(err, context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded):
		return errQueryTimeout()
	case isBadQuery(err):
		return badQuery(err)
	}
	s.Logger.Warn("query failed", "err", err)
	return errStoreFailed()
}