nmon/cmd/nmon-devtools/backfill.go

package main

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"math"
	"net/http"
	"net/url"
	"strings"
	"time"

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

// backfillConfig is what one `devtools backfill` run writes, for every interface in ifaces.
type backfillConfig struct {
	hotURL  string
	device  string
	ifaces  []string
	days    int
	step    time.Duration
	pattern string
	meanBps float64 // bits/s; 0 derives the mean from the live rate, else a stable per-interface value
	seed    uint64
	hc      *http.Client
	logf    func(format string, a ...any)
}

func backfillCmd(ctx context.Context, args []string) error {
	cfg := backfillConfig{hc: &http.Client{Timeout: 2 * time.Minute}, logf: func(f string, a ...any) { fmt.Printf(f+"\n", a...) }}
	fs := devFlagSet("backfill")
	var ifaces stringList
	var mbps float64
	fs.StringVar(&cfg.hotURL, "hot-url", "", "base URL of the hot VictoriaMetrics (required; see \"metric stores up\" in the serve log)")
	fs.StringVar(&cfg.device, "device", "", "device slug (the series' device label)")
	fs.Var(&ifaces, "iface", "interface ULID; repeat the flag or separate with commas")
	fs.IntVar(&cfg.days, "days", 31, "days of history to write, ending where the live data begins")
	fs.DurationVar(&cfg.step, "step", 60*time.Second, "sample spacing")
	fs.StringVar(&cfg.pattern, "pattern", patternDiurnal, "traffic shape: diurnal or flat")
	fs.Float64Var(&mbps, "mean-mbps", 0, "mean rate in Mbit/s; 0 follows the live rate when there is one")
	fs.Uint64Var(&cfg.seed, "seed", 1, "noise seed")
	if err := fs.Parse(args); err != nil {
		return flagErr(err)
	}
	cfg.ifaces, cfg.meanBps = ifaces, mbps*1e6
	if err := cfg.validate(); err != nil {
		return err
	}
	for _, id := range cfg.ifaces {
		if err := cfg.backfillIface(ctx, id); err != nil {
			return fmt.Errorf("iface %s: %w", id, err)
		}
	}
	return nil
}

func (c *backfillConfig) validate() error {
	switch {
	case c.hotURL == "":
		return errors.New("--hot-url is required")
	case c.device == "" || len(c.ifaces) == 0:
		return errors.New("--device and at least one --iface are required")
	case c.days < 1 || c.step < time.Second:
		return errors.New("--days must be at least 1 and --step at least 1s")
	case c.pattern != patternDiurnal && c.pattern != patternFlat:
		return fmt.Errorf("unknown --pattern %q (diurnal or flat)", c.pattern)
	}
	c.hotURL = strings.TrimRight(c.hotURL, "/")
	return nil
}

// liveSeries is what the hot store already holds for one series: the first
// sample (where history must end and which value it must join) and the live rate.
type liveSeries struct {
	ts   time.Time
	val  float64
	rate float64 // units per second between the first two samples; 0 when unknown
	ok   bool
}

// backfillIface writes in/out octet counters and the oper status for one interface.
func (c *backfillConfig) backfillIface(ctx context.Context, iface string) error {
	n := c.days * int(24*time.Hour/c.step)
	var samples []metrics.Sample
	var end time.Time
	for d, dir := range []struct {
		metric string
		salt   uint64
	}{{"ifHCInOctets", 0x1111}, {"ifHCOutOctets", 0x2222}} {
		live, err := c.probe(ctx, dir.metric, iface)
		if err != nil {
			return err
		}
		if d == 0 {
			end = c.endTime(live)
		}
		samples = append(samples, c.counter(dir.metric, iface, end, n, live, c.seed^dir.salt)...)
	}
	status := 1.0
	if live, err := c.probe(ctx, "ifOperStatus", iface); err == nil && live.ok {
		status = live.val
	}
	for i := 0; i < n; i++ {
		ts := end.Add(-time.Duration(n-1-i) * c.step)
		samples = append(samples, metrics.New("ifOperStatus", status, ts, metrics.LabelDevice, c.device, metrics.LabelIface, iface))
	}
	if err := c.send(ctx, samples); err != nil {
		return err
	}
	c.logf("backfilled %s/%s: %d days at %s (%d samples)", c.device, iface, c.days, c.step, len(samples))
	return nil
}

// endTime is one step before the first live sample, or one step ago when there is none.
func (c *backfillConfig) endTime(live liveSeries) time.Time {
	if live.ok {
		return live.ts.Add(-c.step).Truncate(time.Millisecond)
	}
	return time.Now().Add(-c.step).Truncate(time.Millisecond)
}

// counter builds one direction's history ending at end. When the live counter
// is known and large enough, the history is shifted so its last value is where
// the live counter was one step earlier and the join shows no reset.
func (c *backfillConfig) counter(metric, iface string, end time.Time, n int, live liveSeries, salt uint64) []metrics.Sample {
	mean := c.meanBps / 8
	if mean == 0 && live.rate > 0 {
		mean = live.rate
	}
	if mean == 0 {
		mean = defaultMeanBytesPerSec(iface, salt)
	}
	start := end.Add(-time.Duration(n-1) * c.step)
	vals := counterSeries(c.pattern, start, n, c.step, mean, salt)
	var offset float64
	if live.ok {
		offset = live.val - mean*c.step.Seconds() - vals[n-1]
		if offset < 0 {
			c.logf("warning: %s %s: live counter %.3g is below the %.3g bytes of history; the join will look like a counter reset",
				c.device, metric, live.val, vals[n-1])
			offset = 0
		}
	}
	out := make([]metrics.Sample, n)
	for i, v := range vals {
		ts := start.Add(time.Duration(i) * c.step)
		out[i] = metrics.New(metric, math.Floor(v+offset), ts, metrics.LabelDevice, c.device, metrics.LabelIface, iface)
	}
	return out
}

// probe reads the earliest samples of metric{iface=...} from the hot store.
func (c *backfillConfig) probe(ctx context.Context, metric, iface string) (liveSeries, error) {
	q := url.Values{"match[]": {fmt.Sprintf(`%s{iface=%q}`, metric, iface)}}
	var raw bytes.Buffer
	if err := c.get(ctx, c.hotURL+"/api/v1/export?"+q.Encode(), &raw); err != nil {
		return liveSeries{}, err
	}
	return parseExport(raw.Bytes())
}