mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-18 17:30:41 +01:00
Compare commits
4 Commits
nv/caching
...
issue_6090
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5dae0b975a | ||
|
|
d1a382945c | ||
|
|
59af5e0367 | ||
|
|
8286e787b2 |
1
.github/workflows/integrationci.yaml
vendored
1
.github/workflows/integrationci.yaml
vendored
@@ -67,6 +67,7 @@ jobs:
|
||||
- semconvfamilies
|
||||
- serviceaccount
|
||||
- spanmapper
|
||||
- tracedetail
|
||||
- querier_json_body
|
||||
- querier_skip_resource_fingerprint
|
||||
- ttl
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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{}
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
76
pkg/types/spantypes/trace_summary.go
Normal file
76
pkg/types/spantypes/trace_summary.go
Normal 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
|
||||
}
|
||||
1
tests/fixtures/traces.py
vendored
1
tests/fixtures/traces.py
vendored
@@ -895,6 +895,7 @@ _TRACES_TABLES_TO_TRUNCATE = [
|
||||
"span_attributes_keys",
|
||||
"signoz_error_index_v2",
|
||||
"top_level_operations",
|
||||
"trace_summary",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
280
tests/integration/tests/tracedetail/01_summary.py
Normal file
280
tests/integration/tests/tracedetail/01_summary.py
Normal 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
|
||||
Reference in New Issue
Block a user