Compare commits

..

3 Commits

Author SHA1 Message Date
Ashwin Bhatkal
cbbedbaa3c Merge branch 'main' into ci/e2e-node-22 2026-11-06 12:53:25 +05:30
Ashwin Bhatkal
3be463e06a ci(e2e): pin Node 22 2026-11-03 22:58:22 +05:30
Ashwin Bhatkal
e11224105c ci(e2e): run the suite on 10 shards 2026-11-03 22:58:21 +05:30
23 changed files with 64 additions and 909 deletions

View File

@@ -47,7 +47,7 @@ jobs:
- name: node
uses: actions/setup-node@v4
with:
node-version: lts/*
node-version: "22"
- name: install-pnpm
uses: pnpm/action-setup@v6
with:
@@ -105,7 +105,7 @@ jobs:
- name: node
uses: actions/setup-node@v4
with:
node-version: lts/*
node-version: "22"
- name: install-pnpm
uses: pnpm/action-setup@v6
with:

View File

@@ -4027,9 +4027,7 @@ components:
the query they draw.
cursor:
description: The opaque nextCursor value from a previous raw result, used
to fetch the next page. Honored only by raw requests whose order starts
with timestamp, for logs followed by id in the same direction; every other
request ignores it.
to fetch the next page.
type: string
disabled:
description: When true, the result is excluded from the response; the query
@@ -4066,11 +4064,10 @@ components:
are replaced by group-by label values.
type: string
limit:
description: The maximum number of rows for raw and trace requests, or of
series or rows for grouped scalar and time_series requests, keeping the
top entries by order. When 0, scalar and time_series requests are unlimited,
raw requests ordered by timestamp return a page of 10000 rows to continue
with nextCursor, and other raw and trace requests return 100 rows.
description: The maximum number of rows for raw requests, or of series or
rows for grouped aggregations, keeping the top entries by order. When
0, aggregations are unlimited while raw requests ordered by timestamp
return a page of 10000 rows; follow nextCursor for the rest.
type: integer
limitBy:
allOf:
@@ -4082,11 +4079,8 @@ components:
operators and selectedQueryName refer to it.
type: string
offset:
description: The number of rows for raw and trace requests, or of ordered
series or rows for scalar and time_series requests, to skip before the
page starts. Requires limit. On raw requests the skip counts from where
a cursor resumes, and the skipped rows are still read, so prefer cursor
for paging.
description: The number of rows to skip before returning results. Prefer
cursor for raw requests.
type: integer
order:
description: The sort order of the result. Order by a group-by field, an
@@ -10108,10 +10102,6 @@ components:
description: The unique name of the formula, such as F1, used by selectedQueryName
and by panels. Referencing a formula from another formula is not reliable.
type: string
offset:
description: The number of ordered series or rows to skip before the page
starts. Requires limit.
type: integer
order:
description: The sort order of the formula result.
items:
@@ -10137,9 +10127,7 @@ components:
the query they draw.
cursor:
description: The opaque nextCursor value from a previous raw result, used
to fetch the next page. Honored only by raw requests whose order starts
with timestamp, for logs followed by id in the same direction; every other
request ignores it.
to fetch the next page.
type: string
disabled:
description: When true, the result is excluded from the response; the query
@@ -10175,11 +10163,10 @@ components:
are replaced by group-by label values.
type: string
limit:
description: The maximum number of rows for raw and trace requests, or of
series or rows for grouped scalar and time_series requests, keeping the
top entries by order. When 0, scalar and time_series requests are unlimited,
raw requests ordered by timestamp return a page of 10000 rows to continue
with nextCursor, and other raw and trace requests return 100 rows.
description: The maximum number of rows for raw requests, or of series or
rows for grouped aggregations, keeping the top entries by order. When
0, aggregations are unlimited while raw requests ordered by timestamp
return a page of 10000 rows; follow nextCursor for the rest.
type: integer
limitBy:
allOf:
@@ -10191,11 +10178,8 @@ components:
operators and selectedQueryName refer to it.
type: string
offset:
description: The number of rows for raw and trace requests, or of ordered
series or rows for scalar and time_series requests, to skip before the
page starts. Requires limit. On raw requests the skip counts from where
a cursor resumes, and the skipped rows are still read, so prefer cursor
for paging.
description: The number of rows to skip before returning results. Prefer
cursor for raw requests.
type: integer
order:
description: The sort order of the result. Order by a group-by field, an
@@ -10257,9 +10241,7 @@ components:
the query they draw.
cursor:
description: The opaque nextCursor value from a previous raw result, used
to fetch the next page. Honored only by raw requests whose order starts
with timestamp, for logs followed by id in the same direction; every other
request ignores it.
to fetch the next page.
type: string
disabled:
description: When true, the result is excluded from the response; the query
@@ -10296,11 +10278,10 @@ components:
are replaced by group-by label values.
type: string
limit:
description: The maximum number of rows for raw and trace requests, or of
series or rows for grouped scalar and time_series requests, keeping the
top entries by order. When 0, scalar and time_series requests are unlimited,
raw requests ordered by timestamp return a page of 10000 rows to continue
with nextCursor, and other raw and trace requests return 100 rows.
description: The maximum number of rows for raw requests, or of series or
rows for grouped aggregations, keeping the top entries by order. When
0, aggregations are unlimited while raw requests ordered by timestamp
return a page of 10000 rows; follow nextCursor for the rest.
type: integer
limitBy:
allOf:
@@ -10312,11 +10293,8 @@ components:
operators and selectedQueryName refer to it.
type: string
offset:
description: The number of rows for raw and trace requests, or of ordered
series or rows for scalar and time_series requests, to skip before the
page starts. Requires limit. On raw requests the skip counts from where
a cursor resumes, and the skipped rows are still read, so prefer cursor
for paging.
description: The number of rows to skip before returning results. Prefer
cursor for raw requests.
type: integer
order:
description: The sort order of the result. Order by a group-by field, an
@@ -10378,9 +10356,7 @@ components:
the query they draw.
cursor:
description: The opaque nextCursor value from a previous raw result, used
to fetch the next page. Honored only by raw requests whose order starts
with timestamp, for logs followed by id in the same direction; every other
request ignores it.
to fetch the next page.
type: string
disabled:
description: When true, the result is excluded from the response; the query
@@ -10417,11 +10393,10 @@ components:
are replaced by group-by label values.
type: string
limit:
description: The maximum number of rows for raw and trace requests, or of
series or rows for grouped scalar and time_series requests, keeping the
top entries by order. When 0, scalar and time_series requests are unlimited,
raw requests ordered by timestamp return a page of 10000 rows to continue
with nextCursor, and other raw and trace requests return 100 rows.
description: The maximum number of rows for raw requests, or of series or
rows for grouped aggregations, keeping the top entries by order. When
0, aggregations are unlimited while raw requests ordered by timestamp
return a page of 10000 rows; follow nextCursor for the rest.
type: integer
limitBy:
allOf:
@@ -10433,11 +10408,8 @@ components:
operators and selectedQueryName refer to it.
type: string
offset:
description: The number of rows for raw and trace requests, or of ordered
series or rows for scalar and time_series requests, to skip before the
page starts. Requires limit. On raw requests the skip counts from where
a cursor resumes, and the skipped rows are still read, so prefer cursor
for paging.
description: The number of rows to skip before returning results. Prefer
cursor for raw requests.
type: integer
order:
description: The sort order of the result. Order by a group-by field, an
@@ -10538,9 +10510,8 @@ components:
description: The unique name of the trace operator query.
type: string
offset:
description: The number of raw or trace rows, or of ordered series or rows
for time_series and scalar requests, to skip before the page starts. Requires
limit.
description: The number of raw results to skip before returning. Aggregation
requests ignore it.
type: integer
order:
description: The sort order of the matched traces; only span_count and trace_duration

View File

@@ -5273,7 +5273,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
/**
* @type string
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
*/
cursor?: string;
/**
@@ -5306,7 +5306,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
legend?: string;
/**
* @type integer
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
*/
limit?: number;
/**
@@ -5320,7 +5320,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
name?: string;
/**
* @type integer
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. Requires limit. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
*/
offset?: number;
/**
@@ -5665,7 +5665,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
/**
* @type string
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
*/
cursor?: string;
/**
@@ -5698,7 +5698,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
legend?: string;
/**
* @type integer
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
*/
limit?: number;
/**
@@ -5712,7 +5712,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
name?: string;
/**
* @type integer
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. Requires limit. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
*/
offset?: number;
/**
@@ -5845,7 +5845,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
/**
* @type string
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
*/
cursor?: string;
/**
@@ -5878,7 +5878,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
legend?: string;
/**
* @type integer
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
*/
limit?: number;
/**
@@ -5892,7 +5892,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
name?: string;
/**
* @type integer
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. Requires limit. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
*/
offset?: number;
/**
@@ -5944,7 +5944,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
/**
* @type string
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
*/
cursor?: string;
/**
@@ -5977,7 +5977,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
legend?: string;
/**
* @type integer
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
*/
limit?: number;
/**
@@ -5991,7 +5991,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
name?: string;
/**
* @type integer
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. Requires limit. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
*/
offset?: number;
/**
@@ -7064,11 +7064,6 @@ export interface Querybuildertypesv5QueryBuilderFormulaDTO {
* @description The unique name of the formula, such as F1, used by selectedQueryName and by panels. Referencing a formula from another formula is not reliable.
*/
name?: string;
/**
* @type integer
* @description The number of ordered series or rows to skip before the page starts. Requires limit.
*/
offset?: number;
/**
* @type array,null
* @description The sort order of the formula result.
@@ -7151,7 +7146,7 @@ export interface Querybuildertypesv5QueryBuilderTraceOperatorDTO {
name?: string;
/**
* @type integer
* @description The number of raw or trace rows, or of ordered series or rows for time_series and scalar requests, to skip before the page starts. Requires limit.
* @description The number of raw results to skip before returning. Aggregation requests ignore it.
*/
offset?: number;
/**

View File

@@ -182,8 +182,7 @@ func postProcessBuilderQuery[T any](
req *qbtypes.QueryRangeRequest,
) *qbtypes.Result {
// the statement's top-N already skipped offset groups by their whole-window value; only reorder here
result = q.applySeriesLimit(result, query.Limit, 0, query.Order)
result = q.applySeriesLimit(result, query.Limit, query.Order)
// Apply functions
if len(query.Functions) > 0 {
@@ -235,7 +234,7 @@ func postProcessMetricQuery(
reducesToScalar := req.RequestType == qbtypes.RequestTypeScalar && config.ReduceTo != qbtypes.ReduceToUnknown
if !reducesToScalar {
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
result = q.applySeriesLimit(result, query.Limit, query.Order)
}
if len(query.Functions) > 0 {
@@ -247,7 +246,7 @@ func postProcessMetricQuery(
// Apply reduce to for scalar request type
if reducesToScalar {
result = q.applyMetricReduceTo(result, config.ReduceTo)
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
result = q.applySeriesLimit(result, query.Limit, query.Order)
if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
result.Value = convertTimeSeriesDataToScalar(tsData, tsData.QueryName)
}
@@ -264,7 +263,7 @@ func postProcessTraceOperator(
req *qbtypes.QueryRangeRequest,
) *qbtypes.Result {
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
result = q.applySeriesLimit(result, query.Limit, query.Order)
// Apply functions if any
if len(query.Functions) > 0 {
@@ -297,7 +296,7 @@ func (q *querier) applyMetricReduceTo(result *qbtypes.Result, reduceOp qbtypes.R
}
// applySeriesLimit limits the number of series in the result.
func (q *querier) applySeriesLimit(result *qbtypes.Result, limit, offset int, orderBy []qbtypes.OrderBy) *qbtypes.Result {
func (q *querier) applySeriesLimit(result *qbtypes.Result, limit int, orderBy []qbtypes.OrderBy) *qbtypes.Result {
tsData, ok := result.Value.(*qbtypes.TimeSeriesData)
if !ok {
return result
@@ -306,7 +305,7 @@ func (q *querier) applySeriesLimit(result *qbtypes.Result, limit, offset int, or
if tsData != nil {
for _, agg := range tsData.Aggregations {
// Use the ApplySeriesLimit function from querybuildertypes
agg.Series = qbtypes.ApplySeriesLimit(agg.Series, orderBy, limit, offset)
agg.Series = qbtypes.ApplySeriesLimit(agg.Series, orderBy, limit)
}
}
@@ -358,7 +357,7 @@ func (q *querier) applyFormulas(ctx context.Context, results map[string]*qbtypes
case qbtypes.RequestTypeTimeSeries:
result := q.processTimeSeriesFormula(ctx, results, formula, req)
if result != nil {
result = q.applySeriesLimit(result, formula.Limit, formula.Offset, formula.Order)
result = q.applySeriesLimit(result, formula.Limit, formula.Order)
results[name] = result
}
case qbtypes.RequestTypeHeatmap:
@@ -371,7 +370,7 @@ func (q *querier) applyFormulas(ctx context.Context, results map[string]*qbtypes
bucketFormulaOutputAsHeatmap(tsData, bucketing)
resolveHeatmapBucketAxis(tsData, bucketing)
}
result = q.applySeriesLimit(result, formula.Limit, formula.Offset, formula.Order)
result = q.applySeriesLimit(result, formula.Limit, formula.Order)
results[name] = result
}
case qbtypes.RequestTypeScalar:
@@ -652,7 +651,7 @@ func (q *querier) processScalarFormula(
}
// Apply ordering (and limit) before converting to scalar format.
formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit, formula.Offset)
formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit)
// Convert back to scalar format
scalarResult := &qbtypes.ScalarData{

View File

@@ -186,7 +186,6 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
name string
reduceTo qbtypes.ReduceTo
order []qbtypes.OrderBy
offset int
expected [][]any
}{
{
@@ -212,13 +211,6 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
reduceTo: qbtypes.ReduceToMax,
expected: [][]any{{"payment", 40.0}, {"auth", 30.0}},
},
{
name: "Max_Desc_Offset1_SkipsHighestPeak",
reduceTo: qbtypes.ReduceToMax,
order: orderByValue(qbtypes.OrderDirectionDesc),
offset: 1,
expected: [][]any{{"auth", 30.0}, {"checkout", 10.0}},
},
}
for _, testCase := range testCases {
@@ -228,7 +220,6 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
Aggregations: []qbtypes.MetricAggregation{{MetricName: "http_requests_total", ReduceTo: testCase.reduceTo}},
Order: testCase.order,
Limit: 2,
Offset: testCase.offset,
}
req := &qbtypes.QueryRangeRequest{RequestType: qbtypes.RequestTypeScalar}

View File

@@ -536,10 +536,6 @@ func (b *auditQueryStatementBuilder) buildScalarQuery(
sb.Limit(query.Limit)
}
if query.Offset > 0 {
sb.Offset(query.Offset)
}
combinedArgs := append(allGroupByArgs, allAggChArgs...)
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)

View File

@@ -707,10 +707,6 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
sb.Limit(query.Limit)
}
if query.Offset > 0 {
sb.Offset(query.Offset)
}
combinedArgs := allAggChArgs
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)

View File

@@ -722,9 +722,6 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
if query.Limit > 0 {
sb.Limit(query.Limit)
}
if query.Offset > 0 {
sb.Offset(query.Offset)
}
}
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -771,9 +768,6 @@ func outerLimitSQL(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], tr
sb.OrderBy("__result_0 DESC")
}
sb.Limit(query.Limit)
if query.Offset > 0 {
sb.Offset(query.Offset)
}
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}

View File

@@ -835,10 +835,6 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
sb.Limit(query.Limit)
}
if query.Offset > 0 {
sb.Offset(query.Offset)
}
combinedArgs := allAggChArgs
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)

View File

@@ -841,10 +841,6 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
sb.Limit(b.operator.Limit)
}
if b.operator.Offset > 0 {
sb.Offset(b.operator.Offset)
}
combinedArgs := allAggChArgs
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
@@ -926,10 +922,6 @@ func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFr
sb.Limit(b.operator.Limit)
}
if b.operator.Offset > 0 {
sb.Offset(b.operator.Offset)
}
combinedArgs := allAggChArgs
// Add HAVING clause if specified

View File

@@ -34,13 +34,13 @@ type QueryBuilderQuery[T any] struct {
SelectFields []telemetrytypes.TelemetryFieldKey `json:"selectFields,omitzero" description:"The fields to return for each row in raw requests, in addition to the columns always present: timestamp and id for logs, timestamp, trace_id and span_id for traces. Ignored by aggregating requests."`
Limit int `json:"limit,omitempty" description:"The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows."`
Limit int `json:"limit,omitempty" description:"The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest."`
LimitBy *LimitBy `json:"limitBy,omitempty" description:"Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored."`
Offset int `json:"offset,omitempty" description:"The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. Requires limit. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging."`
Offset int `json:"offset,omitempty" description:"The number of rows to skip before returning results. Prefer cursor for raw requests."`
Cursor string `json:"cursor,omitempty" description:"The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it."`
Cursor string `json:"cursor,omitempty" description:"The opaque nextCursor value from a previous raw result, used to fetch the next page."`
Having *Having `json:"having,omitempty" description:"A filter applied to aggregation results, referencing aliases or expressions, for example count() > 100."`

View File

@@ -26,8 +26,6 @@ type QueryBuilderFormula struct {
Limit int `json:"limit,omitempty" description:"The maximum number of series or rows to return. Unlimited when 0."`
Offset int `json:"offset,omitempty" description:"The number of ordered series or rows to skip before the page starts. Requires limit."`
Having *Having `json:"having,omitempty" description:"Accepted but currently ignored; the computed values are returned unfiltered."`
Functions []Function `json:"functions,omitzero" description:"Post-processing functions applied in order to each result series."`
@@ -96,18 +94,6 @@ func (f QueryBuilderFormula) Validate() error {
)
}
if f.Offset < 0 {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"offset must be non-negative, got %d",
f.Offset,
)
}
if f.Offset > 0 && f.Limit == 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset requires a limit")
}
// Validate expression is parseable
if _, err := govaluate.NewEvaluableExpressionWithFunctions(f.Expression, EvalFuncs()); err != nil {
return errors.NewInvalidInputf(

View File

@@ -180,8 +180,6 @@ func (q *QueryEnvelope) GetOffset() int {
return spec.Offset
case QueryBuilderQuery[MetricAggregation]:
return spec.Offset
case QueryBuilderFormula:
return spec.Offset
}
return 0
}

View File

@@ -209,9 +209,6 @@ func (q *QueryEnvelope) SetOffset(offset int) {
case QueryBuilderQuery[MetricAggregation]:
spec.Offset = offset
q.Spec = spec
case QueryBuilderFormula:
spec.Offset = offset
q.Spec = spec
}
}

View File

@@ -13,8 +13,9 @@ const (
DefaultOrderByKey = "__result"
)
// ApplySeriesLimit sorts the series by the order criteria, skips the first offset of them and keeps at most limit.
func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit, offset int) []*TimeSeries {
// ApplyLimit applies limit and ordering to a list of time series
// This function sorts the series based on the provided order criteria and applies the limit.
func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit int) []*TimeSeries {
if len(series) == 0 {
return series
}
@@ -64,10 +65,6 @@ func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit, offset int
return 0
})
if offset > 0 {
series = series[min(offset, len(series)):]
}
// Apply limit if specified
if limit > 0 && len(series) > limit {
return series[:limit]

View File

@@ -84,7 +84,7 @@ func TestApplySeriesLimit(t *testing.T) {
},
}
result := ApplySeriesLimit(series, orderBy, 2, 0)
result := ApplySeriesLimit(series, orderBy, 2)
// Should return top 2 series by value: service-b (50.0), service-c (30.0)
assert.Len(t, result, 2)
@@ -209,7 +209,7 @@ func TestApplySeriesLimit(t *testing.T) {
},
}
result := ApplySeriesLimit(series, orderBy, 3, 0)
result := ApplySeriesLimit(series, orderBy, 3)
// Should return 3 series sorted by service (asc), then environment (desc)
// Expected order:
@@ -233,25 +233,6 @@ func TestApplySeriesLimit(t *testing.T) {
assert.Equal(t, "prod", result[2].Labels[1].Value)
assert.Equal(t, 40.0, result[2].Values[0].Value)
})
t.Run("Offset skips the top series before the limit applies", func(t *testing.T) {
hostKey := telemetrytypes.TelemetryFieldKey{Name: "host", FieldDataType: telemetrytypes.FieldDataTypeString}
series := []*TimeSeries{
{Labels: []*Label{{Key: hostKey, Value: "h1"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 10.0}}},
{Labels: []*Label{{Key: hostKey, Value: "h2"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 50.0}}},
{Labels: []*Label{{Key: hostKey, Value: "h3"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 30.0}}},
{Labels: []*Label{{Key: hostKey, Value: "h4"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 20.0}}},
}
// Default order is value descending: h2, h3, h4, h1. Skipping one leaves h3, h4.
result := ApplySeriesLimit(series, nil, 2, 1)
require.Len(t, result, 2)
assert.Equal(t, "h3", result[0].Labels[0].Value)
assert.Equal(t, "h4", result[1].Labels[0].Value)
assert.Empty(t, ApplySeriesLimit(series, nil, 2, 4))
})
}
func TestApplySeriesLimitRanksHeatmapSeriesByBucketTotals(t *testing.T) {
@@ -290,7 +271,7 @@ func TestApplySeriesLimitRanksHeatmapSeriesByBucketTotals(t *testing.T) {
},
}
result := ApplySeriesLimit(series, nil, 2, 0)
result := ApplySeriesLimit(series, nil, 2)
require.Len(t, result, 2)
assert.Equal(t, "busy", result[0].Labels[0].Value)

View File

@@ -55,7 +55,7 @@ type QueryBuilderTraceOperator struct {
Having *Having `json:"having,omitempty" description:"A condition on the aggregation results, for example count() > 10."`
Limit int `json:"limit,omitempty" description:"The maximum number of traces or groups to return. For raw requests 0 means 100 rows; for aggregations 0 means unlimited."`
Offset int `json:"offset,omitempty" description:"The number of raw or trace rows, or of ordered series or rows for time_series and scalar requests, to skip before the page starts. Requires limit."`
Offset int `json:"offset,omitempty" description:"The number of raw results to skip before returning. Aggregation requests ignore it."`
Cursor string `json:"cursor,omitempty" description:"Accepted but currently ignored by the trace operator; page raw results with offset."`
Legend string `json:"legend" description:"The display name template for each series; {{field}} placeholders are replaced by group-by label values."`
@@ -272,10 +272,6 @@ func (q *QueryBuilderTraceOperator) ValidatePagination() error {
)
}
if q.Offset > 0 && q.Limit == 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset requires a limit")
}
// For production use, you might want to enforce maximum limits
if q.Limit > 10000 {
return errors.WrapInvalidInputf(

View File

@@ -437,10 +437,6 @@ func (q *QueryBuilderQuery[T]) validateLimitAndPagination(cfg validationConfig)
)
}
if q.Offset > 0 && q.Limit == 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset requires a limit")
}
return nil
}

View File

@@ -300,9 +300,6 @@ def build_builder_query(
functions: list[dict] | None = None,
bucket_options: dict | None = None,
disabled: bool = False,
order: list[dict] | None = None,
limit: int | None = None,
offset: int | None = None,
) -> dict:
spec: dict[str, Any] = {
"name": name,
@@ -339,15 +336,6 @@ def build_builder_query(
if functions:
spec["functions"] = functions
if order:
spec["order"] = order
if limit is not None:
spec["limit"] = limit
if offset is not None:
spec["offset"] = offset
return {"type": "builder_query", "spec": spec}
@@ -360,7 +348,6 @@ def build_formula_query(
disabled: bool = False,
order: list[dict] | None = None,
limit: int | None = None,
offset: int | None = None,
) -> dict:
spec: dict[str, Any] = {
"name": name,
@@ -375,8 +362,6 @@ def build_formula_query(
spec["order"] = order
if limit is not None:
spec["limit"] = limit
if offset is not None:
spec["offset"] = offset
return {"type": "builder_formula", "spec": spec}
@@ -683,7 +668,6 @@ def build_scalar_query(
group_by: list[dict] | None = None,
order: list[dict] | None = None,
limit: int | None = None,
offset: int | None = None,
filter_expression: str | None = None,
having_expression: str | None = None,
step_interval: int = DEFAULT_STEP_INTERVAL,
@@ -710,9 +694,6 @@ def build_scalar_query(
if limit is not None:
spec["limit"] = limit
if offset is not None:
spec["offset"] = offset
if filter_expression:
spec["filter"] = {"expression": filter_expression}

View File

@@ -1,145 +0,0 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.logs import Logs
from fixtures.querier import (
Aggregation,
BuilderQuery,
RequestType,
TelemetryFieldKey,
get_all_series,
make_query_request,
)
@pytest.mark.parametrize(
"limit, offset, expected_services",
[
# limit=2 selects the top two groups in SQL first; the skipped group must still count toward that page
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_limited_page"),
pytest.param(2, 4, [], id="offset_past_the_end"),
],
)
def test_logs_time_series_offset_skips_series_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[list[Logs]], None],
limit: int,
offset: int,
expected_services: list[str],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# Ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1, all in the bucket before now.
service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=i + 1),
resources={"service.name": service},
body=f"{service} log {i}",
)
for service, count in service_counts.items()
for i in range(count)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
BuilderQuery(
signal="logs",
name="A",
aggregations=[Aggregation("count()")],
group_by=[TelemetryFieldKey("service.name", "string", "resource")],
step_interval=60,
limit=limit,
offset=offset,
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK, response.text
bucket_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
assert [
(
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
[(point["timestamp"], point["value"]) for point in series["values"]],
)
for series in get_all_series(response.json(), "A")
] == [(service, [(bucket_ms, service_counts[service])]) for service in expected_services]
def test_logs_time_series_offset_pages_are_disjoint(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[list[Logs]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# steady: 2 logs in each of 5 minutes, 10 in all. spike: 6 logs in one minute.
# steady has the larger total, spike the larger single-minute count.
insert_logs(
[
Logs(
timestamp=now - timedelta(minutes=minute, seconds=i + 1),
resources={"service.name": "steady"},
body=f"steady log {minute}-{i}",
)
for minute in range(1, 6)
for i in range(2)
]
+ [
Logs(
timestamp=now - timedelta(minutes=1, seconds=i + 1),
resources={"service.name": "spike"},
body=f"spike log {i}",
)
for i in range(6)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
pages = []
for offset in (0, 1):
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=10)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
BuilderQuery(
signal="logs",
name="A",
aggregations=[Aggregation("count()")],
group_by=[TelemetryFieldKey("service.name", "string", "resource")],
step_interval=60,
limit=1,
offset=offset,
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK, response.text
pages.append(
[
(
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
sorted(point["value"] for point in series["values"]),
)
for series in get_all_series(response.json(), "A")
]
)
# Pages follow the total count over the window: steady (10) first, then spike (6).
assert pages == [[("steady", [2, 2, 2, 2, 2])], [("spike", [6])]]

View File

@@ -1,74 +0,0 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.metrics import Metrics
from fixtures.querier import build_builder_query, get_all_series, make_query_request
@pytest.mark.parametrize(
"limit, offset, expected_services",
[
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
pytest.param(2, 4, [], id="offset_past_the_end"),
],
)
def test_metrics_time_series_offset_skips_series_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
limit: int,
offset: int,
expected_services: list[str],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# Ordered by value, descending: service-c 70, service-a 50, service-b 30, service-d 10, one sample each in the bucket before now.
service_values = {"service-a": 50.0, "service-b": 30.0, "service-c": 70.0, "service-d": 10.0}
insert_metrics(
[
Metrics(
metric_name="offset.gauge",
labels={"service.name": service},
timestamp=now - timedelta(seconds=1),
temporality="Unspecified",
type_="Gauge",
is_monotonic=False,
value=value,
)
for service, value in service_values.items()
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
queries=[
build_builder_query(
"A",
"offset.gauge",
"avg",
"sum",
temporality="unspecified",
group_by=["service.name"],
limit=limit,
offset=offset,
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
assert [
(
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
[point["value"] for point in series["values"]],
)
for series in get_all_series(response.json(), "A")
] == [(service, [service_values[service]]) for service in expected_services]

View File

@@ -1,264 +0,0 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.logs import Logs
from fixtures.metrics import Metrics
from fixtures.querier import (
RequestType,
build_aggregation,
build_formula_query,
build_group_by_field,
build_metrics_aggregation,
build_order_by,
build_scalar_query,
find_named_result,
get_scalar_table_data,
make_query_request,
make_scalar_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
# Four services ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1.
service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
@pytest.mark.parametrize(
"limit, offset, expected_services",
[
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
pytest.param(2, 4, [], id="offset_past_the_end"),
],
)
def test_logs_scalar_offset_skips_rows_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[list[Logs]], None],
limit: int,
offset: int,
expected_services: list[str],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=i + 1),
resources={"service.name": service},
body=f"{service} log {i}",
)
for service, count in service_counts.items()
for i in range(count)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_scalar_query_request(
signoz,
token,
now,
[
build_scalar_query(
name="A",
signal="logs",
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("service.name")],
order=[build_order_by("count()", "desc")],
limit=limit,
offset=offset,
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
assert get_scalar_table_data(response.json()) == [[service, service_counts[service]] for service in expected_services]
@pytest.mark.parametrize(
"limit, offset, expected_services",
[
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
pytest.param(2, 4, [], id="offset_past_the_end"),
],
)
def test_traces_scalar_offset_skips_rows_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
limit: int,
offset: int,
expected_services: list[str],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=i + 1),
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
resources={"service.name": service},
name=f"{service} span {i}",
)
for service, count in service_counts.items()
for i in range(count)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_scalar_query_request(
signoz,
token,
now,
[
build_scalar_query(
name="A",
signal="traces",
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("service.name")],
order=[build_order_by("count()", "desc")],
limit=limit,
offset=offset,
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
assert get_scalar_table_data(response.json()) == [[service, service_counts[service]] for service in expected_services]
def test_metrics_scalar_offset_skips_rows_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# Ordered by value, descending: service-c 70, service-a 50, service-b 30, service-d 10.
insert_metrics(
[
Metrics(
metric_name="offset.gauge",
labels={"service.name": service},
timestamp=now - timedelta(seconds=1),
temporality="Unspecified",
type_="Gauge",
is_monotonic=False,
value=value,
)
for service, value in {"service-a": 50.0, "service-b": 30.0, "service-c": 70.0, "service-d": 10.0}.items()
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_scalar_query_request(
signoz,
token,
now,
[
build_scalar_query(
name="A",
signal="metrics",
aggregations=[build_metrics_aggregation("offset.gauge", "latest", "sum", "unspecified", reduce_to="last")],
group_by=[build_group_by_field("service.name", "string", "attribute")],
order=[build_order_by("sum(offset.gauge)", "desc")],
limit=2,
offset=1,
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
assert get_scalar_table_data(response.json()) == [["service-a", 50.0], ["service-b", 30.0]]
def test_formula_scalar_offset_skips_rows_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[list[Logs]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
insert_logs(
[
Logs(
timestamp=now - timedelta(seconds=i + 1),
resources={"service.name": service},
body=f"{service} log {i}",
)
for service, count in service_counts.items()
for i in range(count)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
request_type=RequestType.SCALAR,
queries=[
build_scalar_query(
name="A",
signal="logs",
aggregations=[build_aggregation("count()")],
group_by=[build_group_by_field("service.name")],
disabled=True,
),
build_formula_query(
"F1",
"A * 2",
order=[build_order_by("__result", "desc")],
limit=2,
offset=1,
),
],
)
assert response.status_code == HTTPStatus.OK, response.text
result = find_named_result(response.json()["data"]["data"]["results"], "F1")
assert result is not None, "formula result F1 not found"
service_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "service.name")
value_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "__result")
assert [[row[service_column], row[value_column]] for row in result["data"]] == [["service-a", 10.0], ["service-b", 6.0]]
def test_negative_offset_is_rejected(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_scalar_query_request(
signoz,
token,
datetime.now(tz=UTC).replace(second=0, microsecond=0),
[build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert response.json()["error"]["message"] == "invalid query 'A': offset must be non-negative, got -1"
def test_offset_without_limit_is_rejected(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
) -> None:
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_scalar_query_request(
signoz,
token,
datetime.now(tz=UTC).replace(second=0, microsecond=0),
[build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=1)],
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
assert response.json()["error"]["message"] == "invalid query 'A': offset requires a limit"

View File

@@ -1,224 +0,0 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import (
Aggregation,
BuilderQuery,
RequestType,
TelemetryFieldKey,
find_named_result,
get_all_series,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
@pytest.mark.parametrize(
"limit, offset, expected_services",
[
# limit=2 selects the top two groups in SQL first; the skipped group must still count toward that page
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_limited_page"),
pytest.param(2, 4, [], id="offset_past_the_end"),
],
)
def test_traces_time_series_offset_skips_series_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
limit: int,
offset: int,
expected_services: list[str],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# Ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1, all in the bucket before now.
service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=i + 1),
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
resources={"service.name": service},
name=f"{service} span {i}",
)
for service, count in service_counts.items()
for i in range(count)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
BuilderQuery(
signal="traces",
name="A",
aggregations=[Aggregation("count()")],
group_by=[TelemetryFieldKey("service.name", "string", "resource")],
step_interval=60,
limit=limit,
offset=offset,
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK, response.text
bucket_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
assert [
(
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
[(point["timestamp"], point["value"]) for point in series["values"]],
)
for series in get_all_series(response.json(), "A")
] == [(service, [(bucket_ms, service_counts[service])]) for service in expected_services]
def test_trace_operator_scalar_offset_skips_rows_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
span_counts = {"offset-a": 4, "offset-b": 3, "offset-c": 2, "offset-d": 1}
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=10),
duration=timedelta(seconds=1),
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
parent_span_id="",
name="scalar-offset",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources={"service.name": service},
attributes={"operation.type": "scalar-offset"},
)
for service, count in span_counts.items()
for _ in range(count)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
request_type=RequestType.SCALAR,
queries=[
{
"type": "builder_query",
"spec": {
"name": "A",
"signal": "traces",
"disabled": True,
"filter": {"expression": "operation.type = 'scalar-offset'"},
"aggregations": [{"expression": "count()"}],
},
},
{
"type": "builder_trace_operator",
"spec": {
"name": "C",
"expression": "A",
"aggregations": [{"expression": "count()"}],
"groupBy": [{"name": "service.name", "fieldContext": "resource"}],
"limit": 2,
"offset": 1,
},
},
],
)
assert response.status_code == HTTPStatus.OK, response.text
result = find_named_result(response.json()["data"]["data"]["results"], "C")
assert result is not None, "trace_operator result C not found"
assert result["data"] == [["offset-b", 3], ["offset-c", 2]]
def test_traces_time_series_offset_pages_are_disjoint(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# steady: 2 spans in each of 5 minutes, 10 in all. spike: 6 spans in one minute.
# steady has the larger total, spike the larger single-minute count.
insert_traces(
[
Traces(
timestamp=now - timedelta(minutes=minute, seconds=i + 1),
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
resources={"service.name": "steady"},
name=f"steady span {minute}-{i}",
)
for minute in range(1, 6)
for i in range(2)
]
+ [
Traces(
timestamp=now - timedelta(minutes=1, seconds=i + 1),
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
resources={"service.name": "spike"},
name=f"spike span {i}",
)
for i in range(6)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
pages = []
for offset in (0, 1):
response = make_query_request(
signoz,
token,
start_ms=int((now - timedelta(minutes=10)).timestamp() * 1000),
end_ms=int(now.timestamp() * 1000),
request_type=RequestType.TIME_SERIES,
queries=[
BuilderQuery(
signal="traces",
name="A",
aggregations=[Aggregation("count()")],
group_by=[TelemetryFieldKey("service.name", "string", "resource")],
step_interval=60,
limit=1,
offset=offset,
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK, response.text
pages.append(
[
(
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
sorted(point["value"] for point in series["values"]),
)
for series in get_all_series(response.json(), "A")
]
)
# Pages follow the total count over the window: steady (10) first, then spike (6).
assert pages == [[("steady", [2, 2, 2, 2, 2])], [("spike", [6])]]