Compare commits

..

4 Commits

Author SHA1 Message Date
Haley Wang
a016eddac0 improve chart 2026-06-11 00:58:27 +08:00
Haley Wang
3aa4a748e7 tidy 2026-06-11 00:51:09 +08:00
Haley Wang
b03ea35f54 add /report page 2026-06-10 23:53:28 +08:00
Haley Wang
a847924000 add sa-tester 2026-06-10 22:00:14 +08:00
64 changed files with 1950 additions and 1078 deletions

View File

@@ -33,8 +33,7 @@ jobs:
name: ${{ matrix.os }}-${{ matrix.arch }}
permissions:
contents: read
# Runs on dedicated runner with extra resources to increase build speed.
runs-on: 'vm-runner'
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:

View File

@@ -30,8 +30,7 @@ jobs:
name: lint
permissions:
contents: read
# Runs on dedicated runner with extra resources since golangci-lint requires extra memory
runs-on: 'vm-runner'
runs-on: ubuntu-latest
steps:
- name: Code checkout
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -65,8 +64,7 @@ jobs:
name: unit
permissions:
contents: read
# Runs on dedicated runner with extra resources to increase tests speed.
runs-on: 'vm-runner'
runs-on: ubuntu-latest
strategy:
matrix:
@@ -97,7 +95,6 @@ jobs:
name: apptest
permissions:
contents: read
# Runs on dedicated runner to isolate app tests from other tests.
runs-on: apptest
steps:

View File

@@ -3,14 +3,27 @@ linters:
settings:
errcheck:
exclude-functions:
- fmt.Fprintf
- fmt.Fprint
- (net/http.ResponseWriter).Write
exclusions:
generated: lax
presets:
- common-false-positives
- legacy
- std-error-handling
rules:
- linters:
- staticcheck
text: 'SA(4003|1019|5011):'
paths:
- ^app/vmui/
- third_party$
- builtin$
- examples$
formatters:
exclusions:
generated: lax
paths:
- third_party$
- builtin$
- examples$

View File

@@ -17,7 +17,7 @@ EXTRA_GO_BUILD_TAGS ?=
GO_BUILDINFO = -X '$(PKG_PREFIX)/lib/buildinfo.Version=$(APP_NAME)-$(DATEINFO_TAG)-$(BUILDINFO_TAG)'
TAR_OWNERSHIP ?= --owner=1000 --group=1000
GOLANGCI_LINT_VERSION := 2.12.2
GOLANGCI_LINT_VERSION := 2.9.0
.PHONY: $(MAKECMDGOALS)
@@ -527,7 +527,7 @@ golangci-lint: install-golangci-lint
golangci-lint run --build-tags 'synctest'
install-golangci-lint:
which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://golangci-lint.run/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
remove-golangci-lint:
rm -rf `which golangci-lint`

View File

@@ -0,0 +1,21 @@
The `sa-tester` provides a stream aggregation config on the `/sa-config` endpoint for an external `vmagent` to read, writes configured mock series to `vmagent` for aggregation, and receives the output aggregation results from `vmagent`. It will print all the input and output samples in its logs, and generate a report with them.
See `app/test/sa_tester/config.yaml` for all supported options.
**Test steps:**
1. Start the `sa-tester` with the local config file `app/test/sa_tester/config.yaml`.
2. Start a `vmagent` instance using the version you want to test:
- 2.1. Use `<sa-tester-addr>/api/v1/write` as one of the `-remoteWrite.url` values. You can also add other `remoteWrite` URLs to `vmsingle` to check the aggregation results.
- 2.2. Use `-streamAggr.config=http://192.168.0.102:8880/sa-config` to make `vmagent` use the stream aggregation config from `sa-tester`.
3. Call `<sa-tester-addr>/start` when you're ready. The `sa-tester` will call the `vmagent` reload endpoint to ensure the stream aggregation config is up to date, and start writing the series to `vmagent` for aggregation.
4. Check results using the `sa-tester` logs, `<sa-tester-addr>/report` page or other remoteWrite destinations.
![alt text](image-1.png)
**Notes:**
1. You can safely rerun the test without restarting the `vmagent` instance if you changed the stream aggregation config (due to the reload call) or just wait for a while (due to the default staleness interval).
2. You can call `<sa-tester-addr>/reset` to update the `/sa-config` endpoint and `input_series` configs, then call `<sa-tester-addr>/start` to start a new test.
3. You can replace the above `vmagent` with `vmsingle` if you want to check results only in `vmsingle` instead of in the `sa-tester` logs.
4. Do not send extra metrics from vmagent to sa-tester, it will generate a wall of logs:)

Binary file not shown.

View File

@@ -0,0 +1,40 @@
# Stream aggregation rules served on GET /sa-config.
# Configure vmagent with:
# -streamAggr.config=http://localhost:8080/sa-config
saRules:
- name: 'foobar'
match: 'test'
interval: 10s
without: [instance]
outputs: [total]
- name: 'foobar'
match: 'test'
interval: 10s
without: [instance]
outputs: [total_prometheus]
# Address of the vmagent that loads SA rules and accepts raw samples.
# POST /start calls <vmagent_address>/-/reload then writes samples to it.
vmagent_address: "http://192.168.0.102:8420"
# Listen address for this tester's HTTP server.
listen_address: ":8880"
# --- Below are for generating input samples for testing. ---
# Time between consecutive sample slots.
# Currently, interval is global, you can have different intervals for different input series by insert null between values.
interval: 10s
input_series:
- series: 'test{env="prod", instance="a"}'
# One value per slot (1-indexed). null means slot is skipped but could be compensated by delays.
values: [null, null, 1, 2, 3, null, null, 6, 7, 8]
# delays can be used to mock delayed or out of order sample cases.
# delays: [originalSlot, sendAtSlot, value] (slots are 1-indexed)
# Sample is timestamped at T+(originalSlot-1)*interval, but delivered to vmagent at T+(sendAtSlot-1)*interval.
delays:
- [6, 8, 4] # slot-6 sample (value=4) sent together at slot 8
- [7, 8, 5] # slot-7 sample (value=5) sent together at slot 8
- series: 'test{env="prod", instance="b"}'
values: [null, null, 1, 2, 3, 4, 5, 6, 7, 8]

Binary file not shown.

After

Width:  |  Height:  |  Size: 495 KiB

622
app/test/sa_tester/main.go Normal file
View File

