jleung
f85c3ef781
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Lx9FNLG2CqeBbVGGd9bFsP
85 lines
2.3 KiB
Go
85 lines
2.3 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/v10networks/nmon/internal/metrics"
|
|
)
|
|
|
|
// parseExport reads VictoriaMetrics' /api/v1/export (JSON lines) and returns
|
|
// the earliest sample of the first series.
|
|
func parseExport(raw []byte) (liveSeries, error) {
|
|
line, _, _ := bytes.Cut(bytes.TrimSpace(raw), []byte("\n"))
|
|
if len(line) == 0 {
|
|
return liveSeries{}, nil
|
|
}
|
|
var s struct {
|
|
Values []float64 `json:"values"`
|
|
Timestamps []int64 `json:"timestamps"`
|
|
}
|
|
if err := json.Unmarshal(line, &s); err != nil {
|
|
return liveSeries{}, fmt.Errorf("decode export: %w", err)
|
|
}
|
|
if len(s.Values) == 0 || len(s.Values) != len(s.Timestamps) {
|
|
return liveSeries{}, nil
|
|
}
|
|
out := liveSeries{ts: time.UnixMilli(s.Timestamps[0]).UTC(), val: s.Values[0], ok: true}
|
|
if len(s.Values) > 1 && s.Timestamps[1] > s.Timestamps[0] {
|
|
out.rate = (s.Values[1] - s.Values[0]) / (float64(s.Timestamps[1]-s.Timestamps[0]) / 1000)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (c *backfillConfig) get(ctx context.Context, u string, dst io.Writer) error {
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = resp.Body.Close() }() // read to the end below
|
|
if resp.StatusCode != http.StatusOK {
|
|
b, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
|
|
return fmt.Errorf("GET %s: %s: %s", u, resp.Status, b)
|
|
}
|
|
_, err = io.Copy(dst, resp.Body)
|
|
return err
|
|
}
|
|
|
|
// importBatch is how many samples go in one POST (about 5 MB of text).
|
|
const importBatch = 50_000
|
|
|
|
// send writes samples through /api/v1/import/prometheus in batches.
|
|
func (c *backfillConfig) send(ctx context.Context, samples []metrics.Sample) error {
|
|
for len(samples) > 0 {
|
|
n := min(len(samples), importBatch)
|
|
body := make([]byte, 0, n*110)
|
|
for _, s := range samples[:n] {
|
|
body = metrics.AppendPromText(body, s)
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.hotURL+"/api/v1/import/prometheus", bytes.NewReader(body))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
resp, err := c.hc.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
|
|
_ = resp.Body.Close() // reply already read
|
|
if resp.StatusCode/100 != 2 {
|
|
return fmt.Errorf("import: %s: %s", resp.Status, msg)
|
|
}
|
|
samples = samples[n:]
|
|
}
|
|
return nil
|
|
}
|