mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-11 23:40:48 +01:00
Compare commits
13 Commits
issue_5602
...
nv/promql-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
25b4a41cd4 | ||
|
|
1002d91de9 | ||
|
|
601b8e82b6 | ||
|
|
eed5020ee4 | ||
|
|
0938381113 | ||
|
|
e61755a143 | ||
|
|
6defe22271 | ||
|
|
87ee5f99af | ||
|
|
97093bf0e4 | ||
|
|
f41f541d2f | ||
|
|
73d40051ac | ||
|
|
025dccec69 | ||
|
|
e7000bbaa6 |
@@ -5,6 +5,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -478,11 +479,19 @@ func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began ti
|
||||
|
||||
for idx := range v.Floats {
|
||||
p := v.Floats[idx]
|
||||
// NaN and +/-Inf have no JSON number form and nothing to plot; the
|
||||
// builder path drops them while scanning rows (see consume.go).
|
||||
if math.IsNaN(p.F) || math.IsInf(p.F, 0) {
|
||||
continue
|
||||
}
|
||||
s.Values = append(s.Values, &qbv5.TimeSeriesValue{
|
||||
Timestamp: p.T,
|
||||
Value: p.F,
|
||||
})
|
||||
}
|
||||
if len(s.Values) == 0 {
|
||||
continue
|
||||
}
|
||||
series = append(series, &s)
|
||||
}
|
||||
|
||||
@@ -494,13 +503,11 @@ func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began ti
|
||||
}
|
||||
statsMu.Unlock()
|
||||
|
||||
tsData := &qbv5.TimeSeriesData{
|
||||
QueryName: q.query.Name,
|
||||
Aggregations: []*qbv5.AggregationBucket{
|
||||
{
|
||||
Series: series,
|
||||
},
|
||||
},
|
||||
tsData := &qbv5.TimeSeriesData{QueryName: q.query.Name}
|
||||
// No bucket at all when nothing survived: a bucket holding no series reads
|
||||
// as "filtered to empty" to the cache, which stores it as a real result.
|
||||
if len(series) > 0 {
|
||||
tsData.Aggregations = []*qbv5.AggregationBucket{{Series: series}}
|
||||
}
|
||||
|
||||
var payload any = tsData
|
||||
|
||||
@@ -2,14 +2,21 @@ package querier
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"math"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
"github.com/prometheus/prometheus/promql"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/prometheustest"
|
||||
qbv5 "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestRemoveAllVarMatchers(t *testing.T) {
|
||||
@@ -453,3 +460,82 @@ func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
|
||||
}
|
||||
assert.Empty(t, q.Fingerprint())
|
||||
}
|
||||
|
||||
func TestToResultDropsNonFiniteValues(t *testing.T) {
|
||||
tests := []struct {
|
||||
description string
|
||||
floats []promql.FPoint
|
||||
expectedTimestamps []int64
|
||||
expectedValues []float64
|
||||
}{
|
||||
{
|
||||
description: "finite values pass through untouched",
|
||||
floats: []promql.FPoint{{T: 1000, F: 1.5}, {T: 2000, F: 2.5}},
|
||||
expectedTimestamps: []int64{1000, 2000},
|
||||
expectedValues: []float64{1.5, 2.5},
|
||||
},
|
||||
{
|
||||
description: "a ratio's 0/0 points are dropped, the rest kept",
|
||||
floats: []promql.FPoint{{T: 1000, F: 1.5}, {T: 2000, F: math.NaN()}, {T: 3000, F: 2.5}},
|
||||
expectedTimestamps: []int64{1000, 3000},
|
||||
expectedValues: []float64{1.5, 2.5},
|
||||
},
|
||||
{
|
||||
description: "both infinities are dropped",
|
||||
floats: []promql.FPoint{{T: 1000, F: math.Inf(1)}, {T: 2000, F: 4.5}, {T: 3000, F: math.Inf(-1)}},
|
||||
expectedTimestamps: []int64{2000},
|
||||
expectedValues: []float64{4.5},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.description, func(t *testing.T) {
|
||||
q := &promqlQuery{query: qbv5.PromQuery{Name: "A"}, requestType: qbv5.RequestTypeTimeSeries}
|
||||
matrix := promql.Matrix{{Metric: labels.FromStrings("job_name", "dbBloatMonitorJob"), Floats: test.floats}}
|
||||
|
||||
var mu sync.Mutex
|
||||
var rows, bytes uint64
|
||||
result := q.toResult(matrix, nil, time.Now(), &mu, &rows, &bytes)
|
||||
|
||||
tsData, ok := result.Value.(*qbv5.TimeSeriesData)
|
||||
require.True(t, ok)
|
||||
require.Len(t, tsData.Aggregations, 1)
|
||||
require.Len(t, tsData.Aggregations[0].Series, 1)
|
||||
|
||||
timestamps := make([]int64, 0, len(test.expectedTimestamps))
|
||||
values := make([]float64, 0, len(test.expectedValues))
|
||||
for _, v := range tsData.Aggregations[0].Series[0].Values {
|
||||
timestamps = append(timestamps, v.Timestamp)
|
||||
values = append(values, v.Value)
|
||||
}
|
||||
assert.Equal(t, test.expectedTimestamps, timestamps)
|
||||
assert.Equal(t, test.expectedValues, values)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A series left with nothing must not surface as an empty series, and a result
|
||||
// left with no series must carry no aggregation bucket at all — the cache reads
|
||||
// a bucket holding no series as a real, filtered-to-empty result and stores it.
|
||||
func TestToResultDropsSeriesAndBucketLeftEmpty(t *testing.T) {
|
||||
q := &promqlQuery{query: qbv5.PromQuery{Name: "A"}, requestType: qbv5.RequestTypeTimeSeries}
|
||||
matrix := promql.Matrix{
|
||||
{Metric: labels.FromStrings("job_name", "idleJob"), Floats: []promql.FPoint{{T: 1000, F: math.NaN()}}},
|
||||
{Metric: labels.FromStrings("job_name", "activeJob"), Floats: []promql.FPoint{{T: 1000, F: 7.5}}},
|
||||
}
|
||||
|
||||
var mu sync.Mutex
|
||||
var rows, bytes uint64
|
||||
tsData, ok := q.toResult(matrix, nil, time.Now(), &mu, &rows, &bytes).Value.(*qbv5.TimeSeriesData)
|
||||
require.True(t, ok)
|
||||
require.Len(t, tsData.Aggregations, 1)
|
||||
require.Len(t, tsData.Aggregations[0].Series, 1, "the all-NaN series is gone")
|
||||
assert.Equal(t, "activeJob", tsData.Aggregations[0].Series[0].Labels[0].Value)
|
||||
|
||||
allNaN := promql.Matrix{
|
||||
{Metric: labels.FromStrings("job_name", "idleJob"), Floats: []promql.FPoint{{T: 1000, F: math.NaN()}}},
|
||||
}
|
||||
tsData, ok = q.toResult(allNaN, nil, time.Now(), &mu, &rows, &bytes).Value.(*qbv5.TimeSeriesData)
|
||||
require.True(t, ok)
|
||||
assert.Empty(t, tsData.Aggregations)
|
||||
}
|
||||
|
||||
116
tests/integration/testdata/promqltestcorpus/nonfinite_exclusions.json
vendored
Normal file
116
tests/integration/testdata/promqltestcorpus/nonfinite_exclusions.json
vendored
Normal file
@@ -0,0 +1,116 @@
|
||||
{
|
||||
"note": "Cases SigNoz intentionally does not match, on every leg: promql_query.go drops NaN and +/-Inf, so these come back short a timestamp or a whole series. Not defects, not for burn-down, but a case that starts matching must be removed.",
|
||||
"cases": [
|
||||
"aggregators.test:630[base]",
|
||||
"aggregators.test:630[instant-coarse]",
|
||||
"aggregators.test:633[base]",
|
||||
"aggregators.test:633[instant-coarse]",
|
||||
"aggregators.test:636[base]",
|
||||
"aggregators.test:636[instant-coarse]",
|
||||
"aggregators.test:639[base]",
|
||||
"aggregators.test:639[instant-coarse]",
|
||||
"aggregators.test:642[base]",
|
||||
"aggregators.test:642[instant-coarse]",
|
||||
"aggregators.test:645[base]",
|
||||
"aggregators.test:645[instant-coarse]",
|
||||
"aggregators.test:648[base]",
|
||||
"aggregators.test:648[instant-coarse]",
|
||||
"aggregators.test:661[base]",
|
||||
"aggregators.test:661[instant-coarse]",
|
||||
"aggregators.test:698[base]",
|
||||
"aggregators.test:698[instant-coarse]",
|
||||
"aggregators.test:702[base]",
|
||||
"aggregators.test:702[instant-coarse]",
|
||||
"aggregators.test:706[base]",
|
||||
"aggregators.test:706[instant-coarse]",
|
||||
"aggregators.test:710[base]",
|
||||
"aggregators.test:710[instant-coarse]",
|
||||
"aggregators.test:714[base]",
|
||||
"aggregators.test:714[instant-coarse]",
|
||||
"aggregators.test:717[base]",
|
||||
"aggregators.test:717[instant-coarse]",
|
||||
"aggregators.test:720[base]",
|
||||
"aggregators.test:720[instant-coarse]",
|
||||
"aggregators.test:724[base]",
|
||||
"aggregators.test:724[instant-coarse]",
|
||||
"aggregators.test:862[base]",
|
||||
"aggregators.test:862[instant-coarse]",
|
||||
"aggregators.test:865[base]",
|
||||
"aggregators.test:865[instant-coarse]",
|
||||
"aggregators.test:868[base]",
|
||||
"aggregators.test:868[instant-coarse]",
|
||||
"aggregators.test:873[base]",
|
||||
"aggregators.test:873[instant-coarse]",
|
||||
"aggregators.test:885[base]",
|
||||
"aggregators.test:885[instant-coarse]",
|
||||
"aggregators.test:888[base]",
|
||||
"aggregators.test:888[instant-coarse]",
|
||||
"aggregators.test:891[base]",
|
||||
"aggregators.test:891[instant-coarse]",
|
||||
"aggregators.test:896[base]",
|
||||
"aggregators.test:896[instant-coarse]",
|
||||
"aggregators.test:906[base]",
|
||||
"aggregators.test:906[instant-coarse]",
|
||||
"aggregators.test:909[base]",
|
||||
"aggregators.test:909[instant-coarse]",
|
||||
"aggregators.test:919[base]",
|
||||
"aggregators.test:919[instant-coarse]",
|
||||
"aggregators.test:922[base]",
|
||||
"aggregators.test:922[instant-coarse]",
|
||||
"aggregators.test:925[base]",
|
||||
"aggregators.test:925[instant-coarse]",
|
||||
"aggregators.test:930[base]",
|
||||
"aggregators.test:930[instant-coarse]",
|
||||
"aggregators.test:942[base]",
|
||||
"aggregators.test:942[instant-coarse]",
|
||||
"aggregators.test:945[base]",
|
||||
"aggregators.test:945[instant-coarse]",
|
||||
"aggregators.test:948[base]",
|
||||
"aggregators.test:948[instant-coarse]",
|
||||
"aggregators.test:953[base]",
|
||||
"aggregators.test:953[instant-coarse]",
|
||||
"aggregators.test:963[base]",
|
||||
"aggregators.test:963[instant-coarse]",
|
||||
"aggregators.test:966[base]",
|
||||
"aggregators.test:966[instant-coarse]",
|
||||
"operators.test:533[base]",
|
||||
"operators.test:533[instant-coarse]",
|
||||
"operators.test:539[base]",
|
||||
"trig_functions.test:13[base]",
|
||||
"trig_functions.test:13[instant-coarse]",
|
||||
"trig_functions.test:18[base]",
|
||||
"trig_functions.test:18[instant-coarse]",
|
||||
"trig_functions.test:23[base]",
|
||||
"trig_functions.test:23[instant-coarse]",
|
||||
"trig_functions.test:28[base]",
|
||||
"trig_functions.test:28[instant-coarse]",
|
||||
"trig_functions.test:33[base]",
|
||||
"trig_functions.test:33[instant-coarse]",
|
||||
"trig_functions.test:38[base]",
|
||||
"trig_functions.test:38[instant-coarse]",
|
||||
"trig_functions.test:43[base]",
|
||||
"trig_functions.test:43[instant-coarse]",
|
||||
"trig_functions.test:48[base]",
|
||||
"trig_functions.test:48[instant-coarse]",
|
||||
"trig_functions.test:53[base]",
|
||||
"trig_functions.test:53[instant-coarse]",
|
||||
"trig_functions.test:58[base]",
|
||||
"trig_functions.test:58[instant-coarse]",
|
||||
"trig_functions.test:63[base]",
|
||||
"trig_functions.test:63[instant-coarse]",
|
||||
"trig_functions.test:68[base]",
|
||||
"trig_functions.test:68[instant-coarse]",
|
||||
"trig_functions.test:73[base]",
|
||||
"trig_functions.test:73[instant-coarse]",
|
||||
"trig_functions.test:78[base]",
|
||||
"trig_functions.test:78[instant-coarse]",
|
||||
"trig_functions.test:83[base]",
|
||||
"trig_functions.test:83[instant-coarse]",
|
||||
"trig_functions.test:88[base]",
|
||||
"trig_functions.test:88[instant-coarse]",
|
||||
"trig_functions.test:8[base]",
|
||||
"trig_functions.test:8[instant-coarse]",
|
||||
"trig_functions.test:93[base]",
|
||||
"trig_functions.test:93[instant-coarse]"
|
||||
]
|
||||
}
|
||||
@@ -26,6 +26,11 @@ LEDGER_FILES = {
|
||||
"clickhousev2": os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences_v2.json"),
|
||||
}
|
||||
|
||||
# Cases SigNoz intentionally does not match, on every leg — kept out of the
|
||||
# ledgers because those track defects to be burned down and these are a product
|
||||
# decision. Enforced in both directions all the same.
|
||||
NONFINITE_EXCLUSIONS_FILE = os.path.join(TESTDATA_DIR, "promqltestcorpus", "nonfinite_exclusions.json")
|
||||
|
||||
# Every case replays on both legs, each asserted against the same frozen
|
||||
# expectations and its own ledger — deliberately never against each other: both
|
||||
# legs can sit within one rounding quantum of the expected value yet differ from
|
||||
@@ -202,6 +207,11 @@ def test_upstream_promqltest_corpus(
|
||||
# divergence is a regression, and a known divergence that starts passing
|
||||
# must be removed from the file. Problems across both legs are collected
|
||||
# before asserting so one leg's failure never hides the other's.
|
||||
nonfinite_exclusions: set[str] = set()
|
||||
if os.path.exists(NONFINITE_EXCLUSIONS_FILE):
|
||||
with open(NONFINITE_EXCLUSIONS_FILE, encoding="utf-8") as f:
|
||||
nonfinite_exclusions = set(json.load(f)["cases"])
|
||||
|
||||
problems: list[str] = []
|
||||
for leg, _ in LEGS:
|
||||
known: dict[str, str] = {}
|
||||
@@ -210,12 +220,15 @@ def test_upstream_promqltest_corpus(
|
||||
known = json.load(f)["divergences"]
|
||||
|
||||
failed_ids = {f_line.split(": ", 1)[0] for f_line in failures[leg]}
|
||||
unexpected = [f_line for f_line in failures[leg] if f_line.split(": ", 1)[0] not in known]
|
||||
unexpected = [f_line for f_line in failures[leg] if f_line.split(": ", 1)[0] not in known and f_line.split(": ", 1)[0] not in nonfinite_exclusions]
|
||||
now_passing = sorted(set(known) - failed_ids)
|
||||
stale_nonfinite_exclusions = sorted(nonfinite_exclusions - failed_ids)
|
||||
|
||||
if unexpected:
|
||||
problems.append(f"[{leg}] {len(unexpected)} corpus cases diverged beyond the known set:\n" + "\n".join(unexpected[:25]))
|
||||
if now_passing:
|
||||
problems.append(f"[{leg}] {len(now_passing)} known divergences now pass — remove them from {os.path.basename(LEDGER_FILES[leg])}: {now_passing[:25]}")
|
||||
if stale_nonfinite_exclusions:
|
||||
problems.append(f"[{leg}] {len(stale_nonfinite_exclusions)} non-finite exclusions no longer diverge, so the promql path has stopped dropping them — remove them from {os.path.basename(NONFINITE_EXCLUSIONS_FILE)}: {stale_nonfinite_exclusions[:25]}")
|
||||
|
||||
assert not problems, "\n\n".join(problems)
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
from uuid import uuid4
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.metrics import Metrics
|
||||
from fixtures.querier import get_all_series, make_query_request
|
||||
|
||||
HOUR_MS = 3_600_000
|
||||
SAMPLE_INTERVAL_MS = 60_000
|
||||
|
||||
|
||||
def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
# 12h ending on an hour boundary 15m ago — old enough to be cached.
|
||||
end_ms = (int((datetime.now(tz=UTC) - timedelta(minutes=15)).timestamp() * 1000) // HOUR_MS) * HOUR_MS
|
||||
start_ms = end_ms - 12 * HOUR_MS
|
||||
|
||||
sum_metric = f"job_duration_sum_{uuid4().hex[:8]}"
|
||||
count_metric = f"job_duration_count_{uuid4().hex[:8]}"
|
||||
|
||||
# active_job divides finite; idle_job is 0/0 at every step.
|
||||
series = {"active_job": (100.0, 4.0), "idle_job": (0.0, 0.0)}
|
||||
metrics: list[Metrics] = []
|
||||
for job_name, (sum_value, count_value) in series.items():
|
||||
for ts_ms in range(start_ms, end_ms + 1, SAMPLE_INTERVAL_MS):
|
||||
timestamp = datetime.fromtimestamp(ts_ms / 1000, tz=UTC)
|
||||
metrics.append(Metrics(metric_name=sum_metric, labels={"job_name": job_name}, timestamp=timestamp, value=sum_value))
|
||||
metrics.append(Metrics(metric_name=count_metric, labels={"job_name": job_name}, timestamp=timestamp, value=count_value))
|
||||
insert_metrics(metrics)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
promql = f"sum by (job_name) ({sum_metric}) / sum by (job_name) ({count_metric})"
|
||||
|
||||
def run() -> tuple[dict[str, dict[int, object]], int]:
|
||||
query = {"type": "promql", "spec": {"name": "A", "query": promql}}
|
||||
response = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
|
||||
assert response.status_code == HTTPStatus.OK, response.text[:300]
|
||||
body = response.json()
|
||||
out: dict[str, dict[int, object]] = {}
|
||||
for entry in get_all_series(body, "A") or []:
|
||||
labels = {l["key"]["name"]: str(l["value"]) for l in entry.get("labels") or []}
|
||||
out[labels["job_name"]] = {v["timestamp"]: v["value"] for v in entry.get("values") or []}
|
||||
return out, int(body["data"]["meta"]["stepIntervals"]["A"])
|
||||
|
||||
# First populates the cache, second must be served from it.
|
||||
first, step_seconds = run()
|
||||
second, _ = run()
|
||||
|
||||
expected_points = (end_ms - start_ms) // (step_seconds * 1000) + 1
|
||||
assert set(first) == {"active_job"}, f"the 0/0 series must not reach the response: {sorted(first)}"
|
||||
assert set(first["active_job"].values()) == {25.0}, sorted(set(first["active_job"].values()))
|
||||
assert len(first["active_job"]) == expected_points, f"expected {expected_points} points, got {len(first['active_job'])}"
|
||||
|
||||
# The cached read excludes end_ms, the one legitimate difference.
|
||||
assert set(second) == set(first), sorted(second)
|
||||
for job_name, points in first.items():
|
||||
expected = {ts: value for ts, value in points.items() if ts < end_ms}
|
||||
assert second[job_name] == expected, f"{job_name}: got {len(second[job_name])} of {len(expected)} points"
|
||||
Reference in New Issue
Block a user