nmon/internal/api/rates.go

package api

import (
	"context"
	"strings"
	"time"

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

// liveWindow is the lookback of the "live" rates on device and search pages: two default poll intervals, so a
// device polled at the default always has two samples inside it (never under the five minutes it used to be).
func (s *server) liveWindow() string { return secs(rateWindowFor(s.pollEvery())) }

// ifRates holds the five-minute rates of one interface by the counter they came
// from; pick chooses between the 64-bit and 32-bit counters.
type ifRates struct {
	hcIn, hcOut, in32, out32, inErr, outErr *float64
}

// rateKinds maps the k label of the batched query to the metric and whether it is a bits/s rate.
var rateKinds = []struct {
	k, metric string
	bits      bool
}{
	{"in_hc", "ifHCInOctets", true}, {"out_hc", "ifHCOutOctets", true},
	{"in_32", "ifInOctets", true}, {"out_32", "ifOutOctets", true},
	{"in_err", "ifInErrors", false}, {"out_err", "ifOutErrors", false},
}

// ratesExpr is one MetricsQL expression yielding every live rate of the interfaces
// matched by match (label matchers), each series tagged with k. When uptime is set
// it also yields the device's last sysUpTime under k="uptime".
func ratesExpr(match string, uptime bool, window string) string {
	parts := make([]string, 0, len(rateKinds)+1)
	for _, k := range rateKinds {
		scale := ""
		if k.bits {
			scale = "*8"
		}
		parts = append(parts, `label_set(rate(`+k.metric+`{`+match+`}[`+window+`])`+scale+`, "k", `+quote(k.k)+`)`)
	}
	if uptime {
		parts = append(parts, `label_set(last_over_time(sysUpTime{`+match+`}[`+window+`]), "k", "uptime")`)
	}
	return strings.Join(parts, " or ")
}

// liveRates runs the batched query. A store that is down, busy or slow yields
// no rates rather than failing the page: the fields are null, as for unknown.
func (s *server) liveRates(ctx context.Context, match string, uptime bool) (map[string]*ifRates, *float64) {
	release, err := s.admit(ctx, false)
	if err != nil {
		return nil, nil
	}
	defer release()
	qctx, cancel := context.WithTimeout(ctx, QueryTimeout)
	defer cancel()
	body, err := s.Querier.Query(qctx, "hot", ratesExpr(match, uptime, s.liveWindow()), s.Clock.Now())
	if err != nil {
		s.Logger.Warn("live rates unavailable", "err", err)
		return nil, nil
	}
	rows, err := parseVM(body)
	if err != nil {
		s.Logger.Warn("live rates unreadable", "err", err)
		return nil, nil
	}
	return indexRates(rows)
}

// indexRates sorts result rows into per-interface rates and the device uptime.
func indexRates(rows []vmSeries) (map[string]*ifRates, *float64) {
	out := map[string]*ifRates{}
	var uptime *float64
	for _, r := range rows {
		if r.Value == nil || !r.Value.OK {
			continue
		}
		v := r.Value.V
		if r.Metric["k"] == "uptime" {
			uptime = &v
			continue
		}
		id := r.Metric["iface"]
		x := out[id]
		if x == nil {
			x = &ifRates{}
			out[id] = x
		}
		switch r.Metric["k"] {
		case "in_hc":
			x.hcIn = &v
		case "out_hc":
			x.hcOut = &v
		case "in_32":
			x.in32 = &v
		case "out_32":
			x.out32 = &v
		case "in_err":
			x.inErr = &v
		case "out_err":
			x.outErr = &v
		}
	}
	return out, uptime
}

// bps returns the in and out rates from the counters the interface really has;
// an unreliable interface has none.
func (x *ifRates) bps(i *store.Interface) (in, out *float64) {
	if x == nil || i.Unreliable {
		return nil, nil
	}
	if i.HasHCCounters {
		return x.hcIn, x.hcOut
	}
	return x.in32, x.out32
}

// errs returns the error rates, none for an unreliable interface.
func (x *ifRates) errs(i *store.Interface) (in, out *float64) {
	if x == nil || i.Unreliable {
		return nil, nil
	}
	return x.inErr, x.outErr
}

// pollEvery is the default poll interval right now: the live cadence when the app gave one, else the fixed option.
func (s *server) pollEvery() time.Duration { return s.Cadence.Or(s.PollInterval) }