Compare commits

...

10 Commits

Author SHA1 Message Date
Naman Verma
eed5020ee4 fix: drop non-finite values for promql, revert cache change 2026-08-10 18:06:41 +05:30
Naman Verma
0938381113 Merge branch 'main' into nv/promql-cache-nan 2026-08-10 16:47:23 +05:30
Naman Verma
e61755a143 Merge branch 'main' into nv/promql-cache-nan 2026-08-06 10:07:12 +05:30
Naman Verma
6defe22271 test: update unit tests 2026-08-04 16:55:19 +05:30
Naman Verma
87ee5f99af chore: remove bucket cache change 2026-08-04 14:49:33 +05:30
Naman Verma
97093bf0e4 Merge branch 'main' into nv/promql-cache-nan 2026-08-04 14:36:50 +05:30
Naman Verma
f41f541d2f test: add integration test 2026-08-04 14:36:22 +05:30
Naman Verma
73d40051ac chore: rearrange methods 2026-08-04 13:29:30 +05:30
Naman Verma
025dccec69 Merge branch 'main' into nv/promql-cache-nan 2026-08-04 12:55:45 +05:30
Naman Verma
e7000bbaa6 chore: first draft of NaN cache fix 2026-08-03 23:06:09 +05:30
3 changed files with 163 additions and 7 deletions

View File

@@ -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

View File

@@ -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)
}

View File

@@ -0,0 +1,63 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
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
SUM_METRIC = "job_duration_sum"
COUNT_METRIC = "job_duration_count"
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
# 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"