jleung
b3aae800bf
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Lx9FNLG2CqeBbVGGd9bFsP
81 lines
3.4 KiB
Go
81 lines
3.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/v10networks/nmon/internal/longq"
|
|
"github.com/v10networks/nmon/internal/metrics"
|
|
"github.com/v10networks/nmon/internal/rollup"
|
|
)
|
|
|
|
// backfillLongCmd writes rollup series straight into the long store, as if the rollup job had been running
|
|
// for years: hourly average and maximum rates for --months months and 5-minute rates for --days days, flat
|
|
// at the given rates so a gate can assert exact values. The data ends --end-offset ago, leaving the recent
|
|
// hours to the real rollup job (its repair writes higher versions, which win).
|
|
func backfillLongCmd(ctx context.Context, args []string) error {
|
|
fs := devFlagSet("backfill-long")
|
|
var ifaces stringList
|
|
long := fs.String("long-url", "", "base URL of the long VictoriaMetrics (see \"metric stores up\" in the serve log)")
|
|
device := fs.String("device", "", "device slug (the series' device label)")
|
|
fs.Var(&ifaces, "iface", "interface ULID; repeat the flag or separate with commas")
|
|
months := fs.Int("months", 25, "months of hourly rollups to write")
|
|
days := fs.Int("days", 70, "days of 5-minute rates to write")
|
|
offset := fs.Duration("end-offset", 3*time.Hour, "stop this long before now")
|
|
in := fs.Float64("in-bps", 8e6, "constant inbound rate")
|
|
out := fs.Float64("out-bps", 2e6, "constant outbound rate")
|
|
if err := fs.Parse(args); err != nil {
|
|
return flagErr(err)
|
|
}
|
|
if *long == "" || *device == "" || len(ifaces) == 0 {
|
|
return fmt.Errorf("backfill-long needs --long-url, --device and --iface: %w", errUsage)
|
|
}
|
|
c := &backfillConfig{hotURL: strings.TrimRight(*long, "/"), hc: &http.Client{Timeout: 5 * time.Minute}}
|
|
end := time.Now().Add(-*offset).Truncate(time.Hour)
|
|
for _, id := range ifaces {
|
|
s := longSamples(*device, id, end, *months, *days, map[string]float64{"in": *in, "out": *out})
|
|
if err := c.send(ctx, s); err != nil {
|
|
return errors.Join(fmt.Errorf("iface %s", id), err)
|
|
}
|
|
fmt.Printf("backfilled long %s/%s: %d months hourly, %d days 5m (%d samples)\n", *device, id, *months, *days, len(s))
|
|
}
|
|
return c.flush(ctx)
|
|
}
|
|
|
|
// flush makes what was just imported searchable at once: VictoriaMetrics shows new rows only after its
|
|
// buffers flush, and a gate that queries right after the import would otherwise see a partial history.
|
|
func (c *backfillConfig) flush(ctx context.Context) error {
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.hotURL+"/internal/force_flush", http.NoBody)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_ = resp.Body.Close() // nothing in the reply matters
|
|
return nil
|
|
}
|
|
|
|
// longSamples builds the hourly series from end back months months and the 5-minute series from end back days.
|
|
func longSamples(device, iface string, end time.Time, months, days int, rate map[string]float64) []metrics.Sample {
|
|
var out []metrics.Sample
|
|
labels := func(dir string, ws time.Time) []string {
|
|
return []string{"device", device, "iface", iface, "dir", dir, "v", rollup.VersionLabel(ws, 1)}
|
|
}
|
|
for dir, bps := range rate {
|
|
for h := end; h.After(end.AddDate(0, -months, 0)); h = h.Add(-time.Hour) {
|
|
out = append(out, metrics.New(longq.MetricRateAvg1h, bps, h, labels(dir, h)...),
|
|
metrics.New(longq.MetricRateMax1h, bps*1.25, h, labels(dir, h)...))
|
|
}
|
|
for t := end; t.After(end.AddDate(0, 0, -days)); t = t.Add(-5 * time.Minute) {
|
|
out = append(out, metrics.New(longq.MetricRate5m, bps, t, labels(dir, t)...))
|
|
}
|
|
}
|
|
return out
|
|
}
|