nmon/internal/backup/vmapi.go

package backup

import (
	"bufio"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"time"
)

// VMStore names one metric store to back up. DataDir is its -storageDataPath; a store nmon does not
// run itself (external URL) has no DataDir here and is skipped with a note in the manifest.
type VMStore struct {
	Name    string
	URL     string
	DataDir string
}

// sampleSeriesCount is how many series per store are fingerprinted into the manifest.
const sampleSeriesCount = 8

type vmAPI struct {
	hc   *http.Client
	base string
}

func (c vmAPI) get(ctx context.Context, path string, q url.Values) (io.ReadCloser, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+path+"?"+q.Encode(), nil)
	if err != nil {
		return nil, err
	}
	resp, err := c.hc.Do(req)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
		_ = resp.Body.Close() // the status is the error
		return nil, fmt.Errorf("%s: %s: %s", path, resp.Status, strings.TrimSpace(string(b)))
	}
	return resp.Body, nil
}

func (c vmAPI) getJSON(ctx context.Context, path string, q url.Values, out any) error {
	body, err := c.get(ctx, path, q)
	if err != nil {
		return err
	}
	defer func() { _ = body.Close() }() // fully read below
	return json.NewDecoder(body).Decode(out)
}

// createSnapshot asks the store for a snapshot and returns its name.
func (c vmAPI) createSnapshot(ctx context.Context) (string, error) {
	var r struct {
		Status   string `json:"status"`
		Snapshot string `json:"snapshot"`
		Msg      string `json:"msg"`
	}
	if err := c.getJSON(ctx, "/snapshot/create", url.Values{}, &r); err != nil {
		return "", err
	}
	if r.Status != "ok" || r.Snapshot == "" {
		return "", fmt.Errorf("snapshot/create: status %q %s", r.Status, r.Msg)
	}
	return r.Snapshot, nil
}

func (c vmAPI) deleteSnapshot(ctx context.Context, name string) error {
	var r struct {
		Status string `json:"status"`
		Msg    string `json:"msg"`
	}
	if err := c.getJSON(ctx, "/snapshot/delete", url.Values{"snapshot": {name}}, &r); err != nil {
		return err
	}
	if r.Status != "ok" {
		return fmt.Errorf("snapshot/delete %s: %s %s", name, r.Status, r.Msg)
	}
	return nil
}

// sampleSelectors picks up to n series that exist at or before end, in a stable order.
func (c vmAPI) sampleSelectors(ctx context.Context, end time.Time, n int) ([]string, error) {
	var r struct {
		Data []map[string]string `json:"data"`
	}
	q := url.Values{"match[]": {`{__name__!=""}`}, "start": {"0"}, "end": {unixSec(end)}, "limit": {"1000"}}
	if err := c.getJSON(ctx, "/api/v1/series", q, &r); err != nil {
		return nil, err
	}
	sels := make([]string, 0, len(r.Data))
	for _, labels := range r.Data {
		sels = append(sels, selectorOf(labels))
	}
	sort.Strings(sels)
	if len(sels) > n {
		sels = sels[:n]
	}
	return sels, nil
}

func selectorOf(labels map[string]string) string {
	names := make([]string, 0, len(labels))
	for k := range labels {
		names = append(names, k)
	}
	sort.Strings(names)
	parts := make([]string, 0, len(names))
	for _, k := range names {
		parts = append(parts, k+"="+strconv.Quote(labels[k]))
	}
	return "{" + strings.Join(parts, ",") + "}"
}

// fingerprint hashes every raw sample of the series selected, up to and including end.
func (c vmAPI) fingerprint(ctx context.Context, selector string, end time.Time) (SeriesCheck, error) {
	body, err := c.get(ctx, "/api/v1/export", url.Values{"match[]": {selector}, "start": {"0"}, "end": {unixSec(end)}})
	if err != nil {
		return SeriesCheck{}, err
	}
	defer func() { _ = body.Close() }() // fully read below
	h := sha256.New()
	var samples int64
	sc := bufio.NewScanner(body)
	sc.Buffer(make([]byte, 1<<20), 64<<20)
	for sc.Scan() {
		var line struct {
			Values     json.RawMessage `json:"values"`
			Timestamps []int64         `json:"timestamps"`
		}
		if err := json.Unmarshal(sc.Bytes(), &line); err != nil {
			return SeriesCheck{}, fmt.Errorf("export %s: %w", selector, err)
		}
		samples += int64(len(line.Timestamps))
		h.Write(line.Values)
		_, _ = fmt.Fprint(h, line.Timestamps)
	}
	if err := sc.Err(); err != nil {
		return SeriesCheck{}, err
	}
	return SeriesCheck{Selector: selector, Samples: samples, SHA256: hex.EncodeToString(h.Sum(nil))}, nil
}

// sampleChecks fingerprints a stable sample of the store's series.
func (c vmAPI) sampleChecks(ctx context.Context, end time.Time) ([]SeriesCheck, error) {
	sels, err := c.sampleSelectors(ctx, end, sampleSeriesCount)
	if err != nil {
		return nil, err
	}
	var out []SeriesCheck
	for _, sel := range sels {
		chk, err := c.fingerprint(ctx, sel, end)
		if err != nil {
			return nil, err
		}
		if chk.Samples > 0 {
			out = append(out, chk)
		}
	}
	return out, nil
}

func unixSec(t time.Time) string { return strconv.FormatInt(t.Unix(), 10) }