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 is the floor of every rate window, for devices polled faster than once a minute. minRateWindow = 2 * time.Minute // maxInnerStep is the finest resolution of the max-over-step subquery. maxInnerStep = time.Minute ) // rateWindowFor is the window of a rate() over a device polled every `every`: two polls, so a window always holds // two samples (VictoriaMetrics also takes the one before the window, which only helps). A zero every means the default. func rateWindowFor(every time.Duration) time.Duration { if every <= 0 { every = store.DefaultPollInterval } return max(minRateWindow, 2*every) } // 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 every time.Duration // the device's effective poll interval; sizes the rate windows } // 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, rateWindowFor(q.every))) + `])*8` } // rawMaxExpr is the highest poll-to-poll rate seen within each step. irate is the rate between the last two // samples in its window, which is the poll-to-poll rate whatever the cadence. func (q *graphQuery) rawMaxExpr(dir string, step time.Duration) string { sel := q.octets(dir) + `{iface=` + quote(q.iface.ID) + `}` return `max_over_time((irate(` + sel + `[` + secs(rateWindowFor(q.every)) + `])*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() }