@@ -0,0 +1,622 @@
// sa_tester is a tool for testing stream aggregation rules.
//
// It does two things:
// 1. Serves a stream aggregation config YAML on GET /sa-config so that
// vmagent can pull it with -streamAggr.config=http://host/sa-config.
// 2. On POST /start it calls vmagent's /-/reload (recording T=now),
// then writes synthetic samples to the same vmagent.
//
// Usage:
//
// go run ./app/test/sa_tester -config app/test/sa_tester/config.yaml
package main
import (
"bytes"
"encoding/json"
"flag"
"fmt"
"html"
"io"
"log"
"math/rand"
"net/http"
"os"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite/stream"
"gopkg.in/yaml.v2"
)
type AppConfig struct {
// SARules holds the stream aggregation rules. It is marshalled back to
// YAML and served verbatim on GET /sa-config.
SARules interface{} `yaml:"saRules"`
// Interval between consecutive sample slots, e.g. "10s".
Interval string `yaml:"interval"`
// InputSeries defines the time series to generate.
InputSeries []InputSeries `yaml:"input_series"`
// VmagentAddress is the vmagent that loads SA rules and accepts raw samples.
// POST /start calls <VmagentAddress>/-/reload, then writes samples to it.
// Default: http://localhost:8429
VmagentAddress string `yaml:"vmagent_address"`
// ListenAddress is the HTTP listen address for this tester.
// Default: :8080
ListenAddress string `yaml:"listen_address"`
}
// InputSeries describes how to generate one time series.
type InputSeries struct {
// Series is a Prometheus-style selector, e.g. 'test1{env="prod",instance="a"}'.
Series string `yaml:"series"`
// Values lists the sample values for consecutive slots (1-indexed).
// A null entry means the slot is skipped or handled by Delays.
Values []*float64 `yaml:"values"`
// Delays is a list of [originalSlot, sendAtSlot, value] triples (1-indexed).
// The sample is timestamped at T+(originalSlot-1)*interval
// but sent to vmagent at T+(sendAtSlot-1)*interval.
//
// Example: [4, 6, 4] means a sample with value=4 whose logical timestamp
// is slot 4 is actually delivered at slot 6 together with the slot-6 sample.
Delays [][]float64 `yaml:"delays"`
}
// scheduledSample is a single data point waiting to be sent.
type scheduledSample struct {
timestamp int64 // sample timestamp in milliseconds
value float64
}
// vmImportLine is one line of the VictoriaMetrics /api/v1/import NDJSON format.
type vmImportLine struct {
Metric map[string]string `json:"metric"`
Values []float64 `json:"values"`
Timestamps []int64 `json:"timestamps"`
}
// --- report data types -------------------------------------------------------
// sentDataPoint is one data point in the sent-series chart.
type sentDataPoint struct {
TsSec float64 `json:"x"` // sample unix timestamp in seconds
Value float64 `json:"y"`
SentAtSec float64 `json:"sentAt"` // wall-clock send time in seconds; equals TsSec if not delayed
Delayed bool `json:"delayed"`
}
// sentSeriesData holds all sent data points for one configured input series.
type sentSeriesData struct {
Name string `json:"name"`
Points []sentDataPoint `json:"points"`
}
// recvDataPoint is one sample received on /api/v1/write.
type recvDataPoint struct {
TsSec float64 `json:"x"` // sample unix timestamp in seconds
Value float64 `json:"y"`
}
// recvSeriesData holds all received samples for one metric series.
type recvSeriesData struct {
Name string `json:"name"`
Points []recvDataPoint `json:"points"`
}
var (
cfg *AppConfig
configPath string // path of the config file, stored at startup for hot-reload
saYAML []byte // SA config YAML to serve
mu sync.Mutex
started bool
reportMu sync.RWMutex
reportT time.Time
reportJitter time.Duration
reportSent []sentSeriesData
reportRecv = make(map[string]*recvSeriesData)
)
func main() {
configFile := flag.String("config", "config.yaml", "path to config YAML file")
flag.Parse()
configPath = *configFile
if err := loadConfig(); err != nil {
log.Fatalf("cannot load config: %v", err)
}
mux := http.NewServeMux()
mux.HandleFunc("/sa-config", handleSAConfig)
mux.HandleFunc("/start", handleStart)
mux.HandleFunc("/reset", handleReset)
mux.HandleFunc("/api/v1/write", handleRemoteWrite)
mux.HandleFunc("/report", handleReport)
log.Printf("HTTP server listening on %s", cfg.ListenAddress)
log.Printf("Endpoints:")
log.Printf(" GET /sa-config — serve SA rules YAML for vmagent")
log.Printf(" POST /start — call vmagent /-/reload then write samples")
log.Printf(" POST /reset — reload config, clear 'started' flag")
log.Printf(" POST /api/v1/write — receive Prometheus remote-write from vmagent SA output")
log.Printf(" GET /report — HTML report with sent/received series charts")
log.Fatalf("server stopped: %v", http.ListenAndServe(cfg.ListenAddress, mux))
}
// handleSAConfig serves the SA rules YAML so that vmagent can fetch it.
func handleSAConfig(w http.ResponseWriter, r *http.Request) {
log.Printf("[sa-config] request from %s", r.RemoteAddr)
w.Header().Set("Content-Type", "text/yaml; charset=utf-8")
if _, err := w.Write(saYAML); err != nil {
log.Printf("[sa-config] write error: %v", err)
}
}
// handleStart triggers vmagent reload and starts the sample writer goroutine.
func handleStart(w http.ResponseWriter, r *http.Request) {
mu.Lock()
if started {
mu.Unlock()
http.Error(w, "test already running; POST /reset to allow re-running", http.StatusConflict)
return
}
started = true
mu.Unlock()
interval, err := time.ParseDuration(cfg.Interval)
if err != nil {
http.Error(w, fmt.Sprintf("invalid interval %q: %v", cfg.Interval, err), http.StatusBadRequest)
mu.Lock()
started = false
mu.Unlock()
return
}
// Call vmagent /-/reload so it picks up the SA config from /sa-config.
reloadURL := cfg.VmagentAddress + "/-/reload"
log.Printf("[start] calling vmagent reload: POST %s", reloadURL)
reloadResp, err := http.Post(reloadURL, "application/json", nil) //nolint:noctx
if err != nil {
log.Printf("[start] reload request failed: %v", err)
http.Error(w, fmt.Sprintf("vmagent reload failed: %v", err), http.StatusBadGateway)
mu.Lock()
started = false
mu.Unlock()
return
}
reloadBody, _ := io.ReadAll(reloadResp.Body)
reloadResp.Body.Close()
log.Printf("[start] reload response: status=%d body=%q", reloadResp.StatusCode, reloadBody)
T := time.Now()
jitter := time.Duration(rand.Int63n(int64(interval / 2))) //nolint:gosec
log.Printf("[start] T=%s jitter=%v first-sample-at T+%v",
T.Format(time.RFC3339Nano), jitter, jitter)
reportMu.Lock()
reportT = T
reportJitter = jitter
reportSent = nil
reportRecv = make(map[string]*recvSeriesData)
reportMu.Unlock()
go runTest(T, interval, jitter)
fmt.Fprintf(w, "test started\nT=%s\njitter=%v\n", T.Format(time.RFC3339Nano), jitter)
}
// handleRemoteWrite accepts Prometheus remote-write requests (e.g. from vmagent's SA output)
// and logs each received time series in a human-readable format for result verification.
func handleRemoteWrite(w http.ResponseWriter, r *http.Request) {
isVMRemoteWrite := r.Header.Get("Content-Encoding") == "zstd"
err := stream.Parse(r.Body, isVMRemoteWrite, func(tss []prompb.TimeSeries, _ []prompb.MetricMetadata) error {
for i := range tss {
ts := &tss[i]
var sb strings.Builder
// Build metric name + labels string.
var metricName string
for _, lbl := range ts.Labels {
if lbl.Name == "__name__" {
metricName = lbl.Value
break
}
}
sb.WriteString(metricName)
sb.WriteByte('{')
first := true
for _, lbl := range ts.Labels {
if lbl.Name == "__name__" {
continue
}
if !first {
sb.WriteByte(',')
}
first = false
sb.WriteString(lbl.Name)
sb.WriteString(`="`)
sb.WriteString(lbl.Value)
sb.WriteByte('"')
}
sb.WriteByte('}')
metricStr := sb.String()
// Log each sample on its own line for easy reading.
for _, s := range ts.Samples {
t := time.UnixMilli(s.Timestamp)
log.Printf("[recv] %-60s value=%-12g ts= %v, ts_human=%s",
metricStr, s.Value, t.UnixMilli(), t.UTC().Format(time.RFC3339Nano))
}
// Record for /report.
reportMu.Lock()
if reportRecv == nil {
reportRecv = make(map[string]*recvSeriesData)
}
rd := reportRecv[metricStr]
if rd == nil {
rd = &recvSeriesData{Name: metricStr}
reportRecv[metricStr] = rd
}
for _, s := range ts.Samples {
rd.Points = append(rd.Points, recvDataPoint{
TsSec: float64(s.Timestamp) / 1000.0,
Value: s.Value,
})
}
reportMu.Unlock()
}
return nil
})
if err != nil {
log.Printf("[recv] parse error: %v", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
w.WriteHeader(http.StatusNoContent)
}
// handleReset re-reads the config file from disk, updates the SA config and
// input_series, and clears the started flag so the test can be triggered again.
func handleReset(w http.ResponseWriter, r *http.Request) {
mu.Lock()
if err := loadConfig(); err != nil {
mu.Unlock()
log.Printf("[reset] failed to reload config: %v", err)
http.Error(w, fmt.Sprintf("config reload failed: %v", err), http.StatusInternalServerError)
return
}
started = false
mu.Unlock()
reportMu.Lock()
reportSent = nil
reportRecv = make(map[string]*recvSeriesData)
reportT = time.Time{}
reportMu.Unlock()
log.Printf("[reset] config reloaded, started flag cleared")
fmt.Fprintln(w, "reset ok")
}
// loadConfig reads configPath from disk, parses it into cfg, and rebuilds saYAML.
// Callers that need thread safety must hold mu.
func loadConfig() error {
data, err := os.ReadFile(configPath)
if err != nil {
return fmt.Errorf("cannot read config file %q: %w", configPath, err)
}
newCfg := &AppConfig{
VmagentAddress: "http://localhost:8429",
ListenAddress: ":8080",
}
if err := yaml.Unmarshal(data, newCfg); err != nil {
return fmt.Errorf("cannot parse config: %w", err)
}
var newSAYAML []byte
if newCfg.SARules != nil {
newSAYAML, err = yaml.Marshal(newCfg.SARules)
if err != nil {
return fmt.Errorf("cannot re-marshal saRules to YAML: %w", err)
}
} else {
newSAYAML = []byte("[]\n")
}
cfg = newCfg
saYAML = newSAYAML
log.Printf("[config] loaded from %q", configPath)
log.Printf("[config] interval : %s", cfg.Interval)
log.Printf("[config] vmagent : %s", cfg.VmagentAddress)
log.Printf("[config] listen : %s", cfg.ListenAddress)
log.Printf("[config] input_series count: %d", len(cfg.InputSeries))
log.Printf("[config] SA config:\n---\n%s---", saYAML)
return nil
}
// --- label parsing -----------------------------------------------------------
// seriesRe matches "metricname" or "metricname{k="v",...}".
var seriesRe = regexp.MustCompile(`^([^{,\s]+?)(?:\{([^}]*)\})?$`)
var labelRe = regexp.MustCompile(`(\w+)="([^"]*)"`)
// parseLabels converts a Prometheus-style selector string into a flat label map.
func parseLabels(series string) (map[string]string, error) {
series = strings.TrimSpace(series)
m := seriesRe.FindStringSubmatch(series)
if m == nil {
return nil, fmt.Errorf("cannot parse series selector %q", series)
}
labels := map[string]string{"__name__": m[1]}
if m[2] != "" {
for _, pair := range labelRe.FindAllStringSubmatch(m[2], -1) {
labels[pair[1]] = pair[2]
}
}
return labels, nil
}
// --- schedule building -------------------------------------------------------
// buildSchedule constructs a map of sendAtMs → []scheduledSample for a series.
//
// Slot numbering is 1-indexed:
// - slot i has sample timestamp T+jitter+(i-1)*interval
// - non-null values[i-1] are sent at their own slot time
// - delay entry [orig, sendAt, val] sends a sample timestamped at slot orig
// but delivered to vmagent at slot sendAt
func buildSchedule(is InputSeries, T time.Time, interval, jitter time.Duration) (map[int64][]scheduledSample, error) {
type delayEntry struct {
sendAtSlot int
value float64
}
delayMap := make(map[int]delayEntry, len(is.Delays))
for _, d := range is.Delays {
if len(d) != 3 {
return nil, fmt.Errorf("each delay must be [originalSlot, sendAtSlot, value]; got %v", d)
}
delayMap[int(d[0])] = delayEntry{sendAtSlot: int(d[1]), value: d[2]}
}
schedule := make(map[int64][]scheduledSample)
// Regular (non-null) values — sent at their natural slot time.
for i, v := range is.Values {
if v == nil {
continue // null → slot is handled by delays or intentionally absent
}
sampleTime := T.Add(jitter + time.Duration(i)*interval)
sendAtMs := sampleTime.UnixMilli()
schedule[sendAtMs] = append(schedule[sendAtMs], scheduledSample{
timestamp: sampleTime.UnixMilli(),
value: *v,
})
}
// Delayed values — timestamped at originalSlot, sent at sendAtSlot.
for origSlot, de := range delayMap {
sampleTime := T.Add(jitter + time.Duration(origSlot-1)*interval)
sendAt := T.Add(jitter + time.Duration(de.sendAtSlot-1)*interval)
sendAtMs := sendAt.UnixMilli()
schedule[sendAtMs] = append(schedule[sendAtMs], scheduledSample{
timestamp: sampleTime.UnixMilli(),
value: de.value,
})
}
// Sort samples within each slot by timestamp for deterministic ordering.
for k, s := range schedule {
sort.Slice(s, func(i, j int) bool { return s[i].timestamp < s[j].timestamp })
schedule[k] = s
}
return schedule, nil
}
// --- test runner -------------------------------------------------------------
type seriesEvent struct {
seriesName string // original series selector string from config
labels map[string]string
samples []scheduledSample
}
func runTest(T time.Time, interval, jitter time.Duration) {
defer func() {
mu.Lock()
started = false
mu.Unlock()
log.Printf("[write] finished")
}()
// Collect all send events across every configured series.
allEvents := make(map[int64][]seriesEvent)
for _, is := range cfg.InputSeries {
labels, err := parseLabels(is.Series)
if err != nil {
log.Printf("[test] cannot parse series %q: %v — skipping", is.Series, err)
continue
}
schedule, err := buildSchedule(is, T, interval, jitter)
if err != nil {
log.Printf("[test] cannot build schedule for %q: %v — skipping", is.Series, err)
continue
}
log.Printf("[test] series %q: %d distinct send-time slots", is.Series, len(schedule))
for sendAtMs, samples := range schedule {
allEvents[sendAtMs] = append(allEvents[sendAtMs], seriesEvent{
seriesName: is.Series,
labels: labels,
samples: samples,
})
}
}
// Sort the unique send times into chronological order.
sendTimes := make([]int64, 0, len(allEvents))
for t := range allEvents {
sendTimes = append(sendTimes, t)
}
sort.Slice(sendTimes, func(i, j int) bool { return sendTimes[i] < sendTimes[j] })
log.Printf("[test] starting: %d distinct send times across %d series",
len(sendTimes), len(cfg.InputSeries))
for _, sendAtMs := range sendTimes {
sendAt := time.UnixMilli(sendAtMs)
if now := time.Now(); sendAt.After(now) {
sleep := sendAt.Sub(now)
time.Sleep(sleep)
}
for _, ev := range allEvents[sendAtMs] {
recordSent(ev.seriesName, ev.samples, sendAtMs)
if err := writeSamples(cfg.VmagentAddress, ev.labels, ev.samples); err != nil {
log.Printf("[test] write error for %v: %v", ev.labels, err)
}
}
}
}
// --- vmagent writer ----------------------------------------------------------
// writeSamples POSTs one NDJSON line to vmagent's /api/v1/import endpoint.
// All samples for the same metric series are batched into a single request.
func writeSamples(addr string, labels map[string]string, samples []scheduledSample) error {
values := make([]float64, len(samples))
timestamps := make([]int64, len(samples))
for i, s := range samples {
values[i] = s.value
timestamps[i] = s.timestamp
}
line := vmImportLine{
Metric: labels,
Values: values,
Timestamps: timestamps,
}
data, err := json.Marshal(line)
if err != nil {
return fmt.Errorf("marshal: %w", err)
}
data = append(data, '\n')
url := addr + "/api/v1/import"
log.Printf("[write] metric=%v values=%v timestamps_ms=%v",
labels, values, timestamps)
resp, err := http.Post(url, "application/json", bytes.NewReader(data)) //nolint:noctx
if err != nil {
return fmt.Errorf("POST %s: %w", url, err)
}
body, _ := io.ReadAll(resp.Body)
resp.Body.Close()
if resp.StatusCode/100 != 2 {
return fmt.Errorf("POST %s: status=%d body=%q", url, resp.StatusCode, body)
}
return nil
}
// --- report ------------------------------------------------------------------
// recordSent stores sent samples into the report data store.
// It must be called without holding reportMu.
func recordSent(seriesName string, samples []scheduledSample, sendAtMs int64) {
reportMu.Lock()
defer reportMu.Unlock()
var sd *sentSeriesData
for i := range reportSent {
if reportSent[i].Name == seriesName {
sd = &reportSent[i]
break
}
}
if sd == nil {
reportSent = append(reportSent, sentSeriesData{Name: seriesName})
sd = &reportSent[len(reportSent)-1]
}
for _, s := range samples {
sd.Points = append(sd.Points, sentDataPoint{
TsSec: float64(s.timestamp) / 1000.0,
Value: s.value,
SentAtSec: float64(sendAtMs) / 1000.0,
Delayed: sendAtMs != s.timestamp,
})
}
}
// handleReport renders and serves the HTML report page with sent/received charts.
func handleReport(w http.ResponseWriter, r *http.Request) {
// saYAML is read without mu, consistent with handleSAConfig.
saYAMLStr := string(saYAML)
reportMu.RLock()
sentSnap := make([]sentSeriesData, len(reportSent))
for i, sd := range reportSent {
pts := make([]sentDataPoint, len(sd.Points))
copy(pts, sd.Points)
sentSnap[i] = sentSeriesData{Name: sd.Name, Points: pts}
}
recvSnap := make([]recvSeriesData, 0, len(reportRecv))
for _, rd := range reportRecv {
pts := make([]recvDataPoint, len(rd.Points))
copy(pts, rd.Points)
recvSnap = append(recvSnap, recvSeriesData{Name: rd.Name, Points: pts})
}
reportMu.RUnlock()
sort.Slice(recvSnap, func(i, j int) bool { return recvSnap[i].Name < recvSnap[j].Name })
sentJSON, _ := json.Marshal(sentSnap)
recvJSON, _ := json.Marshal(recvSnap)
// Single canvas for all sent series combined.
var sentCanvas string
if len(sentSnap) == 0 {
sentCanvas = `<p class="no-data">No data yet — POST /start to run the test.</p>`
} else {
sentCanvas = `<div class="chart-wrap"><canvas id="sent-all"></canvas></div>`
}
var recvCharts strings.Builder
if len(recvSnap) == 0 {
recvCharts.WriteString(`<p class="no-data">No data received yet.</p>`)
} else {
for i, rd := range recvSnap {
fmt.Fprintf(&recvCharts,
"<h3>%s</h3><div class=\"chart-wrap\"><canvas id=\"recv-%d\"></canvas></div>\n",
html.EscapeString(rd.Name), i)
}
}
page := reportPageTemplate
page = strings.ReplaceAll(page, "__GENERATED_AT__", time.Now().UTC().Format(time.RFC3339))
page = strings.ReplaceAll(page, "__SA_YAML__", html.EscapeString(saYAMLStr))
page = strings.ReplaceAll(page, "__SENT_COUNT__", strconv.Itoa(len(sentSnap)))
page = strings.ReplaceAll(page, "__SENT_CANVAS__", sentCanvas)
page = strings.ReplaceAll(page, "__RECV_COUNT__", strconv.Itoa(len(recvSnap)))
page = strings.ReplaceAll(page, "__RECV_CHARTS__", recvCharts.String())
page = strings.ReplaceAll(page, "__SENT_JSON__", string(sentJSON))
page = strings.ReplaceAll(page, "__RECV_JSON__", string(recvJSON))
w.Header().Set("Content-Type", "text/html; charset=utf-8")
fmt.Fprint(w, page)
}

View File

@@ -0,0 +1,222 @@
package main
// reportPageTemplate is the HTML skeleton for GET /report.
// All __TOKEN__ placeholders are replaced at render time by handleReport.
var reportPageTemplate = `<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8">
<title>SA Tester Report</title>
<style>
body{font-family:'Segoe UI',Arial,sans-serif;margin:0;padding:20px;background:#f0f2f5;color:#333}
h1{margin-bottom:4px}
.subtitle{color:#666;font-size:14px;margin-bottom:24px}
.card{background:#fff;padding:20px;margin:16px 0;border-radius:8px;box-shadow:0 1px 4px rgba(0,0,0,.12)}
h2{margin:0 0 12px;color:#444;font-size:18px;border-bottom:1px solid #eee;padding-bottom:8px}
h3{margin:4px 0 8px;font-size:13px;font-family:monospace;color:#555;word-break:break-all}
pre{background:#f7f7f7;padding:12px;border-radius:4px;overflow-x:auto;font-size:13px;margin:0}
.chart-wrap{position:relative;height:340px;margin-bottom:24px}
.no-data{color:#aaa;font-style:italic;font-size:14px}
a{color:#1a73e8}
</style>
</head>
<body>
<h1>SA Tester Report</h1>
<p class="subtitle">Generated: __GENERATED_AT__ &nbsp;|&nbsp; <a href="/report">Refresh</a></p>
<div class="card">
<h2>SA Config</h2>
<pre>__SA_YAML__</pre>
</div>
<div class="card">
<h2>Sent Series (__SENT_COUNT__ series)</h2>
__SENT_CANVAS__
</div>
<div class="card">
<h2>Received Series / SA Output (__RECV_COUNT__ series)</h2>
__RECV_CHARTS__
</div>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.0/dist/chart.umd.min.js"></script>
<script>
(function(){
var SENT = __SENT_JSON__;
var RECV = __RECV_JSON__;
var PALETTE = [
'rgba(54,162,235,0.85)',
'rgba(255,99,132,0.85)',
'rgba(153,102,255,0.85)',
'rgba(255,205,86,0.9)',
'rgba(75,192,192,0.85)',
];
function fmtDate(d) {
return d.getUTCFullYear() + '-' +
(d.getUTCMonth()+1).toString().padStart(2,'0') + '-' +
d.getUTCDate().toString().padStart(2,'0');
}
function fmtTime(d) {
return d.getUTCHours().toString().padStart(2,'0') + ':' +
d.getUTCMinutes().toString().padStart(2,'0') + ':' +
d.getUTCSeconds().toString().padStart(2,'0');
}
// Tooltip: full datetime with milliseconds.
function fmtTs(sec) {
var d = new Date(sec * 1000);
return fmtDate(d) + ' ' + fmtTime(d) + '.' +
d.getUTCMilliseconds().toString().padStart(3,'0');
}
// Tick label: show date only when it changes (or on the first tick).
function fmtTick(v, idx, ticks) {
var d = new Date(v * 1000);
var dateStr = fmtDate(d);
var timeStr = fmtTime(d);
if (idx === 0) { return timeStr + ' ' + dateStr; }
var prev = new Date(ticks[idx - 1].value * 1000);
if (fmtDate(prev) !== dateStr) { return timeStr + ' ' + dateStr; }
return timeStr;
}
// Plugin: draw dashed delay-span lines directly on the canvas so they are
// owned by their series dataset and toggle with it.
var delaySpanPlugin = {
id: 'delaySpan',
afterDatasetsDraw: function(chart) {
var ctx2 = chart.ctx;
chart.data.datasets.forEach(function(ds, dsi) {
if (!chart.isDatasetVisible(dsi)) return;
var meta = chart.getDatasetMeta(dsi);
ds.data.forEach(function(pt, pi) {
if (!pt || !pt.delayed) return;
var el = meta.data[pi];
if (!el) return;
var xScale = chart.scales.x;
var yScale = chart.scales.y;
var x1 = el.x;
var x2 = xScale.getPixelForValue(pt.sentAt);
var y = el.y;
ctx2.save();
ctx2.beginPath();
ctx2.setLineDash([5, 4]);
ctx2.strokeStyle = 'rgba(255,159,64,0.7)';
ctx2.lineWidth = 2;
ctx2.moveTo(x1, y);
ctx2.lineTo(x2, y);
ctx2.stroke();
// Arrow head at sent-at end.
var dir = x2 > x1 ? 1 : -1;
ctx2.setLineDash([]);
ctx2.beginPath();
ctx2.moveTo(x2, y);
ctx2.lineTo(x2 - dir * 8, y - 5);
ctx2.lineTo(x2 - dir * 8, y + 5);
ctx2.closePath();
ctx2.fillStyle = 'rgba(255,159,64,0.7)';
ctx2.fill();
ctx2.restore();
});
});
},
};
var sentEl = document.getElementById('sent-all');
if (sentEl && SENT.length > 0) {
var datasets = [];
SENT.forEach(function(s, si) {
var col = PALETTE[si % PALETTE.length];
var delayCol = 'rgba(255,159,64,0.9)';
var pts = s.points.filter(function(p){ return p !== null; });
if (pts.length === 0) return;
// Attach full metadata (delayed, sentAt) to each chart point so the
// plugin and tooltip can read it without separate datasets.
datasets.push({
type: 'scatter', label: s.name,
data: pts.map(function(p){
return {x: p.x, y: p.y, delayed: p.delayed, sentAt: p.sentAt};
}),
backgroundColor: pts.map(function(p){ return p.delayed ? delayCol : col; }),
pointStyle: pts.map(function(p){ return p.delayed ? 'triangle' : 'circle'; }),
pointRadius: pts.map(function(p){ return p.delayed ? 9 : 7; }),
});
});
new Chart(sentEl, {
type: 'scatter',
data: { datasets: datasets },
plugins: [delaySpanPlugin],
options: {
responsive: true, maintainAspectRatio: false,
plugins: {
legend: {
position: 'bottom',
onClick: function(e, legendItem, legend) {
var chart = legend.chart;
var idx = legendItem.datasetIndex;
var meta = chart.getDatasetMeta(idx);
var allOthersHidden = chart.data.datasets.every(function(_, i) {
return i === idx || !chart.isDatasetVisible(i);
});
if (allOthersHidden && !meta.hidden) {
// Already solo — restore all.
chart.data.datasets.forEach(function(_, i) { chart.show(i); });
} else {
// Solo this series.
chart.data.datasets.forEach(function(_, i) { chart.hide(i); });
chart.show(idx);
}
},
},
tooltip: { callbacks: { label: function(ctx){
var p = ctx.raw;
if (!p) return '';
var msg = 'value=' + p.y + ' ts=' + fmtTs(p.x);
if (p.delayed) {
var delaySec = Math.round((p.sentAt - p.x) * 10) / 10;
msg += ' \u2192 sent at ' + fmtTs(p.sentAt) + ' (+' + delaySec + 's)';
}
return msg;
}}},
},
scales: {
x: { type: 'linear',
title: { display: true, text: 'Unix timestamp (seconds UTC)' },
ticks: { callback: function(v, idx, ticks) { return fmtTick(v, idx, ticks); }, maxRotation: 35, minRotation: 35 } },
y: { title: { display: true, text: 'value' } },
},
},
});
}
RECV.forEach(function(s, i) {
var el = document.getElementById('recv-' + i);
if (!el) return;
var pts = s.points
.filter(function(p){ return p !== null && p !== undefined && typeof p.x === 'number'; })
.sort(function(a, b){ return a.x - b.x; });
new Chart(el, {
type: 'scatter',
data: { datasets: [{
label: s.name, data: pts,
backgroundColor: 'rgba(75,192,75,0.85)',
pointRadius: 6, pointStyle: 'circle',
}]},
options: {
responsive: true, maintainAspectRatio: false,
plugins: {
legend: { display: false },
tooltip: { callbacks: { label: function(ctx){
var p = ctx.raw;
if (p === null || p === undefined) return '';
return 'value=' + p.y + ' ts=' + fmtTs(p.x);
}}},
},
scales: {
x: { type: 'linear',
title: { display: true, text: 'Unix timestamp (seconds UTC)' },
ticks: { callback: function(v, idx, ticks) { return fmtTick(v, idx, ticks); }, maxRotation: 35, minRotation: 35 } },
y: { title: { display: true, text: 'value' } },
},
},
});
});
})();
</script>
</body>
</html>`

View File

@@ -10,7 +10,7 @@ import (
func Compress(wr WriteRequest) []byte {
data, err := wr.Marshal()
if err != nil {
panic(fmt.Errorf("BUG: cannot compress WriteRequest: %w", err))
panic(fmt.Errorf("BUG: cannot compress WriteRequest: %s", err))
}
return snappy.Encode(nil, data)
}

View File

@@ -61,15 +61,15 @@ func parseInputSeries(input []series, interval *promutil.Duration, startStamp ti
for _, data := range input {
expr, err := metricsql.Parse(data.Series)
if err != nil {
return res, fmt.Errorf("failed to parse series %s: %w", data.Series, err)
return res, fmt.Errorf("failed to parse series %s: %v", data.Series, err)
}
promvals, err := parseInputValue(data.Values, true)
if err != nil {
return res, fmt.Errorf("failed to parse input series value %s: %w", data.Values, err)
return res, fmt.Errorf("failed to parse input series value %s: %v", data.Values, err)
}
metricExpr, ok := expr.(*metricsql.MetricExpr)
if !ok || len(metricExpr.LabelFilterss) != 1 {
return res, fmt.Errorf("got invalid input series %s: %w", data.Series, err)
return res, fmt.Errorf("got invalid input series %s: %v", data.Series, err)
}
samples := make([]testutil.Sample, 0, len(promvals))
ts := startStamp

View File

@@ -53,13 +53,13 @@ Outer:
if s.Labels != "" {
metricsqlExpr, err := metricsql.Parse(s.Labels)
if err != nil {
checkErrs = append(checkErrs, fmt.Errorf("\n expr: %q, time: %s, err: %w", mt.Expr,
checkErrs = append(checkErrs, fmt.Errorf("\n expr: %q, time: %s, err: %v", mt.Expr,
mt.EvalTime.Duration().String(), fmt.Errorf("failed to parse labels %q: %w", s.Labels, err)))
continue Outer
}
metricsqlMetricExpr, ok := metricsqlExpr.(*metricsql.MetricExpr)
if !ok || len(metricsqlMetricExpr.LabelFilterss) > 1 {
checkErrs = append(checkErrs, fmt.Errorf("\n expr: %q, time: %s, err: %w", mt.Expr,
checkErrs = append(checkErrs, fmt.Errorf("\n expr: %q, time: %s, err: %v", mt.Expr,
mt.EvalTime.Duration().String(), fmt.Errorf("got invalid exp_samples: %q", s.Labels)))
continue Outer
}

View File

@@ -329,11 +329,11 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
q, err := datasource.Init(nil)
if err != nil {
return []error{fmt.Errorf("failed to init datasource: %w", err)}
return []error{fmt.Errorf("failed to init datasource: %v", err)}
}
rw, err := remotewrite.NewDebugClient()
if err != nil {
return []error{fmt.Errorf("failed to init wr: %w", err)}
return []error{fmt.Errorf("failed to init wr: %v", err)}
}
alertEvalTimesMap := map[time.Duration]struct{}{}

View File

@@ -173,9 +173,9 @@ func (r *Rule) String() string {
if r.Alert != "" {
ruleType = "alerting"
}
var b strings.Builder
fmt.Fprintf(&b, "%s rule %q", ruleType, r.Name())
fmt.Fprintf(&b, "; expr: %q", r.Expr)
b := strings.Builder{}
b.WriteString(fmt.Sprintf("%s rule %q", ruleType, r.Name()))
b.WriteString(fmt.Sprintf("; expr: %q", r.Expr))
kv := sortMap(r.Labels)
for i := range kv {

View File

@@ -89,7 +89,7 @@ func (pi *promInstant) Unmarshal(b []byte) error {
labels.Visit(func(key []byte, v *fastjson.Value) {
lv, errLocal := v.StringBytes()
if errLocal != nil {
err = fmt.Errorf("error when parsing label value %q: %w", v, errLocal)
err = fmt.Errorf("error when parsing label value %q: %s", v, errLocal)
return
}
r.Labels = append(r.Labels, prompb.Label{
@@ -112,7 +112,7 @@ func (pi *promInstant) Unmarshal(b []byte) error {
r.Timestamps = []int64{sample[0].GetInt64()}
val, err := sample[1].StringBytes()
if err != nil {
return fmt.Errorf("error when parsing `value` object %q: %w", sample[1], err)
return fmt.Errorf("error when parsing `value` object %q: %s", sample[1], err)
}
f, err := strconv.ParseFloat(bytesutil.ToUnsafeString(val), 64)
if err != nil {

View File

@@ -601,7 +601,7 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
func (ar *AlertingRule) expandLabelTemplates(m datasource.Metric, qFn templates.QueryFn) (*labelSet, error) {
ls, err := ar.toLabels(m, qFn)
if err != nil {
return ls, fmt.Errorf("failed to expand label templates: %w", err)
return ls, fmt.Errorf("failed to expand label templates: %s", err)
}
return ls, nil
}
@@ -620,7 +620,7 @@ func (ar *AlertingRule) expandAnnotationTemplates(m datasource.Metric, qFn templ
}
as, err := notifier.ExecTemplate(qFn, ar.Annotations, tplData)
if err != nil {
return as, fmt.Errorf("failed to expand annotation templates: %w", err)
return as, fmt.Errorf("failed to expand annotation templates: %s", err)
}
return as, nil
}

View File

@@ -77,7 +77,7 @@ var (
func marshalJson(v any, kind string) ([]byte, *httpserver.ErrorWithStatusCode) {
data, err := json.Marshal(v)
if err != nil {
return nil, errResponse(fmt.Errorf("failed to marshal %s: %w", kind, err), http.StatusInternalServerError)
return nil, errResponse(fmt.Errorf("failed to marshal %s: %s", kind, err), http.StatusInternalServerError)
}
return data, nil
}

View File

@@ -1639,7 +1639,7 @@ func (w *fakeResponseWriter) WriteHeader(statusCode int) {
"X-Content-Type-Options": true,
})
if err != nil {
panic(fmt.Errorf("cannot marshal headers: %w", err))
panic(fmt.Errorf("cannot marshal headers: %s", err))
}
}

View File

@@ -161,7 +161,7 @@ func fetchAndParseJWKs(ctx context.Context, jwksURI string) (*jwt.VerifierPool,
vp, err := jwt.ParseJWKs(b)
if err != nil {
return nil, fmt.Errorf("failed to parse jwks keys from %q: %w", jwksURI, err)
return nil, fmt.Errorf("failed to parse jwks keys from %q: %v", jwksURI, err)
}
return vp, nil
@@ -188,7 +188,7 @@ func getOpenIDConfiguration(ctx context.Context, issuer string) (openidConfig, e
var cfg openidConfig
if err := json.NewDecoder(resp.Body).Decode(&cfg); err != nil {
return openidConfig{}, fmt.Errorf("failed to decode openid config from %q: %w", configURL, err)
return openidConfig{}, fmt.Errorf("failed to decode openid config from %q: %s", configURL, err)
}
return cfg, nil

View File

@@ -43,7 +43,7 @@ func newInfluxProcessor(ic *influx.Client, im *vm.Importer, cc int, separator st
func (ip *influxProcessor) run(ctx context.Context) error {
series, err := ip.ic.Explore()
if err != nil {
return fmt.Errorf("explore query failed: %w", err)
return fmt.Errorf("explore query failed: %s", err)
}
if len(series) < 1 {
return fmt.Errorf("found no timeseries to import")
@@ -71,7 +71,7 @@ func (ip *influxProcessor) run(ctx context.Context) error {
for s := range seriesCh {
if err := ip.do(s); err != nil {
influxErrorsTotal.Inc()
errCh <- fmt.Errorf("request failed for %q.%q: %w", s.Measurement, s.Field, err)
errCh <- fmt.Errorf("request failed for %q.%q: %s", s.Measurement, s.Field, err)
return
}
influxSeriesProcessed.Inc()
@@ -84,10 +84,10 @@ func (ip *influxProcessor) run(ctx context.Context) error {
for _, s := range series {
select {
case infErr := <-errCh:
return fmt.Errorf("influx error: %w", infErr)
return fmt.Errorf("influx error: %s", infErr)
case vmErr := <-ip.im.Errors():
influxErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, ip.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, ip.isVerbose))
case seriesCh <- s:
}
}
@@ -100,11 +100,11 @@ func (ip *influxProcessor) run(ctx context.Context) error {
for vmErr := range ip.im.Errors() {
if vmErr.Err != nil {
influxErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, ip.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, ip.isVerbose))
}
}
for err := range errCh {
return fmt.Errorf("import process failed: %w", err)
return fmt.Errorf("import process failed: %s", err)
}
log.Println("Import finished!")
@@ -119,7 +119,7 @@ const valueField = "value"
func (ip *influxProcessor) do(s *influx.Series) error {
cr, err := ip.ic.FetchDataPoints(s)
if err != nil {
return fmt.Errorf("failed to fetch datapoints: %w", err)
return fmt.Errorf("failed to fetch datapoints: %s", err)
}
defer func() {
_ = cr.Close()

View File

@@ -96,10 +96,10 @@ func NewClient(cfg Config) (*Client, error) {
}
hc, err := influx.NewHTTPClient(c)
if err != nil {
return nil, fmt.Errorf("failed to establish conn: %w", err)
return nil, fmt.Errorf("failed to establish conn: %s", err)
}
if _, _, err := hc.Ping(time.Second); err != nil {
return nil, fmt.Errorf("ping failed: %w", err)
return nil, fmt.Errorf("ping failed: %s", err)
}
chunkSize := cfg.ChunkSize
@@ -155,7 +155,7 @@ func (c *Client) Explore() ([]*Series, error) {
// {"measurement1": ["value1", "value2"]}
mFields, err := c.fieldsByMeasurement()
if err != nil {
return nil, fmt.Errorf("failed to get field keys: %w", err)
return nil, fmt.Errorf("failed to get field keys: %s", err)
}
if len(mFields) < 1 {
@@ -165,12 +165,12 @@ func (c *Client) Explore() ([]*Series, error) {
// {"measurement1": {"tag1", "tag2"}}
measurementTags, err := c.getMeasurementTags()
if err != nil {
return nil, fmt.Errorf("failed to get tags of measurements: %w", err)
return nil, fmt.Errorf("failed to get tags of measurements: %s", err)
}
series, err := c.getSeries()
if err != nil {
return nil, fmt.Errorf("failed to get series: %w", err)
return nil, fmt.Errorf("failed to get series: %s", err)
}
var iSeries []*Series
@@ -237,7 +237,7 @@ func (cr *ChunkedResponse) Next() ([]int64, []float64, error) {
return nil, nil, err
}
if resp.Error() != nil {
return nil, nil, fmt.Errorf("response error for %s: %w", cr.iq.Command, resp.Error())
return nil, nil, fmt.Errorf("response error for %s: %s", cr.iq.Command, resp.Error())
}
if len(resp.Results) != 1 {
return nil, nil, fmt.Errorf("unexpected number of results in response: %d", len(resp.Results))
@@ -265,7 +265,8 @@ func (cr *ChunkedResponse) Next() ([]int64, []float64, error) {
for i, fv := range fieldValues {
v, err := toFloat64(fv)
if err != nil {
return nil, nil, fmt.Errorf("failed to convert value %q.%v to float64: %w", cr.field, v, err)
return nil, nil, fmt.Errorf("failed to convert value %q.%v to float64: %s",
cr.field, v, err)
}
values[i] = v
}
@@ -293,7 +294,7 @@ func (c *Client) FetchDataPoints(s *Series) (*ChunkedResponse, error) {
}
cr, err := c.QueryAsChunk(iq)
if err != nil {
return nil, fmt.Errorf("query %q err: %w", iq.Command, err)
return nil, fmt.Errorf("query %q err: %s", iq.Command, err)
}
return &ChunkedResponse{cr, iq, s.Field}, nil
}
@@ -307,7 +308,7 @@ func (c *Client) fieldsByMeasurement() (map[string][]string, error) {
log.Printf("fetching fields: %s", stringify(q))
qValues, err := c.do(q)
if err != nil {
return nil, fmt.Errorf("error while executing query %q: %w", q.Command, err)
return nil, fmt.Errorf("error while executing query %q: %s", q.Command, err)
}
var total int
@@ -351,7 +352,7 @@ func (c *Client) getSeries() ([]*Series, error) {
log.Printf("fetching series: %s", stringify(q))
cr, err := c.QueryAsChunk(q)
if err != nil {
return nil, fmt.Errorf("error while executing query %q: %w", q.Command, err)
return nil, fmt.Errorf("error while executing query %q: %s", q.Command, err)
}
const key = "key"
@@ -365,7 +366,7 @@ func (c *Client) getSeries() ([]*Series, error) {
return nil, err
}
if resp.Error() != nil {
return nil, fmt.Errorf("response error for query %q: %w", q.Command, resp.Error())
return nil, fmt.Errorf("response error for query %q: %s", q.Command, resp.Error())
}
qValues, err := parseResult(resp.Results[0])
if err != nil {
@@ -416,7 +417,7 @@ func (c *Client) getMeasurementTags() (map[string]map[string]struct{}, error) {
log.Printf("fetching tag keys: %s", stringify(q))
cr, err := c.QueryAsChunk(q)
if err != nil {
return nil, fmt.Errorf("error while executing query %q: %w", q.Command, err)
return nil, fmt.Errorf("error while executing query %q: %s", q.Command, err)
}
const tagKey = "tagKey"
@@ -431,7 +432,7 @@ func (c *Client) getMeasurementTags() (map[string]map[string]struct{}, error) {
return nil, err
}
if resp.Error() != nil {
return nil, fmt.Errorf("response error for query %q: %w", q.Command, resp.Error())
return nil, fmt.Errorf("response error for query %q: %s", q.Command, resp.Error())
}
qValues, err := parseResult(resp.Results[0])
if err != nil {
@@ -454,10 +455,10 @@ func (c *Client) getMeasurementTags() (map[string]map[string]struct{}, error) {
func (c *Client) do(q influx.Query) ([]queryValues, error) {
res, err := c.Query(q)
if err != nil {
return nil, fmt.Errorf("query error: %w", err)
return nil, fmt.Errorf("query error: %s", err)
}
if res.Error() != nil {
return nil, fmt.Errorf("response error: %w", res.Error())
return nil, fmt.Errorf("response error: %s", res.Error())
}
if len(res.Results) < 1 {
return nil, fmt.Errorf("query returned 0 results")

View File

@@ -71,7 +71,7 @@ func toFloat64(v any) (float64, error) {
func parseDate(dateStr string) (int64, error) {
startTime, err := time.Parse(time.RFC3339, dateStr)
if err != nil {
return 0, fmt.Errorf("cannot parse %q: %w", dateStr, err)
return 0, fmt.Errorf("cannot parse %q: %s", dateStr, err)
}
return startTime.UnixNano() / 1e6, nil
}
@@ -92,7 +92,7 @@ func (s *Series) unmarshal(v string) error {
var err error
s.LabelPairs, err = unmarshalTags(v[n+1:], noEscapeChars)
if err != nil {
return fmt.Errorf("failed to unmarhsal tags: %w", err)
return fmt.Errorf("failed to unmarhsal tags: %s", err)
}
return nil
}

View File

@@ -88,7 +88,7 @@ func main() {
tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_opentsdb")
if err != nil {
return fmt.Errorf("failed to create transport for -%s=%q: %w", otsdbAddr, addr, err)
return fmt.Errorf("failed to create transport for -%s=%q: %s", otsdbAddr, addr, err)
}
oCfg := opentsdb.Config{
Addr: addr,
@@ -103,17 +103,17 @@ func main() {
}
otsdbClient, err := opentsdb.NewClient(oCfg)
if err != nil {
return fmt.Errorf("failed to create opentsdb client: %w", err)
return fmt.Errorf("failed to create opentsdb client: %s", err)
}
vmCfg, err := initConfigVM(c)
if err != nil {
return fmt.Errorf("failed to init VM configuration: %w", err)
return fmt.Errorf("failed to init VM configuration: %s", err)
}
importer, err := vm.NewImporter(ctx, vmCfg)
if err != nil {
return fmt.Errorf("failed to create VM importer: %w", err)
return fmt.Errorf("failed to create VM importer: %s", err)
}
otsdbProcessor := newOtsdbProcessor(otsdbClient, importer, c.Int(otsdbConcurrency), c.Bool(globalVerbose))
@@ -137,7 +137,7 @@ func main() {
tc, err := promauth.NewTLSConfig(certFile, keyFile, caFile, serverName, insecureSkipVerify)
if err != nil {
return fmt.Errorf("failed to create TLS Config: %w", err)
return fmt.Errorf("failed to create TLS Config: %s", err)
}
iCfg := influx.Config{
@@ -157,17 +157,17 @@ func main() {
influxClient, err := influx.NewClient(iCfg)
if err != nil {
return fmt.Errorf("failed to create influx client: %w", err)
return fmt.Errorf("failed to create influx client: %s", err)
}
vmCfg, err := initConfigVM(c)
if err != nil {
return fmt.Errorf("failed to init VM configuration: %w", err)
return fmt.Errorf("failed to init VM configuration: %s", err)
}
importer, err = vm.NewImporter(ctx, vmCfg)
if err != nil {
return fmt.Errorf("failed to create VM importer: %w", err)
return fmt.Errorf("failed to create VM importer: %s", err)
}
processor := newInfluxProcessor(
@@ -203,7 +203,7 @@ func main() {
tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_remoteread")
if err != nil {
return fmt.Errorf("failed to create transport for -%s=%q: %w", remoteReadSrcAddr, addr, err)
return fmt.Errorf("failed to create transport for -%s=%q: %s", remoteReadSrcAddr, addr, err)
}
// Backwards compatible default values if none provided by user
@@ -227,17 +227,17 @@ func main() {
DisablePathAppend: c.Bool(remoteReadDisablePathAppend),
})
if err != nil {
return fmt.Errorf("error create remote read client: %w", err)
return fmt.Errorf("error create remote read client: %s", err)
}
vmCfg, err := initConfigVM(c)
if err != nil {
return fmt.Errorf("failed to init VM configuration: %w", err)
return fmt.Errorf("failed to init VM configuration: %s", err)
}
importer, err := vm.NewImporter(ctx, vmCfg)
if err != nil {
return fmt.Errorf("failed to create VM importer: %w", err)
return fmt.Errorf("failed to create VM importer: %s", err)
}
rmp := remoteReadProcessor{
@@ -265,12 +265,12 @@ func main() {
vmCfg, err := initConfigVM(c)
if err != nil {
return fmt.Errorf("failed to init VM configuration: %w", err)
return fmt.Errorf("failed to init VM configuration: %s", err)
}
importer, err = vm.NewImporter(ctx, vmCfg)
if err != nil {
return fmt.Errorf("failed to create VM importer: %w", err)
return fmt.Errorf("failed to create VM importer: %s", err)
}
promCfg := prometheus.Config{
@@ -285,7 +285,7 @@ func main() {
}
cl, err := prometheus.NewClient(promCfg)
if err != nil {
return fmt.Errorf("failed to create prometheus client: %w", err)
return fmt.Errorf("failed to create prometheus client: %s", err)
}
pp := prometheusProcessor{
@@ -307,12 +307,12 @@ func main() {
vmCfg, err := initConfigVM(c)
if err != nil {
return fmt.Errorf("failed to init VM configuration: %w", err)
return fmt.Errorf("failed to init VM configuration: %s", err)
}
importer, err = vm.NewImporter(ctx, vmCfg)
if err != nil {
return fmt.Errorf("failed to create VM importer: %w", err)
return fmt.Errorf("failed to create VM importer: %s", err)
}
mCfg := mimir.Config{
@@ -335,7 +335,7 @@ func main() {
}
cl, err := mimir.NewClient(ctx, mCfg)
if err != nil {
return fmt.Errorf("failed to create mimir client: %w", err)
return fmt.Errorf("failed to create mimir client: %s", err)
}
pp := prometheusProcessor{
@@ -356,12 +356,12 @@ func main() {
fmt.Println("Thanos import mode")
vmCfg, err := initConfigVM(c)
if err != nil {
return fmt.Errorf("failed to init VM configuration: %w", err)
return fmt.Errorf("failed to init VM configuration: %s", err)
}
importer, err = vm.NewImporter(ctx, vmCfg)
if err != nil {
return fmt.Errorf("failed to create VM importer: %w", err)
return fmt.Errorf("failed to create VM importer: %s", err)
}
thanosCfg := thanos.Config{
Snapshot: c.String(thanosSnapshot),
@@ -374,7 +374,7 @@ func main() {
}
cl, err := thanos.NewClient(thanosCfg)
if err != nil {
return fmt.Errorf("failed to create thanos client: %w", err)
return fmt.Errorf("failed to create thanos client: %s", err)
}
var aggrTypes []thanos.AggrType
@@ -382,7 +382,7 @@ func main() {
for _, typeStr := range aggrTypesStr {
aggrType, err := thanos.ParseAggrType(typeStr)
if err != nil {
return fmt.Errorf("failed to parse aggregate type %q: %w", typeStr, err)
return fmt.Errorf("failed to parse aggregate type %q: %s", typeStr, err)
}
aggrTypes = append(aggrTypes, aggrType)
}
@@ -415,7 +415,7 @@ func main() {
bfMinDuration := c.Duration(vmNativeBackoffMinDuration)
bf, err := backoff.New(bfRetries, bfFactor, bfMinDuration)
if err != nil {
return fmt.Errorf("failed to create backoff object: %w", err)
return fmt.Errorf("failed to create backoff object: %s", err)
}
disableKeepAlive := c.Bool(vmNativeDisableHTTPKeepAlive)
@@ -439,7 +439,7 @@ func main() {
srcTC, err := promauth.NewTLSConfig(srcCertFile, srcKeyFile, srcCAFile, srcServerName, srcInsecureSkipVerify)
if err != nil {
return fmt.Errorf("failed to create TLS Config: %w", err)
return fmt.Errorf("failed to create TLS Config: %s", err)
}
trSrc := httputil.NewTransport(false, "vmctl_src")
@@ -469,7 +469,7 @@ func main() {
dstTC, err := promauth.NewTLSConfig(dstCertFile, dstKeyFile, dstCAFile, dstServerName, dstInsecureSkipVerify)
if err != nil {
return fmt.Errorf("failed to create TLS Config: %w", err)
return fmt.Errorf("failed to create TLS Config: %s", err)
}
trDst := httputil.NewTransport(false, "vmctl_dst")
@@ -534,7 +534,7 @@ func main() {
log.Printf("verifying block at path=%q", blockPath)
f, err := os.OpenFile(blockPath, os.O_RDONLY, 0600)
if err != nil {
return cli.Exit(fmt.Errorf("cannot open exported block at path=%q: %w", blockPath, err), 1)
return cli.Exit(fmt.Errorf("cannot open exported block at path=%q err=%w", blockPath, err), 1)
}
defer f.Close()
var blocksCount atomic.Uint64
@@ -542,7 +542,7 @@ func main() {
blocksCount.Add(1)
return nil
}); err != nil {
return cli.Exit(fmt.Errorf("cannot parse block at path=%q, blocksCount=%d: %w", blockPath, blocksCount.Load(), err), 1)
return cli.Exit(fmt.Errorf("cannot parse block at path=%q, blocksCount=%d, err=%w", blockPath, blocksCount.Load(), err), 1)
}
log.Printf("successfully verified block at path=%q, blockCount=%d", blockPath, blocksCount.Load())
return nil
@@ -585,7 +585,7 @@ func initConfigVM(c *cli.Context) (vm.Config, error) {
tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_client")
if err != nil {
return vm.Config{}, fmt.Errorf("failed to create transport for -%s=%q: %w", vmAddr, addr, err)
return vm.Config{}, fmt.Errorf("failed to create transport for -%s=%q: %s", vmAddr, addr, err)
}
bfRetries := c.Int(vmBackoffRetries)
@@ -593,7 +593,7 @@ func initConfigVM(c *cli.Context) (vm.Config, error) {
bfMinDuration := c.Duration(vmBackoffMinDuration)
bf, err := backoff.New(bfRetries, bfFactor, bfMinDuration)
if err != nil {
return vm.Config{}, fmt.Errorf("failed to create backoff object: %w", err)
return vm.Config{}, fmt.Errorf("failed to create backoff object: %s", err)
}
return vm.Config{

View File

@@ -54,7 +54,7 @@ func (lbr *lazyBlockReader) initialize() error {
// fetching block and parse it and store it in lbr.reader
temp, err := lbr.mkTempDir()
if err != nil {
return fmt.Errorf("failed to create temp dir: %w", err)
return fmt.Errorf("failed to create temp dir: %s", err)
}
lbr.tempDirPath = temp
@@ -85,7 +85,7 @@ func (lbr *lazyBlockReader) initialize() error {
return fmt.Errorf("failed to fetch chunk file: %q: %w", chunkName, err)
}
if err := lbr.writeFile(temp, blockChunkPath, chunk); err != nil {
return fmt.Errorf("failed to write chunk file: %q: %w", chunkName, err)
return fmt.Errorf("failed to write chunk file: %q: %s", chunkName, err)
}
}
@@ -135,7 +135,7 @@ func (lbr *lazyBlockReader) Meta() tsdb.BlockMeta {
// Size returns the number of bytes that the block takes up on disk.
func (lbr *lazyBlockReader) Size() int64 {
if err := lbr.initialize(); err != nil {
lbr.err = fmt.Errorf("error get Size of the block: %w, return zero size", err)
lbr.err = fmt.Errorf("error get Size of the block: %s, return zero size", err)
return 0
}
return lbr.reader.Size()
@@ -167,11 +167,11 @@ func (lbr *lazyBlockReader) Close() error {
func (lbr *lazyBlockReader) mkTempDir() (string, error) {
temp, err := os.MkdirTemp("", lbr.ID.String())
if err != nil {
return "", fmt.Errorf("failed to create temp dir: %w", err)
return "", fmt.Errorf("failed to create temp dir: %s", err)
}
err = os.Mkdir(filepath.Join(temp, "chunks"), os.ModePerm)
if err != nil {
return "", fmt.Errorf("failed to create temp dir: %w", err)
return "", fmt.Errorf("failed to create temp dir: %s", err)
}
return temp, nil
}

View File

@@ -133,11 +133,11 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
c.RemoteFS = rfs
timeMin, err := utils.ParseTime(cfg.Filter.TimeMin)
if err != nil {
return nil, fmt.Errorf("failed to parse min time in filter: %w", err)
return nil, fmt.Errorf("failed to parse min time in filter: %s", err)
}
timeMax, err := utils.ParseTime(cfg.Filter.TimeMax)
if err != nil {
return nil, fmt.Errorf("failed to parse max time in filter: %w", err)
return nil, fmt.Errorf("failed to parse max time in filter: %s", err)
}
c.filter = filter{
min: timeMin.UnixMilli(),
@@ -156,7 +156,7 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
indexFile, err := c.fetchIndexFile()
if err != nil {
return nil, fmt.Errorf("failed to fetch index file: %w", err)
return nil, fmt.Errorf("failed to fetch index file: %s", err)
}
var blocksToImport []tsdb.BlockReader
@@ -172,7 +172,7 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
lazyBlockReader, err := newLazyBlockReader(block, c.RemoteFS)
if err != nil {
return nil, fmt.Errorf("failed to create lazy block reader: %w", err)
return nil, fmt.Errorf("failed to create lazy block reader: %s", err)
}
blocksToImport = append(blocksToImport, lazyBlockReader)
}
@@ -185,7 +185,7 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
func (c *Client) Read(ctx context.Context, block tsdb.BlockReader) (*prometheus.CloseableSeriesSet, error) {
meta := block.Meta()
if b, ok := block.(*lazyBlockReader); ok && b.Err() != nil {
return nil, fmt.Errorf("failed to read block: %w", b.Err())
return nil, fmt.Errorf("failed to read block: %s", b.Err())
}
if meta.ULID.String() == "" {
@@ -218,20 +218,20 @@ func (c *Client) fetchIndexFile() (*Index, error) {
file, err := c.ReadFile(bucketIndexCompressedFilename)
if err != nil {
return nil, fmt.Errorf("failed to read bucket index: %w", err)
return nil, fmt.Errorf("failed to read bucket index: %s", err)
}
r := bytes.NewReader(file)
// Read all the content.
gzipReader, err := gzip.NewReader(r)
if err != nil {
return nil, fmt.Errorf("failed to create gzip reader: %w", err)
return nil, fmt.Errorf("failed to create gzip reader: %s", err)
}
var indexFile Index
err = json.NewDecoder(gzipReader).Decode(&indexFile)
if err != nil {
return nil, fmt.Errorf("failed to decode bucket index: %w", err)
return nil, fmt.Errorf("failed to decode bucket index: %s", err)
}
return &indexFile, nil

View File

@@ -47,7 +47,7 @@ func (c *Client) Explore(ctx context.Context, f Filter, tenantID string, start,
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
exploreRequestsErrorsTotal.Inc()
return nil, fmt.Errorf("cannot create request to %q: %w", url, err)
return nil, fmt.Errorf("cannot create request to %q: %s", url, err)
}
params := req.URL.Query()
@@ -60,14 +60,14 @@ func (c *Client) Explore(ctx context.Context, f Filter, tenantID string, start,
if err != nil {
exploreRequestsErrorsTotal.Inc()
exploreDuration.UpdateDuration(startTime)
return nil, fmt.Errorf("series request failed: %w", err)
return nil, fmt.Errorf("series request failed: %s", err)
}
var response Response
if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
exploreRequestsErrorsTotal.Inc()
exploreDuration.UpdateDuration(startTime)
return nil, fmt.Errorf("cannot decode series response: %w", err)
return nil, fmt.Errorf("cannot decode series response: %s", err)
}
exploreDuration.UpdateDuration(startTime)
return response.MetricNames, resp.Body.Close()
@@ -80,19 +80,19 @@ func (c *Client) ImportPipe(ctx context.Context, dstURL string, pr *io.PipeReade
req, err := http.NewRequestWithContext(ctx, http.MethodPost, dstURL, pr)
if err != nil {
importRequestsErrorsTotal.Inc()
return fmt.Errorf("cannot create import request to %q: %w", c.Addr, err)
return fmt.Errorf("cannot create import request to %q: %s", c.Addr, err)
}
importResp, err := c.do(req, http.StatusNoContent)
if err != nil {
importRequestsErrorsTotal.Inc()
importDuration.UpdateDuration(startTime)
return fmt.Errorf("import request failed: %w", err)
return fmt.Errorf("import request failed: %s", err)
}
if err := importResp.Body.Close(); err != nil {
importRequestsErrorsTotal.Inc()
importDuration.UpdateDuration(startTime)
return fmt.Errorf("cannot close import response body: %w", err)
return fmt.Errorf("cannot close import response body: %s", err)
}
importDuration.UpdateDuration(startTime)
return nil
@@ -105,7 +105,7 @@ func (c *Client) ExportPipe(ctx context.Context, url string, f Filter) (io.ReadC
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
exportRequestsErrorsTotal.Inc()
return nil, fmt.Errorf("cannot create request to %q: %w", c.Addr, err)
return nil, fmt.Errorf("cannot create request to %q: %s", c.Addr, err)
}
params := req.URL.Query()
@@ -136,7 +136,7 @@ func (c *Client) GetSourceTenants(ctx context.Context, f Filter) ([]string, erro
u := fmt.Sprintf("%s/%s", c.Addr, nativeTenantsAddr)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return nil, fmt.Errorf("cannot create request to %q: %w", u, err)
return nil, fmt.Errorf("cannot create request to %q: %s", u, err)
}
params := req.URL.Query()
@@ -150,18 +150,18 @@ func (c *Client) GetSourceTenants(ctx context.Context, f Filter) ([]string, erro
resp, err := c.do(req, http.StatusOK)
if err != nil {
return nil, fmt.Errorf("tenants request failed: %w", err)
return nil, fmt.Errorf("tenants request failed: %s", err)
}
var r struct {
Tenants []string `json:"data"`
}
if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
return nil, fmt.Errorf("cannot decode tenants response: %w", err)
return nil, fmt.Errorf("cannot decode tenants response: %s", err)
}
if err := resp.Body.Close(); err != nil {
return nil, fmt.Errorf("cannot close tenants response body: %w", err)
return nil, fmt.Errorf("cannot close tenants response body: %s", err)
}
return r.Tenants, nil
@@ -180,7 +180,7 @@ func (c *Client) do(req *http.Request, expSC int) (*http.Response, error) {
if resp.StatusCode != expSC {
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response body for status code %d: %w", resp.StatusCode, err)
return nil, fmt.Errorf("failed to read response body for status code %d: %s", resp.StatusCode, err)
}
return nil, fmt.Errorf("unexpected response code %d: %s", resp.StatusCode, string(body))
}

View File

@@ -47,7 +47,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
q := fmt.Sprintf("%s/api/suggest?type=metrics&q=%s&max=%d", op.oc.Addr, filter, op.oc.Limit)
m, err := op.oc.FindMetrics(q)
if err != nil {
return fmt.Errorf("metric discovery failed for %q: %w", q, err)
return fmt.Errorf("metric discovery failed for %q: %s", q, err)
}
metrics = append(metrics, m...)
}
@@ -76,7 +76,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
log.Printf("Starting work on %s", metric)
serieslist, err := op.oc.FindSeries(metric)
if err != nil {
return fmt.Errorf("couldn't retrieve series list for %s: %w", metric, err)
return fmt.Errorf("couldn't retrieve series list for %s : %s", metric, err)
}
/*
Create channels for collecting/processing series and errors
@@ -95,7 +95,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
for s := range seriesCh {
if err := op.do(s); err != nil {
otsdbErrorsTotal.Inc()
errCh <- fmt.Errorf("couldn't retrieve series for %s: %w", metric, err)
errCh <- fmt.Errorf("couldn't retrieve series for %s : %s", metric, err)
return
}
otsdbSeriesProcessed.Inc()
@@ -112,7 +112,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
// check for any lingering errors on the query side
for otsdbErr := range errCh {
if runErr == nil {
runErr = fmt.Errorf("import process failed:\n%w", otsdbErr)
runErr = fmt.Errorf("import process failed: \n%s", otsdbErr)
}
}
bar.Finish()
@@ -125,7 +125,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
for vmErr := range op.im.Errors() {
if vmErr.Err != nil {
otsdbErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, op.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, op.isVerbose))
}
}
log.Println("Import finished!")
@@ -141,12 +141,12 @@ func (op *otsdbProcessor) sendQueries(ctx context.Context, serieslist []opentsdb
for _, tr := range rt.QueryRanges {
select {
case <-ctx.Done():
return fmt.Errorf("context canceled: %w", ctx.Err())
return fmt.Errorf("context canceled: %s", ctx.Err())
case otsdbErr := <-errCh:
otsdbErrorsTotal.Inc()
return fmt.Errorf("opentsdb error: %w", otsdbErr)
return fmt.Errorf("opentsdb error: %s", otsdbErr)
case vmErr := <-op.im.Errors():
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, op.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, op.isVerbose))
case seriesCh <- queryObj{
Tr: tr, StartTime: startTime,
Series: series, Rt: opentsdb.RetentionMeta{
@@ -166,7 +166,7 @@ func (op *otsdbProcessor) do(s queryObj) error {
end := s.StartTime - s.Tr.End
data, err := op.oc.GetData(s.Series, s.Rt, start, end, op.oc.MsecsTime)
if err != nil {
return fmt.Errorf("failed to collect data for %v in %v:%v :: %w", s.Series, s.Rt, s.Tr, err)
return fmt.Errorf("failed to collect data for %v in %v:%v :: %v", s.Series, s.Rt, s.Tr, err)
}
if len(data.Timestamps) < 1 || len(data.Values) < 1 {
log.Printf("no data found for %v in %v:%v...skipping", s.Series, s.Rt, s.Tr)

View File

@@ -106,7 +106,7 @@ func (c Client) FindMetrics(q string) ([]string, error) {
resp, err := c.c.Get(q)
if err != nil {
return nil, fmt.Errorf("failed to send GET request to %q: %w", q, err)
return nil, fmt.Errorf("failed to send GET request to %q: %s", q, err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != 200 {
@@ -114,12 +114,12 @@ func (c Client) FindMetrics(q string) ([]string, error) {
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("could not retrieve metric data from %q: %w", q, err)
return nil, fmt.Errorf("could not retrieve metric data from %q: %s", q, err)
}
var metriclist []string
err = json.Unmarshal(body, &metriclist)
if err != nil {
return nil, fmt.Errorf("failed to read response from %q: %w", q, err)
return nil, fmt.Errorf("failed to read response from %q: %s", q, err)
}
return metriclist, nil
}
@@ -130,7 +130,7 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
q := fmt.Sprintf("%s/api/search/lookup?m=%s&limit=%d", c.Addr, metric, c.Limit)
resp, err := c.c.Get(q)
if err != nil {
return nil, fmt.Errorf("failed to send GET request to %q: %w", q, err)
return nil, fmt.Errorf("failed to send GET request to %q: %s", q, err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != 200 {
@@ -138,12 +138,12 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("could not retrieve series data from %q: %w", q, err)
return nil, fmt.Errorf("could not retrieve series data from %q: %s", q, err)
}
var results MetaResults
err = json.Unmarshal(body, &results)
if err != nil {
return nil, fmt.Errorf("failed to read response from %q: %w", q, err)
return nil, fmt.Errorf("failed to read response from %q: %s", q, err)
}
return results.Results, nil
}
@@ -183,7 +183,7 @@ func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64, m
q := fmt.Sprintf("%s/api/query?%s", c.Addr, queryStr)
resp, err := c.c.Get(q)
if err != nil {
return Metric{}, fmt.Errorf("failed to send GET request to %q: %w", q, err)
return Metric{}, fmt.Errorf("failed to send GET request to %q: %s", q, err)
}
defer func() { _ = resp.Body.Close() }()
/*
@@ -303,7 +303,7 @@ func NewClient(cfg Config) (*Client, error) {
for _, r := range cfg.Retentions {
ret, err := convertRetention(r, offsetSecs, cfg.MsecsTime)
if err != nil {
return &Client{}, fmt.Errorf("couldn't parse retention %q :: %w", r, err)
return &Client{}, fmt.Errorf("couldn't parse retention %q :: %v", r, err)
}
retentions = append(retentions, ret)
}

View File

@@ -88,7 +88,7 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
}
queryLengthDuration, err := convertDuration(chunks[2])
if err != nil {
return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %w", chunks[2], err)
return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %s", chunks[2], err)
}
// set ttl in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
queryLength := queryLengthDuration.Milliseconds()
@@ -110,7 +110,7 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
aggTimeDuration, err := convertDuration(aggregates[1])
if err != nil {
return Retention{}, fmt.Errorf("invalid aggregation time duration string: %q: %w", aggregates[1], err)
return Retention{}, fmt.Errorf("invalid aggregation time duration string: %q: %s", aggregates[1], err)
}
aggTime := aggTimeDuration.Milliseconds()
if !msecTime {
@@ -119,7 +119,7 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
rowLengthDuration, err := convertDuration(chunks[1])
if err != nil {
return Retention{}, fmt.Errorf("invalid row length (first order) duration string: %q: %w", chunks[1], err)
return Retention{}, fmt.Errorf("invalid row length (first order) duration string: %q: %s", chunks[1], err)
}
// set length of each row in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
rowLength := rowLengthDuration.Milliseconds()

View File

@@ -46,7 +46,7 @@ type prometheusProcessor struct {
func (pp *prometheusProcessor) run(ctx context.Context) error {
blocks, err := pp.cl.Explore()
if err != nil {
return fmt.Errorf("explore failed: %w", err)
return fmt.Errorf("explore failed: %s", err)
}
if len(blocks) < 1 {
return fmt.Errorf("found no blocks to import")
@@ -57,7 +57,7 @@ func (pp *prometheusProcessor) run(ctx context.Context) error {
}
if err := pp.processBlocks(ctx, blocks); err != nil {
return fmt.Errorf("migration failed: %w", err)
return fmt.Errorf("migration failed: %s", err)
}
log.Println("Import finished!")
@@ -68,7 +68,7 @@ func (pp *prometheusProcessor) run(ctx context.Context) error {
func (pp *prometheusProcessor) do(ctx context.Context, b tsdb.BlockReader) error {
css, err := pp.cl.Read(ctx, b)
if err != nil {
return fmt.Errorf("failed to read block: %w", err)
return fmt.Errorf("failed to read block: %s", err)
}
defer func() {
if err := css.Close(); err != nil {
@@ -146,7 +146,7 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
for br := range blockReadersCh {
if err := pp.do(ctx, br); err != nil {
promErrorsTotal.Inc()
errCh <- fmt.Errorf("cannot read block %q: %w", br.Meta().ULID, err)
errCh <- fmt.Errorf("cannot read block %q: %s", br.Meta().ULID, err)
return
}
if cb, ok := br.(io.Closer); ok {
@@ -164,11 +164,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
select {
case promErr := <-errCh:
close(blockReadersCh)
return fmt.Errorf("prometheus error: %w", promErr)
return fmt.Errorf("prometheus error: %s", promErr)
case vmErr := <-pp.im.Errors():
close(blockReadersCh)
promErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, pp.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
case blockReadersCh <- br:
}
}
@@ -182,11 +182,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
for vmErr := range pp.im.Errors() {
if vmErr.Err != nil {
promErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, pp.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
}
}
for err := range errCh {
return fmt.Errorf("import process failed: %w", err)
return fmt.Errorf("import process failed: %s", err)
}
return nil

View File

@@ -59,12 +59,12 @@ func (f filter) inRange(minV, maxV int64) bool {
func NewClient(cfg Config) (*Client, error) {
db, err := tsdb.OpenDBReadOnly(cfg.Snapshot, cfg.TemporaryDir, nil)
if err != nil {
return nil, fmt.Errorf("failed to open snapshot %q: %w", cfg.Snapshot, err)
return nil, fmt.Errorf("failed to open snapshot %q: %s", cfg.Snapshot, err)
}
c := &Client{DBReadOnly: db}
timeMin, timeMax, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
if err != nil {
return nil, fmt.Errorf("failed to parse time in filter: %w", err)
return nil, fmt.Errorf("failed to parse time in filter: %s", err)
}
c.filter = filter{
min: timeMin,
@@ -83,7 +83,7 @@ func NewClient(cfg Config) (*Client, error) {
func (c *Client) Explore() ([]tsdb.BlockReader, error) {
blocks, err := c.Blocks()
if err != nil {
return nil, fmt.Errorf("failed to fetch blocks: %w", err)
return nil, fmt.Errorf("failed to fetch blocks: %s", err)
}
s := &vmctlutil.Stats{
Filtered: c.filter.min != 0 || c.filter.max != 0 || c.filter.label != "",
@@ -142,14 +142,14 @@ func parseTime(start, end string) (int64, int64, error) {
if start != "" {
v, err := time.Parse(time.RFC3339, start)
if err != nil {
return 0, 0, fmt.Errorf("failed to parse %q: %w", start, err)
return 0, 0, fmt.Errorf("failed to parse %q: %s", start, err)
}
s = v.UnixNano() / int64(time.Millisecond)
}
if end != "" {
v, err := time.Parse(time.RFC3339, end)
if err != nil {
return 0, 0, fmt.Errorf("failed to parse %q: %w", end, err)
return 0, 0, fmt.Errorf("failed to parse %q: %s", end, err)
}
e = v.UnixNano() / int64(time.Millisecond)
}

View File

@@ -44,7 +44,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
ranges, err := stepper.SplitDateRange(*rrp.filter.timeStart, *rrp.filter.timeEnd, rrp.filter.chunk, rrp.filter.timeReverse)
if err != nil {
return fmt.Errorf("failed to create date ranges for the given time filters: %w", err)
return fmt.Errorf("failed to create date ranges for the given time filters: %v", err)
}
question := fmt.Sprintf("Selected time range %q - %q will be split into %d ranges according to %q step. Continue?",
@@ -74,7 +74,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
for r := range rangeC {
if err := rrp.do(ctx, r); err != nil {
remoteReadErrorsTotal.Inc()
errCh <- fmt.Errorf("request failed for: %w", err)
errCh <- fmt.Errorf("request failed for: %s", err)
return
}
remoteReadRangesProcessed.Inc()
@@ -86,10 +86,10 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
for _, r := range ranges {
select {
case infErr := <-errCh:
return fmt.Errorf("remote read error: %w", infErr)
return fmt.Errorf("remote read error: %s", infErr)
case vmErr := <-rrp.dst.Errors():
remoteReadErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, rrp.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
case rangeC <- &remoteread.Filter{
StartTimestampMs: r[0].UnixMilli(),
EndTimestampMs: r[1].UnixMilli(),
@@ -105,11 +105,11 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
for vmErr := range rrp.dst.Errors() {
if vmErr.Err != nil {
remoteReadErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, rrp.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
}
}
for err := range errCh {
return fmt.Errorf("import process failed: %w", err)
return fmt.Errorf("import process failed: %s", err)
}
return nil
@@ -119,7 +119,7 @@ func (rrp *remoteReadProcessor) do(ctx context.Context, filter *remoteread.Filte
return rrp.src.Read(ctx, filter, func(series *vm.TimeSeries) error {
if err := rrp.dst.Input(series); err != nil {
return fmt.Errorf(
"failed to read data for time range start: %d, end: %d: %w",
"failed to read data for time range start: %d, end: %d, %s",
filter.StartTimestampMs, filter.EndTimestampMs, err)
}
return nil

View File

@@ -157,7 +157,7 @@ func (c *Client) Read(ctx context.Context, filter *Filter, streamCb StreamCallba
if errors.Is(err, context.Canceled) {
return fmt.Errorf("fetch request has ben cancelled")
}
return fmt.Errorf("error while fetching data from remote storage: %w", err)
return fmt.Errorf("error while fetching data from remote storage: %s", err)
}
return nil
}

View File

@@ -52,7 +52,7 @@ func (f filter) inRange(minV, maxV int64) bool {
func NewClient(cfg Config) (*Client, error) {
minTime, maxTime, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
if err != nil {
return nil, fmt.Errorf("failed to parse time in filter: %w", err)
return nil, fmt.Errorf("failed to parse time in filter: %s", err)
}
return &Client{
snapshotPath: cfg.Snapshot,
@@ -183,14 +183,14 @@ func parseTime(start, end string) (int64, int64, error) {
if start != "" {
v, err := time.Parse(time.RFC3339, start)
if err != nil {
return 0, 0, fmt.Errorf("failed to parse %q: %w", start, err)
return 0, 0, fmt.Errorf("failed to parse %q: %s", start, err)
}
s = v.UnixNano() / int64(time.Millisecond)
}
if end != "" {
v, err := time.Parse(time.RFC3339, end)
if err != nil {
return 0, 0, fmt.Errorf("failed to parse %q: %w", end, err)
return 0, 0, fmt.Errorf("failed to parse %q: %s", end, err)
}
e = v.UnixNano() / int64(time.Millisecond)
}

View File

@@ -36,7 +36,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
// Use the first aggregate type to explore blocks (block list is the same for all types)
blocks, err := tp.cl.Explore(tp.aggrTypes[0])
if err != nil {
return fmt.Errorf("explore failed: %w", err)
return fmt.Errorf("explore failed: %s", err)
}
if len(blocks) < 1 {
return fmt.Errorf("found no blocks to import")
@@ -84,7 +84,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
log.Println("Processing raw blocks (resolution=0)...")
stats, err := tp.processBlocks(rawBlocks, thanos.AggrTypeNone, bar)
if err != nil {
return fmt.Errorf("migration failed for raw blocks: %w", err)
return fmt.Errorf("migration failed for raw blocks: %s", err)
}
phases = append(phases, phaseStats{
name: "raw",
@@ -108,7 +108,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
aggrBlocks, err := tp.cl.Explore(aggrType)
if err != nil {
return fmt.Errorf("explore failed for aggr type %s: %w", aggrType, err)
return fmt.Errorf("explore failed for aggr type %s: %s", aggrType, err)
}
var downsampledOnly []thanos.BlockInfo
@@ -128,7 +128,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
stats, err := tp.processBlocks(downsampledOnly, aggrType, bar)
thanos.CloseBlocks(aggrBlocks)
if err != nil {
return fmt.Errorf("migration failed for aggr type %s: %w", aggrType, err)
return fmt.Errorf("migration failed for aggr type %s: %s", aggrType, err)
}
phases = append(phases, phaseStats{
name: aggrType.String(),
@@ -153,7 +153,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
for vmErr := range tp.im.Errors() {
if vmErr.Err != nil {
thanosErrorsTotal.Inc()
return fmt.Errorf("import process failed: %w", wrapErr(vmErr, tp.isVerbose))
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, tp.isVerbose))
}
}
@@ -184,7 +184,7 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
seriesCount, samplesCount, err := tp.do(bi, aggrType)
if err != nil {
thanosErrorsTotal.Inc()
errCh <- fmt.Errorf("read failed for block %q with aggr %s: %w", bi.Block.Meta().ULID, aggrType, err)
errCh <- fmt.Errorf("read failed for block %q with aggr %s: %s", bi.Block.Meta().ULID, aggrType, err)
return
}
@@ -209,12 +209,12 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
case thanosErr := <-errCh:
close(blockReadersCh)
wg.Wait()
return processBlocksStats{}, fmt.Errorf("thanos error: %w", thanosErr)
return processBlocksStats{}, fmt.Errorf("thanos error: %s", thanosErr)
case vmErr := <-tp.im.Errors():
close(blockReadersCh)
wg.Wait()
thanosErrorsTotal.Inc()
return processBlocksStats{}, fmt.Errorf("import process failed: %w", wrapErr(vmErr, tp.isVerbose))
return processBlocksStats{}, fmt.Errorf("import process failed: %s", wrapErr(vmErr, tp.isVerbose))
case blockReadersCh <- bi:
}
}
@@ -223,7 +223,7 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
wg.Wait()
close(errCh)
for err := range errCh {
return processBlocksStats{}, fmt.Errorf("import process failed: %w", err)
return processBlocksStats{}, fmt.Errorf("import process failed: %s", err)
}
return processBlocksStats{
@@ -236,7 +236,7 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
func (tp *thanosProcessor) do(bi thanos.BlockInfo, aggrType thanos.AggrType) (uint64, uint64, error) {
ss, err := tp.cl.Read(bi)
if err != nil {
return 0, 0, fmt.Errorf("failed to read block: %w", err)
return 0, 0, fmt.Errorf("failed to read block: %s", err)
}
defer ss.Close() // Ensure querier is closed even on early returns

View File

@@ -163,7 +163,7 @@ func NewImporter(ctx context.Context, cfg Config) (*Importer, error) {
importDuration: metrics.GetOrCreateHistogram(`vmctl_importer_request_duration_seconds`),
}
if err := im.Ping(); err != nil {
return nil, fmt.Errorf("ping to %q failed: %w", addr, err)
return nil, fmt.Errorf("ping to %q failed: %s", addr, err)
}
if cfg.BatchSize < 1 {
@@ -289,7 +289,7 @@ func (im *Importer) flush(ctx context.Context, b []*TimeSeries) error {
retryableFunc := func() error { return im.Import(b) }
attempts, err := im.backoff.Retry(ctx, retryableFunc)
if err != nil {
return fmt.Errorf("import failed with %d retries: %w", attempts, err)
return fmt.Errorf("import failed with %d retries: %s", attempts, err)
}
im.s.Lock()
im.s.retries = attempts
@@ -302,7 +302,7 @@ func (im *Importer) Ping() error {
url := fmt.Sprintf("%s/health", im.addr)
req, err := http.NewRequest(http.MethodGet, url, nil)
if err != nil {
return fmt.Errorf("cannot create request to %q: %w", im.addr, err)
return fmt.Errorf("cannot create request to %q: %s", im.addr, err)
}
if im.user != "" {
req.SetBasicAuth(im.user, im.password)
@@ -332,7 +332,7 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
req, err := http.NewRequest(http.MethodPost, im.importPath, pr)
if err != nil {
im.importRequestsErrorsTotal.Inc()
return fmt.Errorf("cannot create request to %q: %w", im.addr, err)
return fmt.Errorf("cannot create request to %q: %s", im.addr, err)
}
if im.user != "" {
req.SetBasicAuth(im.user, im.password)
@@ -352,7 +352,7 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
zw, err := gzip.NewWriterLevel(w, 1)
if err != nil {
im.importRequestsErrorsTotal.Inc()
return fmt.Errorf("unexpected error when creating gzip writer: %w", err)
return fmt.Errorf("unexpected error when creating gzip writer: %s", err)
}
w = zw
}
@@ -411,7 +411,7 @@ var ErrBadRequest = errors.New("bad request")
func (im *Importer) do(req *http.Request) error {
resp, err := im.client.Do(req)
if err != nil {
return fmt.Errorf("unexpected error when performing request: %w", err)
return fmt.Errorf("unexpected error when performing request: %s", err)
}
defer func() {
_ = resp.Body.Close()
@@ -419,7 +419,7 @@ func (im *Importer) do(req *http.Request) error {
if resp.StatusCode != http.StatusNoContent {
body, err := io.ReadAll(resp.Body)
if err != nil {
return fmt.Errorf("failed to read response body for status code %d: %w", resp.StatusCode, err)
return fmt.Errorf("failed to read response body for status code %d: %s", resp.StatusCode, err)
}
if resp.StatusCode == http.StatusBadRequest {
return fmt.Errorf("%w: unexpected response code %d: %s", ErrBadRequest, resp.StatusCode, string(body))

View File

@@ -55,14 +55,14 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
start, err := vmctlutil.ParseTime(p.filter.TimeStart)
if err != nil {
return fmt.Errorf("failed to parse %s, provided: %s: %w", vmNativeFilterTimeStart, p.filter.TimeStart, err)
return fmt.Errorf("failed to parse %s, provided: %s, error: %w", vmNativeFilterTimeStart, p.filter.TimeStart, err)
}
end := time.Now().In(start.Location())
if p.filter.TimeEnd != "" {
end, err = vmctlutil.ParseTime(p.filter.TimeEnd)
if err != nil {
return fmt.Errorf("failed to parse %s, provided: %s: %w", vmNativeFilterTimeEnd, p.filter.TimeEnd, err)
return fmt.Errorf("failed to parse %s, provided: %s, error: %w", vmNativeFilterTimeEnd, p.filter.TimeEnd, err)
}
}
@@ -91,7 +91,7 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
err := p.runBackfilling(ctx, tenantID, ranges)
if err != nil {
migrationErrorsTotal.Inc()
return fmt.Errorf("migration failed: %w", err)
return fmt.Errorf("migration failed: %s", err)
}
if p.interCluster {
@@ -157,7 +157,7 @@ func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcU
}
default:
}
return fmt.Errorf("failed to write into %q: %w", p.dst.Addr, err)
return fmt.Errorf("failed to write into %q: %s", p.dst.Addr, err)
}
p.s.Lock()
@@ -184,7 +184,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
importAddr, err := vm.AddExtraLabelsToImportPath(importAddr, p.dst.ExtraLabels)
if err != nil {
return fmt.Errorf("failed to add labels to import path: %w", err)
return fmt.Errorf("failed to add labels to import path: %s", err)
}
dstURL := fmt.Sprintf("%s/%s", p.dst.Addr, importAddr)
@@ -222,7 +222,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
format = fmt.Sprintf(nativeWithBackoffTpl, barPrefix)
metricsMap, err = p.explore(ctx, p.src, tenantID, ranges)
if err != nil {
return fmt.Errorf("failed to explore metric names: %w", err)
return fmt.Errorf("failed to explore metric names: %s", err)
}
if len(metricsMap) == 0 {
errMsg := "no metrics found"
@@ -295,7 +295,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
case <-ctx.Done():
return fmt.Errorf("context canceled")
case infErr := <-errCh:
return fmt.Errorf("export/import error: %w", infErr)
return fmt.Errorf("export/import error: %s", infErr)
case filterCh <- native.Filter{
Match: match,
TimeStart: times[0].Format(time.RFC3339),
@@ -313,7 +313,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
close(errCh)
for err := range errCh {
return fmt.Errorf("import process failed: %w", err)
return fmt.Errorf("import process failed: %s", err)
}
return nil

View File

@@ -257,7 +257,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
dealine, err = strconv.Atoi(deadlineStr)
if err != nil {
logger.Errorf("cannot parse `seconds` arg %q: %s", deadlineStr, err)
jsonResponseError(w, fmt.Errorf("cannot parse `seconds` arg %q: %w", deadlineStr, err))
jsonResponseError(w, fmt.Errorf("cannot parse `seconds` arg %q: %s", deadlineStr, err))
return true
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -21,16 +21,16 @@
},
"dependencies": {
"classnames": "^2.5.1",
"dayjs": "^1.11.21",
"dayjs": "^1.11.20",
"lodash.debounce": "^4.0.8",
"marked": "^18.0.5",
"preact": "^10.29.2",
"qs": "^6.15.2",
"marked": "^18.0.2",
"preact": "^10.29.1",
"qs": "^6.15.1",
"react-input-mask": "^2.0.4",
"react-router-dom": "^7.17.0",
"react-router-dom": "^7.14.1",
"uplot": "^1.6.32",
"vite": "^8.0.16",
"web-vitals": "^5.3.0"
"vite": "^8.0.8",
"web-vitals": "^5.2.0"
},
"devDependencies": {
"@eslint/eslintrc": "^3.3.5",
@@ -39,24 +39,24 @@
"@testing-library/jest-dom": "^6.9.1",
"@testing-library/preact": "^3.2.4",
"@types/lodash.debounce": "^4.0.9",
"@types/node": "^25.9.2",
"@types/qs": "^6.15.1",
"@types/react": "^19.2.17",
"@types/node": "^25.6.0",
"@types/qs": "^6.15.0",
"@types/react": "^19.2.14",
"@types/react-input-mask": "^3.0.6",
"@types/react-router-dom": "^5.3.3",
"@typescript-eslint/eslint-plugin": "^8.61.0",
"@typescript-eslint/parser": "^8.61.0",
"@typescript-eslint/eslint-plugin": "^8.58.2",
"@typescript-eslint/parser": "^8.58.2",
"cross-env": "^10.1.0",
"eslint": "^9.39.2",
"eslint-plugin-react": "^7.37.5",
"eslint-plugin-unused-imports": "^4.4.1",
"globals": "^17.6.0",
"http-proxy-middleware": "^4.1.0",
"jsdom": "^29.1.1",
"postcss": "^8.5.15",
"sass-embedded": "^1.100.0",
"typescript": "^6.0.3",
"vitest": "^4.1.8"
"globals": "^17.5.0",
"http-proxy-middleware": "^3.0.5",
"jsdom": "^29.0.2",
"postcss": "^8.5.10",
"sass-embedded": "^1.99.0",
"typescript": "^6.0.2",
"vitest": "^4.1.4"
},
"browserslist": {
"production": [

View File

@@ -156,14 +156,14 @@ func readAllAndClose(t *testing.T, responseBody io.ReadCloser) string {
//
// This type is expected to be embedded by the apps that serve metrics.
type metricsClient struct {
cli *Client
url string
metricsCli *Client
url string
}
func newMetricsClient(cli *Client, addr string) *metricsClient {
return &metricsClient{
cli: cli,
url: fmt.Sprintf("http://%s/metrics", addr),
metricsCli: cli,
url: fmt.Sprintf("http://%s/metrics", addr),
}
}
@@ -179,7 +179,7 @@ func (c *metricsClient) GetIntMetric(t *testing.T, metricName string) int {
func (c *metricsClient) GetMetric(t *testing.T, metricName string) float64 {
t.Helper()
metrics, statusCode := c.cli.Get(t, c.url, nil)
metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -205,7 +205,7 @@ func (c *metricsClient) GetMetricsByPrefix(t *testing.T, prefix string) []float6
values := []float64{}
metrics, statusCode := c.cli.Get(t, c.url, nil)
metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -234,7 +234,7 @@ func (c *metricsClient) GetMetricsByRegexp(t *testing.T, re *regexp.Regexp) []fl
values := []float64{}
metrics, statusCode := c.cli.Get(t, c.url, nil)
metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -270,7 +270,7 @@ func (c *metricsClient) rpcRowsSentTotal(t *testing.T) int {
}
type vmselectClient struct {
cli *Client
vmselectCli *Client
url func(op, path string, opts QueryOpts) string
metricNamesStatsResetURL string
tenantsURL string
@@ -287,7 +287,7 @@ func (c *vmselectClient) PrometheusAPIV1Export(t *testing.T, query string, opts
values := opts.asURLValues()
values.Add("match[]", query)
values.Add("format", "promapi")
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1QueryResponse(t, res)
}
@@ -302,7 +302,7 @@ func (c *vmselectClient) PrometheusAPIV1ExportNative(t *testing.T, query string,
values := opts.asURLValues()
values.Add("match[]", query)
values.Add("format", "promapi")
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return []byte(res)
}
@@ -315,7 +315,7 @@ func (c *vmselectClient) PrometheusAPIV1Query(t *testing.T, query string, opts Q
url := c.url("select", "prometheus/api/v1/query", opts)
values := opts.asURLValues()
values.Add("query", query)
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1QueryResponse(t, res)
}
@@ -329,7 +329,7 @@ func (c *vmselectClient) PrometheusAPIV1QueryRange(t *testing.T, query string, o
url := c.url("select", "prometheus/api/v1/query_range", opts)
values := opts.asURLValues()
values.Add("query", query)
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1QueryResponse(t, res)
}
@@ -342,7 +342,7 @@ func (c *vmselectClient) PrometheusAPIV1Series(t *testing.T, matchQuery string,
url := c.url("select", "prometheus/api/v1/series", opts)
values := opts.asURLValues()
values.Add("match[]", matchQuery)
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1SeriesResponse(t, res)
}
@@ -354,7 +354,7 @@ func (c *vmselectClient) PrometheusAPIV1SeriesCount(t *testing.T, opts QueryOpts
t.Helper()
url := c.url("select", "prometheus/api/v1/series/count", opts)
values := opts.asURLValues()
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1SeriesCountResponse(t, res)
}
@@ -367,7 +367,7 @@ func (c *vmselectClient) PrometheusAPIV1Labels(t *testing.T, matchQuery string,
url := c.url("select", "prometheus/api/v1/labels", opts)
values := opts.asURLValues()
values.Add("match[]", matchQuery)
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1LabelsResponse(t, res)
}
@@ -382,7 +382,7 @@ func (c *vmselectClient) PrometheusAPIV1LabelValues(t *testing.T, labelName, mat
url := c.url("select", path, opts)
values := opts.asURLValues()
values.Add("match[]", matchQuery)
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1LabelValuesResponse(t, res)
}
@@ -394,7 +394,7 @@ func (c *vmselectClient) PrometheusAPIV1Metadata(t *testing.T, metric string, li
values := opts.asURLValues()
values.Add("metric", metric)
values.Add("limit", strconv.Itoa(limit))
res, _ := c.cli.PostForm(t, url, values, opts.Headers)
res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
return NewPrometheusAPIV1Metadata(t, res)
}
@@ -408,7 +408,7 @@ func (c *vmselectClient) PrometheusAPIV1AdminTSDBDeleteSeries(t *testing.T, matc
url := c.url("delete", "prometheus/api/v1/admin/tsdb/delete_series", opts)
values := opts.asURLValues()
values.Add("match[]", matchQuery)
res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusNoContent, res)
}
@@ -426,7 +426,7 @@ func (c *vmselectClient) PrometheusAPIV1StatusMetricNamesStats(t *testing.T, lim
values.Add("limit", limit)
values.Add("le", le)
values.Add("match_pattern", matchPattern)
res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
}
@@ -455,7 +455,7 @@ func (c *vmselectClient) PrometheusAPIV1StatusTSDB(t *testing.T, matchQuery stri
addNonEmpty("match[]", matchQuery)
addNonEmpty("topN", topN)
addNonEmpty("date", date)
res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
}
@@ -476,7 +476,7 @@ func (c *vmselectClient) GraphiteMetricsIndex(t *testing.T, opts QueryOpts) Grap
t.Helper()
url := c.url("select", "graphite/metrics/index.json", opts)
res, statusCode := c.cli.Get(t, url, opts.Headers)
res, statusCode := c.vmselectCli.Get(t, url, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
}
@@ -499,7 +499,7 @@ func (c *vmselectClient) GraphiteMetricsFind(t *testing.T, query string, opts Qu
url := c.url("select", "graphite/metrics/find", opts)
values := opts.asURLValues()
values.Add("query", query)
resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
}
@@ -522,7 +522,7 @@ func (c *vmselectClient) GraphiteMetricsExpand(t *testing.T, query string, opts
url := c.url("select", "graphite/metrics/expand", opts)
values := opts.asURLValues()
values.Add("query", query)
resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
}
@@ -546,7 +546,7 @@ func (c *vmselectClient) GraphiteRender(t *testing.T, target string, opts QueryO
values := opts.asURLValues()
values.Add("format", "json")
values.Add("target", target)
resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
}
@@ -567,7 +567,7 @@ func (c *vmselectClient) GraphiteTagsTagSeries(t *testing.T, record string, opts
url := c.url("select", "graphite/tags/tagSeries", opts)
values := opts.asURLValues()
values.Add("path", record)
_, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
_, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if got, want := statusCode, http.StatusNotImplemented; got != want {
t.Fatalf("unexpected status code: got %d, want %d", got, want)
}
@@ -584,7 +584,7 @@ func (c *vmselectClient) GraphiteTagsTagMultiSeries(t *testing.T, records []stri
for _, rec := range records {
values.Add("path", rec)
}
_, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
_, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
if got, want := statusCode, http.StatusNotImplemented; got != want {
t.Fatalf("unexpected status code: got %d, want %d", got, want)
}
@@ -598,7 +598,7 @@ func (c *vmselectClient) GraphiteTagsTagMultiSeries(t *testing.T, records []stri
func (c *vmselectClient) PrometheusAPIV1AdminStatusMetricNamesStatsReset(t *testing.T, opts QueryOpts) {
t.Helper()
values := opts.asURLValues()
res, statusCode := c.cli.PostForm(t, c.metricNamesStatsResetURL, values, opts.Headers)
res, statusCode := c.vmselectCli.PostForm(t, c.metricNamesStatsResetURL, values, opts.Headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusNoContent, res)
}
@@ -608,7 +608,7 @@ func (c *vmselectClient) PrometheusAPIV1AdminStatusMetricNamesStatsReset(t *test
// /admin/tenants endpoint.
func (c *vmselectClient) APIV1AdminTenants(t *testing.T, opts QueryOpts) *AdminTenantsResponse {
t.Helper()
res, statusCode := c.cli.Get(t, c.tenantsURL, opts.Headers)
res, statusCode := c.vmselectCli.Get(t, c.tenantsURL, opts.Headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
}
@@ -622,7 +622,7 @@ func (c *vmselectClient) APIV1AdminTenants(t *testing.T, opts QueryOpts) *AdminT
}
type vminsertClient struct {
cli *Client
vminsertCli *Client
url func(op, path string, opts QueryOpts) string
openTSDBURL func(op, path string, opts QueryOpts) string
graphiteListenAddr string
@@ -647,7 +647,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportCSV(t *testing.T, records []string
headers := opts.getHeaders()
headers.Set("Content-Type", "text/plain")
c.sendBlocking(t, len(records), func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
}
@@ -671,7 +671,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportNative(t *testing.T, data []byte,
headers := opts.getHeaders()
headers.Set("Content-Type", "text/plain")
c.sendBlocking(t, 1, func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
}
@@ -693,7 +693,7 @@ func (c *vminsertClient) PrometheusAPIV1Write(t *testing.T, wr prompb.WriteReque
headers := opts.getHeaders()
headers.Set("Content-Type", "application/x-protobuf")
c.sendBlocking(t, recordsCount, func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
}
@@ -745,7 +745,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportPrometheus(t *testing.T, records [
headers := opts.getHeaders()
headers.Set("Content-Type", "text/plain")
c.sendBlocking(t, recordsCount, func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
}
@@ -771,7 +771,7 @@ func (c *vminsertClient) InfluxWrite(t *testing.T, records []string, opts QueryO
headers.Set("Content-Type", "text/plain")
c.sendBlocking(t, len(records), func() {
t.Helper()
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
}
@@ -805,7 +805,7 @@ func (c *vminsertClient) OpentelemetryV1Metrics(t *testing.T, md otlppb.MetricsD
headers := opts.getHeaders()
headers.Set("Content-Type", "application/x-protobuf")
c.sendBlocking(t, recordsCount, func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -830,7 +830,7 @@ func (c *vminsertClient) OpenTSDBAPIPut(t *testing.T, records []string, opts Que
headers := opts.getHeaders()
headers.Set("Content-Type", "application/json")
c.sendBlocking(t, len(records), func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusNoContent {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
}
@@ -853,7 +853,7 @@ func (c *vminsertClient) ZabbixConnectorHistory(t *testing.T, records []string,
headers := opts.getHeaders()
headers.Set("Content-Type", "application/json")
c.sendBlocking(t, len(records), func() {
_, statusCode := c.cli.Post(t, url, data, headers)
_, statusCode := c.vminsertCli.Post(t, url, data, headers)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -867,11 +867,11 @@ func (c *vminsertClient) ZabbixConnectorHistory(t *testing.T, records []string,
// See https://docs.victoriametrics.com/victoriametrics/integrations/graphite/#ingesting
func (c *vminsertClient) GraphiteWrite(t *testing.T, records []string, _ QueryOpts) {
t.Helper()
c.cli.Write(t, c.graphiteListenAddr, records)
c.vminsertCli.Write(t, c.graphiteListenAddr, records)
}
type vmstorageClient struct {
cli *Client
vmstorageCli *Client
httpListenAddr string
}
@@ -881,7 +881,7 @@ func (c *vmstorageClient) ForceFlush(t *testing.T) {
t.Helper()
url := fmt.Sprintf("http://%s/internal/force_flush", c.httpListenAddr)
_, statusCode := c.cli.Get(t, url, nil)
_, statusCode := c.vmstorageCli.Get(t, url, nil)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -892,7 +892,7 @@ func (c *vmstorageClient) ForceMerge(t *testing.T) {
t.Helper()
url := fmt.Sprintf("http://%s/internal/force_merge", c.httpListenAddr)
_, statusCode := c.cli.Get(t, url, nil)
_, statusCode := c.vmstorageCli.Get(t, url, nil)
if statusCode != http.StatusOK {
t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
}
@@ -905,7 +905,7 @@ func (c *vmstorageClient) ForceMerge(t *testing.T) {
func (c *vmstorageClient) SnapshotCreate(t *testing.T) *SnapshotCreateResponse {
t.Helper()
data, statusCode := c.cli.Post(t, c.SnapshotCreateURL(), nil, nil)
data, statusCode := c.vmstorageCli.Post(t, c.SnapshotCreateURL(), nil, nil)
if got, want := statusCode, http.StatusOK; got != want {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
}
@@ -931,7 +931,7 @@ func (c *vmstorageClient) APIV1AdminTSDBSnapshot(t *testing.T) *APIV1AdminTSDBSn
t.Helper()
url := fmt.Sprintf("http://%s/api/v1/admin/tsdb/snapshot", c.httpListenAddr)
data, statusCode := c.cli.Post(t, url, nil, nil)
data, statusCode := c.vmstorageCli.Post(t, url, nil, nil)
if got, want := statusCode, http.StatusOK; got != want {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
}
@@ -952,7 +952,7 @@ func (c *vmstorageClient) SnapshotList(t *testing.T) *SnapshotListResponse {
t.Helper()
url := fmt.Sprintf("http://%s/snapshot/list", c.httpListenAddr)
data, statusCode := c.cli.Get(t, url, nil)
data, statusCode := c.vmstorageCli.Get(t, url, nil)
if got, want := statusCode, http.StatusOK; got != want {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
}
@@ -973,7 +973,7 @@ func (c *vmstorageClient) SnapshotDelete(t *testing.T, snapshotName string) *Sna
t.Helper()
url := fmt.Sprintf("http://%s/snapshot/delete?snapshot=%s", c.httpListenAddr, snapshotName)
data, statusCode := c.cli.Delete(t, url)
data, statusCode := c.vmstorageCli.Delete(t, url)
wantStatusCodes := map[int]bool{
http.StatusOK: true,
http.StatusInternalServerError: true,
@@ -998,7 +998,7 @@ func (c *vmstorageClient) SnapshotDeleteAll(t *testing.T) *SnapshotDeleteAllResp
t.Helper()
url := fmt.Sprintf("http://%s/snapshot/delete_all", c.httpListenAddr)
data, statusCode := c.cli.Post(t, url, nil, nil)
data, statusCode := c.vmstorageCli.Post(t, url, nil, nil)
if got, want := statusCode, http.StatusOK; got != want {
t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
}

View File

@@ -77,7 +77,7 @@ type vminsertRuntimeValues struct {
func newVminsert(app *app, cli *Client, rt vminsertRuntimeValues) *Vminsert {
metricsClient := newMetricsClient(cli, rt.httpListenAddr)
vminsertClient := &vminsertClient{
cli: cli,
vminsertCli: cli,
url: func(op, path string, opts QueryOpts) string {
return getClusterPath(rt.httpListenAddr, op, path, opts)
},

View File

@@ -48,7 +48,7 @@ func newVmselect(app *app, cli *Client, rt vmselectRuntimeValues) *Vmselect {
app: app,
metricsClient: newMetricsClient(cli, rt.httpListenAddr),
vmselectClient: &vmselectClient{
cli: cli,
vmselectCli: cli,
url: func(op, path string, opts QueryOpts) string {
return getClusterPath(rt.httpListenAddr, op, path, opts)
},

View File

@@ -58,11 +58,11 @@ func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
app: app,
metricsClient: newMetricsClient(cli, rt.httpListenAddr),
vmstorageClient: &vmstorageClient{
cli: cli,
vmstorageCli: cli,
httpListenAddr: rt.httpListenAddr,
},
vmselectClient: &vmselectClient{
cli: cli,
vmselectCli: cli,
url: func(op, path string, opts QueryOpts) string {
return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
},
@@ -70,7 +70,7 @@ func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
tenantsURL: "vmsingle-does-not-serve-tenants",
},
vminsertClient: &vminsertClient{
cli: cli,
vminsertCli: cli,
url: func(_, path string, _ QueryOpts) string {
return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
},

View File

@@ -63,7 +63,7 @@ func newVmstorage(app *app, cli *Client, rt vmstorageRuntimeValues) *Vmstorage {
app: app,
metricsClient: newMetricsClient(cli, rt.httpListenAddr),
vmstorageClient: &vmstorageClient{
cli: cli,
vmstorageCli: cli,
httpListenAddr: rt.httpListenAddr,
},
storageDataPath: rt.storageDataPath,

View File

@@ -26,8 +26,6 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
## tip
* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): fix issue with producing aggregated samples with identical timestamps between flushes. See PR [#10808](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10808) for details.
## [v1.145.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.145.0)
Released at 2026-06-08

View File

@@ -121,7 +121,7 @@ func (p *Password) initRandomValue() {
_, err := io.ReadFull(rand.Reader, buf[:])
if err != nil {
// cannot use lib/logger here, since it can be uninitialized yet
panic(fmt.Errorf("FATAL: cannot read random data: %w", err))
panic(fmt.Errorf("FATAL: cannot read random data: %s", err))
}
s := string(buf[:])
p.value.Store(&s)

View File

@@ -16,7 +16,7 @@ func ParseKey(key []byte) (any, error) {
k, err := x509.ParsePKIXPublicKey(b.Bytes)
if err != nil {
return nil, fmt.Errorf("failed to parse key %q: %w", key, err)
return nil, fmt.Errorf("failed to parse key %q: %v", key, err)
}
return k, nil

View File

@@ -14,7 +14,7 @@ func BenchmarkWriteRequestUnmarshalProtobuf(b *testing.B) {
wru := &WriteRequestUnmarshaler{}
for pb.Next() {
if _, err := wru.UnmarshalProtobuf(data); err != nil {
panic(fmt.Errorf("unexpected error: %w", err))
panic(fmt.Errorf("unexpected error: %s", err))
}
}
})

View File

@@ -97,12 +97,12 @@ func getDedicatedServerDetails(cfg *apiConfig, dedicatedServerName string) (*ded
request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
})
if err != nil {
return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
return nil, fmt.Errorf("request %s error: %v", reqPath, err)
}
var dedicatedServerDetails dedicatedServer
if err = json.Unmarshal(resp, &dedicatedServerDetails); err != nil {
return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
}
// get IPs for this dedicated server.
@@ -113,12 +113,12 @@ func getDedicatedServerDetails(cfg *apiConfig, dedicatedServerName string) (*ded
request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
})
if err != nil {
return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
return nil, fmt.Errorf("request %s error: %v", reqPath, err)
}
var ips []string
if err = json.Unmarshal(resp, &ips); err != nil {
return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
}
// handle different IP formats
@@ -141,11 +141,11 @@ func getDedicatedServerList(cfg *apiConfig) ([]string, error) {
request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
})
if err != nil {
return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
return nil, fmt.Errorf("request %s error: %v", reqPath, err)
}
if err = json.Unmarshal(resp, &dedicatedServerList); err != nil {
return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
}
return dedicatedServerList, nil

View File

@@ -117,12 +117,12 @@ func getVPSDetails(cfg *apiConfig, vpsName string) (*virtualPrivateServer, error
request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
})
if err != nil {
return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
return nil, fmt.Errorf("request %s error: %v", reqPath, err)
}
var vpsDetails virtualPrivateServer
if err = json.Unmarshal(resp, &vpsDetails); err != nil {
return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
}
// get IPs for this vps.
@@ -133,12 +133,12 @@ func getVPSDetails(cfg *apiConfig, vpsName string) (*virtualPrivateServer, error
request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
})
if err != nil {
return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
return nil, fmt.Errorf("request %s error: %v", reqPath, err)
}
var ips []string
if err = json.Unmarshal(resp, &ips); err != nil {
return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
}
// handle different IP formats
@@ -162,12 +162,12 @@ func getVPSList(cfg *apiConfig) ([]string, error) {
request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
})
if err != nil {
return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
return nil, fmt.Errorf("request %s error: %v", reqPath, err)
}
var vpsList []string
if err = json.Unmarshal(resp, &vpsList); err != nil {
return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
}
return vpsList, nil

View File

@@ -33,7 +33,7 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
}
parsedURL, err := url.Parse(sdc.URL)
if err != nil {
return nil, fmt.Errorf("cannot parse %s: %w", sdc.URL, err)
return nil, fmt.Errorf("parse URL %s error: %v", sdc.URL, err)
}
if parsedURL.Scheme != "http" && parsedURL.Scheme != "https" {
return nil, fmt.Errorf("URL %s scheme must be 'http' or 'https'", sdc.URL)

View File

@@ -221,7 +221,7 @@ func getIAMToken(cfg *apiConfig) (*iamToken, error) {
body := bytes.NewBuffer(passport)
resp, err := cfg.client.Post(iamURL, "application/json", body)
if err != nil {
return nil, fmt.Errorf("cannot send request to yandex cloud iam api %q: %w", iamURL, err)
return nil, fmt.Errorf("cannot send request to yandex cloud iam api %q: %s", iamURL, err)
}
data, err := readResponseBody(resp, iamURL)
if err != nil {

View File

@@ -667,11 +667,11 @@ func TestScrapeWorkScrapeInternalStreamConcurrency(t *testing.T) {
}
generateScrape := func(n int) string {
var w strings.Builder
w := strings.Builder{}
for i := range n {
fmt.Fprintf(&w, "fooooo_%d 1\n", i)
w.WriteString(fmt.Sprintf("fooooo_%d 1\n", i))
if i%100 == 0 {
fmt.Fprintf(&w, "# HELP fooooo_%d This is a test\n", i)
w.WriteString(fmt.Sprintf("# HELP fooooo_%d This is a test\n", i))
}
}
return w.String()
@@ -1005,9 +1005,9 @@ func TestSendStaleSeries(t *testing.T) {
}
}
generateScrape := func(n int) string {
var w strings.Builder
w := strings.Builder{}
for i := range n {
fmt.Fprintf(&w, "foo_%d 1\n", i)
w.WriteString(fmt.Sprintf("foo_%d 1\n", i))
}
return w.String()
}

View File

@@ -186,7 +186,7 @@ func (s *Series) unmarshalProtobuf(src []byte) (err error) {
}
pt := &points[len(points)-1]
if err := pt.unmarshalProtobuf(data); err != nil {
return fmt.Errorf("cannot unmarshal point: %w", err)
return fmt.Errorf("cannot unmarshal point: %s", err)
}
case 1:
data, ok := fc.MessageData()

View File

@@ -30,7 +30,7 @@ func ProcessRequestBody(b []byte) ([]byte, error) {
}
}
if err := json.Unmarshal(b, &req); err != nil {
return nil, fmt.Errorf("cannot unmarshal Firehose JSON in request body: %w", err)
return nil, fmt.Errorf("cannot unmarshal Firehose JSON in request body: %s", err)
}
var dst []byte

View File

@@ -99,17 +99,17 @@ func (r *Row) unmarshal(o *fastjson.Value) error {
n, err := getFloat64(o, "value")
if err != nil {
return fmt.Errorf("missing `value` element: %w", err)
return fmt.Errorf("missing `value` element, %s", err)
}
r.Value = n
cl, err := getInt64(o, "clock")
if err != nil {
return fmt.Errorf("missing `clock` element: %w", err)
return fmt.Errorf("missing `clock` element, %s", err)
}
ns, err := getInt64(o, "ns")
if err != nil {
return fmt.Errorf("missing `ns` element: %w", err)
return fmt.Errorf("missing `ns` element, %s", err)
}
// clock - Number of seconds since Epoch to the moment when value was collected (integer part).
// ns - Number of nanoseconds to be added to clock to get a precise value collection time.
@@ -121,7 +121,7 @@ func (r *Row) unmarshal(o *fastjson.Value) error {
if len(groupValue) != 0 {
groups, err := getArray(o, "groups")
if err != nil {
return fmt.Errorf("missing `groups` element: %w", err)
return fmt.Errorf("missing `groups` element, %s", err)
}
for _, g := range groups {
k := g.GetStringBytes()
@@ -141,7 +141,7 @@ func (r *Row) unmarshal(o *fastjson.Value) error {
itemTags, err := getArray(o, "item_tags")
if err != nil {
return fmt.Errorf("missing `item_tags` element: %w", err)
return fmt.Errorf("missing `item_tags` element, %s", err)
}
if len(duplicateTagsSeparator) == 0 { // Do not merge tags

View File

@@ -71,9 +71,9 @@ func Create(ctx context.Context, createSnapshotURL string) (string, error) {
return snap.Snapshot, nil
}
if snap.Status == "error" {
return "", fmt.Errorf("snapshot status: %q; msg: %q", snap.Status, snap.Msg)
return "", errors.New(snap.Msg)
}
return "", fmt.Errorf("snapshot status unknown: %q", snap.Status)
return "", fmt.Errorf("unknown status: %v", snap.Status)
}
// Delete deletes a snapshot via the provided api endpoint
@@ -121,14 +121,14 @@ func Delete(ctx context.Context, deleteSnapshotURL string, snapshotName string)
if snap.Status == "error" {
return errors.New(snap.Msg)
}
return fmt.Errorf("snapshot status unknown: %q", snap.Status)
return fmt.Errorf("unknown status: %v", snap.Status)
}
// GetHTTPClient returns a new HTTP client configured for snapshot operations.
func GetHTTPClient() (*http.Client, error) {
tr, err := promauth.NewTLSTransport(*tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify, "vm_snapshot_client")
if err != nil {
return nil, fmt.Errorf("failed to create transport: %w", err)
return nil, fmt.Errorf("failed to create transport: %s", err)
}
hc := &http.Client{
Transport: tr,

View File

@@ -649,7 +649,7 @@ func (is *indexSearch) searchLabelNamesWithFiltersOnDate(qt *querytracer.Tracer,
} else {
_, key, err := unmarshalCompositeTagKey(labelName)
if err != nil {
return nil, fmt.Errorf("cannot unmarshal composite tag key: %w", err)
return nil, fmt.Errorf("cannot unmarshal composite tag key: %s", err)
}
lns[string(key)] = struct{}{}
}

View File

@@ -231,7 +231,6 @@ func (d *Deduplicator) flush(pushFunc PushFunc) {
logger.Warnf("deduplication couldn't be finished in the configured dedupInterval=%s; it took %.03fs; "+
"possible solutions: increase dedupInterval; reduce samples' ingestion rate", d.interval, duration.Seconds())
}
deadlineTime = deadlineTime.Add(d.interval)
for time.Now().After(deadlineTime) {
deadlineTime = deadlineTime.Add(d.interval)
}

View File

@@ -845,7 +845,6 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipFlu
} else {
a.flush(pf, flushTime, cs, false)
}
flushTime = flushTime.Add(a.interval)
for time.Now().After(flushTime) {
flushTime = flushTime.Add(a.interval)
}

View File

@@ -1,3 +1,5 @@
//go:build synctest
package streamaggr
import (
@@ -483,8 +485,10 @@ foo 3.3
`, ``, ``, ``, ``}, time.Minute, `foo:1m_count_series 1
foo:1m_count_series{bar="baz"} 1
foo:1m_sum_samples 0
foo:1m_sum_samples 0
foo:1m_sum_samples 4.3
foo:1m_sum_samples{bar="baz"} 0
foo:1m_sum_samples{bar="baz"} 0
foo:1m_sum_samples{bar="baz"} 2
foo:5m_by_bar_sum_samples 4.3
foo:5m_by_bar_sum_samples{bar="baz"} 2
@@ -690,29 +694,21 @@ foo:1m_by_cde_rate_sum{cde="1"} 0.125
// test rate_sum and rate_avg, when two aggregation intervals are empty
f([]string{`
foo{abc="123", cde="1"} 1
foo{abc="123", cde="1"} 2 1
foo{abc="456", cde="1"} 7
foo{abc="456", cde="1"} 8 1
foo{abc="777", cde="1"} 8
foo{abc="777", cde="1"} 9 1
foo{abc="123", cde="1"} 2
foo{abc="456", cde="1"} 8
foo{abc="777", cde="1"} 9 -10
`, ``, ``, `
foo{abc="123", cde="1"} 19
foo{abc="123", cde="1"} 20 1
foo{abc="123", cde="1"} 20
foo{abc="456", cde="1"} 26
foo{abc="456", cde="1"} 27 1
foo{abc="777", cde="1"} 27
foo{abc="777", cde="1"} 28 1
`}, time.Minute, `foo:1m_by_cde_rate_avg{cde="1"} 1
foo:1m_by_cde_rate_avg{cde="1"} 1
foo:1m_by_cde_rate_sum{cde="1"} 3
foo:1m_by_cde_rate_sum{cde="1"} 3
foo{abc="777", cde="1"} 27 -10
`}, time.Minute, `foo:1m_by_cde_rate_avg{cde="1"} 0.1
foo:1m_by_cde_rate_sum{cde="1"} 0.2
`, `
- interval: 1m
by: [cde]
outputs: [rate_sum, rate_avg]
enable_windows: true
`, "111111111111")
`, "111111")
// rate_sum and rate_avg with duplicated events
f([]string{`

View File

@@ -74,7 +74,7 @@ func newBenchAggregators(outputs []string, pushFunc PushFunc) *Aggregators {
`, strings.Join(outputsQuoted, ","))
a, err := LoadFromData([]byte(config), pushFunc, nil, "some_alias")
if err != nil {
panic(fmt.Errorf("unexpected error when initializing aggregators: %w", err))
panic(fmt.Errorf("unexpected error when initializing aggregators: %s", err))
}
return a
}
@@ -133,7 +133,7 @@ func newPerOutputBenchAggregators(outputs []string, pushFunc PushFunc) *Aggregat
a, err := LoadFromData([]byte(config), pushFunc, nil, "some_alias")
if err != nil {
panic(fmt.Errorf("unexpected error when initializing aggregators: %w", err))
panic(fmt.Errorf("unexpected error when initializing aggregators: %s", err))
}
return a
}