mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-10 15:00:47 +01:00
Compare commits
10 Commits
issue_2899
...
nv/promql-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
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)
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user