Compare commits

..

4 Commits

Author SHA1 Message Date
nityanandagohain
5dae0b975a feat: add trace summary endpoint 2026-09-18 17:55:43 +05:30
Naman Verma
d1a382945c fix: read Bearer/bearer/BEARER properly in v2 for webhook notification channels (#12890)
<!--A few plain bullets saying what changed and why, for a reviewer
skimming it - not a wall of text, not a restatement of the diff, not
generated boilerplate.-->
#### Description

Webhook notification channels with a bearer token authorisation work in
v1 with all three spellings `Bearer/bearer/BEARER`, but v2 API was not
accepting anything other than `Bearer`. This PR changes the conversion
from receiver -> gettable flow.

Also, error messages are made better in 2 places.
2026-09-17 11:57:11 +00:00
Nikhil Soni
59af5e0367 refactor(telemetrystore): drop app-side bulk-filtering override (#12871)
#### Description

- Stop managing `secondary_indices_enable_bulk_filtering` from the app.
It was hardcoded to `false` in the query hook as a workaround for
[ClickHouse#82283](https://github.com/ClickHouse/ClickHouse/issues/82283)
(`CANNOT_READ_ALL_DATA` with SET-type skip indexes).
- That bug is fixed
([ClickHouse#87817](https://github.com/ClickHouse/ClickHouse/pull/87817),
backported to 25.7/25.8/25.9) and prod runs 25.12. Verified on a local
25.12.5 container against `signoz_index_v3` that the crash no longer
reproduces with bulk filtering enabled.
- Removes the hook override plus the now-unused config field and
`example.yaml` entry. The setting reverts to being controlled
server-side via ClickHouse profiles (the charts change tracked in the
same issue).

#### Issues closed by this PR

Closes SigNoz/engineering-pod#5915

🤖 Generated with [Claude Code](https://claude.com/claude-code)
2026-09-17 09:40:26 +00:00
Pandey
8286e787b2 fix(tracefunnel): quote step names in slow and error trace queries (#12886)
Some checks failed
build-staging / prepare (push) Has been cancelled
build-staging / js-build (push) Has been cancelled
build-staging / go-build (push) Has been cancelled
build-staging / staging (push) Has been cancelled
cacheci / tests (push) Has been cancelled
Release Drafter / update_release_draft (push) Has been cancelled
#### Description

- `#12593` moved the n-step trace-funnel query builders onto
`clickhousesql.StringLiteral`, but the two-step `slow-traces` and
`error-traces` builders still interpolated `service_name`/`span_name`
into the SQL string literal raw.
- Route those four values through the same helper, so every funnel query
builder quotes step names consistently.

#### Additional Information

- No behaviour change for ordinary names; the
`slow-traces`/`error-traces` funnel queries now handle names containing
a quote the same way the rest of the module already does.
2026-09-17 07:01:21 +00:00
34 changed files with 1056 additions and 1177 deletions

View File

@@ -67,6 +67,7 @@ jobs:
- semconvfamilies
- serviceaccount
- spanmapper
- tracedetail
- querier_json_body
- querier_skip_resource_fingerprint
- ttl

View File

@@ -202,7 +202,6 @@ telemetrystore:
max_bytes_to_read: 0
max_result_rows: 0
ignore_data_skipping_indices: ""
secondary_indices_enable_bulk_filtering: false
##################### Prometheus #####################
prometheus:

View File

@@ -9439,6 +9439,29 @@ components:
required:
- aggregations
type: object
SpantypesGettableTraceSummary:
properties:
ai:
$ref: '#/components/schemas/SpantypesTraceAISummary'
endTimestampMillis:
minimum: 0
type: integer
hasMissingSpans:
type: boolean
rootServiceEntryPoint:
type: string
rootServiceName:
type: string
startTimestampMillis:
minimum: 0
type: integer
totalErrorSpansCount:
minimum: 0
type: integer
totalSpansCount:
minimum: 0
type: integer
type: object
SpantypesGettableWaterfallTrace:
properties:
endTimestampMillis:
@@ -9722,6 +9745,32 @@ components:
nullable: true
type: object
type: object
SpantypesTraceAISummary:
properties:
tokens:
$ref: '#/components/schemas/SpantypesTraceAITokens'
totalCost:
nullable: true
type: number
type: object
SpantypesTraceAITokens:
properties:
cacheRead:
minimum: 0
type: integer
cacheWrite:
minimum: 0
type: integer
input:
minimum: 0
type: integer
output:
minimum: 0
type: integer
reasoning:
minimum: 0
type: integer
type: object
SpantypesUpdatableSpanMapper:
properties:
config:
@@ -15460,6 +15509,66 @@ paths:
summary: Get aggregations for a trace
tags:
- tracedetail
/api/v1/traces/{traceID}/summary:
get:
deprecated: false
description: Returns the trace-level fields of the waterfall (time range, root,
span counts, missing spans) and, when the trace has gen_ai spans, its token
and cost totals. Computed in one aggregate query.
operationId: GetTraceSummary
parameters:
- in: path
name: traceID
required: true
schema:
type: string
responses:
"200":
content:
application/json:
schema:
properties:
data:
$ref: '#/components/schemas/SpantypesGettableTraceSummary'
status:
type: string
required:
- status
- data
type: object
description: OK
"401":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Unauthorized
"403":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Forbidden
"404":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Not Found
"500":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
security:
- api_key:
- VIEWER
- tokenizer:
- VIEWER
summary: Get summary for a trace
tags:
- tracedetail
/api/v1/user/me:
get:
deprecated: true

View File

@@ -10884,6 +10884,78 @@ export interface SpantypesGettableTraceAggregationsDTO {
aggregations: SpantypesSpanAggregationResultDTO[];
}
export interface SpantypesTraceAITokensDTO {
/**
* @type integer
* @minimum 0
*/
cacheRead?: number;
/**
* @type integer
* @minimum 0
*/
cacheWrite?: number;
/**
* @type integer
* @minimum 0
*/
input?: number;
/**
* @type integer
* @minimum 0
*/
output?: number;
/**
* @type integer
* @minimum 0
*/
reasoning?: number;
}
export interface SpantypesTraceAISummaryDTO {
tokens?: SpantypesTraceAITokensDTO;
/**
* @type number,null
*/
totalCost?: number | null;
}
export interface SpantypesGettableTraceSummaryDTO {
ai?: SpantypesTraceAISummaryDTO;
/**
* @type integer
* @minimum 0
*/
endTimestampMillis?: number;
/**
* @type boolean
*/
hasMissingSpans?: boolean;
/**
* @type string
*/
rootServiceEntryPoint?: string;
/**
* @type string
*/
rootServiceName?: string;
/**
* @type integer
* @minimum 0
*/
startTimestampMillis?: number;
/**
* @type integer
* @minimum 0
*/
totalErrorSpansCount?: number;
/**
* @type integer
* @minimum 0
*/
totalSpansCount?: number;
}
export interface SpantypesOtelSpanRefDTO {
/**
* @type string
@@ -12595,6 +12667,17 @@ export type GetTraceAggregations200 = {
status: string;
};
export type GetTraceSummaryPathParameters = {
traceID: string;
};
export type GetTraceSummary200 = {
data: SpantypesGettableTraceSummaryDTO;
/**
* @type string
*/
status: string;
};
export type ListUserPreferences200 = {
/**
* @type array

View File

@@ -4,11 +4,17 @@
* * regenerate with 'pnpm generate:api'
* SigNoz
*/
import { useMutation } from 'react-query';
import { useMutation, useQuery } from 'react-query';
import type {
InvalidateOptions,
MutationFunction,
QueryClient,
QueryFunction,
QueryKey,
UseMutationOptions,
UseMutationResult,
UseQueryOptions,
UseQueryResult,
} from 'react-query';
import type {
@@ -16,6 +22,8 @@ import type {
GetFlamegraphPathParameters,
GetTraceAggregations200,
GetTraceAggregationsPathParameters,
GetTraceSummary200,
GetTraceSummaryPathParameters,
GetWaterfallV4200,
GetWaterfallV4PathParameters,
RenderErrorResponseDTO,
@@ -27,6 +35,26 @@ import type {
import { GeneratedAPIInstance } from '../../../generatedAPIInstance';
import type { ErrorType, BodyType } from '../../../generatedAPIInstance';
const withQueryKey = <T extends object, K>(
query: T,
queryKey: K,
): T & { queryKey: K } => {
const result = { queryKey } as T & { queryKey: K };
for (const key of Object.keys(query)) {
// The explicit queryKey always wins, matching the previous
// `{ ...query, queryKey }` spread where it was set last.
if (key === 'queryKey') {
continue;
}
Object.defineProperty(result, key, {
enumerable: true,
configurable: true,
get: () => (query as Record<string, unknown>)[key],
});
}
return result;
};
/**
* Computes span aggregations grouped by requested field.
* @summary Get aggregations for a trace
@@ -127,6 +155,108 @@ export const useGetTraceAggregations = <
> => {
return useMutation(getGetTraceAggregationsMutationOptions(options));
};
/**
* Returns the trace-level fields of the waterfall (time range, root, span counts, missing spans) and, when the trace has gen_ai spans, its token and cost totals. Computed in one aggregate query.
* @summary Get summary for a trace
*/
export const getTraceSummary = (
{ traceID }: GetTraceSummaryPathParameters,
signal?: AbortSignal,
) => {
return GeneratedAPIInstance<GetTraceSummary200>({
url: `/api/v1/traces/${traceID}/summary`,
method: 'GET',
signal,
});
};
export const getGetTraceSummaryQueryKey = ({
traceID,
}: GetTraceSummaryPathParameters) => {
return [`/api/v1/traces/${traceID}/summary`] as const;
};
export const getGetTraceSummaryQueryOptions = <
TData = Awaited<ReturnType<typeof getTraceSummary>>,
TError = ErrorType<RenderErrorResponseDTO>,
>(
{ traceID }: GetTraceSummaryPathParameters,
options?: {
query?: UseQueryOptions<
Awaited<ReturnType<typeof getTraceSummary>>,
TError,
TData
>;
},
) => {
const { query: queryOptions } = options ?? {};
const queryKey =
queryOptions?.queryKey ?? getGetTraceSummaryQueryKey({ traceID });
const queryFn: QueryFunction<Awaited<ReturnType<typeof getTraceSummary>>> = ({
signal,
}) => getTraceSummary({ traceID }, signal);
return {
queryKey,
queryFn,
enabled: traceID !== null && traceID !== undefined,
...queryOptions,
} as UseQueryOptions<
Awaited<ReturnType<typeof getTraceSummary>>,
TError,
TData
> & { queryKey: QueryKey };
};
export type GetTraceSummaryQueryResult = NonNullable<
Awaited<ReturnType<typeof getTraceSummary>>
>;
export type GetTraceSummaryQueryError = ErrorType<RenderErrorResponseDTO>;
/**
* @summary Get summary for a trace
*/
export function useGetTraceSummary<
TData = Awaited<ReturnType<typeof getTraceSummary>>,
TError = ErrorType<RenderErrorResponseDTO>,
>(
{ traceID }: GetTraceSummaryPathParameters,
options?: {
query?: UseQueryOptions<
Awaited<ReturnType<typeof getTraceSummary>>,
TError,
TData
>;
},
): UseQueryResult<TData, TError> & { queryKey: QueryKey } {
const queryOptions = getGetTraceSummaryQueryOptions({ traceID }, options);
const query = useQuery(queryOptions) as UseQueryResult<TData, TError> & {
queryKey: QueryKey;
};
return withQueryKey(query, queryOptions.queryKey);
}
/**
* @summary Get summary for a trace
*/
export const invalidateGetTraceSummary = async (
queryClient: QueryClient,
{ traceID }: GetTraceSummaryPathParameters,
options?: InvalidateOptions,
): Promise<QueryClient> => {
await queryClient.invalidateQueries(
{ queryKey: getGetTraceSummaryQueryKey({ traceID }) },
options,
);
return queryClient;
};
/**
* Returns the flamegraph view of spans for a given trace ID.
* @summary Get flamegraph view for a trace

View File

@@ -10,6 +10,23 @@ import (
)
func (provider *provider) addTraceDetailRoutes(router *mux.Router) error {
if err := router.Handle("/api/v1/traces/{traceID}/summary", handler.New(
provider.authzMiddleware.ViewAccess(provider.traceDetailHandler.GetTraceSummary),
handler.OpenAPIDef{
ID: "GetTraceSummary",
Tags: []string{"tracedetail"},
Summary: "Get summary for a trace",
Description: "Returns the trace-level fields of the waterfall (time range, root, span counts, missing spans) and, when the trace has gen_ai spans, its token and cost totals. Computed in one aggregate query.",
Response: new(spantypes.GettableTraceSummary),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusNotFound},
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
},
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v4/traces/{traceID}/waterfall", handler.New(
provider.authzMiddleware.ViewAccess(provider.traceDetailHandler.GetWaterfallV4),
handler.OpenAPIDef{

View File

@@ -6,7 +6,9 @@ import (
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/SigNoz/signoz/pkg/modules/tracedetail"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/types/spantypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/gorilla/mux"
)
@@ -18,6 +20,27 @@ func NewHandler(module tracedetail.Module) tracedetail.Handler {
return &handler{module: module}
}
func (h *handler) GetTraceSummary(rw http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(rw, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(rw, err)
return
}
stats, err := h.module.GetTraceStats(r.Context(), orgID, mux.Vars(r)["traceID"])
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, spantypes.NewGettableTraceSummary(stats))
}
func (h *handler) GetWaterfallV4(rw http.ResponseWriter, r *http.Request) {
req := new(spantypes.PostableWaterfall)
if err := binding.JSON.BindBody(r.Body, req); err != nil {

View File

@@ -8,6 +8,7 @@ import (
"github.com/SigNoz/signoz/pkg/modules/tracedetail"
"github.com/SigNoz/signoz/pkg/types/spantypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"go.opentelemetry.io/otel/metric"
)
@@ -39,6 +40,21 @@ func NewModule(traceStore spantypes.TraceStore, providerSettings factory.Provide
return m
}
func (m *module) GetTraceStats(ctx context.Context, orgID valuer.UUID, traceID string) (*spantypes.TraceStats, error) {
summary, err := m.store.GetTraceSummary(ctx, traceID)
if err != nil {
return nil, err
}
stats, err := m.store.GetTraceStats(ctx, orgID, traceID, summary)
if err != nil {
return nil, err
}
if stats.TotalSpans == 0 {
return nil, spantypes.ErrTraceNotFound
}
return stats, nil
}
// GetWaterfallV4 is the OOM-safe V4 waterfall.
// For large traces (NumSpans > effectiveLimit) it uses a two-step fetch:
// minimal fields for all spans to build the tree, then full fields for the

View File

@@ -10,9 +10,15 @@ import (
"github.com/SigNoz/signoz/pkg/clickhousesql"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/spantypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
)
const colServiceName = `resource_string_service$$$$name` // $ gets escaped so $$$$ converts to $$.
@@ -38,10 +44,18 @@ type spanDurationRow struct {
type traceStore struct {
telemetryStore telemetrystore.TelemetryStore
metadataStore telemetrytypes.MetadataStore
storage qbtypes.Storage
flagger flagger.Flagger
}
func NewTraceStore(ts telemetrystore.TelemetryStore) *traceStore {
return &traceStore{telemetryStore: ts}
func NewTraceStore(ts telemetrystore.TelemetryStore, metadataStore telemetrytypes.MetadataStore, fl flagger.Flagger) *traceStore {
return &traceStore{
telemetryStore: ts,
metadataStore: metadataStore,
storage: tracestelemetryschema.NewStorage(),
flagger: fl,
}
}
func (s *traceStore) GetTraceSummary(ctx context.Context, traceID string) (*spantypes.TraceSummary, error) {
@@ -65,6 +79,131 @@ func (s *traceStore) GetTraceSummary(ctx context.Context, traceID string) (*span
return &summary, nil
}
func (s *traceStore) GetTraceStats(ctx context.Context, orgID valuer.UUID, traceID string, summary *spantypes.TraceSummary) (*spantypes.TraceStats, error) {
table := fmt.Sprintf("%s.%s", spantypes.TraceDB, spantypes.TraceTable)
spans := sqlbuilder.NewSelectBuilder()
genAIColumns, err := s.genAISpanColumns(ctx, orgID, summary, spans)
if err != nil {
return nil, err
}
// A span whose parent was never recorded hangs off a synthetic "Missing Span" root in the waterfall.
ids := sqlbuilder.NewSelectBuilder()
ids.Select("span_id")
ids.From(table)
ids.Where(
ids.E("trace_id", traceID),
ids.GE("ts_bucket_start", summary.Start.Unix()-1800),
ids.LE("ts_bucket_start", summary.End.Unix()),
)
missingParent := fmt.Sprintf("parent_span_id <> '' AND parent_span_id GLOBAL NOT IN (%s)", spans.Var(ids))
spans.Select(
"toUnixTimestamp64Nano(timestamp) AS span_start_ns",
"span_start_ns + duration_nano AS span_end_ns",
"span_id",
"has_error",
"("+missingParent+") AS has_missing_parent",
"(parent_span_id = '' OR has_missing_parent) AS is_root",
"if(parent_span_id = '', name, 'Missing Span') AS root_name",
"if(parent_span_id = '', "+colServiceName+", '') AS root_service",
)
spans.SelectMore(genAIColumns...)
spans.From(table)
spans.Where(
spans.E("trace_id", traceID),
spans.GE("ts_bucket_start", summary.Start.Unix()-1800),
spans.LE("ts_bucket_start", summary.End.Unix()),
)
spans.SQL("LIMIT 1 BY span_id")
sb := sqlbuilder.NewSelectBuilder()
sb.Select(
"toUInt64(min(span_start_ns)) AS start_ns",
"toUInt64(max(span_end_ns)) AS end_ns",
"count() AS total_spans",
"countIf(has_error) AS total_error_spans",
"countIf(has_missing_parent) > 0 AS has_missing_spans",
"argMinIf(root_service, (span_start_ns, root_name), is_root) AS root_service_name",
"argMinIf(root_name, (span_start_ns, root_name), is_root) AS root_entry_point",
"countIf(is_gen_ai) AS gen_ai_span_count",
"toUInt64(coalesce(sum(input_tokens_value), 0)) AS input_tokens",
"toUInt64(coalesce(sum(output_tokens_value), 0)) AS output_tokens",
"toUInt64(coalesce(sum(cache_read_tokens_value), 0)) AS cache_read_tokens",
"toUInt64(coalesce(sum(cache_write_tokens_value), 0)) AS cache_write_tokens",
"toUInt64(coalesce(sum(reasoning_tokens_value), 0)) AS reasoning_tokens",
"sum(total_cost_value) AS total_cost",
)
sb.From(sb.BuilderAs(spans, "spans"))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
var stats spantypes.TraceStats
err = s.telemetryStore.ClickhouseDB().QueryRow(ctx, query, args...).Scan(
&stats.StartNs, &stats.EndNs, &stats.TotalSpans, &stats.TotalErrorSpans, &stats.HasMissingSpans,
&stats.RootServiceName, &stats.RootEntryPoint, &stats.GenAISpanCount,
&stats.Tokens.Input, &stats.Tokens.Output, &stats.Tokens.CacheRead, &stats.Tokens.CacheWrite, &stats.Tokens.Reasoning,
&stats.TotalCost,
)
if err != nil {
return nil, errors.WrapInternalf(err, errors.CodeInternal, "error querying trace stats")
}
return &stats, nil
}
// genAISpanColumns renders the per-span gen_ai gate and value reads through the shared
// traces storage, so each attribute is read from the column its evolutions place it in
// over the trace's own time window. Exists predicates bind their args into sb.
func (s *traceStore) genAISpanColumns(ctx context.Context, orgID valuer.UUID, summary *spantypes.TraceSummary, sb *sqlbuilder.SelectBuilder) ([]string, error) {
// no data type: metadata reports token counts as number, so a float64 request would
// miss them and fall back to a map read without evolutions
attributeKey := func(name string) *telemetrytypes.TelemetryFieldKey {
return &telemetrytypes.TelemetryFieldKey{Name: name, Signal: telemetrytypes.SignalTraces, FieldContext: telemetrytypes.FieldContextAttribute}
}
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(aiobservabilitytypes.GenAISpanGateKeys)+len(spantypes.TraceStatsGenAIColumns))
addSelector := func(name string) {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextAttribute,
SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
})
}
for _, name := range aiobservabilitytypes.GenAISpanGateKeys {
addSelector(name)
}
for _, col := range spantypes.TraceStatsGenAIColumns {
addSelector(col.Key)
}
keys, _, err := s.metadataStore.GetKeysMulti(ctx, orgID, querybuilder.ExpandKeySelectorsForFamilies(ctx, orgID, s.flagger, selectors))
if err != nil {
return nil, err
}
q := querybuilder.NewQueryInfo(ctx, orgID, s.flagger, telemetrytypes.SignalTraces, nil, uint64(summary.Start.UnixNano()), uint64(summary.End.UnixNano()))
gate := make([]string, 0, len(aiobservabilitytypes.GenAISpanGateKeys))
for _, name := range aiobservabilitytypes.GenAISpanGateKeys {
conds, _, err := querybuilder.Conditions(ctx, q, s.storage, attributeKey(name), qbtypes.FilterOperatorExists, nil, keys, false, sb)
if err != nil {
return nil, err
}
gate = append(gate, conds...)
}
columns := []string{sb.Or(gate...) + " AS is_gen_ai"}
for _, col := range spantypes.TraceStatsGenAIColumns {
expr, err := querybuilder.ResolveColumn(ctx, q, s.storage, attributeKey(col.Key), telemetrytypes.FieldDataTypeFloat64, keys)
if err != nil {
return nil, err
}
// a materialized column name carries `$$`, which Build would otherwise unescape
columns = append(columns, sqlbuilder.Escape(expr)+" AS "+col.Column+"_value")
}
return columns, nil
}
func (s *traceStore) GetTraceSpans(ctx context.Context, traceID string, summary *spantypes.TraceSummary) ([]spantypes.StorableSpan, error) {
// DISTINCT ON (span_id) is ClickHouse-specific syntax not supported by sqlbuilder
query := fmt.Sprintf(`

File diff suppressed because one or more lines are too long

View File

@@ -6,10 +6,12 @@ import (
"github.com/SigNoz/signoz/pkg/types/spantypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
)
// Handler exposes HTTP handlers for trace detail APIs.
type Handler interface {
GetTraceSummary(http.ResponseWriter, *http.Request)
GetWaterfallV4(http.ResponseWriter, *http.Request)
GetTraceAggregations(http.ResponseWriter, *http.Request)
GetFlamegraph(http.ResponseWriter, *http.Request)
@@ -17,6 +19,7 @@ type Handler interface {
// Module defines the business logic for trace detail operations.
type Module interface {
GetTraceStats(ctx context.Context, orgID valuer.UUID, traceID string) (*spantypes.TraceStats, error)
GetWaterfallV4(ctx context.Context, traceID string, selectedSpanID string, uncollapsedSpans []string) (*spantypes.GettableWaterfallTrace, error)
GetTraceAggregations(ctx context.Context, traceID string, req *spantypes.PostableTraceAggregations) (*spantypes.GettableTraceAggregations, error)
GetFlamegraph(ctx context.Context, traceID string, selectedSpanID string, selectFields []telemetrytypes.TelemetryFieldKey) (*spantypes.GettableFlamegraphTrace, error)

View File

@@ -495,8 +495,8 @@ WITH
toDateTime64(%[3]d/1e9, 9) AS start_ts,
toDateTime64(%[4]d/1e9, 9) AS end_ts,
('%[5]s','%[6]s') AS step1,
('%[7]s','%[8]s') AS step2
(%[5]s,%[6]s) AS step1,
(%[7]s,%[8]s) AS step2
SELECT
trace_id,
@@ -527,10 +527,10 @@ LIMIT 5;
containsErrorT2,
startTs,
endTs,
serviceNameT1,
spanNameT1,
serviceNameT2,
spanNameT2,
clickhousesql.StringLiteral(serviceNameT1),
clickhousesql.StringLiteral(spanNameT1),
clickhousesql.StringLiteral(serviceNameT2),
clickhousesql.StringLiteral(spanNameT2),
clauseStep1,
clauseStep2,
t1TimeExpr,
@@ -571,8 +571,8 @@ WITH
toDateTime64(%[3]d/1e9, 9) AS start_ts,
toDateTime64(%[4]d/1e9, 9) AS end_ts,
('%[5]s','%[6]s') AS step1,
('%[7]s','%[8]s') AS step2
(%[5]s,%[6]s) AS step1,
(%[7]s,%[8]s) AS step2
SELECT
trace_id,
@@ -607,10 +607,10 @@ LIMIT 5;
containsErrorT2,
startTs,
endTs,
serviceNameT1,
spanNameT1,
serviceNameT2,
spanNameT2,
clickhousesql.StringLiteral(serviceNameT1),
clickhousesql.StringLiteral(spanNameT1),
clickhousesql.StringLiteral(serviceNameT2),
clickhousesql.StringLiteral(spanNameT2),
clauseStep1,
clauseStep2,
t1TimeExpr,

View File

@@ -55,9 +55,6 @@ func (bc *bucketCache) GetMissRanges(
// Get query window
startMs, endMs := q.Window()
stepMs := uint64(step.Milliseconds())
startOffsetMs := calculateStartOffset(q, startMs, stepMs)
bc.logger.DebugContext(ctx, "getting miss ranges", slog.String("fingerprint", q.Fingerprint()), slog.Uint64("start", startMs), slog.Uint64("end", endMs))
// Generate cache key
@@ -77,8 +74,11 @@ func (bc *bucketCache) GetMissRanges(
return nil, missing
}
// Extract step interval if this is a builder query
stepMs := uint64(step.Milliseconds())
// Find missing ranges with step alignment
missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs, startOffsetMs)
missing = bc.findMissingRangesWithStep(data.Buckets, startMs, endMs, stepMs)
bc.logger.DebugContext(ctx, "missing ranges", slog.Any("missing", missing), slog.Uint64("step", stepMs))
// If no cached data overlaps with requested range, return empty result
@@ -95,7 +95,8 @@ func (bc *bucketCache) GetMissRanges(
// Merge buckets into a single result
mergedResult := bc.mergeBuckets(ctx, relevantBuckets, data.Warnings)
mergedResult = bc.filterResultToTimeRange(mergedResult, q, startMs, endMs, stepMs)
// Filter the merged result to only include values within the requested time range
mergedResult = bc.filterResultToTimeRange(mergedResult, startMs, endMs)
return mergedResult, missing
}
@@ -105,9 +106,6 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
// Get query window
startMs, endMs := q.Window()
stepMs := uint64(step.Milliseconds())
startOffsetMs := calculateStartOffset(q, startMs, stepMs)
// Calculate the flux boundary - data after this point should not be cached
currentMs := uint64(time.Now().UnixMilli())
fluxBoundary := currentMs - uint64(bc.fluxInterval.Milliseconds())
@@ -148,14 +146,19 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
// Adjust start and end times to only cache complete intervals
cachableStartMs := startMs
stepMs := uint64(step.Milliseconds())
// If we have a step interval, adjust boundaries to only cache complete intervals
if stepMs > 0 {
// If start is not aligned, round up to next step boundary (first complete interval)
cachableStartMs = alignUpToStep(startMs, stepMs, startOffsetMs)
if startMs%stepMs != 0 {
cachableStartMs = ((startMs / stepMs) + 1) * stepMs
}
// If end is not aligned, round down to previous step boundary (last complete interval)
cachableEndMs = alignDownToStep(cachableEndMs, stepMs, startOffsetMs)
if cachableEndMs%stepMs != 0 {
cachableEndMs = (cachableEndMs / stepMs) * stepMs
}
// If after adjustment we have no complete intervals, don't cache
if cachableStartMs >= cachableEndMs {
@@ -203,9 +206,8 @@ func (bc *bucketCache) generateCacheKey(q qbtypes.Query) string {
return fmt.Sprintf("v5:query:%s", fingerprint)
}
// findMissingRangesWithStep identifies time ranges not covered by cached buckets
// with step alignment. Boundaries are whole steps from startOffsetMs.
func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket, startMs, endMs uint64, stepMs uint64, startOffsetMs uint64) []*qbtypes.TimeRange {
// findMissingRangesWithStep identifies time ranges not covered by cached buckets with step alignment.
func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket, startMs, endMs uint64, stepMs uint64) []*qbtypes.TimeRange {
// When step is 0 or window is too small to be cached, use simple algorithm
if stepMs == 0 || (startMs+stepMs) > endMs {
return bc.findMissingRangesBasic(buckets, startMs, endMs)
@@ -218,7 +220,8 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
currentMs := startMs
// Check if start is not aligned - add partial window
if nextAggStart := alignUpToStep(startMs, stepMs, startOffsetMs); nextAggStart != startMs {
if startMs%stepMs != 0 {
nextAggStart := startMs - (startMs % stepMs) + stepMs
missing = append(missing, &qbtypes.TimeRange{
From: startMs,
To: min(nextAggStart, endMs),
@@ -264,7 +267,8 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
currentMs := startMs
// Check if start is not aligned - add partial window
if nextAggStart := alignUpToStep(startMs, stepMs, startOffsetMs); nextAggStart != startMs {
if startMs%stepMs != 0 {
nextAggStart := startMs - (startMs % stepMs) + stepMs
missing = append(missing, &qbtypes.TimeRange{
From: startMs,
To: min(nextAggStart, endMs),
@@ -283,7 +287,11 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
}
// Align bucket boundaries to step intervals
alignedBucketStart := alignUpToStep(bucket.StartMs, stepMs, startOffsetMs)
alignedBucketStart := bucket.StartMs
if bucket.StartMs%stepMs != 0 {
// Round up to next step boundary
alignedBucketStart = bucket.StartMs - (bucket.StartMs % stepMs) + stepMs
}
// Add gap before this bucket if needed
if currentMs < alignedBucketStart && currentMs < endMs {
@@ -296,12 +304,9 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
// Update current position to the end of this bucket
// But ensure it's aligned to step boundary
bucketEnd := min(bucket.EndMs, endMs)
// The step the window ends inside reaches past it, so that stretch is
// missing however far the bucket runs.
bucketEnd = min(bucketEnd, alignDownToStep(endMs, stepMs, startOffsetMs))
if bucketEnd < endMs {
if bucketEnd%stepMs != 0 && bucketEnd < endMs {
// Round down to step boundary
bucketEnd = alignDownToStep(bucketEnd, stepMs, startOffsetMs)
bucketEnd = bucketEnd - (bucketEnd % stepMs)
}
currentMs = max(currentMs, bucketEnd)
}
@@ -318,42 +323,6 @@ func (bc *bucketCache) findMissingRangesWithStep(buckets []*qbtypes.CachedBucket
return missing
}
// calculateStartOffset returns how far into a step a query's values sit. Only
// promql reports at the window start and every step after it; the rest report
// on absolute step boundaries.
func calculateStartOffset(q qbtypes.Query, startMs, stepMs uint64) uint64 {
if _, isPromQL := q.(*promqlQuery); !isPromQL || stepMs == 0 {
return 0
}
return startMs % stepMs
}
// With a 5m step and no offset the times seen by a query are 10:00, 10:05, 10:10. So 10:07
// is at an offset of 2m, and 10:05 is at 0.
//
// With a 1m step and a 30s offset the times seen are 10:00:30, 10:01:30, 10:02:30. So 10:01:00
// is at an offset of 30s.
func calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
if stepMs == 0 {
return 0
}
return ((timestampMs % stepMs) + stepMs - startOffsetMs%stepMs) % stepMs
}
// alignUpToStep returns the first time seen by a query at or after timestampMs.
func alignUpToStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
offset := calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs)
if offset == 0 {
return timestampMs
}
return timestampMs - offset + stepMs
}
// alignDownToStep returns the last time seen by a query at or before timestampMs.
func alignDownToStep(timestampMs, stepMs, startOffsetMs uint64) uint64 {
return timestampMs - calculateOffsetIntoStep(timestampMs, stepMs, startOffsetMs)
}
// findMissingRangesBasic is the simple algorithm without step alignment.
func (bc *bucketCache) findMissingRangesBasic(buckets []*qbtypes.CachedBucket, startMs, endMs uint64) []*qbtypes.TimeRange {
// Check if already sorted before sorting
@@ -822,26 +791,12 @@ func max(a, b uint64) uint64 {
return b
}
// filterResultToTimeRange narrows the cached result to the requested window, both
// the values in it and the heatmap axis under them.
func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, q qbtypes.Query, startMs, endMs, stepMs uint64) *qbtypes.Result {
// filterResultToTimeRange filters the result to only include values within the requested time range.
func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, startMs, endMs uint64) *qbtypes.Result {
if result == nil || result.Value == nil {
return result
}
_, isPromQL := q.(*promqlQuery)
maxTimestampMs := endMs
// A promql value at T is the query evaluated at T, so T == endMs is inside the
// requested range. For every other query type the value at T aggregates
// [T, T+stepMs), which the requested range contains only when T <= endMs-stepMs.
if !isPromQL {
if stepMs > 0 {
maxTimestampMs = endMs - stepMs
} else {
maxTimestampMs = endMs - 1
}
}
switch result.Type {
case qbtypes.RequestTypeTimeSeries, qbtypes.RequestTypeHeatmap:
if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
@@ -866,7 +821,7 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, q qbtypes
// Filter values to only include those within the requested time range
for _, value := range series.Values {
timestampMs := uint64(value.Timestamp)
if timestampMs >= startMs && timestampMs <= maxTimestampMs {
if timestampMs >= startMs && timestampMs < endMs {
filteredSeries.Values = append(filteredSeries.Values, value)
}
}
@@ -881,8 +836,6 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, q qbtypes
}
}
bc.trimHeatmapAxisToTheWindow(q, filteredData)
// Create a new result with the filtered data
return &qbtypes.Result{
Type: result.Type,
@@ -896,20 +849,3 @@ func (bc *bucketCache) filterResultToTimeRange(result *qbtypes.Result, q qbtypes
// For non-time series data, return as is
return result
}
// a cached range covers more than the window now being asked for, so its axis
// carries buckets only the dropped columns reached. Left there, they show as
// empty rows the same window never has when the cache did not answer it.
func (bc *bucketCache) trimHeatmapAxisToTheWindow(q qbtypes.Query, tsData *qbtypes.TimeSeriesData) {
// promql and clickhouse name their own buckets, and an empty one of theirs
// still belongs on the axis
switch q.(type) {
case *builderQuery[qbtypes.MetricAggregation], *builderQuery[qbtypes.LogAggregation], *builderQuery[qbtypes.TraceAggregation]:
default:
return
}
for _, aggBucket := range tsData.Aggregations {
aggBucket.TrimAxisToCountedBuckets()
}
}

View File

@@ -201,7 +201,7 @@ func BenchmarkBucketCache_FindMissingRangesWithStep(b *testing.B) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
missing := bc.findMissingRangesWithStep(buckets, startMs, endMs, stepMs, 0)
missing := bc.findMissingRangesWithStep(buckets, startMs, endMs, stepMs)
_ = missing
}
})
@@ -327,7 +327,7 @@ func BenchmarkBucketCache_FilterResultToTimeRange(b *testing.B) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
filtered := bc.filterResultToTimeRange(result, &promqlQuery{}, startMs, endMs, 0)
filtered := bc.filterResultToTimeRange(result, startMs, endMs)
_ = filtered
}
})

View File

@@ -3,7 +3,6 @@ package querier
import (
"context"
"fmt"
"log/slog"
"testing"
"time"
@@ -530,7 +529,7 @@ func TestBucketCache_FindMissingRanges_EdgeCases(t *testing.T) {
}
// Query range that spans all buckets
missing := bc.findMissingRangesWithStep(buckets, 500, 6500, 500, 0)
missing := bc.findMissingRangesWithStep(buckets, 500, 6500, 500)
// Expected missing ranges: 500-1000, 2000-2500, 4000-5000, 6000-6500
assert.Len(t, missing, 4)
@@ -1070,11 +1069,8 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
// Get cached data - should be filtered to requested range
cached, missing := bc.GetMissRanges(ctx, orgID, query2, qbtypes.Step{Duration: 1000 * time.Millisecond})
// The value at 3000 stands for the whole step to 4000, which reaches past the
// window, so it is left to be recomputed as a partial rather than served.
require.Len(t, missing, 1)
assert.Equal(t, uint64(3000), missing[0].From)
assert.Equal(t, uint64(3500), missing[0].To)
// Should have no missing ranges
assert.Len(t, missing, 0)
assert.NotNil(t, cached)
// Verify the cached result only contains values within the requested range
@@ -1084,77 +1080,29 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
require.Len(t, tsData.Aggregations[0].Series, 1)
series := tsData.Aggregations[0].Series[0]
require.Len(t, series.Values, 1)
assert.Len(t, series.Values, 2) // Only values at 2000 and 3000 should be included
// Verify the exact values
assert.Equal(t, int64(2000), series.Values[0].Timestamp)
assert.Equal(t, float64(20), series.Values[0].Value)
assert.Equal(t, int64(3000), series.Values[1].Timestamp)
assert.Equal(t, float64(30), series.Values[1].Value)
// Value at 1000 should not be included (before requested range)
// Value at 4000 should not be included (after requested range)
}
// A promql value is the query evaluated at a single moment rather than over a
// span, so the one at the window's end belongs to it and has to survive caching.
func TestBucketCache_PromQLKeepsTheValueAtTheWindowEnd(t *testing.T) {
bc := createTestBucketCache(t)
ctx := context.Background()
orgID := valuer.UUID{}
step := qbtypes.Step{Duration: time.Minute}
query := &promqlQuery{
logger: slog.Default(),
query: qbtypes.PromQuery{Query: "up", Step: step},
tr: qbtypes.TimeRange{From: 600_000, To: 780_000},
requestType: qbtypes.RequestTypeTimeSeries,
}
bc.Put(ctx, orgID, query, step, &qbtypes.Result{
Type: qbtypes.RequestTypeTimeSeries,
Value: &qbtypes.TimeSeriesData{
QueryName: "A",
Aggregations: []*qbtypes.AggregationBucket{{
Series: []*qbtypes.TimeSeries{{
Values: []*qbtypes.TimeSeriesValue{
{Timestamp: 600_000, Value: 1},
{Timestamp: 660_000, Value: 2},
{Timestamp: 720_000, Value: 3},
{Timestamp: 780_000, Value: 4},
},
}},
}},
},
})
time.Sleep(10 * time.Millisecond)
cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
assert.Empty(t, missing)
require.NotNil(t, cached)
tsData, ok := cached.Value.(*qbtypes.TimeSeriesData)
require.True(t, ok)
require.Len(t, tsData.Aggregations, 1)
require.Len(t, tsData.Aggregations[0].Series, 1)
timestamps := []int64{}
for _, value := range tsData.Aggregations[0].Series[0].Values {
timestamps = append(timestamps, value.Timestamp)
}
assert.Equal(t, []int64{600_000, 660_000, 720_000, 780_000}, timestamps)
}
func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
bc := createTestBucketCache(t)
tests := []struct {
name string
buckets []*qbtypes.CachedBucket
startMs uint64
endMs uint64
stepMs uint64
startOffsetMs uint64
expectedMiss []*qbtypes.TimeRange
description string
name string
buckets []*qbtypes.CachedBucket
startMs uint64
endMs uint64
stepMs uint64
expectedMiss []*qbtypes.TimeRange
description string
}{
{
name: "start_not_aligned_to_step",
@@ -1204,32 +1152,6 @@ func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
},
description: "Window smaller than step should use basic algorithm",
},
{
name: "start_aligned_to_its_own_offset",
buckets: []*qbtypes.CachedBucket{},
startMs: 1500,
endMs: 5000,
stepMs: 1000,
startOffsetMs: 500,
expectedMiss: []*qbtypes.TimeRange{
{From: 1500, To: 5000},
},
description: "A query reporting every 1000ms from 1500 needs no partial window at its own start",
},
{
name: "gap_lands_on_the_offset",
buckets: []*qbtypes.CachedBucket{
{StartMs: 1500, EndMs: 3500},
},
startMs: 1500,
endMs: 5500,
stepMs: 1000,
startOffsetMs: 500,
expectedMiss: []*qbtypes.TimeRange{
{From: 3500, To: 5500},
},
description: "The refetched range starts where the cached one ends, on an instant the query reports at",
},
{
name: "zero_step_uses_basic_algorithm",
buckets: []*qbtypes.CachedBucket{},
@@ -1246,7 +1168,7 @@ func TestBucketCache_FindMissingRangesWithStep(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Mock current time for flux boundary tests
result := bc.findMissingRangesWithStep(tt.buckets, tt.startMs, tt.endMs, tt.stepMs, tt.startOffsetMs)
result := bc.findMissingRangesWithStep(tt.buckets, tt.startMs, tt.endMs, tt.stepMs)
// Compare lengths first
assert.Len(t, result, len(tt.expectedMiss), tt.description)

View File

@@ -22,7 +22,7 @@ import (
const promHistogramBucketLabel = "le"
// cumulativeColumn maps a bucket's upper bound to the cumulative count at it.
// Differencing turns it into the per-bucket counts a heatmapColumn holds.
// Differencing turns it into the per-band counts a heatmapColumn holds.
type cumulativeColumn map[float64]float64
// promHeatmapGroup assembles one group across the several matrix series its `le`
@@ -34,8 +34,8 @@ type promHeatmapGroup struct {
}
// foldMatrixAsHeatmap folds a matrix of one cumulative series per (group, `le`)
// into one series per group whose points hold a count per bucket.
func foldMatrixAsHeatmap(matrix promql.Matrix, queryName string) (*qbv5.TimeSeriesData, error) {
// into one series per group whose points hold a count per band.
func foldMatrixAsHeatmap(matrix promql.Matrix, queryWindow *qbv5.TimeRange, stepMs uint64, queryName string) (*qbv5.TimeSeriesData, error) {
groups, groupOrder := collectCumulativeGroups(matrix)
// An empty matrix is only ever the window having no data, but series that
@@ -53,12 +53,11 @@ func foldMatrixAsHeatmap(matrix promql.Matrix, queryName string) (*qbv5.TimeSeri
}
}
// a promql data point can never be partial, hence nil and 0 are sent here
return accumulator.foldSeries(nil, 0, queryName)
return accumulator.foldSeries(queryWindow, stepMs, queryName)
}
// collectCumulativeGroups reads the matrix into one group per label set. A series
// without `le` has no bucket to sit in, so an expression that dropped the label
// without `le` has no band to sit in, so an expression that dropped the label
// draws nothing.
func collectCumulativeGroups(matrix promql.Matrix) (groups map[string]*promHeatmapGroup, groupOrder []string) {
groups = map[string]*promHeatmapGroup{}

View File

@@ -16,7 +16,7 @@ import (
// The cache key is the fingerprint alone, so two request types over one
// expression must not produce the same one — a time series payload served to a
// heatmap request has no axis and reads back as a single collapsed bucket.
// heatmap request has no axis and reads back as a single collapsed band.
func TestFingerprintSeparatesHeatmapFromTimeSeries(t *testing.T) {
fingerprintFor := func(requestType qbv5.RequestType) string {
q := &promqlQuery{
@@ -50,7 +50,7 @@ func TestFoldMatrixAsHeatmapClampsADecreasingCumulativeCount(t *testing.T) {
},
}
data, err := foldMatrixAsHeatmap(matrix, "A")
data, err := foldMatrixAsHeatmap(matrix, &qbv5.TimeRange{From: 1710000000000, To: 1710000060000}, uint64(time.Minute.Milliseconds()), "A")
require.NoError(t, err)
require.Len(t, data.Aggregations, 1)
@@ -76,7 +76,7 @@ func TestFoldMatrixAsHeatmapWidensTheBandOverAMissingUpperBound(t *testing.T) {
},
}
data, err := foldMatrixAsHeatmap(matrix, "A")
data, err := foldMatrixAsHeatmap(matrix, &qbv5.TimeRange{From: 1710000000000, To: 1710000060000}, uint64(time.Minute.Milliseconds()), "A")
require.NoError(t, err)
require.Len(t, data.Aggregations, 1)

View File

@@ -175,12 +175,6 @@ func (q *promqlQuery) Fingerprint() string {
q.query.Step.String(),
}
// Two windows a fraction of a step apart describe different instants, so
// they must not share an entry.
if stepMs := uint64(q.query.Step.Milliseconds()); stepMs > 0 && q.tr.From%stepMs != 0 {
parts = append(parts, fmt.Sprintf("offset=%d", q.tr.From%stepMs))
}
return strings.Join(parts, "&")
}
@@ -491,7 +485,7 @@ func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began ti
}
func (q *promqlQuery) toResultForHeatmap(matrix promql.Matrix, warnings []string, began time.Time, statsMu *sync.Mutex, rowsScanned, bytesScanned *uint64) (*qbv5.Result, error) {
tsData, err := foldMatrixAsHeatmap(matrix, q.query.Name)
tsData, err := foldMatrixAsHeatmap(matrix, &q.tr, uint64(q.query.Step.Milliseconds()), q.query.Name)
if err != nil {
return nil, err
}

View File

@@ -461,37 +461,6 @@ func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
assert.Empty(t, q.Fingerprint())
}
// promql reports at the window start and every step after it, so a window
// starting later inside the step describes instants the earlier one never does.
func TestFingerprintSeparatesWindowsInsideAStep(t *testing.T) {
minuteStep := qbv5.Step{Duration: time.Minute}
onTheMinute := (&promqlQuery{
logger: slog.Default(),
query: qbv5.PromQuery{Query: "up", Step: minuteStep},
tr: qbv5.TimeRange{From: 600_000, To: 1_200_000},
requestType: qbv5.RequestTypeTimeSeries,
}).Fingerprint()
halfAStepLater := (&promqlQuery{
logger: slog.Default(),
query: qbv5.PromQuery{Query: "up", Step: minuteStep},
tr: qbv5.TimeRange{From: 630_000, To: 1_230_000},
requestType: qbv5.RequestTypeTimeSeries,
}).Fingerprint()
aWholeMinuteLater := (&promqlQuery{
logger: slog.Default(),
query: qbv5.PromQuery{Query: "up", Step: minuteStep},
tr: qbv5.TimeRange{From: 900_000, To: 1_500_000},
requestType: qbv5.RequestTypeTimeSeries,
}).Fingerprint()
require.NotEmpty(t, onTheMinute)
assert.NotEqual(t, onTheMinute, halfAStepLater, "windows half a step apart share no instants")
assert.Equal(t, onTheMinute, aWholeMinuteLater, "windows whole steps apart report at the same instants")
}
func TestToResultDropsNonFiniteValues(t *testing.T) {
tests := []struct {
description string

View File

@@ -161,7 +161,7 @@ func NewModules(
LogsPipeline: impllogspipeline.NewModule(sqlstore),
RuleStateHistory: implrulestatehistory.NewModule(implrulestatehistory.NewStore(telemetryStore, telemetryMetadataStore, providerSettings.Logger), ruleStore),
CloudIntegration: cloudIntegrationModule,
TraceDetail: impltracedetail.NewModule(impltracedetail.NewTraceStore(telemetryStore), providerSettings, config.TraceDetail),
TraceDetail: impltracedetail.NewModule(impltracedetail.NewTraceStore(telemetryStore, telemetryMetadataStore, fl), providerSettings, config.TraceDetail),
SpanMapper: implspanmapper.NewModule(implspanmapper.NewStore(sqlstore), fl),
LLMPricingRule: impllmpricingrule.NewModule(impllmpricingrule.NewStore(sqlstore), fl, querier),
Tag: tagModule,

View File

@@ -19,6 +19,7 @@ var (
aiobservabilitytypes.GenAIUsageOutputTokens: genAIAttribute(aiobservabilitytypes.GenAIUsageOutputTokens, telemetrytypes.FieldDataTypeFloat64),
aiobservabilitytypes.GenAIUsageCacheReadInputTokens: genAIAttribute(aiobservabilitytypes.GenAIUsageCacheReadInputTokens, telemetrytypes.FieldDataTypeFloat64),
aiobservabilitytypes.GenAIUsageCacheCreationInputTokens: genAIAttribute(aiobservabilitytypes.GenAIUsageCacheCreationInputTokens, telemetrytypes.FieldDataTypeFloat64),
aiobservabilitytypes.GenAIUsageReasoningOutputTokens: genAIAttribute(aiobservabilitytypes.GenAIUsageReasoningOutputTokens, telemetrytypes.FieldDataTypeFloat64),
aiobservabilitytypes.SignozGenAITotalCost: genAIAttribute(aiobservabilitytypes.SignozGenAITotalCost, telemetrytypes.FieldDataTypeFloat64),
aiobservabilitytypes.GenAIInputMessages: genAIAttribute(aiobservabilitytypes.GenAIInputMessages, telemetrytypes.FieldDataTypeString),

View File

@@ -46,7 +46,6 @@ type QuerySettings struct {
MaxBytesToRead int `mapstructure:"max_bytes_to_read"`
MaxResultRows int `mapstructure:"max_result_rows"`
IgnoreDataSkippingIndices string `mapstructure:"ignore_data_skipping_indices"`
SecondaryIndicesEnableBulkFiltering bool `mapstructure:"secondary_indices_enable_bulk_filtering"`
}
func NewConfigFactory() factory.ConfigFactory {

View File

@@ -72,10 +72,6 @@ func (h *provider) BeforeQuery(ctx context.Context, _ *telemetrystore.QueryEvent
settings["result_overflow_mode"] = ctx.Value("result_overflow_mode")
}
// TODO(srikanthccv): enable it when the "Cannot read all data" issue is fixed
// https://github.com/ClickHouse/ClickHouse/issues/82283
settings["secondary_indices_enable_bulk_filtering"] = false
ctx = clickhouse.Context(ctx, clickhouse.WithSettings(settings))
return ctx
}

View File

@@ -15,6 +15,7 @@ const (
GenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
GenAIUsageCacheReadInputTokens = "gen_ai.usage.cache_read.input_tokens"
GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
GenAIUsageReasoningOutputTokens = "gen_ai.usage.reasoning.output_tokens"
GenAIInputMessages = "gen_ai.input.messages"
GenAIOutputMessages = "gen_ai.output.messages"

View File

@@ -1014,7 +1014,7 @@ func rejectHTTPBasicAuthBeyondPassword(channelName string, httpConfig *commoncfg
basicAuth := httpConfig.BasicAuth
if *basicAuth != (commoncfg.BasicAuth{Username: basicAuth.Username, Password: basicAuth.Password}) {
return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.basic_auth, which is not supported", channelName)
return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.basic_auth with fields other than username and password, which is not supported", channelName)
}
return nil
@@ -1026,8 +1026,8 @@ func rejectHTTPAuthorizationBeyondBearer(channelName string, httpConfig *commonc
}
authorization := httpConfig.Authorization
if *authorization != (commoncfg.Authorization{Type: bearerAuthorizationType, Credentials: authorization.Credentials}) {
return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.authorization, which is not supported", channelName)
if !strings.EqualFold(authorization.Type, bearerAuthorizationType) || *authorization != (commoncfg.Authorization{Type: authorization.Type, Credentials: authorization.Credentials}) {
return errors.NewInvalidInputf(ErrCodeAlertmanagerChannelInvalid, "channel %q sets http_config.authorization with fields other than a bearer token, which is not supported", channelName)
}
return nil

View File

@@ -542,3 +542,42 @@ func TestChannelToPostableChannelRejectsUnrepresentableChannels(t *testing.T) {
})
}
}
// The HTTP auth scheme is case-insensitive (RFC 7235) and Alertmanager sends
// the stored spelling verbatim, so a hand-written receiver may carry any casing.
func TestChannelToPostableChannelReadsWebhookBearerSchemeCaseInsensitively(t *testing.T) {
sendResolved := config.DefaultWebhookConfig.VSendResolved
testCases := []struct {
name string
storedChannelData string
expectedWebhookSpec *ChannelWebhookConfig
}{
{
name: "CanonicalBearer",
storedChannelData: `{"name":"hook","webhook_configs":[{"send_resolved":true,"url":"https://a","http_config":{"authorization":{"type":"Bearer","credentials":"tok"},"follow_redirects":true,"enable_http2":true}}]}`,
expectedWebhookSpec: &ChannelWebhookConfig{SendResolved: &sendResolved, URL: "https://a", BearerToken: "tok"},
},
{
name: "LowercaseBearer",
storedChannelData: `{"name":"hook","webhook_configs":[{"send_resolved":true,"url":"https://b","http_config":{"authorization":{"type":"bearer","credentials":"lower"},"follow_redirects":true,"enable_http2":true}}]}`,
expectedWebhookSpec: &ChannelWebhookConfig{SendResolved: &sendResolved, URL: "https://b", BearerToken: "lower"},
},
{
name: "UppercaseBearer",
storedChannelData: `{"name":"hook","webhook_configs":[{"send_resolved":true,"url":"https://c","http_config":{"authorization":{"type":"BEARER","credentials":"upper"},"follow_redirects":true,"enable_http2":true}}]}`,
expectedWebhookSpec: &ChannelWebhookConfig{SendResolved: &sendResolved, URL: "https://c", BearerToken: "upper"},
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
channel := Channel{DisplayName: "hook", Data: testCase.storedChannelData}
postable, err := channel.toPostableNotificationChannel()
require.NoError(t, err)
assert.Equal(t, ChannelKindWebhook, postable.Config.Kind)
assert.Equal(t, testCase.expectedWebhookSpec, postable.Config.Spec)
})
}
}

View File

@@ -189,50 +189,6 @@ func (a *AggregationBucket) ReindexValuesToNewUpperBounds(onto []float64) {
a.Meta.Buckets = onto
}
// TrimAxisToCountedBuckets drops the buckets at either end of Meta.Buckets that hold
// no counts, since an axis runs from the lowest value in the window to the highest.
// Not for a query that chose its own buckets: an empty `le` is still one it reported.
func (a *AggregationBucket) TrimAxisToCountedBuckets() {
if a == nil || len(a.Meta.Buckets) == 0 {
return
}
lowestCounted, highestCounted := len(a.Meta.Buckets), -1
for _, series := range a.Series {
for _, point := range series.Values {
for slot := 0; slot < len(a.Meta.Buckets) && slot < len(point.Values); slot++ {
if point.Values[slot] != 0 {
lowestCounted = min(lowestCounted, slot)
highestCounted = max(highestCounted, slot)
}
}
}
}
if highestCounted < 0 {
return
}
if lowestCounted == 0 && highestCounted == len(a.Meta.Buckets)-1 {
return
}
trimmed := a.Meta.Buckets[lowestCounted : highestCounted+1]
for _, series := range a.Series {
for _, point := range series.Values {
if len(point.Values) == 0 {
continue
}
counts := make([]float64, len(trimmed)+1)
for slot, count := range point.Values {
counts[min(max(slot-lowestCounted, 0), len(trimmed))] += count
}
point.Values = counts
}
}
a.Meta.Buckets = trimmed
}
type AggregationMeta struct {
Unit string `json:"unit,omitempty"`
// Buckets holds ascending upper bounds shared by every series in the

View File

@@ -27,6 +27,7 @@ type SpanMapperStore interface {
// TraceStore defines the data access interface for trace detail queries.
type TraceStore interface {
GetTraceSummary(ctx context.Context, traceID string) (*TraceSummary, error)
GetTraceStats(ctx context.Context, orgID valuer.UUID, traceID string, summary *TraceSummary) (*TraceStats, error)
GetTraceSpans(ctx context.Context, traceID string, summary *TraceSummary) ([]StorableSpan, error)
GetMinimalSpans(ctx context.Context, traceID string, start, end time.Time) ([]MinimalSpan, error)
GetTraceSpansByIDs(ctx context.Context, traceID string, start, end time.Time, spanIDs []string) ([]StorableSpan, error)

View File

@@ -0,0 +1,76 @@
package spantypes
import "github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
// TraceStatsGenAIColumns pairs each summed TraceStats column with the gen_ai attribute it sums.
var TraceStatsGenAIColumns = []TraceStatsGenAIColumn{
{Column: "input_tokens", Key: aiobservabilitytypes.GenAIUsageInputTokens},
{Column: "output_tokens", Key: aiobservabilitytypes.GenAIUsageOutputTokens},
{Column: "cache_read_tokens", Key: aiobservabilitytypes.GenAIUsageCacheReadInputTokens},
{Column: "cache_write_tokens", Key: aiobservabilitytypes.GenAIUsageCacheCreationInputTokens},
{Column: "reasoning_tokens", Key: aiobservabilitytypes.GenAIUsageReasoningOutputTokens},
{Column: "total_cost", Key: aiobservabilitytypes.SignozGenAITotalCost},
}
type TraceStatsGenAIColumn struct {
Column string
Key string
}
// TraceStats is the single-row result of the trace summary aggregate query.
type TraceStats struct {
StartNs uint64
EndNs uint64
RootServiceName string
RootEntryPoint string
TotalSpans uint64
TotalErrorSpans uint64
HasMissingSpans bool
GenAISpanCount uint64
Tokens TraceAITokens
TotalCost *float64
}
// GettableTraceSummary is the response for the trace summary API; the trace-level
// fields match the waterfall response.
type GettableTraceSummary struct {
StartTimestampMillis uint64 `json:"startTimestampMillis"`
EndTimestampMillis uint64 `json:"endTimestampMillis"`
RootServiceName string `json:"rootServiceName"`
RootServiceEntryPoint string `json:"rootServiceEntryPoint"`
TotalSpansCount uint64 `json:"totalSpansCount"`
TotalErrorSpansCount uint64 `json:"totalErrorSpansCount"`
HasMissingSpans bool `json:"hasMissingSpans"`
AI *TraceAISummary `json:"ai,omitempty"`
}
// TraceAISummary is present when any span carries a gen_ai gate key.
type TraceAISummary struct {
Tokens TraceAITokens `json:"tokens"`
// TotalCost is null when no span carries a cost attribute.
TotalCost *float64 `json:"totalCost" nullable:"true"`
}
type TraceAITokens struct {
Input uint64 `json:"input"`
Output uint64 `json:"output"`
CacheRead uint64 `json:"cacheRead"`
CacheWrite uint64 `json:"cacheWrite"`
Reasoning uint64 `json:"reasoning"`
}
func NewGettableTraceSummary(stats *TraceStats) *GettableTraceSummary {
summary := &GettableTraceSummary{
StartTimestampMillis: stats.StartNs / 1_000_000,
EndTimestampMillis: stats.EndNs / 1_000_000,
RootServiceName: stats.RootServiceName,
RootServiceEntryPoint: stats.RootEntryPoint,
TotalSpansCount: stats.TotalSpans,
TotalErrorSpansCount: stats.TotalErrorSpans,
HasMissingSpans: stats.HasMissingSpans,
}
if stats.GenAISpanCount > 0 {
summary.AI = &TraceAISummary{Tokens: stats.Tokens, TotalCost: stats.TotalCost}
}
return summary
}

View File

@@ -895,6 +895,7 @@ _TRACES_TABLES_TO_TRUNCATE = [
"span_attributes_keys",
"signoz_error_index_v2",
"top_level_operations",
"trace_summary",
]

View File

@@ -58,7 +58,8 @@ def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
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'])}"
# Both reads must agree exactly, including the point promql reports at end_ms.
# The cached read excludes end_ms, the one legitimate difference.
assert set(second) == set(first), sorted(second)
for job_name, points in first.items():
assert second[job_name] == points, f"{job_name}: got {len(second[job_name])} of {len(points)} points"
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"

View File

@@ -1,325 +0,0 @@
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 (
assert_results_equal,
build_builder_query,
get_series_values,
make_query_request,
)
MINUTE_MS = 60_000
def test_builder_shortening_the_time_range_at_the_end(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
# the cache outlives the run, so a fixed name would serve the previous run's
# points back to this one
metric_name = f"cache_end_shortened_{uuid4().hex[:8]}"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache. Flooring to a multiple of the 5m step makes the base query span
# two whole steps, so both its points are complete
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
end_time_ms_base_query = start_time_ms + 10 * MINUTE_MS
end_time_ms_shortened_query = start_time_ms + 7 * MINUTE_MS
query = [build_builder_query("A", metric_name, "max", "max", step_interval=300)]
# the 5m step splits the ten minutes into two points, each the max over its
# own step: minutes 0-4 and minutes 5-9. The second changes partway through,
# 256 until minute 7 and then 4096, so ending the range at minute 7 has to
# reach a different value than ending it at minute 10
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=(16, 16, 16, 16, 16, 256, 256, 4096, 4096, 4096)[minute],
type_="Gauge",
is_monotonic=False,
)
for minute in range(10)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
assert base_query.status_code == HTTPStatus.OK, base_query.text
points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["value"], point.get("partial", False)) for point in points]
assert returned_points == [(16, False), (4096, False)]
from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened end")
# the shortened end reaches only minutes 5-6 of the second point, so it comes
# back as 256 and partial, where the cached one spans all five minutes
for label, response in (("from cache", from_cache), ("uncached", uncached)):
points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["value"], point.get("partial", False)) for point in points]
assert returned_points == [(16, False), (256, True)], label
def test_builder_shortening_the_time_range_at_the_start(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"cache_start_shortened_{uuid4().hex[:8]}"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache. Flooring to a multiple of the 5m step makes the base query span
# two whole steps, so both its points are complete
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
start_time_ms_base_query = int(start_time.timestamp() * 1000)
start_time_ms_shortened_query = start_time_ms_base_query + 3 * MINUTE_MS
end_time_ms = start_time_ms_base_query + 10 * MINUTE_MS
query = [build_builder_query("A", metric_name, "max", "max", step_interval=300)]
# the 5m step splits the ten minutes into two points, each the max over its
# own step: minutes 0-4 and minutes 5-9. Only minute 0 holds 65536, so a first
# point reaching it says the whole step was read even though the shortened
# range opens at minute 3
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=(65536, 16, 16, 16, 16, 4096, 4096, 4096, 4096, 4096)[minute],
type_="Gauge",
is_monotonic=False,
)
for minute in range(10)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False)
assert base_query.status_code == HTTPStatus.OK, base_query.text
points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["value"], point.get("partial", False)) for point in points]
assert returned_points == [(65536, False), (4096, False)]
from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened start")
# starting inside the first point's step flags that point partial without
# clipping its value, which still covers the whole step and so reaches the
# 65536 at minute 0
for label, response in (("from cache", from_cache), ("uncached", uncached)):
points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["value"], point.get("partial", False)) for point in points]
assert returned_points == [(65536, True), (4096, False)], label
def test_promql_running_the_same_query_twice(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"cache_repeat_total_{uuid4().hex[:8]}"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
end_time_ms = start_time_ms + 2 * MINUTE_MS
query = [{"type": "promql", "spec": {"name": "A", "query": f"sum(increase({metric_name}[2m]))", "step": 60}}]
# the counter opens a minute before the query so its first point has something
# to increase over, and starts far above its own rise across the range, below
# which increase clips its back-extrapolation at the counter's zero point. It
# rises by a different amount each minute, so every point is its own number
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=(1000, 1010, 1030, 1060, 1100)[minute + 1],
temporality="Cumulative",
type_="Sum",
is_monotonic=True,
)
for minute in range(-1, 4)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
assert first.status_code == HTTPStatus.OK, first.text
second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
assert second.status_code == HTTPStatus.OK, second.text
assert_results_equal(first.json(), second.json(), "A", "the same query twice")
# promql reports a point at the instant the range closes, and the second run,
# answered out of what the first one cached, has to keep it
for run, response in (("first", first), ("second", second)):
points = sorted(get_series_values(response.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["timestamp"], point["value"]) for point in points]
## at each timestamp t, promql looks at points in (t-2minutes, t].
assert returned_points == [
(start_time_ms, 20), # t = 0, points taken 1000, 1010. hence diff over 1m is 10, extrapolated to 20.
(start_time_ms + MINUTE_MS, 40), # t = 1m, points taken 1010, 1030. hence diff over 1m is 20, extrapolated to 40.
(end_time_ms, 60), # t = 2m, points taken 1030, 1060. hence diff over 1m is 30, extrapolated to 60.
], f"{run} run"
def test_promql_shifting_the_time_range(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"cache_shift_gauge_{uuid4().hex[:8]}"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache. Flooring to a whole minute is what makes the first query aligned
# to its 1m step, and the unaligned one half a step off it
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
aligned_start_time_ms = int(start_time.timestamp() * 1000)
aligned_end_time_ms = aligned_start_time_ms + 3 * MINUTE_MS
unaligned_start_time_ms = aligned_start_time_ms + MINUTE_MS // 2
unaligned_end_time_ms = aligned_end_time_ms + MINUTE_MS // 2
query = [{"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric_name}[2m])", "step": 60}}]
# a sample every 30s, rising by 100 each time. The two queries report 30s
# apart, so they land on different samples and share no value between them
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(seconds=30 * half_minute),
value=100 * (half_minute + 4),
type_="Gauge",
is_monotonic=False,
)
for half_minute in range(-3, 8)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False)
assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
# what the cache now holds, and what the unaligned query must not be served
points = sorted(get_series_values(aligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["timestamp"], point["value"]) for point in points]
## at each timestamp t, promql takes the highest sample in (t-2minutes, t],
## which is the one at t itself since the gauge only rises.
assert returned_points == [
(aligned_start_time_ms, 400), # t = 0
(aligned_start_time_ms + MINUTE_MS, 600), # t = 1m
(aligned_start_time_ms + 2 * MINUTE_MS, 800), # t = 2m
(aligned_end_time_ms, 1000), # t = 3m
]
unaligned_and_uncached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=True)
assert unaligned_and_uncached.status_code == HTTPStatus.OK, unaligned_and_uncached.text
# promql reports at the range start plus whole steps, so these points sit 30s
# off the cached ones. The first run stores them, the second reads them back
for run in ("first", "second"):
unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False)
assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
assert_results_equal(unaligned_and_cached.json(), unaligned_and_uncached.json(), "A", f"unaligned query, {run} run")
points = sorted(get_series_values(unaligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["timestamp"], point["value"]) for point in points]
## every point falls on a sample the aligned run never reported, so being
## served the cached run's answer shows up in the values and not only the
## timestamps.
assert returned_points == [
(unaligned_start_time_ms, 500), # t = 30s
(unaligned_start_time_ms + MINUTE_MS, 700), # t = 1m30s
(unaligned_start_time_ms + 2 * MINUTE_MS, 900), # t = 2m30s
(unaligned_end_time_ms, 1100), # t = 3m30s
], f"unaligned query, {run} run"
def test_builder_refreshing_a_sliding_time_range(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"cache_sliding_{uuid4().hex[:8]}"
# 90 minutes back so even the twentieth refresh closes clear of the flux
# interval, which holds recent data out of the cache
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=90)).timestamp()) // 60 * 60, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
query = [build_builder_query("A", metric_name, "max", "max")]
# the 1m step gives one point per seeded minute, and a value no other minute
# carries, so a point stitched in from the wrong range reads as the wrong minute
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=1000 + minute,
type_="Gauge",
is_monotonic=False,
)
for minute in range(80)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# a dashboard left open on a one hour range, re-running a minute later each time
for refresh in range(20):
refresh_start_ms = start_time_ms + refresh * MINUTE_MS
from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
# each refresh is stitched out of overlapping cached ranges, so this catches
# a point served twice, dropped, or carried over from an earlier refresh
points = sorted(get_series_values(from_cache.json(), "A"), key=lambda point: point["timestamp"])
returned_points = [(point["timestamp"], point["value"], point.get("partial", False)) for point in points]
expected_points = [(start_time_ms + minute * MINUTE_MS, 1000 + minute, False) for minute in range(refresh, refresh + 60)]
assert returned_points == expected_points, f"refresh {refresh} did not return the minutes it covers"
last_refresh_start_ms = start_time_ms + 19 * MINUTE_MS
uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")

View File

@@ -1,546 +0,0 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
from uuid import uuid4
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 (
RequestType,
assert_identical_query_response,
build_builder_query,
build_linear_bucket_options,
get_heatmap_buckets,
get_heatmap_columns,
make_query_request,
)
MINUTE_MS = 60_000
@pytest.mark.parametrize(
"first_minute, expected_buckets",
[
pytest.param(0, [100, 200], id="the_lower_half"),
pytest.param(5, [800, 900], id="the_upper_half"),
],
)
def test_builder_narrowing_to_half_the_range(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
first_minute: int,
expected_buckets: list[int],
) -> None:
metric_name = f"heatmap_cache_narrowed_{uuid4().hex[:8]}"
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
end_time_ms = start_time_ms + 10 * MINUTE_MS
# 100 wide buckets, and the first five minutes sit seven buckets under the
# last five, so the axis over all ten covers a stretch neither half reaches
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=150 if minute < 5 else 850,
type_="Gauge",
is_monotonic=False,
)
for minute in range(10)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
query = [build_builder_query("A", metric_name, "max", "max", bucket_options=build_linear_bucket_options(1000, 10))]
# the whole range first, which is what puts its axis in the cache
whole_range = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert whole_range.status_code == HTTPStatus.OK, whole_range.text
assert get_heatmap_buckets(whole_range.json(), "A") == pytest.approx([100, 200, 300, 400, 500, 600, 700, 800, 900])
assert [column["values"] for column in get_heatmap_columns(whole_range.json(), "A")] == [
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
]
half_start_ms = start_time_ms + first_minute * MINUTE_MS
half_end_ms = half_start_ms + 5 * MINUTE_MS
from_cache = make_query_request(signoz, token, half_start_ms, half_end_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
uncached = make_query_request(signoz, token, half_start_ms, half_end_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
for source, response in (("uncached", uncached), ("from cache", from_cache)):
assert get_heatmap_buckets(response.json(), "A") == pytest.approx(expected_buckets), source
assert [column["values"] for column in get_heatmap_columns(response.json(), "A")] == [
[0, 1, 0],
[0, 1, 0],
[0, 1, 0],
[0, 1, 0],
[0, 1, 0],
], source
assert_identical_query_response(from_cache, uncached)
def test_builder_narrowing_a_histogram(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"heatmap_cache_histogram_{uuid4().hex[:8]}_bucket"
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
end_time_ms = start_time_ms + 10 * MINUTE_MS
# the count each `le` reports every minute, cumulative across `le` as a
# histogram is. For the first five minutes the ten arrivals are all at or
# below 1, for the last five they are all between 4 and 8, and the buckets
# holding none of them report a count of 0 rather than going unreported
le_to_counts = {
"1": [10, 10, 10, 10, 10, 0, 0, 0, 0, 0],
"2": [10, 10, 10, 10, 10, 0, 0, 0, 0, 0],
"4": [10, 10, 10, 10, 10, 0, 0, 0, 0, 0],
"8": [10, 10, 10, 10, 10, 10, 10, 10, 10, 10],
"+Inf": [10, 10, 10, 10, 10, 10, 10, 10, 10, 10],
}
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"le": le},
timestamp=start_time + timedelta(minutes=minute),
value=count,
temporality="Delta",
type_="Histogram",
)
for le, counts in le_to_counts.items()
for minute, count in enumerate(counts)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
query = [build_builder_query("A", metric_name, "increase", "p50", temporality="delta", group_by=["le"])]
# the whole range first, which is what puts its axis in the cache
whole_range = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert whole_range.status_code == HTTPStatus.OK, whole_range.text
assert get_heatmap_buckets(whole_range.json(), "A") == [1, 2, 4, 8]
assert [column["values"] for column in get_heatmap_columns(whole_range.json(), "A")] == [
[10, 0, 0, 0, 0],
[10, 0, 0, 0, 0],
[10, 0, 0, 0, 0],
[10, 0, 0, 0, 0],
[10, 0, 0, 0, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
]
# even though this shortened time range has no data below 4, all histogram
# buckets are still returned back
half_start_ms = start_time_ms + 5 * MINUTE_MS
from_cache = make_query_request(signoz, token, half_start_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
uncached = make_query_request(signoz, token, half_start_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
for source, response in (("uncached", uncached), ("from cache", from_cache)):
assert get_heatmap_buckets(response.json(), "A") == [1, 2, 4, 8], source
assert [column["values"] for column in get_heatmap_columns(response.json(), "A")] == [
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
[0, 0, 0, 10, 0],
], source
assert_identical_query_response(from_cache, uncached)
def test_builder_shortening_the_time_range_at_the_end(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"heatmap_cache_end_shortened_{uuid4().hex[:8]}"
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
end_time_ms_base_query = start_time_ms + 10 * MINUTE_MS
end_time_ms_shortened_query = start_time_ms + 7 * MINUTE_MS
query = [build_builder_query("A", metric_name, "max", "max", step_interval=300, bucket_options=build_linear_bucket_options(1000, 10))]
# the 5m step splits the ten minutes into two columns, each the max over its
# own step: minutes 0-4 and minutes 5-9. The second changes partway through,
# 250 until minute 7 and then 850, which fall six buckets apart, so ending
# the range at minute 7 has to reach a different bucket than ending it at
# minute 10 and an axis that stops well below it
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=(150, 150, 150, 150, 150, 250, 250, 850, 850, 850)[minute],
type_="Gauge",
is_monotonic=False,
)
for minute in range(10)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, request_type=RequestType.HEATMAP, no_cache=False)
assert base_query.status_code == HTTPStatus.OK, base_query.text
# 100 wide buckets, and the two maxes are 150 and 850, so the axis runs from
# the bottom of (100, 200] to the top of (800, 900]
assert get_heatmap_buckets(base_query.json(), "A") == pytest.approx([100, 200, 300, 400, 500, 600, 700, 800, 900])
base_columns = get_heatmap_columns(base_query.json(), "A")
assert [column["values"] for column in base_columns] == [
[0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
[0, 0, 0, 0, 0, 0, 0, 0, 1, 0],
]
assert [column.get("partial", False) for column in base_columns] == [False, False]
from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, request_type=RequestType.HEATMAP, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, request_type=RequestType.HEATMAP, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
# the shortened end reaches only minutes 5-6 of the second column, whose max
# is 250 and which comes back partial. Nothing in this window passes 300, so
# the axis stops there rather than carrying the buckets above it
for label, response in (("uncached", uncached), ("from cache", from_cache)):
assert get_heatmap_buckets(response.json(), "A") == pytest.approx([100, 200, 300]), label
columns = get_heatmap_columns(response.json(), "A")
assert [column["values"] for column in columns] == [
[0, 1, 0, 0],
[0, 0, 1, 0],
], label
assert [column.get("partial", False) for column in columns] == [False, True], label
assert_identical_query_response(from_cache, uncached)
def test_builder_shortening_the_time_range_at_the_start(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"heatmap_cache_start_shortened_{uuid4().hex[:8]}"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache. Flooring to a multiple of the 5m step makes the base query span
# two whole steps, so both its columns are complete
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 300 * 300, tz=UTC)
start_time_ms_base_query = int(start_time.timestamp() * 1000)
start_time_ms_shortened_query = start_time_ms_base_query + 3 * MINUTE_MS
end_time_ms = start_time_ms_base_query + 10 * MINUTE_MS
query = [build_builder_query("A", metric_name, "max", "max", step_interval=300, bucket_options=build_linear_bucket_options(1000, 10))]
# the 5m step splits the ten minutes into two columns, each the max over its
# own step: minutes 0-4 and minutes 5-9. Only minute 0 reaches 950, so a
# first column counted in (900, 1000] says the whole step was read even
# though the shortened range opens at minute 3
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=(950, 150, 150, 150, 150, 350, 350, 350, 350, 350)[minute],
type_="Gauge",
is_monotonic=False,
)
for minute in range(10)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert base_query.status_code == HTTPStatus.OK, base_query.text
# 100 wide buckets, and the two maxes are 950 and 350, so the axis runs from
# the bottom of (300, 400] to the top of (900, 1000]
assert get_heatmap_buckets(base_query.json(), "A") == pytest.approx([300, 400, 500, 600, 700, 800, 900, 1000])
base_columns = get_heatmap_columns(base_query.json(), "A")
assert [column["values"] for column in base_columns] == [
[0, 0, 0, 0, 0, 0, 0, 1, 0],
[0, 1, 0, 0, 0, 0, 0, 0, 0],
]
assert [column.get("partial", False) for column in base_columns] == [False, False]
from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
# starting inside the first column's step flags that column partial without
# clipping its counts, which still cover the whole step and so reach the 950
# at minute 0, leaving the axis where the base query drew it
for label, response in (("uncached", uncached), ("from cache", from_cache)):
assert get_heatmap_buckets(response.json(), "A") == pytest.approx([300, 400, 500, 600, 700, 800, 900, 1000]), label
columns = get_heatmap_columns(response.json(), "A")
assert [column["values"] for column in columns] == [
[0, 0, 0, 0, 0, 0, 0, 1, 0],
[0, 1, 0, 0, 0, 0, 0, 0, 0],
], label
assert [column.get("partial", False) for column in columns] == [True, False], label
assert_identical_query_response(from_cache, uncached)
def test_builder_refreshing_a_sliding_time_range(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"heatmap_cache_sliding_{uuid4().hex[:8]}"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
query = [build_builder_query("A", metric_name, "max", "max", bucket_options=build_linear_bucket_options(1000, 10))]
# the 1m step gives one column per seeded minute, and 100 wide buckets give
# every minute a bucket no other minute reaches, so a column stitched in from
# the wrong range is counted in the wrong bucket
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"service": "api"},
timestamp=start_time + timedelta(minutes=minute),
value=100 * minute + 50,
type_="Gauge",
is_monotonic=False,
)
for minute in range(7)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
# the window slides onto a bucket a minute higher each refresh, so an axis
# carried over from an earlier one is off by as many buckets
expected_buckets_by_refresh = [
[0, 100, 200, 300, 400],
[100, 200, 300, 400, 500],
[200, 300, 400, 500, 600],
[300, 400, 500, 600, 700],
]
# whichever four minutes a refresh reads, each is in a bucket of its own and
# they arrive in order, so the counts run down the diagonal
expected_columns = [
[0, 1, 0, 0, 0, 0],
[0, 0, 1, 0, 0, 0],
[0, 0, 0, 1, 0, 0],
[0, 0, 0, 0, 1, 0],
]
# a dashboard left open on a four minute range, re-running a minute later each
# time, so every refresh is stitched out of the ranges the ones before it cached
for refresh, expected_buckets in enumerate(expected_buckets_by_refresh):
refresh_start_ms = start_time_ms + refresh * MINUTE_MS
from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 4 * MINUTE_MS, query, request_type=RequestType.HEATMAP, no_cache=False)
assert from_cache.status_code == HTTPStatus.OK, from_cache.text
assert get_heatmap_buckets(from_cache.json(), "A") == pytest.approx(expected_buckets), f"refresh {refresh}"
# a column served twice, dropped, or carried over from an earlier refresh
# breaks the diagonal or the run of timestamps
columns = get_heatmap_columns(from_cache.json(), "A")
assert [column["timestamp"] for column in columns] == [
refresh_start_ms,
refresh_start_ms + MINUTE_MS,
refresh_start_ms + 2 * MINUTE_MS,
refresh_start_ms + 3 * MINUTE_MS,
], f"refresh {refresh}"
assert [column["values"] for column in columns] == expected_columns, f"refresh {refresh}"
uncached = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 4 * MINUTE_MS, query, request_type=RequestType.HEATMAP, no_cache=True)
assert uncached.status_code == HTTPStatus.OK, uncached.text
assert_identical_query_response(from_cache, uncached)
def test_promql_running_the_same_query_twice(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"heatmap_cache_repeat_{uuid4().hex[:8]}_bucket"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
start_time_ms = int(start_time.timestamp() * 1000)
end_time_ms = start_time_ms + 2 * MINUTE_MS
query = [{"type": "promql", "spec": {"name": "A", "query": f"sum by (le) (increase({metric_name}[2m]))", "step": 60}}]
# the cumulative count of each `le`, one entry per minute. The counters open
# a minute before the query so its first column has something to increase
# over, and start far above their own rise across the range, below which
# increase clips its back-extrapolation at a counter's zero point
le_to_counts = {
"1": [1000, 1005, 1010, 1020],
"2": [2000, 2010, 2025, 2040],
"4": [3000, 3015, 3040, 3070],
"+Inf": [4000, 4022, 4050, 4090],
}
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"__temporality__": "Cumulative", "service": "api", "le": le},
timestamp=start_time + timedelta(minutes=minute),
value=count,
temporality="Cumulative",
type_="Histogram",
)
for le, counts in le_to_counts.items()
for minute, count in enumerate(counts, start=-1)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert first.status_code == HTTPStatus.OK, first.text
second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert second.status_code == HTTPStatus.OK, second.text
# promql reports a column at the instant the range closes, and the second
# run, answered out of what the first one cached, has to keep it
for run, response in (("first", first), ("second", second)):
assert get_heatmap_buckets(response.json(), "A") == [1, 2, 4], run
## what the query returns per `le` is cumulative across `le`, so each
## count is its own minus the one below it, and `le=+Inf` has no finite
## bound to sit on and lands in the trailing slot. increase over a 2m
## window of minutely samples extrapolates one minute's rise to two.
assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(response.json(), "A")] == [
(start_time_ms, [10, 10, 10, 14]), # t = 0, the minute brings 5, 10, 15 and 22 arrivals at or below each `le`
(start_time_ms + MINUTE_MS, [10, 20, 20, 6]), # t = 1m, 5, 15, 25 and 28
(end_time_ms, [20, 10, 30, 20]), # t = 2m, 10, 15, 30 and 40
], f"{run} run"
assert_identical_query_response(first, second)
def test_promql_shifting_the_time_range(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
metric_name = f"heatmap_cache_shift_{uuid4().hex[:8]}_bucket"
# 40 minutes back clears the flux interval, which holds recent data out of
# the cache. Flooring to a whole minute is what makes the first query aligned
# to its 1m step, and the unaligned one half a step off it
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=40)).timestamp()) // 60 * 60, tz=UTC)
aligned_start_time_ms = int(start_time.timestamp() * 1000)
aligned_end_time_ms = aligned_start_time_ms + 3 * MINUTE_MS
unaligned_start_time_ms = aligned_start_time_ms + MINUTE_MS // 2
unaligned_end_time_ms = aligned_end_time_ms + MINUTE_MS // 2
query = [{"type": "promql", "spec": {"name": "A", "query": f"sum by (le) (max_over_time({metric_name}[2m]))", "step": 60}}]
# a sample every 30s, each `le` counting up by its own fixed amount every
# time. The two queries report 30s apart, so they land on different samples
# and share no count between them
le_to_arrivals_per_sample = {"1": 100, "2": 300, "4": 600, "+Inf": 1000}
insert_metrics(
[
Metrics(
metric_name=metric_name,
labels={"__temporality__": "Cumulative", "service": "api", "le": le},
timestamp=start_time + timedelta(seconds=30 * half_minute),
value=arrivals_per_sample * (half_minute + 4),
temporality="Cumulative",
type_="Histogram",
)
for le, arrivals_per_sample in le_to_arrivals_per_sample.items()
for half_minute in range(-3, 8)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
# what the cache now holds, and what the unaligned query must not be served
assert get_heatmap_buckets(aligned_and_cached.json(), "A") == [1, 2, 4]
## each column reads the counters at their latest sample at or before its
## timestamp, and a bucket holds its own `le`'s count less the one below it.
assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(aligned_and_cached.json(), "A")] == [
(aligned_start_time_ms, [400, 800, 1200, 1600]), # t = 0, the fourth sample
(aligned_start_time_ms + MINUTE_MS, [600, 1200, 1800, 2400]), # t = 1m, the sixth
(aligned_start_time_ms + 2 * MINUTE_MS, [800, 1600, 2400, 3200]), # t = 2m, the eighth
(aligned_end_time_ms, [1000, 2000, 3000, 4000]), # t = 3m, the tenth
]
## every column falls on a sample the aligned run never reported, so being
## served the cached run's answer shows up in the counts and not only the
## timestamps.
unaligned_columns = [
(unaligned_start_time_ms, [500, 1000, 1500, 2000]), # t = 30s, the fifth sample
(unaligned_start_time_ms + MINUTE_MS, [700, 1400, 2100, 2800]), # t = 1m30s, the seventh
(unaligned_start_time_ms + 2 * MINUTE_MS, [900, 1800, 2700, 3600]), # t = 2m30s, the ninth
(unaligned_end_time_ms, [1100, 2200, 3300, 4400]), # t = 3m30s, the eleventh
]
unaligned_and_uncached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=True)
assert unaligned_and_uncached.status_code == HTTPStatus.OK, unaligned_and_uncached.text
assert get_heatmap_buckets(unaligned_and_uncached.json(), "A") == [1, 2, 4], "unaligned query, uncached"
assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(unaligned_and_uncached.json(), "A")] == unaligned_columns, "unaligned query, uncached"
# promql reports at the range start plus whole steps, so these columns sit
# 30s off the cached ones. The first run stores them, the second reads them back
for run in ("first", "second"):
unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, request_type=RequestType.HEATMAP, no_cache=False)
assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
assert get_heatmap_buckets(unaligned_and_cached.json(), "A") == [1, 2, 4], f"unaligned query, {run} run"
assert [(column["timestamp"], column["values"]) for column in get_heatmap_columns(unaligned_and_cached.json(), "A")] == unaligned_columns, f"unaligned query, {run} run"
assert_identical_query_response(unaligned_and_cached, unaligned_and_uncached)

View File

@@ -0,0 +1,280 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
import requests
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querierai import root_span
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
WATERFALL_FIELDS = (
"startTimestampMillis",
"endTimestampMillis",
"rootServiceName",
"rootServiceEntryPoint",
"totalSpansCount",
"totalErrorSpansCount",
"hasMissingSpans",
)
@pytest.mark.parametrize("attribute_backend", ["map", "json"])
def test_summary_ai_trace(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
use_attribute_backend: Callable[[str], None],
attribute_backend: str,
) -> None:
"""The summary carries the waterfall's trace-level fields and, for a trace with gen_ai
spans, token totals over every LLM span and the cost summed over the spans that carry it.
Spans are written to one layout only, so a read from the wrong column sums to zero."""
use_attribute_backend(attribute_backend)
write_mode = "json_only" if attribute_backend == "json" else "legacy_only"
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = f"td-summary-{attribute_backend}"
resources = {"service.name": service}
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
insert_traces(
[
root_span(now=now, trace_id=trace_id, span_id=root_id, resources=resources, duration_s=4),
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.usage.input_tokens": 100,
"gen_ai.usage.output_tokens": 20,
"gen_ai.usage.cache_read.input_tokens": 7,
"_signoz.gen_ai.total_cost": 0.01,
},
attribute_write_mode=write_mode,
),
# a failed LLM call: counted in tokens and errors, but priced by nobody
Traces(
timestamp=now - timedelta(seconds=3),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_ERROR,
resources=resources,
attributes={
"gen_ai.request.model": "gpt-4o-mini",
"gen_ai.usage.input_tokens": 50,
"gen_ai.usage.output_tokens": 5,
},
attribute_write_mode=write_mode,
),
Traces(
timestamp=now - timedelta(seconds=2),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="execute_tool",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.tool.name": "get_weather"},
attribute_write_mode=write_mode,
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
headers = {"authorization": f"Bearer {token}", "content-type": "application/json"}
summary = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/traces/{trace_id}/summary"), timeout=10, headers=headers)
assert summary.status_code == HTTPStatus.OK, summary.text
summary = summary.json()["data"]
waterfall = requests.post(
signoz.self.host_configs["8080"].get(f"/api/v4/traces/{trace_id}/waterfall"),
timeout=10,
headers=headers,
json={"selectedSpanId": "", "uncollapsedSpans": []},
)
assert waterfall.status_code == HTTPStatus.OK, waterfall.text
waterfall = waterfall.json()["data"]
assert {k: summary[k] for k in WATERFALL_FIELDS} == {k: waterfall[k] for k in WATERFALL_FIELDS}
assert summary["rootServiceName"] == service
assert summary["rootServiceEntryPoint"] == "POST /api/chat"
assert summary["totalSpansCount"] == 4
assert summary["totalErrorSpansCount"] == 1
assert summary["hasMissingSpans"] is False
assert summary["ai"]["tokens"] == {"input": 150, "output": 25, "cacheRead": 7, "cacheWrite": 0, "reasoning": 0}
assert summary["ai"]["totalCost"] == pytest.approx(0.01)
def test_summary_ai_trace_across_json_rollout(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
seed_attribute_evolution: Callable[[str, datetime], None],
) -> None:
"""A trace that straddles the attribute JSON rollout has LLM spans written only to the legacy
maps before it and to the JSON column after it. The summary window covers both, so the gen_ai
reads must fall back across columns and sum every span."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
rollout = now - timedelta(minutes=30)
seed_attribute_evolution("traces", rollout)
service = "td-summary-rollout"
resources = {"service.name": service}
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
insert_traces(
[
Traces(
timestamp=rollout - timedelta(minutes=10),
duration=timedelta(minutes=15),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="long agent run",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attribute_write_mode="legacy_only",
),
Traces(
timestamp=rollout - timedelta(minutes=5),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.request.model": "gpt-4o-mini", "gen_ai.usage.input_tokens": 100, "gen_ai.usage.output_tokens": 20, "_signoz.gen_ai.total_cost": 0.01},
attribute_write_mode="legacy_only",
),
Traces(
timestamp=rollout + timedelta(minutes=4),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat gpt-4o-mini",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.request.model": "gpt-4o-mini", "gen_ai.usage.input_tokens": 50, "gen_ai.usage.output_tokens": 5, "_signoz.gen_ai.total_cost": 0.02},
attribute_write_mode="json_only",
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
summary = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/traces/{trace_id}/summary"),
timeout=10,
headers={"authorization": f"Bearer {token}"},
)
assert summary.status_code == HTTPStatus.OK, summary.text
summary = summary.json()["data"]
assert summary["totalSpansCount"] == 3
assert summary["rootServiceEntryPoint"] == "long agent run"
assert summary["ai"]["tokens"] == {"input": 150, "output": 25, "cacheRead": 0, "cacheWrite": 0, "reasoning": 0}
assert summary["ai"]["totalCost"] == pytest.approx(0.03)
def test_summary_non_ai_trace_with_missing_root(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""A trace whose recorded spans all hang off an unrecorded parent reports the synthetic
"Missing Span" root exactly as the waterfall does, and a trace without gen_ai spans has
no `ai` block."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
resources = {"service.name": "td-summary-orphan"}
trace_id = TraceIdGenerator.trace_id()
missing_parent_id = TraceIdGenerator.span_id()
insert_traces(
[
Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=missing_parent_id,
name="SELECT users",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
),
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=missing_parent_id,
name="publish event",
kind=TracesKind.SPAN_KIND_PRODUCER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
headers = {"authorization": f"Bearer {token}", "content-type": "application/json"}
summary = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/traces/{trace_id}/summary"), timeout=10, headers=headers)
assert summary.status_code == HTTPStatus.OK, summary.text
summary = summary.json()["data"]
waterfall = requests.post(
signoz.self.host_configs["8080"].get(f"/api/v4/traces/{trace_id}/waterfall"),
timeout=10,
headers=headers,
json={"selectedSpanId": "", "uncollapsedSpans": []},
)
assert waterfall.status_code == HTTPStatus.OK, waterfall.text
waterfall = waterfall.json()["data"]
assert {k: summary[k] for k in WATERFALL_FIELDS} == {k: waterfall[k] for k in WATERFALL_FIELDS}
assert summary["hasMissingSpans"] is True
assert summary["rootServiceName"] == ""
assert summary["rootServiceEntryPoint"] == "Missing Span"
assert summary["totalSpansCount"] == 2
assert "ai" not in summary
def test_summary_unknown_trace(
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 = requests.get(
signoz.self.host_configs["8080"].get(f"/api/v1/traces/{TraceIdGenerator.trace_id()}/summary"),
timeout=10,
headers={"authorization": f"Bearer {token}"},
)
assert response.status_code == HTTPStatus.NOT_FOUND, response.text