nmon/cmd/nmon-devtools/pushreplay.go

package main

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

// pushReplayCmd sends a recorded Influx line-protocol file to a running nmon's POST /write, shifting every
// timestamp so the newest line is "now": a recording is days old by the time a gate replays it, and ingest
// refuses timestamps far from the clock.
func pushReplayCmd(ctx context.Context, args []string) error {
	fs := devFlagSet("push-replay")
	url := fs.String("url", "", "ingest URL, for example http://127.0.0.1:8443/write")
	token := fs.String("token", "", "push token secret (omit to test the unauthenticated case)")
	file := fs.String("file", "", "Influx line-protocol file")
	extra := fs.String("append", "", "one more line to send after the file (a raw line, rebased like the others)")
	if err := fs.Parse(args); err != nil {
		return flagErr(err)
	}
	raw, err := os.ReadFile(*file)
	if err != nil || *url == "" {
		return fmt.Errorf("push-replay needs --url and --file: %w", errors.Join(err, errUsage))
	}
	body, n, err := rebaseInflux(string(raw)+"\n"+*extra, time.Now())
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, *url, bytes.NewReader([]byte(body)))
	if err != nil {
		return err
	}
	if *token != "" {
		req.Header.Set("Authorization", "Bearer "+*token)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	defer func() { _ = resp.Body.Close() }() // body read below
	msg, _ := io.ReadAll(io.LimitReader(resp.Body, 500))
	fmt.Printf("%d lines -> HTTP %d %s\n", n, resp.StatusCode, strings.TrimSpace(string(msg)))
	if resp.StatusCode/100 != 2 {
		return fmt.Errorf("push refused: HTTP %d", resp.StatusCode)
	}
	return nil
}

// rebaseInflux shifts the nanosecond timestamp at the end of every line by the same amount so the newest one
// equals now. Lines without a timestamp, blank lines and comments pass through.
func rebaseInflux(text string, now time.Time) (string, int, error) {
	lines := strings.Split(text, "\n")
	var newest int64
	for _, l := range lines {
		if ts, ok := lineTS(l); ok {
			newest = max(newest, ts)
		}
	}
	shift := now.UnixNano() - newest
	var out strings.Builder
	n := 0
	for _, l := range lines {
		l = strings.TrimSpace(l)
		if l == "" || strings.HasPrefix(l, "#") {
			continue
		}
		n++
		if ts, ok := lineTS(l); ok {
			l = l[:strings.LastIndexByte(l, ' ')+1] + strconv.FormatInt(ts+shift, 10)
		}
		out.WriteString(l + "\n")
	}
	if n == 0 {
		return "", 0, errors.New("no lines to send")
	}
	return out.String(), n, nil
}

// lineTS returns the trailing nanosecond timestamp of an Influx line, if it has one.
func lineTS(l string) (int64, bool) {
	l = strings.TrimSpace(l)
	i := strings.LastIndexByte(l, ' ')
	if l == "" || strings.HasPrefix(l, "#") || i < 0 {
		return 0, false
	}
	ts, err := strconv.ParseInt(l[i+1:], 10, 64)
	return ts, err == nil && ts > 1e15
}