Compare commits

..

16 Commits

Author SHA1 Message Date
Artem Fetishev
b6e70f5859 extend existing unit tests with cases when global index is disabled
Signed-off-by: Artem Fetishev <rtm@victoriametrics.com>
2026-07-31 14:11:50 +02:00
Artem Fetishev
4cb3f50477 ib/storage: initial implementation of disabling global index
Existing tests pass, all except TestUnitTest in app/vmalert-tool/unittest/unittest_test.go.
Some changes to existing unit and apptests tests were necessary (such as explicitly enabling global index).
Tests that verify the behavior with disabled global index will be added in subsequent commits.

Signed-off-by: Artem Fetishev <rtm@victoriametrics.com>
2026-07-31 13:59:14 +02:00
Max Kotliar
a8759a539c docs/changelog: cut release v1.149.0
Signed-off-by: Max Kotliar <mkotlyar@victoriametrics.com>
2026-07-31 13:39:51 +03:00
Max Kotliar
f32b743efe docs: update version to v1.149.0
Signed-off-by: Max Kotliar <mkotlyar@victoriametrics.com>
2026-07-31 13:39:32 +03:00
Max Kotliar
8fbf865d9e app/vmselect: run make vmui-update
Signed-off-by: Max Kotliar <mkotlyar@victoriametrics.com>
2026-07-31 13:37:07 +03:00
Max Kotliar
80b6b56028 apptest: fix flaky TestClusterSearchWithDisabledPerDayIndex (#11336)
The test used hardcoded ports for vmstorage instances, which could
already be in use by other processes or tests, causing intermittent
failures like:

    cannot create a server with -vminsertAddr=127.0.0.1:62002:
    unable to listen vminsertAddr 127.0.0.1:62002:
    listen tcp4 127.0.0.1:62002: bind: address already in use

The ports were hardcoded to ensure consistent sharding across cluster
restarts so that the same metrics land on the same storage. However, the
test only verifies query correctness when the per-day index is disabled,
not sharding behavior. Sharding is already covered by
`TestClusterVminsertShardsDataVmselectBuildsFullResultFromShards`.

Switch to a single vmstorage with dynamic ports to eliminate the
flakiness without losing test coverage.
2026-07-31 12:55:21 +03:00
Artem Fetishev
5bdcc5050e lib/storage: reserve 1970-01-01 date for global index search (#11326)
VictoriaMetrics does not support samples with negative timestamps and
limits the min timestamp to `0` (i.e. `1970-01-01T00:00:00Z`).

While samples from the first day (`1970-01-01`) are currently valid (can
be ingested and retrieved), this first day has a special meaning in
`vmstorage` and is used to indicate the search in `global index` instead
of `per-day index`.

This will stop working once the `global index` will be disabled
(https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11196).
I.e. when search is performed on `1970-01-01`, `vmstorage`
will return no results even if the samples exist.

To fix this, we reserve `1970-01-01` for internal use, and allow samples
to have timestamps starting from `1970-01-02`:
- Ingested samples with timestamps from `1970-01-01` will be rejected
and will increment the `vm_rows_ignored_total{reason="small_timestamp"}`
metric.
- Searches whose time range falls completely within the `1970-01-01`
will return empty results.

---------

Signed-off-by: Artem Fetishev <rtm@victoriametrics.com>
2026-07-31 11:19:00 +02:00
Hill Patel
e425aebbc2 lib/timeutil: accept scientific-notation timestamps with sub-second precision (#11278)
Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11268

`/api/v1/query` and `/api/v1/query_range` reject `start`/`end` values written in scientific notation whose mantissa carries more fractional digits than the exponent shifts — e.g. `start=1.784144612388E9` returns
`HTTP 422`, while the mathematically equal `start=1784144612.388` returns `200`. Prometheus accepts both (it parses the timestamp with `strconv.ParseFloat`), so this is a Prometheus-compatibility gap.

**Root cause:** in `lib/timeutil/time.go`,
`tryParseScientificNumberForUnixTimestamp` had a guard `if decimalExp <
len(fracStr) { return 0, false }`. For `1.784144612388E9` the mantissa
has 12 fractional digits and the exponent is 9, so `9 < 12` rejected it
— even though the value (`1784144612.388`) is a valid sub-second
timestamp.

**Fix:** when the exponent leaves sub-second fractional digits (`0 <=
decimalExp < len(fracStr)`), keep the mantissa (already
decimal-point-removed) and pad it up to the nearest milli/micro/nano
boundary so `getUnixTimestampNanoseconds` classifies its unit correctly
— exactly how the equivalent plain fractional timestamp is already
parsed by `TryParseUnixTimestamp`.

Kept deliberately in scope, per the discussion on the issue:
- **Negative exponents** on a fractional mantissa (e.g.
`17841446121e-1`) remain unsupported, as @JayiceZ and @valyala decided.
- Sub-second scientific values too small to be a millisecond-scale
timestamp (e.g. `1.23e1`) remain rejected as before, so **no
previously-rejected input changes behaviour** — the change is purely
additive for the reported class of large sub-second timestamps.

## Test plan

- Added success cases to `TestTryParseUnixTimestamp_Success` covering
`1.784144612388E9`, its lowercase and negative forms, and
`1.5000000005e9`, asserting they equal the corresponding
plain-fractional result.
- Verified `TestTryParseUnixTimestamp_Failure` still passes unchanged
(`1.23e1`, `1.234e0`, `1E-1`, negative-exponent forms all still
rejected).
- `go test ./lib/timeutil/`, `go vet ./lib/timeutil/`, and `gofmt` all
clean.

## Disclosure

This change was prepared with AI assistance; I have reviewed the diff,
verified the root cause against the source, and am responsible for the
change and able to explain it.

---------

Co-authored-by: Max Kotliar <mkotlyar@victoriametrics.com>
2026-07-30 21:02:33 +03:00
beyond-infra
f52771ceaf lib/streamaggr: fix sum_samples_total non-monotonic output with enable_windows (#11262)
Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11261

`sum_samples_total` produces non-monotonic and undercounted output when`enable_windows: true` is set.


When windows are enabled, blue and green window values are initialized via:
```go
nv.blue[idx]  = ac.getValue(nil)
nv.green[idx] = ac.getValue(nv.blue[idx].state())
```

`sum_samples_total`'s `state()` returned `nil` and `getValue()` ignored its argument, so each window maintained an **independent** cumulative sum. They alternated flushing under the same metric name, producing
repeated, undercounted, and non-monotonic values — e.g. `1, 1, 2, 2` instead of `1, 2, 3, 4`.

## Fix

Introduced `sumSamplesAggrValueShared`, mirroring the pattern already used in `total.go`:
- The shared struct holds the cross-window cumulative `total`
- Both windows receive a pointer to the **same** shared instance via
`state()` / `getValue(s)`
- Each flush adds the window-local `delta` into `shared.total` and outputs that value

`sum_samples` (`resetTotalOnFlush=true`) is unchanged — it has no
cumulative state and uses a plain per-window value as before.

## Test

Added a regression test in `streamaggr_synctest_test.go` that sends 4
batches of `delta=1` with `enable_windows: true` and asserts the output
is monotonically `1, 2, 3, 4`. Before the fix the test fails with `1, 1,
2, 2`.

PR https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11262

---------

Co-authored-by: Hui Wang <haley@victoriametrics.com>
Co-authored-by: Max Kotliar <mkotlyar@victoriametrics.com>
2026-07-30 20:41:12 +03:00
Max Kotliar
eeef07836e docs/changelog: chore changelog 2026-07-30 20:32:29 +03:00
面壁
f1a9c61ba0 lib/promscrape: ignore proxies for Unix socket targets
Unix socket scrape targets inherit `http.DefaultTransport.Proxy` through
`httputil.NewTransport`. When proxy environment variables are
configured, HTTP requests over Unix sockets are altered as proxy
requests, while HTTPS scrapes send `CONNECT` to the Unix socket and fail
the TLS handshake.

Unix sockets are explicit local transport endpoints and cannot be
reached through HTTP proxies. Clear the transport proxy for targets
configured with `__unix_socket__`, matching the existing rejection of
`proxy_url` for these targets.
The bug was introduced with Unix socket scraping support in v1.148.0  at https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11193

Related PR https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11318
2026-07-29 18:33:40 +02:00
f41gh7
6cb014fde5 docs/changelog: re-order entries 2026-07-29 15:42:42 +02:00
刘旭
565ecdc4fb app/vmctl: support Prometheus native histograms in remote read mode
This commit add support migrating [Prometheus native
histograms](https://prometheus.io/docs/specs/native_histograms/) in
`vmctl` remote read mode.

- `SAMPLES` mode: process `TimeSeries.Histograms` in addition to
`TimeSeries.Samples`. Previously native histogram samples were silently
ignored.
- `STREAMED_XOR_CHUNKS` mode: dispatch on the chunk encoding and decode
`HISTOGRAM` / `FLOAT_HISTOGRAM` chunks. Previously all chunks were
parsed as XOR, which failed with `EOF` error on native histogram chunks.
Unknown chunk encodings now return an explicit error. `UNKNOWN` (unset)
chunk type is parsed as XOR for compatibility with senders predating
native histograms support.

Native histograms are converted into `_count`, `_sum` and `_bucket`
series with `vmrange` labels in the same way as VictoriaMetrics converts
native histograms received via Prometheus remote write protocol
(`lib/prompb`), so the migration result matches direct remote write
ingestion.

Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11292
2026-07-29 15:23:15 +02:00
Hui Wang
06f4fde931 app/vmalert: add flag -replay.continueWithExecutionErr to allow continuing with evaluation errors
Previously, if a rule has a bad syntax, or a query fails due to a bad
expression that causes a duplicate time series on the left side of
error, or the datasource enforces a resource limit that rejects the
request, the replay process exits immediately and the user must fix the
rule before proceeding.

However, since rules are often managed by different teams, bad rules are
not always easy to fix promptly. In such cases, users may want to
continue replaying other rules even when specific rules are problematic.

 This commit adds new `-replay.continueWithExecutionErr` flag to tolerate the 422 response
code from datasource, which indicates that an expression was executed
but failed, see
https://prometheus.io/docs/prometheus/latest/querying/api/#format-overview.
Unlike https://github.com/VictoriaMetrics/VictoriaMetrics/pull/8746,
other errors, such as an unreachable datasource or 5xx responses, are
not covered by this flag, since they indicate a datasource or network
issue that is global in nature rather than specific to a given rule.

Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11313
2026-07-29 15:20:09 +02:00
Phuong Le
203eb3a2b4 README: remove retired Go Report Card badge (#11321)
Go Report Card has been sunset: https://goreportcard.com/. Code quality
checks remain covered by the existing CI workflow.
2026-07-29 15:18:34 +02:00
Solomon Wakhungu
7827647b96 app/vmalert: expose evaluation interval as template variable
This commit exposes the alert group's evaluation interval as a new `.Interval`
template variable in vmalert, making it available for use in alert
annotations, labels, and dashboard links without hardcoding.

## Problem

Previously, vmalert templates had access to variables like `.Expr`,
`.ActiveAt`, `.Labels`, `.Value`, `.For`, etc., but not the evaluation
interval. Users generating dashboard links from alert templates need to
know the interval to set the correct time range. Without it, they must
hardcode the interval value, which breaks when the interval changes.

For example, with a 1h evaluation interval, a dashboard link starting
from `.ActiveAt` shows data for the next hour instead of the hour before
the alert became active. With `.Interval`, the link can be generated
relative to the interval.

Fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11232
2026-07-29 15:13:26 +02:00
40 changed files with 1675 additions and 448 deletions

View File

@@ -62,6 +62,6 @@ jobs:
uses: github/codeql-action/autobuild@54f647b7e1bb85c95cddabcd46b0c578ec92bc1a # v4.36.3
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@99df26d4f13ea111d4ec1a7dddef6063f76b97e9 # v4.37.0
uses: github/codeql-action/analyze@54f647b7e1bb85c95cddabcd46b0c578ec92bc1a # v4.36.3
with:
category: 'language:go'

View File

@@ -2,7 +2,6 @@
[![Latest Release](https://img.shields.io/github/v/release/VictoriaMetrics/VictoriaMetrics?sort=semver&label=&filter=!*-victorialogs&logo=github&labelColor=gray&color=gray&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Freleases%2Flatest)](https://github.com/VictoriaMetrics/VictoriaMetrics/releases)
[![Docker Pulls](https://img.shields.io/docker/pulls/victoriametrics/victoria-metrics?label=&logo=docker&logoColor=white&labelColor=2496ED&color=2496ED&link=https%3A%2F%2Fhub.docker.com%2Fr%2Fvictoriametrics%2Fvictoria-metrics)](https://hub.docker.com/u/victoriametrics)
[![Go Report](https://goreportcard.com/badge/github.com/VictoriaMetrics/VictoriaMetrics?link=https%3A%2F%2Fgoreportcard.com%2Freport%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics)](https://goreportcard.com/report/github.com/VictoriaMetrics/VictoriaMetrics)
[![Build Status](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/build.yml/badge.svg?branch=master&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Factions)](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/build.yml)
[![License](https://img.shields.io/github/license/VictoriaMetrics/VictoriaMetrics?labelColor=green&label=&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Fblob%2Fmaster%2FLICENSE)](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/LICENSE)
[![Join Slack](https://img.shields.io/badge/Join%20Slack-4A154B?logo=slack)](https://slack.victoriametrics.com)

View File

@@ -11,6 +11,7 @@ import (
"strings"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
@@ -267,7 +268,11 @@ func (c *Client) do(req *http.Request) (*http.Response, error) {
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
_ = resp.Body.Close()
return nil, fmt.Errorf("unexpected response code %d for %s. Response body %s", resp.StatusCode, ru, body)
err = &httpserver.ErrorWithStatusCode{
StatusCode: resp.StatusCode,
Err: fmt.Errorf("unexpected response code %d for %s. Response body %s", resp.StatusCode, ru, body),
}
return nil, err
}
return resp, nil
}

View File

@@ -31,6 +31,8 @@ type Alert struct {
State AlertState
// Expr contains expression that was executed to generate the Alert
Expr string
// Interval contains the evaluation interval of the Alert's group
Interval time.Duration
// ActiveAt defines the moment of time when Alert has become active
ActiveAt time.Time
// Start defines the moment of time when Alert has become firing
@@ -84,6 +86,7 @@ type AlertTplData struct {
Labels map[string]string
Value float64
Expr string
Interval time.Duration
AlertID uint64
GroupID uint64
ActiveAt time.Time
@@ -96,6 +99,7 @@ var tplHeaders = []string{
"{{ $type := .Type }}",
"{{ $labels := .Labels }}",
"{{ $expr := .Expr }}",
"{{ $interval := .Interval }}",
"{{ $externalLabels := .ExternalLabels }}",
"{{ $externalURL := .ExternalURL }}",
"{{ $alertID := .AlertID }}",
@@ -115,6 +119,7 @@ func (a *Alert) ExecTemplate(q templates.QueryFn, labels, annotations map[string
Type: a.Type,
Labels: labels,
Expr: a.Expr,
Interval: a.Interval,
AlertID: a.ID,
GroupID: a.GroupID,
ActiveAt: a.ActiveAt,

View File

@@ -129,6 +129,17 @@ func TestAlertExecTemplate(t *testing.T) {
"exprEscapedHTML": "vm_rows{&quot;label&quot;=&quot;bar&quot;}&lt;0",
})
// interval-template
f(&Alert{
Interval: 10 * time.Second,
}, map[string]string{
"interval": "{{ .Interval }}",
"intervalVariable": "{{ $interval }}",
}, map[string]string{
"interval": "10s",
"intervalVariable": "10s",
})
// query
f(&Alert{
Expr: `vm_rows{"label"="bar"}>0`,

View File

@@ -25,11 +25,12 @@ var (
replayMaxDatapoints = flag.Int("replay.maxDatapointsPerQuery", 1e3,
"Max number of data points expected in one request. It affects the max time range for every '/query_range' request during the replay. The higher the value, the less requests will be made during replay.")
replayRuleRetryAttempts = flag.Int("replay.ruleRetryAttempts", 5,
"Defines how many retries to make before giving up on rule if request for it returns an error.")
"Defines how many retries to make before giving up on rule if request for it returns a retriable error.")
disableProgressBar = flag.Bool("replay.disableProgressBar", false, "Whether to disable rendering progress bars during the replay. "+
"Progress bar rendering might be verbose or break the logs parsing, so it is recommended to be disabled when not used in interactive mode.")
ruleEvaluationConcurrency = flag.Int("replay.ruleEvaluationConcurrency", 1, "The maximum number of concurrent '/query_range' requests when replay recording rule or alerting rule with for=0. "+
"Increasing this value when replaying for a long time, since each request is limited by -replay.maxDatapointsPerQuery.")
continueWithExecutionErr = flag.Bool("replay.continueWithExecutionErr", false, "Whether to continue replaying other rules if a rule execution fails with a 422 response code, which can happen due to an expression syntax error or a resource limit being hit.")
)
func replay(groupsCfg []config.Group, qb datasource.QuerierBuilder, rw remotewrite.RWClient) (totalRows, droppedRows int, err error) {
@@ -73,7 +74,7 @@ func replay(groupsCfg []config.Group, qb datasource.QuerierBuilder, rw remotewri
for _, cfg := range groupsCfg {
ng := rule.NewGroup(cfg, qb, *evaluationInterval, labels)
totalRows += ng.Replay(tFrom, tTo, rw, *replayMaxDatapoints, *replayRuleRetryAttempts, *replayRulesDelay, *disableProgressBar, *ruleEvaluationConcurrency)
totalRows += ng.Replay(tFrom, tTo, rw, *replayMaxDatapoints, *replayRuleRetryAttempts, *replayRulesDelay, *disableProgressBar, *ruleEvaluationConcurrency, *continueWithExecutionErr)
}
logger.Infof("replay evaluation finished, generated %d samples", totalRows)
if err := rw.Close(); err != nil {

View File

@@ -8,8 +8,10 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
"github.com/VictoriaMetrics/metricsql"
)
type fakeReplayQuerier struct {
@@ -32,6 +34,14 @@ func (fc *fakeRWClient) Close() error {
}
func (fr *fakeReplayQuerier) QueryRange(_ context.Context, q string, from, to time.Time) (res datasource.Result, err error) {
_, err = metricsql.Parse(q)
if err != nil {
return res, &httpserver.ErrorWithStatusCode{
StatusCode: 422,
Err: err,
}
}
key := fmt.Sprintf("%s+%s", from.Format("15:04:05"), to.Format("15:04:05"))
dps, ok := fr.registry[q]
if !ok {
@@ -275,4 +285,28 @@ func TestReplay(t *testing.T) {
},
},
}, 10)
// rule with wrong expression won't break the other rule with continueWithExecutionErr
continueWithExecutionErrOld := *continueWithExecutionErr
defer func() {
*continueWithExecutionErr = continueWithExecutionErrOld
}()
*continueWithExecutionErr = true
f("2021-01-01T12:00:00.000Z", "2021-01-01T12:02:30.000Z", 1, 1, time.Millisecond, []config.Group{
{Rules: []config.Rule{{Record: "foo", Expr: "sum(up)"}}},
{Rules: []config.Rule{{Record: "bar", Expr: "up ++"}}},
}, &fakeReplayQuerier{
registry: map[string]map[string][]datasource.Metric{
"sum(up)": {
"12:00:00+12:01:00": {
{
Timestamps: []int64{1, 2},
Values: []float64{1, 2},
},
},
"12:01:00+12:02:00": {},
"12:02:00+12:02:30": {},
},
},
}, 2)
}

View File

@@ -530,6 +530,7 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
ar.logDebugf(ts, a, "INACTIVE => PENDING")
}
a.Value = m.Values[0]
a.Interval = ar.EvalInterval
a.Annotations = annotations
a.KeepFiringSince = time.Time{}
continue
@@ -612,6 +613,7 @@ func (ar *AlertingRule) expandAnnotationTemplates(m datasource.Metric, qFn templ
Type: ar.Type.String(),
Labels: ls.origin,
Expr: ar.Expr,
Interval: ar.EvalInterval,
AlertID: hash(ls.processed),
GroupID: ar.GroupID,
ActiveAt: activeAt,
@@ -673,6 +675,7 @@ func (ar *AlertingRule) newAlert(m datasource.Metric, start time.Time, labels, a
Name: ar.Name,
Type: ar.Type.String(),
Expr: ar.Expr,
Interval: ar.EvalInterval,
For: ar.For,
ActiveAt: start,
Value: m.Values[0],

View File

@@ -662,6 +662,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
GroupID: fakeGroup.GetID(),
Name: "for-pending",
Type: config.NewPrometheusType().String(),
Interval: time.Second,
Labels: map[string]string{"alertname": "for-pending"},
Annotations: map[string]string{},
State: notifier.StatePending,
@@ -682,6 +683,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
GroupID: fakeGroup.GetID(),
Name: "for-firing",
Type: config.NewPrometheusType().String(),
Interval: 3 * time.Second,
Labels: map[string]string{"alertname": "for-firing"},
Annotations: map[string]string{},
State: notifier.StateFiring,
@@ -703,6 +705,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
GroupID: fakeGroup.GetID(),
Name: "for-hold-pending",
Type: config.NewPrometheusType().String(),
Interval: time.Second,
Labels: map[string]string{"alertname": "for-hold-pending"},
Annotations: map[string]string{},
State: notifier.StatePending,
@@ -759,6 +762,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
GroupID: fakeGroup.GetID(),
Name: "multi-series",
Type: config.NewPrometheusType().String(),
Interval: 3 * time.Second,
Labels: map[string]string{"alertname": "multi-series"},
Annotations: map[string]string{},
State: notifier.StateFiring,
@@ -771,6 +775,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
GroupID: fakeGroup.GetID(),
Name: "multi-series",
Type: config.NewPrometheusType().String(),
Interval: 3 * time.Second,
Labels: map[string]string{"alertname": "multi-series", "foo": "bar"},
Annotations: map[string]string{},
State: notifier.StatePending,
@@ -1134,7 +1139,8 @@ func TestAlertingRule_Template(t *testing.T) {
fq.Add(metrics...)
fq.SetPartialResponse(isResponsePartial)
if _, err := rule.exec(context.TODO(), time.Now(), 0); err != nil {
ts := time.Unix(3600, 0)
if _, err := rule.exec(context.TODO(), ts, 0); err != nil {
t.Fatalf("unexpected error: %s", err)
}
for hash, expAlert := range alertsExpected {
@@ -1152,12 +1158,14 @@ func TestAlertingRule_Template(t *testing.T) {
}
f(&AlertingRule{
Name: "common",
Name: "common",
EvalInterval: time.Hour,
Labels: map[string]string{
"region": "east",
},
Annotations: map[string]string{
"summary": `{{ $labels.alertname }}: Too high connection number for "{{ $labels.instance }}"`,
"summary": `{{ $labels.alertname }}: Too high connection number for "{{ $labels.instance }}"`,
"dashboard": `&from={{ ($activeAt.Add (parseDurationTime (printf "-%s" .Interval))).UnixMilli }}&to={{ $activeAt.UnixMilli }}`,
},
alerts: make(map[uint64]*notifier.Alert),
}, []datasource.Metric{
@@ -1166,7 +1174,8 @@ func TestAlertingRule_Template(t *testing.T) {
}, false, map[uint64]*notifier.Alert{
hash(map[string]string{alertNameLabel: "common", "region": "east", "instance": "foo"}): {
Annotations: map[string]string{
"summary": `common: Too high connection number for "foo"`,
"summary": `common: Too high connection number for "foo"`,
"dashboard": "&from=0&to=3600000",
},
Labels: map[string]string{
alertNameLabel: "common",
@@ -1176,7 +1185,8 @@ func TestAlertingRule_Template(t *testing.T) {
},
hash(map[string]string{alertNameLabel: "common", "region": "east", "instance": "bar"}): {
Annotations: map[string]string{
"summary": `common: Too high connection number for "bar"`,
"summary": `common: Too high connection number for "bar"`,
"dashboard": "&from=0&to=3600000",
},
Labels: map[string]string{
alertNameLabel: "common",
@@ -1388,7 +1398,7 @@ func TestAlertingRule_ToLabels(t *testing.T) {
"alertname": "ConfigurationReloadFailure",
"alertgroup": "vmalert",
"pod": "vmalert-0",
"invalid_label": `error evaluating template: template: :1:298: executing "" at <.Values.mustRuntimeFail>: can't evaluate field Values in type notifier.tplData`,
"invalid_label": `error evaluating template: template: :1:326: executing "" at <.Values.mustRuntimeFail>: can't evaluate field Values in type notifier.tplData`,
}
expectedProcessedLabels := map[string]string{
@@ -1398,7 +1408,7 @@ func TestAlertingRule_ToLabels(t *testing.T) {
"exported_alertname": "ConfigurationReloadFailure",
"group": "vmalert",
"alertgroup": "vmalert",
"invalid_label": `error evaluating template: template: :1:298: executing "" at <.Values.mustRuntimeFail>: can't evaluate field Values in type notifier.tplData`,
"invalid_label": `error evaluating template: template: :1:326: executing "" at <.Values.mustRuntimeFail>: can't evaluate field Values in type notifier.tplData`,
}
ls, err := ar.toLabels(metric, nil)

View File

@@ -548,7 +548,7 @@ func (g *Group) infof(format string, args ...any) {
}
// Replay performs group replay
func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoint, replayRuleRetryAttempts int, replayDelay time.Duration, disableProgressBar bool, ruleEvaluationConcurrency int) int {
func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoint, replayRuleRetryAttempts int, replayDelay time.Duration, disableProgressBar bool, ruleEvaluationConcurrency int, continueWithExecutionErr bool) int {
var total int
step := g.Interval * time.Duration(maxDataPoint)
ri := rangeIterator{start: start, end: end, step: step}
@@ -576,7 +576,7 @@ func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoi
if !disableProgressBar {
bar = pb.StartNew(iterations)
}
total += replayRuleRange(rule, ri, bar, rw, replayRuleRetryAttempts, ruleEvaluationConcurrency)
total += replayRuleRange(rule, ri, bar, rw, replayRuleRetryAttempts, ruleEvaluationConcurrency, continueWithExecutionErr)
if bar != nil {
bar.Finish()
}
@@ -598,7 +598,7 @@ func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoi
rule := g.Rules[i]
sem <- struct{}{}
wg.Go(func() {
res <- replayRuleRange(rule, ri, bar, rw, replayRuleRetryAttempts, ruleEvaluationConcurrency)
res <- replayRuleRange(rule, ri, bar, rw, replayRuleRetryAttempts, ruleEvaluationConcurrency, continueWithExecutionErr)
<-sem
})
}
@@ -618,7 +618,7 @@ func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoi
return total
}
func replayRuleRange(r Rule, ri rangeIterator, bar *pb.ProgressBar, rw remotewrite.RWClient, replayRuleRetryAttempts, ruleEvaluationConcurrency int) int {
func replayRuleRange(r Rule, ri rangeIterator, bar *pb.ProgressBar, rw remotewrite.RWClient, replayRuleRetryAttempts, ruleEvaluationConcurrency int, continueWithExecutionErr bool) int {
fmt.Printf("> Rule %q (ID: %d)\n", r, r.ID())
// alerting rule with for>0 can't be replayed concurrently, since the status change might depend on the previous evaluation
// see https://github.com/VictoriaMetrics/VictoriaMetrics/commit/abcb21aa5ee918ba9a4e9cde495dba06e1e9564c
@@ -633,7 +633,7 @@ func replayRuleRange(r Rule, ri rangeIterator, bar *pb.ProgressBar, rw remotewri
start := ri.s
end := ri.e
wg.Go(func() {
n, err := replayRule(r, start, end, rw, replayRuleRetryAttempts)
n, err := replayRule(r, start, end, rw, replayRuleRetryAttempts, continueWithExecutionErr)
if err != nil {
logger.Fatalf("rule %q: %s", r, err)
}

View File

@@ -4,12 +4,14 @@ import (
"context"
"errors"
"fmt"
"net/http"
"sync"
"time"
"github.com/VictoriaMetrics/metrics"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
)
@@ -118,7 +120,7 @@ func (s *ruleState) add(e StateEntry) {
s.entries[s.cur] = e
}
func replayRule(r Rule, start, end time.Time, rw remotewrite.RWClient, replayRuleRetryAttempts int) (int, error) {
func replayRule(r Rule, start, end time.Time, rw remotewrite.RWClient, replayRuleRetryAttempts int, continueWithExecutionErr bool) (int, error) {
var err error
var tss []prompb.TimeSeries
for i := range replayRuleRetryAttempts {
@@ -126,6 +128,21 @@ func replayRule(r Rule, start, end time.Time, rw remotewrite.RWClient, replayRul
if err == nil {
break
}
// retry request if possible to tolerate temporary network or datasource unavailability issues
var esc *httpserver.ErrorWithStatusCode
if errors.As(err, &esc) {
statusCode := esc.StatusCode
// if the status code is 422, it means that the query was executed but failed due to an expression syntax error or a the resource limit being hit,
// continue replaying but skip the problematic execution if continueWithExecutionErr is true, otherwise, return the error without retry.
if statusCode == http.StatusUnprocessableEntity {
if continueWithExecutionErr {
logger.Errorf("rule %q: %s", r, err)
return 0, nil
} else {
return 0, err
}
}
}
logger.Errorf("attempt %d to execute rule %q failed: %s", i+1, r, err)
time.Sleep(time.Second)
}

View File

@@ -8,12 +8,14 @@ import (
"io"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"github.com/gogo/protobuf/proto"
"github.com/golang/snappy"
"github.com/prometheus/prometheus/config"
"github.com/prometheus/prometheus/model/histogram"
"github.com/prometheus/prometheus/prompb"
"github.com/prometheus/prometheus/storage/remote"
"github.com/prometheus/prometheus/tsdb/chunkenc"
@@ -234,9 +236,29 @@ func processResponse(body io.ReadCloser, callback StreamCallback) error {
// shouldn't be accounted as an error.
for _, res := range readResp.Results {
for _, ts := range res.Timeseries {
vmTs := convertSamples(ts.Samples, ts.Labels)
if err := callback(vmTs); err != nil {
return err
// A series contains either float samples or native histogram samples.
// Both fields are processed independently, since a series may switch
// from float to native histogram representation at some point in time,
// so the requested time range may contain samples of both types.
if len(ts.Samples) > 0 {
vmTs := convertSamples(ts.Samples, ts.Labels)
if err := callback(vmTs); err != nil {
return err
}
}
if len(ts.Histograms) > 0 {
hSamples := make([]histogramSample, 0, len(ts.Histograms))
for _, h := range ts.Histograms {
hSamples = append(hSamples, histogramSample{
timestamp: h.Timestamp,
fh: h.ToFloatHistogram(),
})
}
for _, vmTs := range convertHistograms(hSamples, ts.Labels) {
if err := callback(vmTs); err != nil {
return err
}
}
}
}
}
@@ -263,17 +285,45 @@ func processStreamResponse(body io.ReadCloser, callback StreamCallback) error {
for _, series := range res.ChunkedSeries {
samples := make([]prompb.Sample, 0)
var hSamples []histogramSample
for _, chunk := range series.Chunks {
s, err := parseSamples(chunk.Data)
if err != nil {
return err
switch chunk.Type {
case prompb.Chunk_XOR, prompb.Chunk_UNKNOWN:
// In proto3 the `type` field may be left unset (UNKNOWN) for XOR chunks.
// Prometheus remote.proto: "REQUIREMENT: when using proto3, this field
// MUST be set when using anything else than XOR". Senders before native
// histograms support (Prometheus < 2.40) do not set this field at all,
// so UNKNOWN chunks must be parsed as XOR ones.
s, err := parseSamples(chunk.Data)
if err != nil {
return err
}
samples = append(samples, s...)
case prompb.Chunk_HISTOGRAM, prompb.Chunk_FLOAT_HISTOGRAM:
hs, err := parseHistograms(chunk.Type, chunk.Data)
if err != nil {
return err
}
hSamples = append(hSamples, hs...)
default:
return fmt.Errorf("unsupported chunk encoding %q", chunk.Type)
}
samples = append(samples, s...)
}
ts := convertSamples(samples, series.Labels)
if err := callback(ts); err != nil {
return err
// A series contains either XOR chunks or native histogram chunks.
// Both are processed independently, since a series may switch
// from float to native histogram representation at some point in time,
// so the requested time range may contain chunks of both types.
if len(samples) > 0 {
ts := convertSamples(samples, series.Labels)
if err := callback(ts); err != nil {
return err
}
}
for _, ts := range convertHistograms(hSamples, series.Labels) {
if err := callback(ts); err != nil {
return err
}
}
}
}
@@ -312,6 +362,151 @@ func parseSamples(chunk []byte) ([]prompb.Sample, error) {
return samples, it.Err()
}
// histogramSample represents a single native histogram sample.
type histogramSample struct {
timestamp int64
fh *histogram.FloatHistogram
}
func parseHistograms(encoding prompb.Chunk_Encoding, chunk []byte) ([]histogramSample, error) {
var enc chunkenc.Encoding
switch encoding {
case prompb.Chunk_HISTOGRAM:
enc = chunkenc.EncHistogram
case prompb.Chunk_FLOAT_HISTOGRAM:
enc = chunkenc.EncFloatHistogram
default:
return nil, fmt.Errorf("unsupported histogram chunk encoding %q", encoding)
}
c, err := chunkenc.FromData(enc, chunk)
if err != nil {
return nil, fmt.Errorf("error read chunk: %w", err)
}
var hSamples []histogramSample
it := c.Iterator(nil)
for {
typ := it.Next()
if typ == chunkenc.ValNone {
break
}
switch typ {
case chunkenc.ValHistogram:
ts, h := it.AtHistogram(nil)
hSamples = append(hSamples, histogramSample{
timestamp: ts,
fh: h.ToFloat(nil),
})
case chunkenc.ValFloatHistogram:
ts, fh := it.AtFloatHistogram(nil)
hSamples = append(hSamples, histogramSample{
timestamp: ts,
fh: fh,
})
default:
// Skip unsupported values
continue
}
}
if err := it.Err(); err != nil {
return nil, fmt.Errorf("error iterate over chunks: %w", err)
}
return hSamples, nil
}
// convertHistograms converts native histogram samples into VictoriaMetrics histogram
// time series in the same way as VictoriaMetrics converts native histograms
// received via Prometheus remote write protocol: every native histogram sample
// is converted into `<name>_count` and `<name>_sum` series plus a set of
// `<name>_bucket` series with `vmrange` labels containing non-cumulative bucket counts.
// The only difference is that for native histograms with custom buckets (NHCB)
// bucket bounds are taken from the custom values, while the remote write protocol
// parser ignores custom values and estimates the bounds with the exponential formula.
// See https://prometheus.io/docs/specs/native_histograms/#data-model
func convertHistograms(hSamples []histogramSample, labels []prompb.Label) []*vm.TimeSeries {
if len(hSamples) == 0 {
return nil
}
labelPairs := make([]vm.LabelPair, 0, len(labels))
nameValue := ""
for _, label := range labels {
if label.Name == "__name__" {
nameValue = label.Value
continue
}
labelPairs = append(labelPairs, vm.LabelPair{Name: label.Name, Value: label.Value})
}
// the metric has no name, skip it in the same way as VictoriaMetrics does
// when it receives a native histogram without the metric name via remote write protocol.
if nameValue == "" {
return nil
}
countSeries := &vm.TimeSeries{
Name: nameValue + "_count",
LabelPairs: labelPairs,
}
sumSeries := &vm.TimeSeries{
Name: nameValue + "_sum",
LabelPairs: labelPairs,
}
bucketSeries := make(map[string]*vm.TimeSeries)
// vmranges preserves the order of bucketSeries creation
// in order to get deterministic results.
var vmranges []string
for _, hs := range hSamples {
fh := hs.fh
countSeries.Timestamps = append(countSeries.Timestamps, hs.timestamp)
countSeries.Values = append(countSeries.Values, fh.Count)
sumSeries.Timestamps = append(sumSeries.Timestamps, hs.timestamp)
sumSeries.Values = append(sumSeries.Values, fh.Sum)
it := fh.AllBucketIterator()
for it.Next() {
b := it.At()
if b.Count <= 0 {
continue
}
vmrange := formatVmrange(b.Lower, b.Upper)
s := bucketSeries[vmrange]
if s == nil {
bucketLabelPairs := make([]vm.LabelPair, len(labelPairs), len(labelPairs)+1)
copy(bucketLabelPairs, labelPairs)
bucketLabelPairs = append(bucketLabelPairs, vm.LabelPair{Name: "vmrange", Value: vmrange})
s = &vm.TimeSeries{
Name: nameValue + "_bucket",
LabelPairs: bucketLabelPairs,
}
bucketSeries[vmrange] = s
vmranges = append(vmranges, vmrange)
}
s.Timestamps = append(s.Timestamps, hs.timestamp)
s.Values = append(s.Values, b.Count)
}
}
tss := make([]*vm.TimeSeries, 0, 2+len(vmranges))
tss = append(tss, countSeries, sumSeries)
for _, vmrange := range vmranges {
tss = append(tss, bucketSeries[vmrange])
}
return tss
}
// formatVmrange formats the given bucket bounds into `vmrange` label value
// in the same way as VictoriaMetrics does for native histograms
// received via Prometheus remote write protocol.
func formatVmrange(lower, upper float64) string {
b := make([]byte, 0, 24)
b = strconv.AppendFloat(b, lower, 'e', 3, 64)
b = append(b, "..."...)
b = strconv.AppendFloat(b, upper, 'e', 3, 64)
return string(b)
}
type keyValue struct {
key string
value string

View File

@@ -0,0 +1,334 @@
package remoteread
import (
"bytes"
"io"
"reflect"
"testing"
"github.com/gogo/protobuf/proto"
"github.com/golang/snappy"
"github.com/prometheus/prometheus/model/histogram"
"github.com/prometheus/prometheus/prompb"
"github.com/prometheus/prometheus/storage/remote"
"github.com/prometheus/prometheus/tsdb/chunkenc"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
)
func testHistogram(mul int64) *histogram.Histogram {
return &histogram.Histogram{
Schema: 0,
Count: uint64(10 * mul),
Sum: 25.5 * float64(mul),
ZeroThreshold: 0.001,
ZeroCount: uint64(2 * mul),
PositiveSpans: []histogram.Span{{Offset: 0, Length: 2}},
PositiveBuckets: []int64{1 * mul, 2 * mul},
NegativeSpans: []histogram.Span{{Offset: 0, Length: 1}},
NegativeBuckets: []int64{4 * mul},
}
}
func TestConvertHistograms(t *testing.T) {
f := func(hSamples []histogramSample, labels []prompb.Label, expected []*vm.TimeSeries) {
t.Helper()
tss := convertHistograms(hSamples, labels)
if !reflect.DeepEqual(tss, expected) {
t.Fatalf("unexpected result\ngot:\n%v\nwant:\n%v", tss, expected)
}
}
// series without samples
f(nil, []prompb.Label{{Name: "__name__", Value: "foo"}}, nil)
// series without the metric name must be skipped
f([]histogramSample{
{timestamp: 1000, fh: testHistogram(1).ToFloat(nil)},
}, []prompb.Label{{Name: "job", Value: "bar"}}, nil)
// native histogram must be converted to _count, _sum and _bucket series
// in the same way as VictoriaMetrics does for Prometheus remote write protocol
labels := []prompb.Label{
{Name: "__name__", Value: "request_duration_seconds"},
{Name: "job", Value: "bar"},
}
jobLabel := []vm.LabelPair{{Name: "job", Value: "bar"}}
bucketLabels := func(vmrange string) []vm.LabelPair {
return []vm.LabelPair{
{Name: "job", Value: "bar"},
{Name: "vmrange", Value: vmrange},
}
}
f([]histogramSample{
{timestamp: 1000, fh: testHistogram(1).ToFloat(nil)},
{timestamp: 2000, fh: testHistogram(2).ToFloat(nil)},
}, labels, []*vm.TimeSeries{
{
Name: "request_duration_seconds_count",
LabelPairs: jobLabel,
Timestamps: []int64{1000, 2000},
Values: []float64{10, 20},
},
{
Name: "request_duration_seconds_sum",
LabelPairs: jobLabel,
Timestamps: []int64{1000, 2000},
Values: []float64{25.5, 51},
},
{
Name: "request_duration_seconds_bucket",
LabelPairs: bucketLabels("-1.000e+00...-5.000e-01"),
Timestamps: []int64{1000, 2000},
Values: []float64{4, 8},
},
{
Name: "request_duration_seconds_bucket",
LabelPairs: bucketLabels("-1.000e-03...1.000e-03"),
Timestamps: []int64{1000, 2000},
Values: []float64{2, 4},
},
{
Name: "request_duration_seconds_bucket",
LabelPairs: bucketLabels("5.000e-01...1.000e+00"),
Timestamps: []int64{1000, 2000},
Values: []float64{1, 2},
},
{
Name: "request_duration_seconds_bucket",
LabelPairs: bucketLabels("1.000e+00...2.000e+00"),
Timestamps: []int64{1000, 2000},
Values: []float64{3, 6},
},
})
}
func TestParseHistograms(t *testing.T) {
c := chunkenc.NewHistogramChunk()
app, err := c.Appender()
if err != nil {
t.Fatalf("cannot create chunk appender: %s", err)
}
if _, _, _, err := app.AppendHistogram(nil, 0, 1000, testHistogram(1), true); err != nil {
t.Fatalf("cannot append histogram: %s", err)
}
if _, _, _, err := app.AppendHistogram(nil, 0, 2000, testHistogram(2), true); err != nil {
t.Fatalf("cannot append histogram: %s", err)
}
hSamples, err := parseHistograms(prompb.Chunk_HISTOGRAM, c.Bytes())
if err != nil {
t.Fatalf("cannot parse histogram chunk: %s", err)
}
if len(hSamples) != 2 {
t.Fatalf("unexpected number of histogram samples; got %d; want 2", len(hSamples))
}
for i, expected := range []struct {
timestamp int64
count float64
sum float64
}{
{timestamp: 1000, count: 10, sum: 25.5},
{timestamp: 2000, count: 20, sum: 51},
} {
if hSamples[i].timestamp != expected.timestamp {
t.Fatalf("unexpected timestamp; got %d; want %d", hSamples[i].timestamp, expected.timestamp)
}
if hSamples[i].fh.Count != expected.count {
t.Fatalf("unexpected count; got %f; want %f", hSamples[i].fh.Count, expected.count)
}
if hSamples[i].fh.Sum != expected.sum {
t.Fatalf("unexpected sum; got %f; want %f", hSamples[i].fh.Sum, expected.sum)
}
}
// unsupported chunk encoding must return error
if _, err := parseHistograms(prompb.Chunk_XOR, c.Bytes()); err == nil {
t.Fatalf("expecting non-nil error for unsupported chunk encoding")
}
}
func TestProcessResponse(t *testing.T) {
readResp := &prompb.ReadResponse{
Results: []*prompb.QueryResult{
{
Timeseries: []*prompb.TimeSeries{
{
Labels: []prompb.Label{
{Name: "__name__", Value: "cpu_usage"},
{Name: "job", Value: "bar"},
},
Samples: []prompb.Sample{
{Timestamp: 1000, Value: 1.5},
},
},
{
Labels: []prompb.Label{
{Name: "__name__", Value: "request_duration_seconds"},
{Name: "job", Value: "bar"},
},
Histograms: []prompb.Histogram{
prompb.FromIntHistogram(1000, testHistogram(1)),
},
},
},
},
},
}
data, err := proto.Marshal(readResp)
if err != nil {
t.Fatalf("cannot marshal ReadResponse: %s", err)
}
compressed := snappy.Encode(nil, data)
var tss []*vm.TimeSeries
err = processResponse(io.NopCloser(bytes.NewReader(compressed)), func(ts *vm.TimeSeries) error {
tss = append(tss, ts)
return nil
})
if err != nil {
t.Fatalf("cannot process response: %s", err)
}
// 1 float series + _count + _sum + 4 buckets
if len(tss) != 7 {
t.Fatalf("unexpected number of time series; got %d; want 7", len(tss))
}
if tss[0].Name != "cpu_usage" || !reflect.DeepEqual(tss[0].Values, []float64{1.5}) {
t.Fatalf("unexpected float series: %v", tss[0])
}
if tss[1].Name != "request_duration_seconds_count" || !reflect.DeepEqual(tss[1].Values, []float64{10}) {
t.Fatalf("unexpected _count series: %v", tss[1])
}
if tss[2].Name != "request_duration_seconds_sum" || !reflect.DeepEqual(tss[2].Values, []float64{25.5}) {
t.Fatalf("unexpected _sum series: %v", tss[2])
}
for _, ts := range tss[3:] {
if ts.Name != "request_duration_seconds_bucket" {
t.Fatalf("unexpected bucket series name %q", ts.Name)
}
}
}
type nopFlusher struct{}
func (nopFlusher) Flush() {}
func TestProcessStreamResponse(t *testing.T) {
// build a histogram chunk
hc := chunkenc.NewHistogramChunk()
hApp, err := hc.Appender()
if err != nil {
t.Fatalf("cannot create histogram chunk appender: %s", err)
}
if _, _, _, err := hApp.AppendHistogram(nil, 0, 1000, testHistogram(1), true); err != nil {
t.Fatalf("cannot append histogram: %s", err)
}
// build a float chunk
xc := chunkenc.NewXORChunk()
xApp, err := xc.Appender()
if err != nil {
t.Fatalf("cannot create xor chunk appender: %s", err)
}
xApp.Append(0, 1000, 1.5)
res := &prompb.ChunkedReadResponse{
ChunkedSeries: []*prompb.ChunkedSeries{
{
Labels: []prompb.Label{
{Name: "__name__", Value: "request_duration_seconds"},
{Name: "job", Value: "bar"},
},
Chunks: []prompb.Chunk{
{Type: prompb.Chunk_HISTOGRAM, Data: hc.Bytes()},
},
},
{
Labels: []prompb.Label{
{Name: "__name__", Value: "cpu_usage"},
},
Chunks: []prompb.Chunk{
{Type: prompb.Chunk_XOR, Data: xc.Bytes()},
},
},
{
Labels: []prompb.Label{
{Name: "__name__", Value: "memory_usage"},
},
Chunks: []prompb.Chunk{
// the `type` field may be unset for XOR chunks,
// such chunks must be parsed as XOR ones
{Type: prompb.Chunk_UNKNOWN, Data: xc.Bytes()},
},
},
},
}
data, err := proto.Marshal(res)
if err != nil {
t.Fatalf("cannot marshal ChunkedReadResponse: %s", err)
}
var buf bytes.Buffer
cw := remote.NewChunkedWriter(&buf, nopFlusher{})
if _, err := cw.Write(data); err != nil {
t.Fatalf("cannot write chunked response: %s", err)
}
var tss []*vm.TimeSeries
err = processStreamResponse(io.NopCloser(&buf), func(ts *vm.TimeSeries) error {
tss = append(tss, ts)
return nil
})
if err != nil {
t.Fatalf("cannot process stream response: %s", err)
}
// _count + _sum + 4 buckets + 1 float series + 1 float series from UNKNOWN chunk
if len(tss) != 8 {
t.Fatalf("unexpected number of time series; got %d; want 8", len(tss))
}
if tss[0].Name != "request_duration_seconds_count" || !reflect.DeepEqual(tss[0].Values, []float64{10}) {
t.Fatalf("unexpected _count series: %v", tss[0])
}
if tss[1].Name != "request_duration_seconds_sum" || !reflect.DeepEqual(tss[1].Values, []float64{25.5}) {
t.Fatalf("unexpected _sum series: %v", tss[1])
}
for _, ts := range tss[2:6] {
if ts.Name != "request_duration_seconds_bucket" {
t.Fatalf("unexpected bucket series name %q", ts.Name)
}
}
if tss[6].Name != "cpu_usage" || !reflect.DeepEqual(tss[6].Values, []float64{1.5}) {
t.Fatalf("unexpected float series: %v", tss[6])
}
if tss[7].Name != "memory_usage" || !reflect.DeepEqual(tss[7].Values, []float64{1.5}) {
t.Fatalf("unexpected float series from UNKNOWN chunk: %v", tss[7])
}
}
func TestParseFloatHistograms(t *testing.T) {
c := chunkenc.NewFloatHistogramChunk()
app, err := c.Appender()
if err != nil {
t.Fatalf("cannot create chunk appender: %s", err)
}
fh := testHistogram(1).ToFloat(nil)
if _, _, _, err := app.AppendFloatHistogram(nil, 0, 1000, fh, true); err != nil {
t.Fatalf("cannot append float histogram: %s", err)
}
hSamples, err := parseHistograms(prompb.Chunk_FLOAT_HISTOGRAM, c.Bytes())
if err != nil {
t.Fatalf("cannot parse float histogram chunk: %s", err)
}
if len(hSamples) != 1 {
t.Fatalf("unexpected number of histogram samples; got %d; want 1", len(hSamples))
}
if hSamples[0].timestamp != 1000 {
t.Fatalf("unexpected timestamp; got %d; want 1000", hSamples[0].timestamp)
}
if hSamples[0].fh.Count != 10 {
t.Fatalf("unexpected count; got %f; want 10", hSamples[0].fh.Count)
}
}

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@@ -37,7 +37,7 @@
<meta property="og:title" content="UI for VictoriaMetrics">
<meta property="og:url" content="https://victoriametrics.com/">
<meta property="og:description" content="Explore and troubleshoot your VictoriaMetrics data">
<script type="module" crossorigin src="./assets/index-D5egN2id.js"></script>
<script type="module" crossorigin src="./assets/index-B1dXK3k7.js"></script>
<link rel="modulepreload" crossorigin href="./assets/rolldown-runtime-CNC7AqOf.js">
<link rel="modulepreload" crossorigin href="./assets/vendor-DwJYpOdw.js">
<link rel="stylesheet" crossorigin href="./assets/vendor-CnsZ1jie.css">

View File

@@ -94,9 +94,13 @@ var (
cacheSizeIndexDBTagFilters = flagutil.NewBytes("storage.cacheSizeIndexDBTagFilters", 0, "Overrides max size for indexdb/tagFiltersToMetricIDs cache. "+
"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#cache-tuning")
enableGlobalIndex = flag.Bool("enableGlobalIndex", false, "Enable global index. "+
"Deployments with high churn rate should have this index disabled as this decreases disk space usage. "+
"Such deployments may enable global index if the dominant query time range is > 1m as it may slightly improve query performance. "+
"Also see -disablePerDayIndex and https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#index-tuning")
disablePerDayIndex = flag.Bool("disablePerDayIndex", false, "Disable per-day index and use global index for all searches. "+
"This may improve performance and decrease disk space usage for the use cases with fixed set of timeseries scattered across a "+
"big time range (for example, when loading years of historical data). "+
"This may improve performance and decrease disk space usage for deployment with no/low churn rate. "+
"Disabling per-day index forces enabling global index and the -enableGlobalIndex flag value is ignored."+
"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#index-tuning")
trackMetricNamesStats = flag.Bool("storage.trackMetricNamesStats", true, "Whether to track ingest and query requests for timeseries metric names. "+
"This feature allows to track metric names unused at query requests. "+
@@ -148,6 +152,12 @@ func Init(vmselectMaxConcurrentRequests int, vmselectMaxQueueDuration time.Durat
if *idbPrefillStart > 23*time.Hour {
logger.Panicf("-storage.idbPrefillStart cannot exceed 23 hours; got %s", idbPrefillStart)
}
disableGlobalIndex := !*enableGlobalIndex
if *disablePerDayIndex {
// In case if per-day index has been disabled, forcibly enable global
// index even if -enableGlobalIndex flag is false.
disableGlobalIndex = false
}
fs.RegisterPathFsMetrics(*storageDataPath)
logger.Infof("opening storage at %q with -retentionPeriod=%s", *storageDataPath, retentionPeriod)
startTime := time.Now()
@@ -158,6 +168,7 @@ func Init(vmselectMaxConcurrentRequests int, vmselectMaxQueueDuration time.Durat
DenyQueriesOutsideRetention: *denyQueriesOutsideRetention,
MaxHourlySeries: getMaxHourlySeries(),
MaxDailySeries: getMaxDailySeries(),
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: *disablePerDayIndex,
TrackMetricNamesStats: *trackMetricNamesStats,
IDBPrefillStart: *idbPrefillStart,

View File

@@ -16,6 +16,7 @@ func TestSingleSearchWithDisabledPerDayIndex(t *testing.T) {
"-storageDataPath=" + tc.Dir() + "/vmsingle",
"-retentionPeriod=100y",
"-search.maxStalenessInterval=1m",
"-enableGlobalIndex=true",
fmt.Sprintf("-disablePerDayIndex=%t", disablePerDayIndex),
})
})
@@ -26,33 +27,21 @@ func TestClusterSearchWithDisabledPerDayIndex(t *testing.T) {
defer tc.Stop()
testSearchWithDisabledPerDayIndex(tc, func(name string, disablePerDayIndex bool) apptest.PrometheusWriteQuerier {
// Using static ports for vmstorage because random ports may cause
// changes in how data is sharded.
vmstorage1 := tc.MustStartVmstorage("vmstorage1-"+name, []string{
"-storageDataPath=" + tc.Dir() + "/vmstorage1",
vmstorage := tc.MustStartVmstorage("vmstorage-"+name, []string{
"-storageDataPath=" + tc.Dir() + "/vmstorage",
"-retentionPeriod=100y",
"-httpListenAddr=127.0.0.1:61001",
"-vminsertAddr=127.0.0.1:61002",
"-vmselectAddr=127.0.0.1:61003",
fmt.Sprintf("-disablePerDayIndex=%t", disablePerDayIndex),
})
vmstorage2 := tc.MustStartVmstorage("vmstorage2-"+name, []string{
"-storageDataPath=" + tc.Dir() + "/vmstorage2",
"-retentionPeriod=100y",
"-httpListenAddr=127.0.0.1:62001",
"-vminsertAddr=127.0.0.1:62002",
"-vmselectAddr=127.0.0.1:62003",
"-enableGlobalIndex=true",
fmt.Sprintf("-disablePerDayIndex=%t", disablePerDayIndex),
})
vminsert := tc.MustStartVminsert("vminsert-"+name, []string{
"-storageNode=" + vmstorage1.VminsertAddr() + "," + vmstorage2.VminsertAddr(),
"-storageNode=" + vmstorage.VminsertAddr(),
})
vmselect := tc.MustStartVmselect("vmselect"+name, []string{
"-storageNode=" + vmstorage1.VmselectAddr() + "," + vmstorage2.VmselectAddr(),
"-storageNode=" + vmstorage.VmselectAddr(),
"-search.maxStalenessInterval=1m",
})
return &apptest.Vmcluster{
Vmstorages: []*apptest.Vmstorage{vmstorage1, vmstorage2},
Vmstorages: []*apptest.Vmstorage{vmstorage},
Vminsert: vminsert,
Vmselect: vmselect,
}

View File

@@ -59,7 +59,7 @@ func (ms *PrometheusMockStorage) Read(_ context.Context, query *prompb.Query, so
}
if !notMatch {
q.Timeseries = append(q.Timeseries, &prompb.TimeSeries{Labels: s.Labels, Samples: s.Samples})
q.Timeseries = append(q.Timeseries, &prompb.TimeSeries{Labels: s.Labels, Samples: s.Samples, Histograms: s.Histograms})
}
}

View File

@@ -12,6 +12,7 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/golang/snappy"
"github.com/prometheus/prometheus/model/histogram"
"github.com/prometheus/prometheus/model/labels"
"github.com/prometheus/prometheus/prompb"
"github.com/prometheus/prometheus/storage/remote"
@@ -86,10 +87,17 @@ func (rrs *RemoteReadServer) getReadHandler(t *testing.T) http.Handler {
samples = append(samples, sample)
}
}
var histograms []prompb.Histogram
for _, h := range s.Histograms {
if h.Timestamp >= startTs && h.Timestamp < endTs {
histograms = append(histograms, h)
}
}
var series prompb.TimeSeries
if len(samples) > 0 {
if len(samples) > 0 || len(histograms) > 0 {
series.Labels = s.Labels
series.Samples = samples
series.Histograms = histograms
}
ts[i] = &series
}
@@ -317,6 +325,37 @@ func generateRemoteReadSamples(idx int, startTime, endTime, numOfSamples int64)
return samples
}
// GenerateRemoteReadHistogramSeries generates a remote read series
// with native histogram samples within the given time range.
func GenerateRemoteReadHistogramSeries(start, end, numOfSamples int64) []*prompb.TimeSeries {
timeSeries := &prompb.TimeSeries{
Labels: []prompb.Label{
{Name: labels.MetricName, Value: "vm_histogram_metric"},
{Name: "job", Value: "0"},
},
}
delta := (end - start) / numOfSamples
mul := int64(0)
for t := start; t != end; t += delta {
mul++
h := &histogram.Histogram{
Schema: 0,
Count: uint64(10 * mul),
Sum: 25.5 * float64(mul),
ZeroThreshold: 0.001,
ZeroCount: uint64(2 * mul),
PositiveSpans: []histogram.Span{{Offset: 0, Length: 2}},
PositiveBuckets: []int64{mul, 2 * mul},
NegativeSpans: []histogram.Span{{Offset: 0, Length: 1}},
NegativeBuckets: []int64{4 * mul},
}
timeSeries.Histograms = append(timeSeries.Histograms, prompb.FromIntHistogram(t*1000, h))
}
return []*prompb.TimeSeries{timeSeries}
}
func labelsToLabelsProto(ls labels.Labels) []prompb.Label {
result := make([]prompb.Label, 0, ls.Len())
ls.Range(func(l labels.Label) {

View File

@@ -75,6 +75,118 @@ func TestClusterVmctlRemoteReadProtocol(t *testing.T) {
testRemoteReadProtocol(tc, clusterDst, newRemoteReadServer, vmctlFlags)
}
func TestSingleVmctlRemoteReadNativeHistograms(t *testing.T) {
fs.MustRemoveDir(t.Name())
tc := apptest.NewTestCase(t)
defer tc.Stop()
vmsingleDst := tc.MustStartDefaultVmsingle()
vmAddr := fmt.Sprintf("http://%s/", vmsingleDst.HTTPAddr())
vmctlFlags := []string{
`remote-read`,
`--remote-read-filter-time-start=2025-06-11T15:31:10Z`,
`--remote-read-filter-time-end=2025-06-11T15:31:20Z`,
`--remote-read-step-interval=minute`,
`--vm-addr=` + vmAddr,
`--disable-progress-bar=true`,
}
testRemoteReadNativeHistograms(tc, vmsingleDst, NewRemoteReadServer, vmctlFlags)
}
func TestSingleVmctlRemoteReadStreamNativeHistograms(t *testing.T) {
fs.MustRemoveDir(t.Name())
tc := apptest.NewTestCase(t)
defer tc.Stop()
vmsingleDst := tc.MustStartDefaultVmsingle()
vmAddr := fmt.Sprintf("http://%s/", vmsingleDst.HTTPAddr())
vmctlFlags := []string{
`remote-read`,
`--remote-read-filter-time-start=2025-06-11T15:31:10Z`,
`--remote-read-filter-time-end=2025-06-11T15:31:20Z`,
`--remote-read-step-interval=minute`,
`--vm-addr=` + vmAddr,
`--remote-read-use-stream=true`,
`--disable-progress-bar=true`,
}
testRemoteReadNativeHistograms(tc, vmsingleDst, NewRemoteReadStreamServer, vmctlFlags)
}
// testRemoteReadNativeHistograms verifies that native histograms are migrated
// as _count, _sum and _bucket series with vmrange labels in the same way
// as VictoriaMetrics converts native histograms received via Prometheus remote write protocol.
func testRemoteReadNativeHistograms(tc *apptest.TestCase, sut apptest.PrometheusWriteQuerier, newRemoteReadServer func(t *testing.T, series []*prompb.TimeSeries) *RemoteReadServer, vmctlFlags []string) {
t := tc.T()
t.Helper()
series := GenerateRemoteReadHistogramSeries(1749655870, 1749655880, 2)
rrs := newRemoteReadServer(t, series)
defer rrs.Close()
vmctlFlags = append(vmctlFlags, `--remote-read-src-addr=`+rrs.HTTPAddr())
tc.MustStartVmctl("vmctl", vmctlFlags)
sut.ForceFlush(t)
tc.Assert(&apptest.AssertOptions{
Retries: 300,
Msg: `unexpected native histogram metrics stored on vmsingle via the prometheus protocol`,
Got: func() any {
got := sut.PrometheusAPIV1Export(t, `{__name__=~".*"}`, apptest.QueryOpts{
Start: "2025-06-11T15:31:10Z",
End: "2025-06-11T15:32:20Z",
})
got.Sort()
return got.Data.Result
},
Want: expectedNativeHistogramQueryResult(),
CmpOpts: []cmp.Option{
cmpopts.IgnoreFields(apptest.PrometheusAPIV1QueryResponse{}, "Status", "Data.ResultType"),
},
})
}
// expectedNativeHistogramQueryResult returns the series expected to be stored in VictoriaMetrics
// after migrating the series generated by GenerateRemoteReadHistogramSeries(1749655870, 1749655880, 2).
func expectedNativeHistogramQueryResult() []*apptest.QueryResult {
metric := func(name, vmrange string) map[string]string {
m := map[string]string{
"__name__": name,
"job": "0",
}
if vmrange != "" {
m["vmrange"] = vmrange
}
return m
}
samples := func(v1, v2 float64) []*apptest.Sample {
return []*apptest.Sample{
{Timestamp: 1749655870000, Value: v1},
{Timestamp: 1749655875000, Value: v2},
}
}
resp := &apptest.PrometheusAPIV1QueryResponse{
Data: &apptest.QueryData{
Result: []*apptest.QueryResult{
{Metric: metric("vm_histogram_metric_count", ""), Samples: samples(10, 20)},
{Metric: metric("vm_histogram_metric_sum", ""), Samples: samples(25.5, 51)},
{Metric: metric("vm_histogram_metric_bucket", "-1.000e+00...-5.000e-01"), Samples: samples(4, 8)},
{Metric: metric("vm_histogram_metric_bucket", "-1.000e-03...1.000e-03"), Samples: samples(2, 4)},
{Metric: metric("vm_histogram_metric_bucket", "5.000e-01...1.000e+00"), Samples: samples(1, 2)},
{Metric: metric("vm_histogram_metric_bucket", "1.000e+00...2.000e+00"), Samples: samples(3, 6)},
},
},
}
// sort in the same way as the exported result
resp.Sort()
return resp.Data.Result
}
func testRemoteReadProtocol(tc *apptest.TestCase, sut apptest.PrometheusWriteQuerier, newRemoteReadServer func(t *testing.T) *RemoteReadServer, vmctlFlags []string) {
t := tc.T()
t.Helper()

View File

@@ -829,7 +829,7 @@ See also [minimum downtime strategy](#minimum-downtime-strategy).
## Slowness-based re-routing
By default{{% available_from "#" %}}, `vminsert` automatically re-routes writes away from the slowest `vmstorage` node
By default{{% available_from "v1.149.0" %}}, `vminsert` automatically re-routes writes away from the slowest `vmstorage` node
to preserve maximum ingestion throughput. This prevents a single slow `vmstorage` node
from throttling the entire cluster.
@@ -843,7 +843,7 @@ Disable slowness-based re-routing with `-disableRerouting=true` when keeping met
perfectly balanced across nodes or minimizing the number of [active time series](https://docs.victoriametrics.com/victoriametrics/faq/#what-is-an-active-time-series)
matters more than peak write throughput.
Slowness-based re-routing is automatically disabled{{% available_from "#" %}} when `-replicationFactor` is greater than `1`,
Slowness-based re-routing is automatically disabled{{% available_from "v1.149.0" %}} when `-replicationFactor` is greater than `1`,
because rerouting does not guarantee that replicated copies land on distinct storage nodes,
which violates the replication contract.

View File

@@ -39,6 +39,8 @@ VictoriaMetrics has the following prominent features:
* Easy and fast backups from [instant snapshots](https://medium.com/@valyala/how-victoriametrics-makes-instant-snapshots-for-multi-terabyte-time-series-data-e1f3fb0e0282)
can be done with [vmbackup](https://docs.victoriametrics.com/victoriametrics/vmbackup/) / [vmrestore](https://docs.victoriametrics.com/victoriametrics/vmrestore/) tools.
See [this article](https://medium.com/@valyala/speeding-up-backups-for-big-time-series-databases-533c1a927883) for more details.
* It supports storage and retrieval of samples with timestamps that fall within the `[1970-01-02T00:00:00.000Z, 2262-03-31T23:59:59.999Z]` time range with millisecond precision.
See [Retention](#retention) for details.
* It implements a PromQL-like query language - [MetricsQL](https://docs.victoriametrics.com/victoriametrics/metricsql/), which provides improved functionality on top of PromQL.
* It provides a global query view. Multiple Prometheus instances or any other data sources may ingest data into VictoriaMetrics. Later this data may be queried via a single query.
* It provides high performance and good vertical and horizontal scalability for both
@@ -1540,6 +1542,9 @@ It is safe to extend `-retentionPeriod` on existing data. If `-retentionPeriod`
value than before, then data outside the configured period will be eventually deleted.
VictoriaMetrics does not support indefinite retention, but you can specify an arbitrarily high duration, e.g. `-retentionPeriod=100y`.
Just keep in mind that VictoriaMetrics does not support samples with negative timestamps. Timestamps at `1970-01-01` are also not
supported because this date has a special meaning internally. It therefore rejects samples with timestamps before
`1970-01-02T00:00:00.000Z`.
By default, VictoriaMetrics doesn't accept samples with timestamps bigger than `now+2d`, e.g. 2 days in the future.
If you need accepting samples with bigger timestamps, then specify the desired "future retention" via `-futureRetention` command-line flag.
@@ -1551,6 +1556,9 @@ For example, the following command starts VictoriaMetrics, which accepts samples
/path/to/victoria-metrics -futureRetention=1y
```
VictoriaMetrics does not support stamples after `2262-03-31T23:59:59.999Z`. If the future retention includes dates after this timestamp,
the samples for those dates will be rejected.
By default, VictoriaMetrics accepts samples with timestamps as old as the configured `-retentionPeriod` allows, e.g. it accepts backfilled
historical data as long as it fits into the retention. If you need rejecting samples with historical timestamps older than the specified
duration, then specify the desired duration via the `-maxBackfillAge` command-line flag. This can be useful for limiting ingestion of

View File

@@ -26,22 +26,32 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
## tip
## [v1.149.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.149.0)
Release candidate
**Update Note 1:** `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): the default value of `-disableRerouting` flag has changed from `true` to `false`, enabling [slowness-based re-routing](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#slowness-based-re-routing) by default. Slowness re-routing is automatically disabled when `-replicationFactor` is greater than 1. If you rely on the old behavior, pass `-disableRerouting` command-line flag to `vminsert`. See [#11287](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11287).
**Update Note 2:** [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): the `/api/v1/admin/tsdb/delete_series`, `/tags/delSeries` endpoints now require `POST` method. Previously, it also accepted `GET` requests. If you use `GET` requests for this endpoint, update your scripts or tooling to use `POST` instead. See [#5552](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5552).
* SECURITY: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): restrict `/api/v1/admin/tsdb/delete_series`, `/tags/delSeries` endpoints to `POST` method only to prevent some [SSRF](https://en.wikipedia.org/wiki/Server-side_request_forgery)-based data deletion attacks. See [#5552](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5552).
* FEATURE: [dashboards](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/dashboards): add `Fsync avg duration` panel to the Troubleshooting section of the single-node, cluster, and vmagent dashboards. This panel surfaces degradation of IO operation for faster incident triage. See [#10432](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10432).
* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): add `name` label identifying the corresponding `-remoteWrite.url` target to the `vm_persistentqueue_*` metrics exposed by persistent queue. See [#7944](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7944). Thanks to @tIGO for contribution.
* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): add `-remoteWrite.obfuscateLabels` flag for hashing values of the specified labels before sending metrics to the corresponding `-remoteWrite.url`. This allows sharing metrics with external systems while keeping sensitive label values hidden. See [#10599](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10599).
* FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): add `-replay.continueWithExecutionErr` flag to allow continuing to replay other rules when a rule execution fails with a 422 response code, which can happen due to an expression syntax error or a resource limit being hit. See [11313](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11313).
* FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): add template variable `$interval` to expose the alerting rule group's evaluation interval. This allows generating dashboard links with a lookback window relative to the rule's interval, for example `&from={{ ($activeAt.Add (parseDurationTime (printf "-%s" .Interval))).UnixMilli }}&to={{ $activeAt.UnixMilli }}`. See [#11232](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11232). Thanks to @1solomonwakhungu for contribution.
* FEATURE: [vmbackupmanager](https://docs.victoriametrics.com/victoriametrics/vmbackupmanager/) and [alerts](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules): introduce `vm_backup_last_success_at` metric to track the last successful backup by type. Add [alerting rules](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules/alerts-vmbackupmanager.yml) `NoLatestBackupWithinLastDay`, `NoHourlyBackupWithinLastDay`, `NoDailyBackupWithinLast3Days`, `NoWeeklyBackupWithinLast14Days` and `NoMonthlyBackupWithinLast62Days` to remind users about the missing backups. See [#11217](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11217).
* FEATURE: [vmctl](https://docs.victoriametrics.com/victoriametrics/vmctl/): support [Prometheus native histograms](https://prometheus.io/docs/specs/native_histograms/) migration in [remote read mode](https://docs.victoriametrics.com/victoriametrics/vmctl/remoteread/). Native histograms are converted into `_count`, `_sum` and `_bucket` series with `vmrange` labels in the same way as VictoriaMetrics [converts native histograms received via Prometheus remote write protocol](https://docs.victoriametrics.com/victoriametrics/integrations/prometheus/#native-histograms), except that for native histograms with custom buckets the original bucket bounds are preserved instead of being estimated with the exponential formula. Previously native histograms were silently ignored in `SAMPLES` mode, while in stream mode the migration failed with `EOF` error. See [#11292](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11292). Thanks to @liuxu623 for contribution.
* FEATURE: `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): enable [slowness-based re-routing](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#slowness-based-re-routing) by default. Previously, `-disableRerouting` defaulted to `true`, which limited ingestion throughput to the slowest `vmstorage` node. Now `-disableRerouting` defaults to `false`, so `vminsert` automatically routes data away from the slowest `vmstorage` node, improving overall ingestion performance. Slowness re-routing is automatically disabled when `-replicationFactor` is greater than 1. See [#11287](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11287).
* FEATURE: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): persist the selected auto-refresh interval in the URL. See [VictoriaLogs#1310](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1310).
* FEATURE: [dashboards](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/dashboards): add `Fsync avg duration` panel to the Troubleshooting section of the single-node, cluster, and vmagent dashboards. This panel surfaces degradation of IO operation for faster incident triage. See [#10432](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10432).
* FEATURE: [vmbackupmanager](https://docs.victoriametrics.com/victoriametrics/vmbackupmanager/) and [alerts](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules): introduce `vm_backup_last_success_at` metric to track the last successful backup by type. Add [alerting rules](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules/alerts-vmbackupmanager.yml) `NoLatestBackupWithinLastDay`, `NoHourlyBackupWithinLastDay`, `NoDailyBackupWithinLast3Days`, `NoWeeklyBackupWithinLast14Days` and `NoMonthlyBackupWithinLast62Days` to remind users about the missing backups. See [#11217](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11217).
* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): add `-remoteWrite.obfuscateLabels` flag for hashing values of the specified labels before sending metrics to the corresponding `-remoteWrite.url`. This allows sharing metrics with external systems while keeping sensitive label values hidden. See [#10599](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10599).
* BUGFIX: [MetricsQL](https://docs.victoriametrics.com/victoriametrics/metricsql/): properly drop data points filtered out by an inner [comparison operation](https://prometheus.io/docs/prometheus/latest/querying/operators/#comparison-binary-operators) when its result is used on the right side of another comparison. Previously, queries like `foo != (bar > 100)` could return unexpected results because filtered-out data points are represented internally as `NaN`, and `value != NaN` evaluates to `true`. Comparisons against explicitly present `NaN` values keep the previous behavior. See [#10018](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10018). Thanks to @zasdaym for contribution.
* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): ignore HTTP proxy environment variables when scraping targets over Unix domain sockets. See [#11318](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11318). Thanks to @lwmacct for contribution.
* BUGFIX: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): fixed the display of rule state badges on the `Groups` page in the web UI. See [#11160](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11160).
* BUGFIX: [vmbackupmanager](https://docs.victoriametrics.com/victoriametrics/vmbackupmanager/): previously, `vmbackupmanager` was crashing on startup when it failed to restore backup state from remote storage, causing a crash loop. Now it logs the error and continues running, retrying the state restore before each scheduled backup. Added `vm_backup_errors_total{type="restoreState"}` metric to track backup state restore failures. See [#11217](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11217).
* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): fix incorrect [sum_samples_total](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/configuration/#sum_samples_total) results when `enable_windows: true` is set. See [#11261](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11261). Thanks to @beyond-infra for contribution.
* BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/), `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) and [vmctl](https://docs.victoriametrics.com/victoriametrics/vmctl/): accept scientific notation with sub-second precision (e.g. `1.784144612388E9`) for timestamp args such as `start` and `end` in `/api/v1/query_range` and `--vm-native-filter-time-start` and `--vm-native-filter-time-end` in `vmctl`. Previously, values with this pattern were rejected, which is incompatible with Prometheus. See [#11268](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11268). Thanks to @STiFLeR7 for contribution.
## [v1.148.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.148.0)

View File

@@ -688,7 +688,7 @@ Extra labels can be added to metrics collected by `vmagent` via the following me
## Obfuscating label values
`vmagent` can obfuscate the values of specified labels before sending metrics to `-remoteWrite.url`
via `-remoteWrite.obfuscateLabels`{{% available_from "#" %}}.
via `-remoteWrite.obfuscateLabels`{{% available_from "v1.149.0" %}}.
This is useful when one or more `-remoteWrite.url` endpoints point to external monitoring services
outside the organization, and sensitive label values such as `ip`, `host`, `instance`, or `datacenter`

View File

@@ -341,6 +341,7 @@ The following variables are available in templating:
| $externalLabels or .ExternalLabels | List of labels configured via `-external.label` command-line flag. | `Issues with {{ $labels.instance }} (datacenter-{{ $externalLabels.dc }})` |
| $externalURL or .ExternalURL | URL configured via `-external.url` command-line flag. Used for cases when vmalert is hidden behind proxy. | `Visit {{ $externalURL }} for more details` |
| $isPartial or .IsPartial | Indicates whether the latest rule query response from the datasource(that supports returning `isPartial` option, such as vmcluster) could be partial. | `{{ if $isPartial }}WARNING: The latest alert state may be a false alarm due to a partial response from the datasource.{{ end }}` |
| $interval or .Interval | Alerting rule group's evaluation interval. | `http://vm-grafana.com/<dashboard-id>?viewPanel=<panel-id>&from={{ ($activeAt.Add (parseDurationTime (printf "-%s" .Interval))).UnixMilli }}&to={{ $activeAt.UnixMilli }}` |
Additionally, `vmalert` provides some extra templating functions listed in [template functions](#template-functions) and [reusable templates](#reusable-templates).

View File

@@ -404,6 +404,8 @@ See the docs at https://docs.victoriametrics.com/victoriametrics/vmalert/ .
Optional TLS server name to use for connections to -remoteWrite.url. By default, the server name from -remoteWrite.url is used
-remoteWrite.url string
Optional URL to persist alerts state and recording rules results in form of timeseries. It must support either VictoriaMetrics remote write protocol or Prometheus remote_write protocol. Supports address in the form of IP address with a port (e.g., http://127.0.0.1:8428) or DNS SRV record. For example, if -remoteWrite.url=http://127.0.0.1:8428 is specified, then the alerts state will be written to http://127.0.0.1:8428/api/v1/write . See also -remoteWrite.disablePathAppend, '-remoteWrite.showURL'.
-replay.continueWithExecutionErr bool
Whether to continue replaying other rules if a rule execution fails with a 422 response code, which can happen due to an expression syntax error or a resource limit being hit.
-replay.disableProgressBar
Whether to disable rendering progress bars during the replay. Progress bar rendering might be verbose or break the logs parsing, so it is recommended to be disabled when not used in interactive mode.
-replay.maxDatapointsPerQuery int
@@ -411,7 +413,7 @@ See the docs at https://docs.victoriametrics.com/victoriametrics/vmalert/ .
-replay.ruleEvaluationConcurrency int
The maximum number of concurrent '/query_range' requests when replay recording rule or alerting rule with for=0. Increasing this value when replaying for a long time, since each request is limited by -replay.maxDatapointsPerQuery. (default 1)
-replay.ruleRetryAttempts int
Defines how many retries to make before giving up on rule if request for it returns an error. (default 5)
Defines how many retries to make before giving up on rule if request for it returns a retriable error. (default 5)
-replay.rulesDelay duration
Delay before evaluating the next rule within the group. Is important for chained rules. Keep it equal or bigger than -remoteWrite.flushInterval. When set to >0, replay ignores group's concurrency setting. (default 1s)
-replay.timeFrom string

View File

@@ -480,7 +480,7 @@ Clusters here are referred to as `source` and `destination`.
To verify that `vmbackupmanager` is executing backup tasks normally, the following metrics can help:
* `vm_backup_last_success_at{type="<backup_type>"}` - unix timestamp of the last successful backup{{% available_from "#" %}}. Remains `0` if no backup has completed successfully since startup. Check error logs and verify remote storage accessibility if this persists.
* `vm_backup_last_success_at{type="<backup_type>"}` - unix timestamp of the last successful backup{{% available_from "v1.149.0" %}}. Remains `0` if no backup has completed successfully since startup. Check error logs and verify remote storage accessibility if this persists.
* `vm_backup_last_run_failed{type="<backup_type>"}` - whether the last backup task for the given backup type failed. The value `1` means the last task failed. Check the error logs of `vmbackupmanager` for the root cause
* `vm_backup_errors_total{type="<backup_type>"}` - total number of backup errors for the given backup type.

View File

@@ -90,7 +90,11 @@ func newClient(ctx context.Context, sw *ScrapeWork) (*client, error) {
}
tr := httputil.NewTransport(false, "vm_promscrape")
if proxyURLFunc != nil {
if sw.UnixSocket != "" {
// Unix sockets are direct local endpoints and cannot be reached via HTTP proxies.
// Proxy could be set implicitly via global env variable HTTP_PROXY
tr.Proxy = nil
} else if proxyURLFunc != nil {
tr.Proxy = proxyURLFunc
}
tr.TLSHandshakeTimeout = 10 * time.Second

View File

@@ -436,18 +436,6 @@ func (db *indexDB) createGlobalIndexes(tsid *TSID, mn *MetricName) {
ii := getIndexItems()
defer putIndexItems(ii)
if db.s.disablePerDayIndex {
// Create metricName -> TSID entry.
// This index is used for searching a TSID by metric name during data
// ingestion or metric name registration when -disablePerDayIndex flag
// is set.
ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricNameToTSID)
ii.B = mn.Marshal(ii.B)
ii.B = append(ii.B, kvSeparatorChar)
ii.B = tsid.Marshal(ii.B)
ii.Next()
}
// Create metricID -> metricName entry.
ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricIDToMetricName)
ii.B = encoding.MarshalUint64(ii.B, tsid.MetricID)
@@ -460,11 +448,20 @@ func (db *indexDB) createGlobalIndexes(tsid *TSID, mn *MetricName) {
ii.B = tsid.Marshal(ii.B)
ii.Next()
// Create tag -> metricID entries for every tag in mn.
kb := kbPool.Get()
kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
ii.registerTagIndexes(kb.B, mn, tsid.MetricID)
kbPool.Put(kb)
if !db.s.disableGlobalIndex {
// Create metricName -> TSID entry.
ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricNameToTSID)
ii.B = mn.Marshal(ii.B)
ii.B = append(ii.B, kvSeparatorChar)
ii.B = tsid.Marshal(ii.B)
ii.Next()
// Create tag -> metricID entries for every tag in mn.
kb := kbPool.Get()
kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
ii.registerTagIndexes(kb.B, mn, tsid.MetricID)
kbPool.Put(kb)
}
db.tb.AddItems(ii.Items)
}
@@ -759,6 +756,7 @@ func (db *indexDB) SearchLabelValues(qt *querytracer.Tracer, labelName string, t
func filterLabelValues(lvs map[string]struct{}, tf *tagFilter, key string) {
var b []byte
for lv := range lvs {
// TODO
b = marshalCommonPrefix(b[:0], nsPrefixTagToMetricIDs)
b = marshalTagValue(b, bytesutil.ToUnsafeBytes(key))
b = marshalTagValue(b, bytesutil.ToUnsafeBytes(lv))
@@ -1517,10 +1515,11 @@ func (db *indexDB) DeleteSeries(qt *querytracer.Tracer, tfss []*TagFilters, maxM
is := db.getIndexSearch(noDeadline)
defer db.putIndexSearch(is)
// Unconditionally search global index since a given day in per-day
// index may not contain the full set of metricIDs that correspond
// to the tfss.
metricIDs, err := is.searchMetricIDs(qt, tfss, globalIndexTimeRange, maxMetrics)
tr := globalIndexTimeRange
if db.s.disableGlobalIndex {
tr = db.tr
}
metricIDs, err := is.searchMetricIDs(qt, tfss, tr, maxMetrics)
if err != nil {
return nil, db.wrapError("delete series", err)
}
@@ -1967,6 +1966,7 @@ func (is *indexSearch) updateMetricIDsByMetricNameMatch(qt *querytracer.Tracer,
qt.Printf("sort %d metric ids", len(sortedMetricIDs))
kb := &is.kb
// TODO
kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
tfs = removeCompositeTagFilters(tfs, kb.B)
@@ -2077,6 +2077,7 @@ func hasCompositeTagFilters(tfs []*tagFilter, prefix []byte) bool {
}
func matchTagFilters(mn *MetricName, tfs []*tagFilter, kb *bytesutil.ByteBuffer) (bool, error) {
// TODO
kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
for i, tf := range tfs {
if bytes.Equal(tf.key, graphiteReverseTagKey) {
@@ -3236,6 +3237,7 @@ func (mp *tagToMetricIDsRowParser) GetMatchingSeriesCount(filter, negativeFilter
}
func mergeTagToMetricIDsRows(data []byte, items []mergeset.Item) ([]byte, []mergeset.Item) {
// TODO
data, items = mergeTagToMetricIDsRowsInternal(data, items, nsPrefixTagToMetricIDs)
data, items = mergeTagToMetricIDsRowsInternal(data, items, nsPrefixDateTagToMetricIDs)
return data, items

View File

@@ -2,10 +2,10 @@ package storage
import (
"bytes"
"math"
"path/filepath"
"strconv"
"sync/atomic"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
@@ -79,7 +79,7 @@ func mustOpenLegacyIndexDB(path string, s *Storage) *legacyIndexDB {
tr := TimeRange{
MinTimestamp: 0,
MaxTimestamp: math.MaxInt64,
MaxTimestamp: time.Now().UnixMilli(),
}
idb := mustOpenIndexDB(id, tr, name, path, s, &s.isReadOnly, true)
legacyIDB := &legacyIndexDB{idb: idb}

View File

@@ -486,7 +486,7 @@ func TestIndexDBOpenClose(t *testing.T) {
func TestIndexDB(t *testing.T) {
defer testRemoveAll(t)
f := func(t *testing.T, concurrency int, disablePerDayIndex bool) {
f := func(t *testing.T, concurrency int, disableGlobalIndex, disablePerDayIndex bool) {
const metricGroups = 10
now := time.Now().UTC()
timestamp := now.UnixMilli()
@@ -500,6 +500,7 @@ func TestIndexDB(t *testing.T) {
}
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -531,13 +532,19 @@ func TestIndexDB(t *testing.T) {
}
for _, concurrency := range []int{1, 4} {
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("concurrency=%d/disablePerDayIndex=%t", concurrency, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
f(t, concurrency, disablePerDayIndex)
// Repeat the same test on non-empty reopened storage.
f(t, concurrency, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("concurrency=%d/disableGlobalIndex=%t/disablePerDayIndex=%t", concurrency, disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
f(t, concurrency, disableGlobalIndex, disablePerDayIndex)
// Repeat the same test on non-empty reopened storage.
f(t, concurrency, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
}
@@ -1400,15 +1407,21 @@ func TestMatchTagFilters(t *testing.T) {
func TestIndexDBSearchTSIDs(t *testing.T) {
defer testRemoveAll(t)
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBSearchTSIDs(t, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBSearchTSIDs(t, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
func testIndexDBSearchTSIDs(t *testing.T, disablePerDayIndex bool) {
func testIndexDBSearchTSIDs(t *testing.T, disableGlobalIndex, disablePerDayIndex bool) {
const days = 5
const metricsPerDay = 1000
timestamp := time.Date(2019, time.October, 15, 5, 1, 0, 0, time.UTC).UnixMilli()
@@ -1417,6 +1430,7 @@ func testIndexDBSearchTSIDs(t *testing.T, disablePerDayIndex bool) {
allMetricIDs := &uint64set.Set{}
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -1474,8 +1488,10 @@ func testIndexDBSearchTSIDs(t *testing.T, disablePerDayIndex bool) {
assertMetricIDs(date, metricsPerDay, perDayMetricIDs[date])
}
}
// Check that all the metrics are found in global index
assertMetricIDs(globalIndexDate, metricsPerDay*days, allMetricIDs)
if !disableGlobalIndex {
// Check that all the metrics are found in global index
assertMetricIDs(globalIndexDate, metricsPerDay*days, allMetricIDs)
}
db.putIndexSearch(is2)
@@ -1522,21 +1538,28 @@ func testIndexDBSearchTSIDs(t *testing.T, disablePerDayIndex bool) {
func TestIndexDBSearchLabelNames(t *testing.T) {
defer testRemoveAll(t)
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBSearchLabelNames(t, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBSearchLabelNames(t, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
func testIndexDBSearchLabelNames(t *testing.T, disablePerDayIndex bool) {
func testIndexDBSearchLabelNames(t *testing.T, disableGlobalIndex, disablePerDayIndex bool) {
const days = 5
const metricsPerDay = 1000
timestamp := time.Date(2019, time.October, 15, 5, 1, 0, 0, time.UTC).UnixMilli()
baseDate := uint64(timestamp) / msecPerDay
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -1633,16 +1656,21 @@ func testIndexDBSearchLabelNames(t *testing.T, disablePerDayIndex bool) {
func TestIndexDBSearchLabelValues(t *testing.T) {
defer testRemoveAll(t)
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBSearchLabelValues(t, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBSearchLabelValues(t, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
func testIndexDBSearchLabelValues(t *testing.T, disablePerDayIndex bool) {
func testIndexDBSearchLabelValues(t *testing.T, disableGlobalIndex, disablePerDayIndex bool) {
const days = 5
const metricsPerDay = 1000
timestamp := time.Date(2019, time.October, 15, 5, 1, 0, 0, time.UTC).UnixMilli()
@@ -1650,6 +1678,7 @@ func testIndexDBSearchLabelValues(t *testing.T, disablePerDayIndex bool) {
var allMetricNames []string
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -1816,15 +1845,21 @@ func TestFilterLabelValues(t *testing.T) {
func TestIndexDBDeleteSeries(t *testing.T) {
defer testRemoveAll(t)
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBDeleteSeries(t, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBDeleteSeries(t, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
func testIndexDBDeleteSeries(t *testing.T, disablePerDayIndex bool) {
func testIndexDBDeleteSeries(t *testing.T, disableGlobalIndex, disablePerDayIndex bool) {
const days = 5
const metricsPerDay = 1000
trAllDays := TimeRange{
@@ -1837,6 +1872,7 @@ func testIndexDBDeleteSeries(t *testing.T, disablePerDayIndex bool) {
metricNamesByDate := make(map[uint64][]string)
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -1922,21 +1958,28 @@ func testIndexDBDeleteSeries(t *testing.T, disablePerDayIndex bool) {
func TestIndexDBGetTSDBStatus(t *testing.T) {
defer testRemoveAll(t)
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBGetTSDBStatus(t, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testIndexDBGetTSDBStatus(t, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
func testIndexDBGetTSDBStatus(t *testing.T, disablePerDayIndex bool) {
func testIndexDBGetTSDBStatus(t *testing.T, disableGlobalIndex, disablePerDayIndex bool) {
const days = 5
const metricsPerDay = 1000
timestamp := time.Date(2019, time.October, 15, 5, 1, 0, 0, time.UTC).UnixMilli()
baseDate := uint64(timestamp) / msecPerDay
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()

View File

@@ -80,6 +80,7 @@ type Storage struct {
// compatibility with partition index.
legacyIndexDBs atomic.Pointer[legacyIndexDBs]
disableGlobalIndex bool
disablePerDayIndex bool
tb *table
@@ -169,6 +170,7 @@ type OpenOptions struct {
DenyQueriesOutsideRetention bool
MaxHourlySeries int
MaxDailySeries int
DisableGlobalIndex bool
DisablePerDayIndex bool
TrackMetricNamesStats bool
IDBPrefillStart time.Duration
@@ -269,6 +271,10 @@ func MustOpenStorage(path string, opts OpenOptions) *Storage {
fs.MustMkdirIfNotExist(metadataDir)
s.minTimestampForCompositeIndex = mustGetMinTimestampForCompositeIndex(metadataDir, isEmptyDB)
if opts.DisableGlobalIndex && opts.DisablePerDayIndex {
logger.Panicf("BUG: global and per-day indexes cannot be disabled at the same time")
}
s.disableGlobalIndex = opts.DisableGlobalIndex
s.disablePerDayIndex = opts.DisablePerDayIndex
// Load legacy indexDBs.
@@ -1118,6 +1124,14 @@ func searchAndMerge[T any](qt *querytracer.Tracer, s *Storage, tr TimeRange, sea
qt = qt.NewChild("search indexDBs: timeRange=%v", &tr)
defer qt.Done()
var zeroValue T
if tr.MinTimestamp < minUnixMilli {
tr.MinTimestamp = minUnixMilli
}
if tr.MaxTimestamp < tr.MinTimestamp {
return zeroValue, nil
}
var idbts []indexDBWithType
ptws := s.tb.GetPartitions(tr)
@@ -1724,6 +1738,12 @@ const maxDaysForPerDaySearch = 40
// globalIndexTimeRange based on the time range length and -disablePerDayIndex
// flag.
func (s *Storage) adjustTimeRange(searchTR, idbTR TimeRange) TimeRange {
// if the global index is disabled and the searchTR is the
// globalIndexTimeRange, use idbTR.
if s.disableGlobalIndex && searchTR == globalIndexTimeRange {
return idbTR
}
// If the per day index is disabled, unconditionally search global index.
if s.disablePerDayIndex {
return globalIndexTimeRange
@@ -1740,7 +1760,7 @@ func (s *Storage) adjustTimeRange(searchTR, idbTR TimeRange) TimeRange {
// For legacy IndexDBs only, partition indexDBs can't span more than a
// month.
minDate, maxDate := tr.DateRange()
if maxDate-minDate > maxDaysForPerDaySearch {
if !s.disableGlobalIndex && maxDate-minDate > maxDaysForPerDaySearch {
return globalIndexTimeRange
}
@@ -1748,7 +1768,7 @@ func (s *Storage) adjustTimeRange(searchTR, idbTR TimeRange) TimeRange {
// the idb time range, then return globalIndexTimeRange to indicate that we
// want to search the global index since the entire index db needs to be
// searched anyway.
if tr == idbTR {
if !s.disableGlobalIndex && tr == idbTR {
return globalIndexTimeRange
}

View File

@@ -402,6 +402,10 @@ func TestStorageDeletePendingSeries(t *testing.T) {
defer testRemoveAll(t)
const numMonths = 10
start := time.Date(1971, 1, 1, 0, 0, 0, 0, time.UTC)
middle := start.AddDate(0, (numMonths-1)/2, 0)
end := start.AddDate(0, numMonths-1, 0)
s := MustOpenStorage(t.Name(), OpenOptions{})
var metricGroupName = []byte("metric")
@@ -456,7 +460,7 @@ func TestStorageDeletePendingSeries(t *testing.T) {
assertCountMonthsWithLabels := func(count int) {
t.Helper()
ts := time.Unix(0, 0)
ts := start
n := 0
for range numMonths {
lns, err := s.SearchLabelNames(nil, nil, TimeRange{ts.UnixMilli(), ts.UnixMilli()}, 1e5, 1e9, noDeadline)
@@ -481,7 +485,7 @@ func TestStorageDeletePendingSeries(t *testing.T) {
var search Search
defer search.MustClose()
search.Init(nil, s, []*TagFilters{tfs}, TimeRange{0, math.MaxInt64}, 1e5, noDeadline)
search.Init(nil, s, []*TagFilters{tfs}, TimeRange{start.UnixMilli(), math.MaxInt64}, 1e5, noDeadline)
n := 0
for search.NextMetricBlock() {
var b Block
@@ -498,10 +502,6 @@ func TestStorageDeletePendingSeries(t *testing.T) {
// Verify no metrics exist
assertCountRows(0)
start := time.Unix(0, 0)
middle := start.AddDate(0, (numMonths-1)/2, 0)
end := start.AddDate(0, numMonths-1, 0)
// Add some rows and flush, so next DeleteSeries() can delete them
addRows(start, middle, false)
s.DebugFlush()
@@ -531,22 +531,29 @@ func TestStorageDeleteSeries(t *testing.T) {
defer testRemoveAll(t)
for _, concurrency := range []int{1, 4} {
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("concurrency=%d/disablePerDayIndex=%t", concurrency, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testStorageDeleteSeries(t, concurrency, disablePerDayIndex)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("concurrency=%d/disableGlobalIndex=%t/disablePerDayIndex=%t", concurrency, disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testStorageDeleteSeries(t, concurrency, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
}
func testStorageDeleteSeries(t *testing.T, concurrency int, disablePerDayIndex bool) {
func testStorageDeleteSeries(t *testing.T, concurrency int, disableGlobalIndex, disablePerDayIndex bool) {
tr := TimeRange{
MinTimestamp: time.Date(2026, 1, 15, 0, 0, 0, 0, time.UTC).UnixMilli(),
MaxTimestamp: time.Date(2026, 1, 15, 23, 59, 59, 999_999_999, time.UTC).UnixMilli(),
}
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -2821,10 +2828,11 @@ func TestStorageGetTSDBStatus(t *testing.T) {
func TestStorageAdjustTimeRange(t *testing.T) {
defer testRemoveAll(t)
f := func(disablePerDayIndex bool, searchTR, idbTR, want TimeRange) {
f := func(disableGlobalIndex, disablePerDayIndex bool, searchTR, idbTR, want TimeRange) {
t.Helper()
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
@@ -2835,7 +2843,7 @@ func TestStorageAdjustTimeRange(t *testing.T) {
legacyIDBTimeRange := TimeRange{
MinTimestamp: 0,
MaxTimestamp: math.MaxInt64,
MaxTimestamp: time.Now().UnixMilli(),
}
partitionIDBTimeRange := TimeRange{
MinTimestamp: time.Date(2025, 2, 1, 0, 0, 0, 0, time.UTC).UnixMilli(),
@@ -2845,10 +2853,12 @@ func TestStorageAdjustTimeRange(t *testing.T) {
// Search time range is the same as globalIndexTimeRange.
searchTimeRange = globalIndexTimeRange
f(false, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, legacyIDBTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, partitionIDBTimeRange)
// The search time range is smaller than a month (and therefore < 40 days)
// and is fully included into the partition idb time range.
@@ -2856,17 +2866,21 @@ func TestStorageAdjustTimeRange(t *testing.T) {
MinTimestamp: partitionIDBTimeRange.MinTimestamp + msecPerDay,
MaxTimestamp: partitionIDBTimeRange.MaxTimestamp - msecPerDay,
}
f(false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, searchTimeRange)
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, searchTimeRange)
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, searchTimeRange)
// The search time range is the same as partition idb time range.
searchTimeRange = partitionIDBTimeRange
f(false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, searchTimeRange)
// The search time range is smaller than 40 days and fully includes the
// partition idb time range.
@@ -2874,10 +2888,12 @@ func TestStorageAdjustTimeRange(t *testing.T) {
MinTimestamp: partitionIDBTimeRange.MinTimestamp - msecPerDay,
MaxTimestamp: partitionIDBTimeRange.MaxTimestamp + msecPerDay,
}
f(false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, partitionIDBTimeRange)
// The search time range is 41 days and fully includes the partition idb
// time range.
@@ -2885,10 +2901,12 @@ func TestStorageAdjustTimeRange(t *testing.T) {
MinTimestamp: partitionIDBTimeRange.MinTimestamp - msecPerDay,
MaxTimestamp: partitionIDBTimeRange.MinTimestamp + 41*msecPerDay,
}
f(false, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, partitionIDBTimeRange)
// The search time range is smaller than 40 days and overlaps with partition
// idb time range on the left.
@@ -2896,13 +2914,18 @@ func TestStorageAdjustTimeRange(t *testing.T) {
MinTimestamp: partitionIDBTimeRange.MinTimestamp - msecPerDay,
MaxTimestamp: partitionIDBTimeRange.MinTimestamp + msecPerDay,
}
f(false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, TimeRange{
f(false, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, TimeRange{
MinTimestamp: partitionIDBTimeRange.MinTimestamp,
MaxTimestamp: searchTimeRange.MaxTimestamp,
})
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, TimeRange{
MinTimestamp: partitionIDBTimeRange.MinTimestamp,
MaxTimestamp: searchTimeRange.MaxTimestamp,
})
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
// The search time range is smaller than 40 days and overlaps with partition
// idb time range on the right.
@@ -2910,13 +2933,18 @@ func TestStorageAdjustTimeRange(t *testing.T) {
MinTimestamp: partitionIDBTimeRange.MaxTimestamp - msecPerDay,
MaxTimestamp: partitionIDBTimeRange.MaxTimestamp + msecPerDay,
}
f(false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, searchTimeRange, partitionIDBTimeRange, TimeRange{
f(false, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(false, false, searchTimeRange, partitionIDBTimeRange, TimeRange{
MinTimestamp: searchTimeRange.MinTimestamp,
MaxTimestamp: partitionIDBTimeRange.MaxTimestamp,
})
f(false, true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(false, true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
f(true, false, searchTimeRange, legacyIDBTimeRange, searchTimeRange)
f(true, false, searchTimeRange, partitionIDBTimeRange, TimeRange{
MinTimestamp: searchTimeRange.MinTimestamp,
MaxTimestamp: partitionIDBTimeRange.MaxTimestamp,
})
f(true, searchTimeRange, legacyIDBTimeRange, globalIndexTimeRange)
f(true, searchTimeRange, partitionIDBTimeRange, globalIndexTimeRange)
}
type testStorageSearchWithoutIndexOptions struct {
@@ -2935,9 +2963,11 @@ type testStorageSearchWithoutIndexOptions struct {
func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutIndexOptions) {
defer testRemoveAll(t)
// The data is inserted and the search is performed when per-day index is enabled.
// The data is inserted and the search is performed when both global and
// per-day indexes are enabled.
t.Run("Add-Global-PerDay/Search-Global-PerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: false,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
@@ -2948,9 +2978,11 @@ func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutI
s.MustClose()
})
// The data is inserted and the search is performed when per-day index is disabled.
// The data is inserted and the search is performed when global index is
// enabled and per-day index is disabled.
t.Run("Add-Global-noPerDay/Search-Global-noPerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: true,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
@@ -2964,10 +2996,12 @@ func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutI
s.MustClose()
})
// The data is inserted when per-day index are enabled.
// The search is performed when per-day index is disabled.
// The data is inserted when both global and per-day indexes are enabled.
// The search is performed when the global index is enabled and per-day
// index is disabled.
t.Run("Add-Global-PerDay/Search-Global-noPerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: false,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
@@ -2975,6 +3009,7 @@ func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutI
s.MustClose()
s = MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: true,
})
for tr, want := range opts.wantPerTimeRange {
@@ -2986,12 +3021,14 @@ func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutI
s.MustClose()
})
// The data is inserted when per-day index is disabled.
// The search is performed when per-day index is enabled.
// The data is inserted when global index is enabled and per-day index is
// disabled.
// The search is performed when both global and per-day index is enabled.
// This case also shows that registering metric names recovers the per-day
// index.
t.Run("Add-Global-noPerDay/Search-Global-PerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: true,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
@@ -2999,6 +3036,7 @@ func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutI
s.MustClose()
s = MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: false,
})
@@ -3015,6 +3053,111 @@ func testStorageSearchWithoutIndex(t *testing.T, opts *testStorageSearchWithoutI
}
s.MustClose()
})
// The data is inserted and the search is performed when global index is
// disabled and per-day index is enabled.
t.Run("Add-noGlobal-PerDay/Search-noGlobal-PerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: true,
DisablePerDayIndex: false,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
s.DebugFlush()
for tr, want := range opts.wantPerTimeRange {
opts.assertSearchResult(t, s, tr, want)
}
s.MustClose()
})
// The data is inserted when both global and per-day indexes are enabled.
// The search is performed when the global index is disabled and per-day
// index is enabled.
t.Run("Add-Global-PerDay/Search-noGlobal-PerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: false,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
s.DebugFlush()
s.MustClose()
s = MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: true,
DisablePerDayIndex: false,
})
for tr, want := range opts.wantPerTimeRange {
opts.assertSearchResult(t, s, tr, want)
}
s.MustClose()
})
// The data is inserted when global index is disabled and per-day index is
// enabled.
// The search is performed when both global and per-day indexes are enabled.
t.Run("Add-noGlobal-PerDay/Search-Global-PerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: true,
DisablePerDayIndex: false,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
s.DebugFlush()
s.MustClose()
s = MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: false,
})
for tr, want := range opts.wantPerTimeRange {
opts.assertSearchResult(t, s, tr, want)
}
s.MustClose()
})
// The data is inserted when global index is disabled and per-day index is
// enabled.
// The search is performed when global index is enabled and per-day index is
// disabled.
t.Run("Add-noGlobal-PerDay/Search-Global-noPerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: true,
DisablePerDayIndex: false,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
s.DebugFlush()
s.MustClose()
s = MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: true,
})
for tr := range opts.wantPerTimeRange {
opts.assertSearchResult(t, s, tr, opts.wantEmpty)
}
s.MustClose()
})
// The data is inserted when global index is enabled and per-day index is
// disabled.
// The search is performed when the global index is disabled and per-day
// index is enabled.
t.Run("Add-Global-noPerDay/Search-noGlobal-PerDay", func(t *testing.T) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: false,
DisablePerDayIndex: true,
})
s.AddRows(opts.mrs, defaultPrecisionBits)
s.DebugFlush()
s.MustClose()
s = MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: true,
DisablePerDayIndex: false,
})
for tr := range opts.wantPerTimeRange {
opts.assertSearchResult(t, s, tr, opts.wantEmpty)
}
s.MustClose()
})
}
func TestStorageGetTSDBStatusWithoutIndex(t *testing.T) {
@@ -3385,51 +3528,195 @@ func TestStorageQueryWithoutIndex(t *testing.T) {
testStorageSearchWithoutIndex(t, &opts)
}
func TestStorageAddRows_SamplesWithZeroDate(t *testing.T) {
func TestStorageAddRowsWithZeroDate(t *testing.T) {
defer testRemoveAll(t)
f := func(t *testing.T, disablePerDayIndex bool) {
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
name := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(name, func(t *testing.T) {
testStorageAddRowsWithZeroDate(t, disableGlobalIndex, disablePerDayIndex)
})
}
}
}
func testStorageAddRowsWithZeroDate(t *testing.T, disableGlobalIndex, disablePerDayIndex bool) {
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
const numDays = 4
var metricNamesAll []string
labelNamesAll := []string{"__name__", "label"}
var labelValuesAll []string
mrs := make([]MetricRow, numDays)
for day := range numDays {
metricName := fmt.Sprintf("metric_%02d", day)
labelName := fmt.Sprintf("label_%02d", day)
labelValue := fmt.Sprintf("value_%02d", day)
if day != 0 {
metricNamesAll = append(metricNamesAll, metricName)
labelNamesAll = append(labelNamesAll, labelName)
labelValuesAll = append(labelValuesAll, labelValue)
}
mn := MetricName{
MetricGroup: []byte(metricName),
Tags: []Tag{
{Key: []byte(labelName), Value: []byte("value")},
{Key: []byte("label"), Value: []byte(labelValue)},
},
}
mn.sortTags()
mrs[day].MetricNameRaw = mn.marshalRaw(nil)
mrs[day].Timestamp = int64(day * msecPerDay)
}
s.AddRows(mrs, defaultPrecisionBits)
s.DebugFlush()
if got, want := s.newTimeseriesCreated.Load(), uint64(numDays-1); got != want {
t.Fatalf("unexpected new timeseries count: got %d, want %d", got, want)
}
if got, want := s.tooSmallTimestampRows.Load(), uint64(1); got != want {
t.Fatalf("unexpected rows with too small timestamp: got %d, want %d", got, want)
}
assertMetricNames := func(tr TimeRange, want []string) {
t.Helper()
s := MustOpenStorage(t.Name(), OpenOptions{
DisablePerDayIndex: disablePerDayIndex,
})
defer s.MustClose()
mn := MetricName{MetricGroup: []byte("metric")}
mr := MetricRow{MetricNameRaw: mn.marshalRaw(nil)}
for range 10 {
mr.Timestamp = rand.Int63n(msecPerDay)
mr.Value = float64(rand.Intn(1000))
s.AddRows([]MetricRow{mr}, defaultPrecisionBits)
s.DebugFlush()
// Reset TSID cache so that insertion takes the path that involves
// checking whether the index contains metricName->TSID mapping.
s.resetAndSaveTSIDCache()
tfs := NewTagFilters()
if err := tfs.Add(nil, []byte("metric_.*"), false, true); err != nil {
t.Fatalf("unexpected error in TagFilters.Add: %v", err)
}
want := 1
firstUnixDay := TimeRange{
MinTimestamp: 0,
MaxTimestamp: msecPerDay - 1,
got, err := s.SearchMetricNames(nil, []*TagFilters{tfs}, tr, 1e9, noDeadline)
if err != nil {
t.Fatalf("SearchMetricNames(%v, %v) failed unexpectedly: %v", tfs, &tr, err)
}
if got := s.newTimeseriesCreated.Load(); got != uint64(want) {
t.Errorf("unexpected new timeseries count: got %d, want %d", got, want)
for i, name := range got {
var mn MetricName
if err := mn.UnmarshalString(name); err != nil {
t.Fatalf("Could not unmarshal metric name %q: %v", name, err)
}
got[i] = string(mn.MetricGroup)
}
if got := testCountAllMetricNames(s, firstUnixDay); got != want {
t.Errorf("unexpected metric name count: got %d, want %d", got, want)
slices.Sort(got)
slices.Sort(want)
if diff := cmp.Diff(want, got); diff != "" {
t.Fatalf("unexpected metric names (-want, +got):\n%s", diff)
}
if got := testCountAllMetricIDs(s, firstUnixDay); got != want {
t.Errorf("unexpected metric id count: got %d, want %d", got, want)
}
assertLabelNames := func(tr TimeRange, want []string) {
t.Helper()
tfs := NewTagFilters()
if err := tfs.Add(nil, []byte("metric_.*"), false, true); err != nil {
t.Fatalf("unexpected error in TagFilters.Add: %v", err)
}
got, err := s.SearchLabelNames(nil, []*TagFilters{tfs}, tr, 1e9, 1e9, noDeadline)
if err != nil {
t.Fatalf("SearchLabelNames(%v, %v) failed unexpectedly: %s", tfs, &tr, err)
}
slices.Sort(got)
slices.Sort(want)
if diff := cmp.Diff(want, got); diff != "" {
t.Fatalf("unexpected label names (-want, +got):\n%s", diff)
}
}
assertLabelValues := func(tr TimeRange, want []string) {
t.Helper()
tfs := NewTagFilters()
if err := tfs.Add([]byte("label"), []byte("value_.*"), false, true); err != nil {
t.Fatalf("unexpected error in TagFilters.Add: %v", err)
}
got, err := s.SearchLabelValues(nil, "label", []*TagFilters{tfs}, tr, 1e9, 1e9, noDeadline)
if err != nil {
t.Fatalf("SearchLabelValues(%v, %v) failed unexpectedly: %s", tfs, tr, err)
}
slices.Sort(got)
slices.Sort(want)
if diff := cmp.Diff(want, got); diff != "" {
t.Fatalf("unexpected label values (-want, +got):\n%s", diff)
}
}
assertData := func(tr TimeRange, want []MetricRow) {
t.Helper()
tfs := NewTagFilters()
if err := tfs.Add(nil, []byte("metric_.*"), false, true); err != nil {
t.Fatalf("TagFilters.Add() failed unexpectedly: %v", err)
}
if err := testAssertSearchResult(s, tr, tfs, want); err != nil {
t.Fatalf("Search(%v, %v) failed unexpectedly: %v", tfs, tr, err)
}
}
for _, disablePerDayIndex := range []bool{false, true} {
name := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(name, func(t *testing.T) {
f(t, disablePerDayIndex)
})
var tr TimeRange
// Empty time range.
// Expect empty search results
tr = TimeRange{}
assertMetricNames(tr, nil)
assertLabelNames(tr, []string{})
assertLabelValues(tr, []string{})
assertData(tr, nil)
// First day time range.
// Expect empty search results
tr = TimeRange{
MinTimestamp: 0,
MaxTimestamp: msecPerDay - 1,
}
assertMetricNames(tr, nil)
assertLabelNames(tr, []string{})
assertLabelValues(tr, []string{})
assertData(tr, nil)
// Second day time range.
tr = TimeRange{
MinTimestamp: msecPerDay,
MaxTimestamp: 2*msecPerDay - 1,
}
if disablePerDayIndex {
// Expect index search results for all days if per-day index is
// disabled.
assertMetricNames(tr, metricNamesAll)
assertLabelNames(tr, labelNamesAll)
assertLabelValues(tr, labelValuesAll)
} else {
// Expect index search results on second day only if per-day index is
// enabled.
assertMetricNames(tr, []string{"metric_01"})
assertLabelNames(tr, []string{"__name__", "label", "label_01"})
assertLabelValues(tr, []string{"value_01"})
}
assertData(tr, mrs[1:2])
// First two days time range.
// Expect results on second day only.
tr = TimeRange{
MinTimestamp: 0,
MaxTimestamp: 2*msecPerDay - 1,
}
if disablePerDayIndex {
// Expect index search results for all days if per-day index is
// disabled.
assertMetricNames(tr, metricNamesAll)
assertLabelNames(tr, labelNamesAll)
assertLabelValues(tr, labelValuesAll)
} else {
// Expect index search results on second day only if per-day index is
// enabled.
assertMetricNames(tr, []string{"metric_01"})
assertLabelNames(tr, []string{"__name__", "label", "label_01"})
assertLabelValues(tr, []string{"value_01"})
}
assertData(tr, mrs[1:2])
}
// testSearchMetricIDs returns metricIDs for the given tfss and tr.
@@ -3489,17 +3776,23 @@ func testStorageVariousDataPatternsConcurrently(t *testing.T, registerOnly bool,
const concurrency = 4
for _, disablePerDayIndex := range []bool{false, true} {
prefix := fmt.Sprintf("disablePerDayIndex=%t", disablePerDayIndex)
t.Run(prefix+"/serial", func(t *testing.T) {
testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, 1, false)
})
t.Run(prefix+"/concurrentRows", func(t *testing.T) {
testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, concurrency, true)
})
t.Run(prefix+"/concurrentBatches", func(t *testing.T) {
testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, concurrency, false)
})
for _, disableGlobalIndex := range []bool{false, true} {
for _, disablePerDayIndex := range []bool{false, true} {
if disableGlobalIndex && disablePerDayIndex {
// Both indexes cannot be disabled at the same time.
continue
}
prefix := fmt.Sprintf("disableGlobalIndex=%t/disablePerDayIndex=%t", disableGlobalIndex, disablePerDayIndex)
t.Run(prefix+"/serial", func(t *testing.T) {
testStorageVariousDataPatterns(t, disableGlobalIndex, disablePerDayIndex, registerOnly, op, 1, false)
})
t.Run(prefix+"/concurrentRows", func(t *testing.T) {
testStorageVariousDataPatterns(t, disableGlobalIndex, disablePerDayIndex, registerOnly, op, concurrency, true)
})
t.Run(prefix+"/concurrentBatches", func(t *testing.T) {
testStorageVariousDataPatterns(t, disableGlobalIndex, disablePerDayIndex, registerOnly, op, concurrency, false)
})
}
}
}
@@ -3509,7 +3802,7 @@ func testStorageVariousDataPatternsConcurrently(t *testing.T, registerOnly bool,
// The function is intended to be used by other tests that define the
// concurrency, the per-day index setting, and the operation (AddRows or
// RegisterMetricNames) under test.
func testStorageVariousDataPatterns(t *testing.T, disablePerDayIndex, registerOnly bool, op func(s *Storage, mrs []MetricRow), concurrency int, splitBatches bool) {
func testStorageVariousDataPatterns(t *testing.T, disableGlobalIndex, disablePerDayIndex, registerOnly bool, op func(s *Storage, mrs []MetricRow), concurrency int, splitBatches bool) {
f := func(t *testing.T, sameBatchMetricNames, sameRowMetricNames, sameBatchDates, sameRowDates bool) {
batches, wantCounts := testGenerateMetricRowBatches(&batchOptions{
numBatches: 3,
@@ -3526,6 +3819,7 @@ func testStorageVariousDataPatterns(t *testing.T, disablePerDayIndex, registerOn
rowsAddedTotal := wantCounts.metrics.RowsAddedTotal
s := MustOpenStorage(t.Name(), OpenOptions{
DisableGlobalIndex: disableGlobalIndex,
DisablePerDayIndex: disablePerDayIndex,
})

View File

@@ -429,9 +429,8 @@ func (tb *table) getMinMaxIngestionTimestamps() (int64, int64) {
func (tb *table) getMinMaxTimestampsForAge(minAgeMsecs int64) (int64, int64) {
now := int64(fasttime.UnixTimestamp() * 1000)
minTimestamp := now - minAgeMsecs
if minTimestamp < 0 {
// Negative timestamps aren't supported by the storage.
minTimestamp = 0
if minTimestamp < minUnixMilli {
minTimestamp = minUnixMilli
}
maxTimestamp := int64(maxUnixMilli)
if maxUnixMilli-now > tb.s.futureRetentionMsecs {

View File

@@ -40,12 +40,6 @@ type TimeRange struct {
MaxTimestamp int64
}
// Zero time range and zero date are used to force global index search.
var (
globalIndexTimeRange = TimeRange{}
globalIndexDate = uint64(0)
)
// DateRange returns the date range for the given time range.
func (tr *TimeRange) DateRange() (uint64, uint64) {
minDate := uint64(tr.MinTimestamp) / msecPerDay
@@ -117,10 +111,29 @@ func (tr *TimeRange) contains(timestamp int64) bool {
return tr.MinTimestamp <= timestamp && timestamp <= tr.MaxTimestamp
}
// Zero time range and zero date are used to force global index search.
var (
globalIndexDate = uint64(0)
globalIndexTimeRange = TimeRange{}
)
const (
msecPerDay = 24 * 3600 * 1000
msecPerHour = 3600 * 1000
// minUnixMilli is the min millisecond that is allowed to be used as the
// sample timestamp.
//
// It corresponds to the first millisecond of the second day of the Unix
// Epoch, i.e. 1970-01-02T00:00:00.000Z.
//
// The first day of the Unix Epoch is reserved: zero date and zero time
// range are used for indicating that the the global index search is
// required. See globalIndexDate and globalIndexTimeRange above.
//
// Negative timestamps aren't supported.
minUnixMilli = msecPerDay
// maxUnixMilli is the max millisecond that is allowed to be used as the
// sample timestamp.
//
@@ -130,6 +143,6 @@ const (
// time.UnixMicro(math.MaxInt64/1000) == 2262-04-11 23:47:16.854775 UTC.
//
// Round it to the last millisecond of the last complete partition:
// 2262-03-31 23:59:59.999 UTC.
// 2262-03-31T23:59:59.999Z.
maxUnixMilli = 9222422399999
)

View File

@@ -935,4 +935,25 @@ foo:1m_sum_samples{baz="qwe"} 10
dedup_interval: 30s
outputs: [sum_samples]
`, "11111111")
// Reproduce issue #11261: sum_samples_total must be monotonic with enable_windows: true
// See https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11262
f([]string{`
test_delta 1
`, `
test_delta 1
`, `
test_delta 1
`, `
test_delta 1
`}, time.Minute, `test_delta 1
test_delta 2
test_delta 3
test_delta 4
`, `
- interval: 1m
keep_metric_names: true
outputs: [sum_samples_total]
enable_windows: true
`, "1111")
}

View File

@@ -4,30 +4,39 @@ import (
"math"
)
type sumSamplesAggrValueShared struct {
total float64
}
type sumSamplesAggrValue struct {
sum float64
delta float64
shared *sumSamplesAggrValueShared
}
func (av *sumSamplesAggrValue) pushSample(_ aggrConfig, sample *pushSample, _ string, _ int64) {
if math.Abs(av.sum) >= (1 << 53) {
// It is time to reset the entry, since it starts losing float64 precision
av.sum = 0
}
av.sum += sample.value
av.delta += sample.value
}
func (av *sumSamplesAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, _ bool) {
ac := c.(*sumSamplesAggrConfig)
if ac.resetTotalOnFlush {
ctx.appendSeries(key, "sum_samples", av.sum)
av.sum = 0
ctx.appendSeries(key, "sum_samples", av.delta)
av.delta = 0
return
}
ctx.appendSeries(key, "sum_samples_total", av.sum)
total := av.shared.total + av.delta
av.delta = 0
if math.Abs(total) >= (1 << 53) {
// It is time to reset the entry, since it starts losing float64 precision
av.shared.total = 0
} else {
av.shared.total = total
}
ctx.appendSeries(key, "sum_samples_total", total)
}
func (*sumSamplesAggrValue) state() any {
return nil
func (av *sumSamplesAggrValue) state() any {
return av.shared
}
func newSumSamplesAggrConfig(resetTotalOnFlush bool) aggrConfig {
@@ -40,6 +49,14 @@ type sumSamplesAggrConfig struct {
resetTotalOnFlush bool
}
func (*sumSamplesAggrConfig) getValue(_ any) aggrValue {
return &sumSamplesAggrValue{}
func (*sumSamplesAggrConfig) getValue(s any) aggrValue {
var shared *sumSamplesAggrValueShared
if s == nil {
shared = &sumSamplesAggrValueShared{}
} else {
shared = s.(*sumSamplesAggrValueShared)
}
return &sumSamplesAggrValue{
shared: shared,
}
}

View File

@@ -206,17 +206,36 @@ func tryParseScientificNumberForUnixTimestamp(s string, decimalExp int64) (int64
return multiplyByDecimalExp(n, decimalExp)
}
intStr := s[:dotIdx]
fracStr := s[dotIdx+1:]
if decimalExp < int64(len(fracStr)) {
if decimalExp < 0 {
// Negative exponents on a fractional mantissa are intentionally not
// supported. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11268
return 0, false
}
intStr := s[:dotIdx]
fracStr := s[dotIdx+1:]
n, ok := tryParseFractionalNumberForUnixTimestamp(intStr, fracStr)
if !ok {
return 0, false
}
decimalExp -= int64(len(fracStr))
return multiplyByDecimalExp(n, decimalExp)
if decimalExp >= int64(len(fracStr)) {
// The exponent shifts the decimal point past every fractional digit,
// so the value is an integer number of seconds (or coarser).
decimalExp -= int64(len(fracStr))
return multiplyByDecimalExp(n, decimalExp)
}
// The exponent leaves fractional digits, e.g. 1.784144612388E9 == 1784144612.388
// Pad n as plain fractional timestamps do.
fracDigits := int64(len(fracStr)) - decimalExp
for fracDigits%3 != 0 {
if n >= 0 && n > math.MaxInt64/10 || n < 0 && n < math.MinInt64/10 {
return 0, false
}
n *= 10
fracDigits++
}
return n, true
}
func tryParseFractionalNumberForUnixTimestamp(intStr, fracStr string) (int64, bool) {

View File

@@ -69,6 +69,17 @@ func TestTryParseUnixTimestamp_Success(t *testing.T) {
f("1.23e2", 123000000000)
f("1.2e1", 12000000000)
f("1123.456789123456789E15", 1123456789123456789)
// scientific notation with sub-second precision, i.e. more fractional digits
// than the exponent shifts (https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11268).
// These must match the equivalent plain fractional form.
f("1.784144612388E9", 1784144612388000000) // == 1784144612.388
f("1.784144612388e9", 1784144612388000000)
f("-1.784144612388e9", -1784144612388000000)
f("1.5000000005e9", 1500000000500000000) // == 1500000000.5
f("1.23456789e9", 1234567890000000000) // exponent consumes all frac digits (integer result)
f("1.23e1", 12300000000000) // == 12.3
f("1.234e0", 1234000000000) // == 1.234
}
func TestTryParseUnixTimestamp_Failure(t *testing.T) {
@@ -115,9 +126,7 @@ func TestTryParseUnixTimestamp_Failure(t *testing.T) {
f("1e19")
f("1.3e123456789090123")
// too small decimal exponent
f("1.23e1")
f("1.234e0")
// negative decimal exponent
f("1E-1")
f("1.3e-123456789090123")
}