mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-13 08:20:36 +01:00
Compare commits
1 Commits
feat/span-
...
issue-4293
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a077ec03f4 |
1
.github/workflows/integrationci.yaml
vendored
1
.github/workflows/integrationci.yaml
vendored
@@ -58,6 +58,7 @@ jobs:
|
||||
- querierai
|
||||
- rawexportdata
|
||||
- promqlconformance
|
||||
- promapiconformance
|
||||
- querierauthz
|
||||
- role
|
||||
- rootuser
|
||||
|
||||
@@ -1617,13 +1617,13 @@ export const getNamespaceMetricsQueryPayload = (
|
||||
reduceTo: ReduceOperators.LAST,
|
||||
spaceAggregation: 'max',
|
||||
stepInterval: 60,
|
||||
timeAggregation: 'latest',
|
||||
timeAggregation: 'avg',
|
||||
},
|
||||
],
|
||||
queryFormulas: [
|
||||
{
|
||||
disabled: false,
|
||||
expression: '(B/A) * 100',
|
||||
expression: 'A/B',
|
||||
legend: 'util %',
|
||||
queryName: 'F1',
|
||||
},
|
||||
|
||||
@@ -1521,9 +1521,9 @@ const onboardingConfigWithLinks = [
|
||||
},
|
||||
{
|
||||
dataSource: 'nginx-tracing',
|
||||
label: 'Nginx - OpenTelemetry',
|
||||
label: 'Nginx - Tracing',
|
||||
imgUrl: nginxUrl,
|
||||
tags: ['apm/traces', 'logs', 'metrics'],
|
||||
tags: ['apm/traces'],
|
||||
module: 'apm',
|
||||
relatedSearchKeywords: [
|
||||
'apm',
|
||||
@@ -1626,7 +1626,7 @@ const onboardingConfigWithLinks = [
|
||||
dataSource: 'cloudflare-workers',
|
||||
label: 'Cloudflare Workers',
|
||||
imgUrl: cloudflareUrl,
|
||||
tags: ['apm/traces', 'logs'],
|
||||
tags: ['apm/traces'],
|
||||
module: 'apm',
|
||||
relatedSearchKeywords: [
|
||||
'cloudflare',
|
||||
@@ -5346,17 +5346,13 @@ const onboardingConfigWithLinks = [
|
||||
dataSource: 'temporal',
|
||||
label: 'Temporal',
|
||||
imgUrl: temporalUrl,
|
||||
tags: ['apm/traces', 'logs', 'metrics'],
|
||||
tags: ['apm/traces'],
|
||||
module: 'apm',
|
||||
relatedSearchKeywords: [
|
||||
'apm',
|
||||
'application performance monitoring',
|
||||
'integrations',
|
||||
'logs',
|
||||
'metrics',
|
||||
'temporal',
|
||||
'temporal logs',
|
||||
'temporal metrics',
|
||||
'temporal traces',
|
||||
'traces',
|
||||
'tracing',
|
||||
@@ -5482,7 +5478,7 @@ const onboardingConfigWithLinks = [
|
||||
dataSource: 'dbos',
|
||||
label: 'DBOS',
|
||||
imgUrl: dbosUrl,
|
||||
tags: ['apm/traces', 'logs'],
|
||||
tags: ['apm/traces'],
|
||||
module: 'apm',
|
||||
relatedSearchKeywords: [
|
||||
'database oriented',
|
||||
@@ -6626,7 +6622,7 @@ const onboardingConfigWithLinks = [
|
||||
dataSource: 'opentelemetry-ebpf',
|
||||
label: 'OpenTelemetry eBPF (OBI)',
|
||||
imgUrl: opentelemetryUrl,
|
||||
tags: ['apm/traces', 'metrics'],
|
||||
tags: ['apm/traces'],
|
||||
module: 'apm',
|
||||
relatedSearchKeywords: [
|
||||
'auto instrumentation',
|
||||
|
||||
@@ -42,7 +42,7 @@ function SpanPercentilePanel({
|
||||
selectedTimeRange,
|
||||
setSelectedTimeRange,
|
||||
showResourceAttributesSelector,
|
||||
toggleResourceAttributesSelector,
|
||||
setShowResourceAttributesSelector,
|
||||
resourceAttributesSearchQuery,
|
||||
setResourceAttributesSearchQuery,
|
||||
spanResourceAttributes,
|
||||
@@ -72,7 +72,9 @@ function SpanPercentilePanel({
|
||||
variant="link"
|
||||
color="secondary"
|
||||
size="icon"
|
||||
onClick={toggleResourceAttributesSelector}
|
||||
onClick={(): void =>
|
||||
setShowResourceAttributesSelector(!showResourceAttributesSelector)
|
||||
}
|
||||
prefix={
|
||||
showResourceAttributesSelector ? <Check size={16} /> : <Plus size={16} />
|
||||
}
|
||||
|
||||
@@ -8,11 +8,6 @@ import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
|
||||
import { USER_PREFERENCES } from 'constants/userPreferences';
|
||||
import dayjs from 'dayjs';
|
||||
import useClickOutside from 'hooks/useClickOutside';
|
||||
import {
|
||||
TraceDetailEventKeys,
|
||||
TraceDetailEvents,
|
||||
} from 'pages/TraceDetailsV3/events';
|
||||
import { useTraceDetailLogEvent } from 'pages/TraceDetailsV3/hooks/useTraceDetailLogEvent';
|
||||
import { SpanV3 } from 'types/api/trace/getTraceV3';
|
||||
|
||||
export interface IResourceAttribute {
|
||||
@@ -42,7 +37,7 @@ export interface UseSpanPercentileReturn {
|
||||
selectedTimeRange: number;
|
||||
setSelectedTimeRange: (range: number) => void;
|
||||
showResourceAttributesSelector: boolean;
|
||||
toggleResourceAttributesSelector: () => void;
|
||||
setShowResourceAttributesSelector: (show: boolean) => void;
|
||||
resourceAttributesSearchQuery: string;
|
||||
setResourceAttributesSearchQuery: (query: string) => void;
|
||||
spanResourceAttributes: IResourceAttribute[];
|
||||
@@ -81,8 +76,6 @@ function useSpanPercentile(selectedSpan: SpanV3): UseSpanPercentileReturn {
|
||||
|
||||
const resourceAttributesSelectorRef = useRef<HTMLDivElement | null>(null);
|
||||
|
||||
const logTraceEvent = useTraceDetailLogEvent('v3', selectedSpan.trace_id);
|
||||
|
||||
useClickOutside({
|
||||
ref: resourceAttributesSelectorRef,
|
||||
onClickOutside: () => {
|
||||
@@ -264,12 +257,6 @@ function useSpanPercentile(selectedSpan: SpanV3): UseSpanPercentileReturn {
|
||||
|
||||
const handleResourceAttributeChange = useCallback(
|
||||
(key: string, value: string, isSelected: boolean): void => {
|
||||
logTraceEvent(TraceDetailEvents.SpanPercentileAttributeChanged, {
|
||||
[TraceDetailEventKeys.SpanId]: selectedSpan.span_id,
|
||||
[TraceDetailEventKeys.ResourceAttributeKey]: key,
|
||||
[TraceDetailEventKeys.Selected]: isSelected,
|
||||
});
|
||||
|
||||
updateSpanResourceAttributes((prev) =>
|
||||
prev.map((attr) => (attr.key === key ? { ...attr, isSelected } : attr)),
|
||||
);
|
||||
@@ -284,7 +271,7 @@ function useSpanPercentile(selectedSpan: SpanV3): UseSpanPercentileReturn {
|
||||
setShouldFetchData(true);
|
||||
setShouldUpdateUserPreference(true);
|
||||
},
|
||||
[selectedResourceAttributes, logTraceEvent, selectedSpan.span_id],
|
||||
[selectedResourceAttributes],
|
||||
);
|
||||
|
||||
useEffect(() => {
|
||||
@@ -306,37 +293,12 @@ function useSpanPercentile(selectedSpan: SpanV3): UseSpanPercentileReturn {
|
||||
'ms',
|
||||
);
|
||||
|
||||
const toggleOpen = useCallback(() => {
|
||||
const nextOpen = !isOpen;
|
||||
setIsOpen(nextOpen);
|
||||
logTraceEvent(TraceDetailEvents.SpanPercentileToggled, {
|
||||
[TraceDetailEventKeys.SpanId]: selectedSpan.span_id,
|
||||
[TraceDetailEventKeys.Open]: nextOpen,
|
||||
[TraceDetailEventKeys.PercentileValue]: percentileValue,
|
||||
});
|
||||
}, [isOpen, logTraceEvent, selectedSpan.span_id, percentileValue]);
|
||||
const toggleOpen = useCallback(() => setIsOpen((prev) => !prev), []);
|
||||
|
||||
const toggleResourceAttributesSelector = useCallback(() => {
|
||||
const nextOpen = !showResourceAttributesSelector;
|
||||
setShowResourceAttributesSelector(nextOpen);
|
||||
logTraceEvent(TraceDetailEvents.SpanPercentileAttributesSelectorToggled, {
|
||||
[TraceDetailEventKeys.SpanId]: selectedSpan.span_id,
|
||||
[TraceDetailEventKeys.Open]: nextOpen,
|
||||
});
|
||||
}, [showResourceAttributesSelector, logTraceEvent, selectedSpan.span_id]);
|
||||
|
||||
const handleTimeRangeChange = useCallback(
|
||||
(range: number): void => {
|
||||
logTraceEvent(TraceDetailEvents.SpanPercentileTimeRangeChanged, {
|
||||
[TraceDetailEventKeys.SpanId]: selectedSpan.span_id,
|
||||
[TraceDetailEventKeys.From]: selectedTimeRange,
|
||||
[TraceDetailEventKeys.To]: range,
|
||||
});
|
||||
setShouldFetchData(true);
|
||||
setSelectedTimeRange(range);
|
||||
},
|
||||
[logTraceEvent, selectedSpan.span_id, selectedTimeRange],
|
||||
);
|
||||
const handleTimeRangeChange = useCallback((range: number): void => {
|
||||
setShouldFetchData(true);
|
||||
setSelectedTimeRange(range);
|
||||
}, []);
|
||||
|
||||
return {
|
||||
isOpen,
|
||||
@@ -350,7 +312,7 @@ function useSpanPercentile(selectedSpan: SpanV3): UseSpanPercentileReturn {
|
||||
selectedTimeRange,
|
||||
setSelectedTimeRange: handleTimeRangeChange,
|
||||
showResourceAttributesSelector,
|
||||
toggleResourceAttributesSelector,
|
||||
setShowResourceAttributesSelector,
|
||||
resourceAttributesSearchQuery,
|
||||
setResourceAttributesSearchQuery,
|
||||
spanResourceAttributes,
|
||||
|
||||
@@ -8,10 +8,6 @@ export enum TraceDetailEvents {
|
||||
SpanPanelTabChanged = 'Trace Detail: Span panel tab changed',
|
||||
DownloadTriggered = 'Trace Detail: Download triggered',
|
||||
DownloadCancelled = 'Trace Detail: Download cancelled',
|
||||
SpanPercentileToggled = 'Trace Detail: Span percentile toggled',
|
||||
SpanPercentileTimeRangeChanged = 'Trace Detail: Span percentile time range changed',
|
||||
SpanPercentileAttributesSelectorToggled = 'Trace Detail: Span percentile attributes selector toggled',
|
||||
SpanPercentileAttributeChanged = 'Trace Detail: Span percentile attribute changed',
|
||||
}
|
||||
|
||||
export enum TraceDetailEventKeys {
|
||||
@@ -40,10 +36,6 @@ export enum TraceDetailEventKeys {
|
||||
SpanId = 'spanId',
|
||||
// Download triggered (reuses TotalSpansCount for trace size)
|
||||
Format = 'format',
|
||||
// Span percentile (reuses Open, SpanId, From, To)
|
||||
PercentileValue = 'percentileValue',
|
||||
ResourceAttributeKey = 'resourceAttributeKey',
|
||||
Selected = 'selected',
|
||||
}
|
||||
|
||||
export type TraceDetailView = 'v2' | 'v3';
|
||||
|
||||
2
go.mod
2
go.mod
@@ -4,7 +4,7 @@ go 1.25.7
|
||||
|
||||
require (
|
||||
dario.cat/mergo v1.0.2
|
||||
github.com/AfterShip/clickhouse-sql-parser v0.5.6
|
||||
github.com/AfterShip/clickhouse-sql-parser v0.5.5
|
||||
github.com/ClickHouse/clickhouse-go/v2 v2.44.0
|
||||
github.com/DATA-DOG/go-sqlmock v1.5.2
|
||||
github.com/SigNoz/clickhouse-go-mock v0.14.0
|
||||
|
||||
4
go.sum
4
go.sum
@@ -66,8 +66,8 @@ dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
|
||||
dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
|
||||
filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
|
||||
filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
|
||||
github.com/AfterShip/clickhouse-sql-parser v0.5.6 h1:njgRLxQz/pE16ZO1MSjWadIzabwqsjDMMX8RR5Dbv7Y=
|
||||
github.com/AfterShip/clickhouse-sql-parser v0.5.6/go.mod h1:Qi3qvPTfZb/aFwI5V4WFOahgjsLJa4MzVijIAfwOhDw=
|
||||
github.com/AfterShip/clickhouse-sql-parser v0.5.5 h1:LCA23yAA4GgF73PoYXb67yzCdC4sXsj4geQz1Oij3U8=
|
||||
github.com/AfterShip/clickhouse-sql-parser v0.5.5/go.mod h1:Qi3qvPTfZb/aFwI5V4WFOahgjsLJa4MzVijIAfwOhDw=
|
||||
github.com/Azure/azure-sdk-for-go v68.0.0+incompatible h1:fcYLmCpyNYRnvJbPerq7U0hS+6+I79yEDJBqVNcqUzU=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0 h1:fou+2+WFTib47nS+nz/ozhEBnvU96bKHy6LjRsY4E28=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0/go.mod h1:t76Ruy8AHvUAC8GfMWJMa0ElSbuIcO03NLpynfbgsPA=
|
||||
|
||||
9
pkg/prometheus/handler.go
Normal file
9
pkg/prometheus/handler.go
Normal file
@@ -0,0 +1,9 @@
|
||||
package prometheus
|
||||
|
||||
import "net/http"
|
||||
|
||||
type Handler interface {
|
||||
Query(http.ResponseWriter, *http.Request)
|
||||
|
||||
QueryRange(http.ResponseWriter, *http.Request)
|
||||
}
|
||||
259
pkg/prometheus/promapi/handler.go
Normal file
259
pkg/prometheus/promapi/handler.go
Normal file
@@ -0,0 +1,259 @@
|
||||
// Package promapi serves the Prometheus HTTP query API over a
|
||||
// prometheus.Prometheus provider: /query and /query_range in the shape of
|
||||
// Prometheus' /api/v1 endpoints (https://prometheus.io/docs/prometheus/latest/querying/api/),
|
||||
// intended to be mounted under a distinguishing prefix (/prometheus/api/v1)
|
||||
// so PromQL-only endpoints are separate from the SigNoz query APIs. The
|
||||
// request and response contracts follow Prometheus: form-encoded GET/POST
|
||||
// params, {"status":"success","data":{resultType,result}} on success and
|
||||
// {"status":"error","errorType","error"} with Prometheus' status codes on
|
||||
// failure — so Prometheus-compatible clients can point at the prefix.
|
||||
package promapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"math"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
promModel "github.com/prometheus/common/model"
|
||||
"github.com/prometheus/prometheus/promql"
|
||||
"github.com/prometheus/prometheus/promql/parser"
|
||||
"github.com/prometheus/prometheus/util/stats"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
)
|
||||
|
||||
type handler struct {
|
||||
logger *slog.Logger
|
||||
prom prometheus.Prometheus
|
||||
}
|
||||
|
||||
func NewHandler(logger *slog.Logger, prom prometheus.Prometheus) prometheus.Handler {
|
||||
return &handler{logger: logger, prom: prom}
|
||||
}
|
||||
|
||||
type errorType string
|
||||
|
||||
const (
|
||||
errBadData errorType = "bad_data"
|
||||
errExec errorType = "execution"
|
||||
errCanceled errorType = "canceled"
|
||||
errTimeout errorType = "timeout"
|
||||
errInternal errorType = "internal"
|
||||
)
|
||||
|
||||
type queryData struct {
|
||||
ResultType parser.ValueType `json:"resultType"`
|
||||
Result parser.Value `json:"result"`
|
||||
Stats stats.QueryStats `json:"stats,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
Status string `json:"status"`
|
||||
Data *queryData `json:"data,omitempty"`
|
||||
ErrorType errorType `json:"errorType,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Warnings []string `json:"warnings,omitempty"`
|
||||
Infos []string `json:"infos,omitempty"`
|
||||
}
|
||||
|
||||
// QueryRange evaluates an expression over a grid: query, start, end, step,
|
||||
// and optional timeout/stats params, all in Prometheus' formats.
|
||||
func (h *handler) QueryRange(w http.ResponseWriter, r *http.Request) {
|
||||
start, err := parseTime(r.FormValue("start"))
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
end, err := parseTime(r.FormValue("end"))
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
if end.Before(start) {
|
||||
h.respondError(r.Context(), w, errBadData, errors.NewInvalidInputf(errors.CodeInvalidInput, "end timestamp must not be before start time"))
|
||||
return
|
||||
}
|
||||
step, err := parseDuration(r.FormValue("step"))
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
if step <= 0 {
|
||||
h.respondError(r.Context(), w, errBadData, errors.NewInvalidInputf(errors.CodeInvalidInput, "zero or negative query resolution step widths are not accepted. Try a positive integer"))
|
||||
return
|
||||
}
|
||||
// The engine materializes every point of every series; an unbounded
|
||||
// grid is an unbounded allocation. 11,000 points covers 60s resolution
|
||||
// for a week or 1h resolution for a year.
|
||||
if end.Sub(start)/step > 11000 {
|
||||
h.respondError(r.Context(), w, errBadData, errors.NewInvalidInputf(errors.CodeInvalidInput, "exceeded maximum resolution of 11,000 points per timeseries. Try decreasing the query resolution (?step=XX)"))
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel, err := h.contextWithTimeout(r)
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
defer cancel()
|
||||
|
||||
if h.tryRangeExecutor(ctx, w, r, start, end, step) {
|
||||
return
|
||||
}
|
||||
|
||||
qry, err := h.prom.Engine().NewRangeQuery(ctx, h.prom.Storage(), nil, r.FormValue("query"), start, end, step)
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
h.exec(ctx, w, r, qry)
|
||||
}
|
||||
|
||||
// tryRangeExecutor serves the query the way a RangeExecutor provider is
|
||||
// designed to serve: evaluated inside the datastore when the shape allows.
|
||||
// It reports whether the response was written.
|
||||
func (h *handler) tryRangeExecutor(ctx context.Context, w http.ResponseWriter, r *http.Request, start, end time.Time, step time.Duration) bool {
|
||||
re, ok := h.prom.(prometheus.RangeExecutor)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
matrix, served, err := re.TryExecuteRange(ctx, r.FormValue("query"), start, end, step)
|
||||
if err != nil {
|
||||
h.respondError(ctx, w, errExec, err)
|
||||
return true
|
||||
}
|
||||
if !served {
|
||||
return false
|
||||
}
|
||||
h.respond(ctx, w, &queryData{ResultType: matrix.Type(), Result: matrix}, nil, nil)
|
||||
return true
|
||||
}
|
||||
|
||||
// Query evaluates an expression at a single instant: query and optional
|
||||
// time/timeout/stats params. A missing time evaluates at the server's now,
|
||||
// as in Prometheus.
|
||||
func (h *handler) Query(w http.ResponseWriter, r *http.Request) {
|
||||
ts := time.Now()
|
||||
if t := r.FormValue("time"); t != "" {
|
||||
var err error
|
||||
ts, err = parseTime(t)
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
ctx, cancel, err := h.contextWithTimeout(r)
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
defer cancel()
|
||||
|
||||
qry, err := h.prom.Engine().NewInstantQuery(ctx, h.prom.Storage(), nil, r.FormValue("query"), ts)
|
||||
if err != nil {
|
||||
h.respondError(r.Context(), w, errBadData, err)
|
||||
return
|
||||
}
|
||||
h.exec(ctx, w, r, qry)
|
||||
}
|
||||
|
||||
func (h *handler) exec(ctx context.Context, w http.ResponseWriter, r *http.Request, qry promql.Query) {
|
||||
defer qry.Close()
|
||||
res := qry.Exec(ctx)
|
||||
if res.Err != nil {
|
||||
h.logger.ErrorContext(ctx, "error evaluating promql query", errors.Attr(res.Err))
|
||||
switch res.Err.(type) {
|
||||
case promql.ErrQueryCanceled:
|
||||
h.respondError(ctx, w, errCanceled, res.Err)
|
||||
case promql.ErrQueryTimeout:
|
||||
h.respondError(ctx, w, errTimeout, res.Err)
|
||||
case promql.ErrStorage:
|
||||
h.respondError(ctx, w, errInternal, res.Err)
|
||||
default:
|
||||
h.respondError(ctx, w, errExec, res.Err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
data := &queryData{ResultType: res.Value.Type(), Result: res.Value}
|
||||
if r.FormValue("stats") != "" {
|
||||
data.Stats = stats.NewQueryStats(qry.Stats())
|
||||
}
|
||||
warnings, infos := res.Warnings.AsStrings(r.FormValue("query"), 10, 10)
|
||||
h.respond(ctx, w, data, warnings, infos)
|
||||
}
|
||||
|
||||
func (h *handler) contextWithTimeout(r *http.Request) (context.Context, context.CancelFunc, error) {
|
||||
ctx := r.Context()
|
||||
if to := r.FormValue("timeout"); to != "" {
|
||||
timeout, err := parseDuration(to)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
return ctx, cancel, nil
|
||||
}
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
return ctx, cancel, nil
|
||||
}
|
||||
|
||||
func (h *handler) respond(ctx context.Context, w http.ResponseWriter, data *queryData, warnings, infos []string) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
if err := json.NewEncoder(w).Encode(&response{Status: "success", Data: data, Warnings: warnings, Infos: infos}); err != nil {
|
||||
h.logger.ErrorContext(ctx, "error writing prometheus api response", errors.Attr(err))
|
||||
}
|
||||
}
|
||||
|
||||
// respondError follows Prometheus' status-code mapping: bad_data 400,
|
||||
// execution 422, canceled/timeout 503, internal 500.
|
||||
func (h *handler) respondError(ctx context.Context, w http.ResponseWriter, typ errorType, err error) {
|
||||
code := http.StatusInternalServerError
|
||||
switch typ {
|
||||
case errBadData:
|
||||
code = http.StatusBadRequest
|
||||
case errExec:
|
||||
code = http.StatusUnprocessableEntity
|
||||
case errCanceled, errTimeout:
|
||||
code = http.StatusServiceUnavailable
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(code)
|
||||
if encErr := json.NewEncoder(w).Encode(&response{Status: "error", ErrorType: typ, Error: err.Error()}); encErr != nil {
|
||||
h.logger.ErrorContext(ctx, "error writing prometheus api error response", errors.Attr(encErr))
|
||||
}
|
||||
}
|
||||
|
||||
// parseTime accepts Prometheus' time formats: float unix seconds or RFC3339.
|
||||
func parseTime(s string) (time.Time, error) {
|
||||
if t, err := strconv.ParseFloat(s, 64); err == nil {
|
||||
sec, ns := math.Modf(t)
|
||||
return time.Unix(int64(sec), int64(ns*float64(time.Second))), nil
|
||||
}
|
||||
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
|
||||
return t, nil
|
||||
}
|
||||
return time.Time{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "cannot parse %q to a valid timestamp", s)
|
||||
}
|
||||
|
||||
// parseDuration accepts Prometheus' duration formats: float seconds or a
|
||||
// duration string like 5m.
|
||||
func parseDuration(s string) (time.Duration, error) {
|
||||
if d, err := strconv.ParseFloat(s, 64); err == nil {
|
||||
ts := d * float64(time.Second)
|
||||
if ts > float64(math.MaxInt64) || ts < float64(math.MinInt64) {
|
||||
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "cannot parse %q to a valid duration. It overflows int64", s)
|
||||
}
|
||||
return time.Duration(ts), nil
|
||||
}
|
||||
if d, err := promModel.ParseDuration(s); err == nil {
|
||||
return time.Duration(d), nil
|
||||
}
|
||||
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "cannot parse %q to a valid duration", s)
|
||||
}
|
||||
@@ -484,6 +484,9 @@ func (aH *APIHandler) Respond(w http.ResponseWriter, data interface{}) {
|
||||
func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
|
||||
router.HandleFunc("/api/v1/query_range", am.ViewAccess(aH.queryRangeMetrics)).Methods(http.MethodGet)
|
||||
router.HandleFunc("/api/v1/query", am.ViewAccess(aH.queryMetrics)).Methods(http.MethodGet)
|
||||
|
||||
router.HandleFunc("/prometheus/api/v1/query_range", am.ViewAccess(aH.Signoz.Handlers.PrometheusHandler.QueryRange)).Methods(http.MethodGet, http.MethodPost)
|
||||
router.HandleFunc("/prometheus/api/v1/query", am.ViewAccess(aH.Signoz.Handlers.PrometheusHandler.Query)).Methods(http.MethodGet, http.MethodPost)
|
||||
router.HandleFunc("/api/v1/rules", am.ViewAccess(aH.listRules)).Methods(http.MethodGet)
|
||||
router.HandleFunc("/api/v1/rules/{id}", am.ViewAccess(aH.getRule)).Methods(http.MethodGet)
|
||||
router.HandleFunc("/api/v1/rules", am.EditAccess(aH.createRule)).Methods(http.MethodPost)
|
||||
|
||||
@@ -25,10 +25,6 @@ func TestErrIfStatementIsNotValid_Pass(t *testing.T) {
|
||||
{"GlobalNotIn", "SELECT a FROM t WHERE a GLOBAL NOT IN (SELECT b FROM t2)"},
|
||||
{"Union", "SELECT * FROM t UNION ALL SELECT * FROM t2"},
|
||||
{"Intersect", "SELECT * FROM t INTERSECT SELECT * FROM t2"},
|
||||
// A parenthesised left operand of a set operator. https://github.com/AfterShip/clickhouse-sql-parser/pull/312
|
||||
{"ParenthesisedUnionLeftOperand", "SELECT a FROM ((SELECT 1 AS a) UNION ALL (SELECT 2 AS a))"},
|
||||
{"ParenthesisedExceptLeftOperand", "SELECT a FROM ((SELECT 1 AS a) EXCEPT (SELECT 2 AS a))"},
|
||||
{"ParenthesisedUnionLeftOperandAtStatementLevel", "(SELECT 1 AS a) UNION ALL (SELECT 2 AS a)"},
|
||||
{"WindowFunction", "SELECT sum(v) OVER (PARTITION BY a ORDER BY t) FROM t"},
|
||||
{"UnrelatedSetting", "SELECT * FROM t SETTINGS max_threads = 4"},
|
||||
{"TerminatedBlockComment", "SELECT /* keep me */ count() FROM t"},
|
||||
@@ -187,10 +183,12 @@ func TestErrIfStatementIsNotValid_ShouldPassButFails(t *testing.T) {
|
||||
query string
|
||||
expectedCode errors.Code
|
||||
}{
|
||||
// The left operand commits the parser to a subquery, leaving the operator nowhere to bind. Parenthesising only the right operand is fine.
|
||||
{"ParenthesisedUnionLeftOperand", "SELECT a FROM ((SELECT 1 AS a) UNION ALL (SELECT 2 AS a))", CodeClickHouseSQLUnparseable},
|
||||
{"ParenthesisedExceptLeftOperand", "SELECT a FROM ((SELECT 1 AS a) EXCEPT (SELECT 2 AS a))", CodeClickHouseSQLUnparseable},
|
||||
{"ParenthesisedUnionLeftOperandAtStatementLevel", "(SELECT 1 AS a) UNION ALL (SELECT 2 AS a)", CodeClickHouseSQLUnparseable},
|
||||
// The one keyword PR 305 left behind, because ON also opens a join condition.
|
||||
{"UnquotedOnAsColumnName", "SELECT on + 1 FROM t", CodeClickHouseSQLUnparseable},
|
||||
// ClickHouse accepts NULLS FIRST|LAST as an ORDER BY modifier; the parser's grammar has no rule for it.
|
||||
{"OrderByNullsLast", "SELECT x FROM t ORDER BY x DESC NULLS LAST", CodeClickHouseSQLUnparseable},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
|
||||
@@ -48,6 +48,8 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/modules/tracedetail/impltracedetail"
|
||||
"github.com/SigNoz/signoz/pkg/modules/tracefunnel"
|
||||
"github.com/SigNoz/signoz/pkg/modules/tracefunnel/impltracefunnel"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/promapi"
|
||||
"github.com/SigNoz/signoz/pkg/querier"
|
||||
"github.com/SigNoz/signoz/pkg/ruler"
|
||||
"github.com/SigNoz/signoz/pkg/ruler/signozruler"
|
||||
@@ -81,6 +83,7 @@ type Handlers struct {
|
||||
RuleStateHistory rulestatehistory.Handler
|
||||
SpanMapperHandler spanmapper.Handler
|
||||
AlertmanagerHandler alertmanager.Handler
|
||||
PrometheusHandler prometheus.Handler
|
||||
TraceDetail tracedetail.Handler
|
||||
RulerHandler ruler.Handler
|
||||
LLMPricingRuleHandler llmpricingrule.Handler
|
||||
@@ -101,6 +104,7 @@ func NewHandlers(
|
||||
zeusService zeus.Zeus,
|
||||
registryHandler factory.Handler,
|
||||
alertmanagerService alertmanager.Alertmanager,
|
||||
prometheusService prometheus.Prometheus,
|
||||
rulerService ruler.Ruler,
|
||||
statsAggregator statsreporter.Aggregator,
|
||||
) Handlers {
|
||||
@@ -129,6 +133,7 @@ func NewHandlers(
|
||||
CloudIntegrationHandler: implcloudintegration.NewHandler(modules.CloudIntegration),
|
||||
SpanMapperHandler: implspanmapper.NewHandler(modules.SpanMapper),
|
||||
AlertmanagerHandler: signozalertmanager.NewHandler(alertmanagerService),
|
||||
PrometheusHandler: promapi.NewHandler(providerSettings.Logger, prometheusService),
|
||||
TraceDetail: impltracedetail.NewHandler(modules.TraceDetail),
|
||||
RulerHandler: signozruler.NewHandler(rulerService),
|
||||
LLMPricingRuleHandler: impllmpricingrule.NewHandler(modules.LLMPricingRule),
|
||||
|
||||
@@ -63,7 +63,7 @@ func TestNewHandlers(t *testing.T) {
|
||||
|
||||
querierHandler := querier.NewHandler(providerSettings, nil, nil)
|
||||
registryHandler := factory.NewHandler(nil)
|
||||
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil, registryHandler, alertmanager, nil, nil)
|
||||
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil, registryHandler, alertmanager, nil, nil, nil)
|
||||
reflectVal := reflect.ValueOf(handlers)
|
||||
for i := 0; i < reflectVal.NumField(); i++ {
|
||||
f := reflectVal.Field(i)
|
||||
|
||||
@@ -617,7 +617,7 @@ func New(
|
||||
|
||||
// Initialize all handlers for the modules
|
||||
registryHandler := factory.NewHandler(registry)
|
||||
handlers := NewHandlers(modules, providerSettings, analytics, querierHandler, licensing, global, flagger, gateway, telemetryMetadataStore, authz, zeus, registryHandler, alertmanager, rulerInstance, statsAggregator)
|
||||
handlers := NewHandlers(modules, providerSettings, analytics, querierHandler, licensing, global, flagger, gateway, telemetryMetadataStore, authz, zeus, registryHandler, alertmanager, prometheus, rulerInstance, statsAggregator)
|
||||
|
||||
// Initialize the API server (after registry so it can access service health)
|
||||
apiserverInstance, err := factory.NewProviderFromNamedMap(
|
||||
|
||||
66
tests/fixtures/promqltestcorpus.py
vendored
Normal file
66
tests/fixtures/promqltestcorpus.py
vendored
Normal file
@@ -0,0 +1,66 @@
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from fixtures.metrics import Metrics
|
||||
|
||||
TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "integration", "testdata", "promqltestcorpus")
|
||||
CORPUS_FILE = os.path.join(TESTDATA_DIR, "corpus.json")
|
||||
|
||||
# Datasets sit on disjoint time windows (2h gaps, far beyond the 5m lookback)
|
||||
# so one bulk ingest serves every case without cross-talk.
|
||||
ISOLATION_GAP_MS = 2 * 3600 * 1000
|
||||
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "-Inf": -math.inf}
|
||||
|
||||
|
||||
def ingest_promqltest_corpus(insert_metrics: Callable[[list[Metrics]], None]) -> tuple[dict, dict[int, int]]:
|
||||
"""Loads the frozen corpus, lays its datasets end to end on the timeline
|
||||
(newest last, ending safely in the past), ingests every sample, and
|
||||
returns (corpus, dataset base timestamps).
|
||||
|
||||
Dataset bases are hour-aligned: registration rows are hour-bucketed, so
|
||||
behavior depends on where samples fall relative to hour boundaries, and
|
||||
exact known-divergences enforcement needs identical placement every run."""
|
||||
with open(CORPUS_FILE, encoding="utf-8") as f:
|
||||
corpus = json.load(f)
|
||||
|
||||
cases_by_dataset: dict[int, list[dict]] = {}
|
||||
for case in corpus["cases"]:
|
||||
cases_by_dataset.setdefault(case["dataset"], []).append(case)
|
||||
|
||||
spans = {}
|
||||
for ds in corpus["datasets"]:
|
||||
sample_max = max((s["samples"][-1][0] for s in ds["series"] if s["samples"]), default=0)
|
||||
case_max = max((c["end_ms"] for c in cases_by_dataset.get(ds["id"], [])), default=0)
|
||||
spans[ds["id"]] = max(sample_max, case_max) + corpus["meta"]["lookback_ms"]
|
||||
|
||||
hour_ms = 3_600_000
|
||||
advances = {ds["id"]: -(-(spans[ds["id"]] + ISOLATION_GAP_MS) // hour_ms) * hour_ms for ds in corpus["datasets"]}
|
||||
total = sum(advances.values())
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
cursor = (int((now - timedelta(hours=1)).timestamp() * 1000) - total) // hour_ms * hour_ms
|
||||
|
||||
bases: dict[int, int] = {}
|
||||
metrics: list[Metrics] = []
|
||||
for ds in corpus["datasets"]:
|
||||
bases[ds["id"]] = cursor
|
||||
for series in ds["series"]:
|
||||
labels = dict(series["labels"])
|
||||
metric_name = labels.pop("__name__")
|
||||
for off_ms, raw in series["samples"]:
|
||||
stale = raw == "stale"
|
||||
metrics.append(
|
||||
Metrics(
|
||||
metric_name=metric_name,
|
||||
labels=labels,
|
||||
timestamp=datetime.fromtimestamp((cursor + off_ms) / 1000, tz=UTC),
|
||||
value=0.0 if stale else (SPECIALS[raw] if isinstance(raw, str) else float(raw)),
|
||||
flags=1 if stale else 0,
|
||||
)
|
||||
)
|
||||
cursor += advances[ds["id"]]
|
||||
|
||||
insert_metrics(metrics)
|
||||
return corpus, bases
|
||||
@@ -0,0 +1,138 @@
|
||||
import json
|
||||
import math
|
||||
from collections.abc import Callable
|
||||
from http import HTTPStatus
|
||||
|
||||
import requests
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.metrics import Metrics
|
||||
from fixtures.promqltestcorpus import ingest_promqltest_corpus
|
||||
|
||||
# The same frozen corpus the promqlconformance package replays through
|
||||
# /api/v5/query_range, here replayed against the /prometheus/api/v1 endpoints
|
||||
# with clickhousev2 as the serving provider (see conftest.py) — the two paths
|
||||
# nothing else exercises. Range cases go to query_range, where a
|
||||
# RangeExecutor provider serves transpiled statements when the shape allows.
|
||||
# Instant cases go to /query with a real `time` parameter, so they need no
|
||||
# grid encoding.
|
||||
#
|
||||
# Prometheus API sample values are strings, "NaN"/"+Inf"/"-Inf" included.
|
||||
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "+Inf": math.inf, "-Inf": -math.inf}
|
||||
QUERY_TIMEOUT = 30
|
||||
|
||||
|
||||
def test_prometheus_api_corpus(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
corpus, bases = ingest_promqltest_corpus(insert_metrics)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
failures: list[str] = []
|
||||
for case in corpus["cases"]:
|
||||
# instant-coarse variants encode an instant eval as a coarse-step
|
||||
# range because the v5 API cannot run true instants. This API can:
|
||||
# the [base] form of the same eval goes through /query below, and the
|
||||
# transpiled coarse-step serving the encoding exercises is covered
|
||||
# (and its known divergences ledgered) by promqlconformance's
|
||||
# clickhousev2 leg.
|
||||
if case["variant"] == "instant-coarse":
|
||||
continue
|
||||
|
||||
base = bases[case["dataset"]]
|
||||
start_ms = base + case["start_ms"]
|
||||
end_ms = base + case["end_ms"]
|
||||
step_s = max(1, case["step_ms"] // 1000)
|
||||
case_id = f"{case['source']}[{case['variant']}]"
|
||||
|
||||
if case["instant"]:
|
||||
path, params = "/prometheus/api/v1/query", {"query": case["expr"], "time": end_ms / 1000}
|
||||
else:
|
||||
path, params = (
|
||||
"/prometheus/api/v1/query_range",
|
||||
{
|
||||
"query": case["expr"],
|
||||
"start": start_ms / 1000,
|
||||
"end": end_ms / 1000,
|
||||
"step": step_s,
|
||||
},
|
||||
)
|
||||
response = requests.get(
|
||||
signoz.self.host_configs["8080"].get(path),
|
||||
params=params,
|
||||
timeout=QUERY_TIMEOUT,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
)
|
||||
if response.status_code != HTTPStatus.OK:
|
||||
failures.append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
|
||||
continue
|
||||
body = response.json()
|
||||
if body.get("status") != "success":
|
||||
failures.append(f"{case_id}: status {body.get('status')!r} for {case['expr']!r}: {json.dumps(body)[:200]}")
|
||||
continue
|
||||
|
||||
result_type, result = body["data"]["resultType"], body["data"]["result"]
|
||||
actual: dict[tuple, dict[int, float]] = {}
|
||||
if result_type == "matrix":
|
||||
for series in result:
|
||||
points = {round(float(ts) * 1000): SPECIALS[v] if v in SPECIALS else float(v) for ts, v in series.get("values") or []}
|
||||
actual[tuple(sorted((series.get("metric") or {}).items()))] = points
|
||||
elif result_type == "vector":
|
||||
for series in result:
|
||||
ts, v = series["value"]
|
||||
actual[tuple(sorted((series.get("metric") or {}).items()))] = {round(float(ts) * 1000): SPECIALS[v] if v in SPECIALS else float(v)}
|
||||
elif result_type == "scalar":
|
||||
ts, v = result
|
||||
actual[()] = {round(float(ts) * 1000): SPECIALS[v] if v in SPECIALS else float(v)}
|
||||
|
||||
expected: dict[tuple, dict[int, float]] = {}
|
||||
for res in case["expected"]:
|
||||
points = {base + off_ms: SPECIALS[v] if isinstance(v, str) else float(v) for off_ms, v in res["points"]}
|
||||
expected[tuple(sorted(res["labels"].items()))] = points
|
||||
|
||||
if set(actual) != set(expected):
|
||||
missing = set(expected) - set(actual)
|
||||
extra = set(actual) - set(expected)
|
||||
failures.append(f"{case_id}: series mismatch for {case['expr']!r} (missing={sorted(missing)[:3]} extra={sorted(extra)[:3]})")
|
||||
continue
|
||||
|
||||
mismatch = None
|
||||
for lset, exp_points in expected.items():
|
||||
act_points = actual[lset]
|
||||
if set(act_points) != set(exp_points):
|
||||
mismatch = f"{case_id}: timestamp mismatch for {case['expr']!r} series {dict(lset)} (expected {len(exp_points)} points, got {len(act_points)})"
|
||||
break
|
||||
for ts, exp_v in exp_points.items():
|
||||
act_v = act_points[ts]
|
||||
if math.isnan(act_v) or math.isnan(exp_v):
|
||||
close = math.isnan(act_v) and math.isnan(exp_v)
|
||||
elif math.isinf(act_v) or math.isinf(exp_v):
|
||||
close = act_v == exp_v
|
||||
elif act_v == exp_v:
|
||||
close = True
|
||||
else:
|
||||
# Expected values carry the v5 API's rounding (>=1: three
|
||||
# decimal places; <1: three significant digits); this API
|
||||
# returns raw floats. One rounding quantum covers the
|
||||
# largest possible rounding difference.
|
||||
scale = max(abs(act_v), abs(exp_v))
|
||||
if scale >= 1:
|
||||
quantum = max(1e-3, scale * 1e-9)
|
||||
else:
|
||||
quantum = 10 ** (math.floor(math.log10(scale)) - 2)
|
||||
close = abs(act_v - exp_v) <= quantum + 1e-12
|
||||
if not close:
|
||||
mismatch = f"{case_id}: value mismatch for {case['expr']!r} series {dict(lset)} at {ts}: expected {exp_v}, got {act_v}"
|
||||
break
|
||||
if mismatch:
|
||||
break
|
||||
if mismatch:
|
||||
failures.append(mismatch)
|
||||
|
||||
for f_line in failures:
|
||||
print("DIVERGED", f_line)
|
||||
assert not failures, f"{len(failures)} corpus cases diverged:\n" + "\n".join(failures[:25])
|
||||
37
tests/integration/tests/promapiconformance/conftest.py
Normal file
37
tests/integration/tests/promapiconformance/conftest.py
Normal file
@@ -0,0 +1,37 @@
|
||||
import pytest
|
||||
from testcontainers.core.container import Network
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.signoz import create_signoz
|
||||
|
||||
|
||||
@pytest.fixture(name="signoz", scope="package")
|
||||
def signoz_promapi_v2(
|
||||
network: Network,
|
||||
migrator: types.Operation, # pylint: disable=unused-argument
|
||||
zeus: types.TestContainerDocker,
|
||||
gateway: types.TestContainerDocker,
|
||||
sqlstore: types.TestContainerSQL,
|
||||
clickhouse: types.TestContainerClickhouse,
|
||||
request: pytest.FixtureRequest,
|
||||
pytestconfig: pytest.Config,
|
||||
) -> types.SigNoz:
|
||||
"""
|
||||
SigNoz with clickhousev2 as the serving prometheus provider. The corpus
|
||||
replays against the /prometheus/api/v1 endpoints, so this package covers
|
||||
the two paths nothing else serves: v2 as the provider (range queries
|
||||
transpile when the shape allows), and the Prometheus HTTP API contract.
|
||||
"""
|
||||
return create_signoz(
|
||||
network=network,
|
||||
zeus=zeus,
|
||||
gateway=gateway,
|
||||
sqlstore=sqlstore,
|
||||
clickhouse=clickhouse,
|
||||
request=request,
|
||||
pytestconfig=pytestconfig,
|
||||
cache_key="signoz-promapi-v2",
|
||||
env_overrides={
|
||||
"SIGNOZ_PROMETHEUS_PROVIDER": "clickhousev2",
|
||||
},
|
||||
)
|
||||
@@ -2,21 +2,21 @@ import json
|
||||
import math
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.metrics import Metrics
|
||||
from fixtures.promqltestcorpus import ingest_promqltest_corpus
|
||||
from fixtures.querier import get_all_series, make_query_request
|
||||
|
||||
TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "testdata")
|
||||
# Frozen corpus extracted from Prometheus' own promql/promqltest testdata by
|
||||
# scripts/promqltestcorpus (upstream load scripts + the vendored reference engine).
|
||||
# Unlike live-vs-live parity suites, the oracle is this committed file, so the suite
|
||||
# keeps working when the serving path itself is the thing being changed — the one
|
||||
# situation where comparing two live paths against each other is blind.
|
||||
CORPUS_FILE = os.path.join(TESTDATA_DIR, "promqltestcorpus", "corpus.json")
|
||||
# The corpus (see fixtures/promqltestcorpus.py) is frozen from Prometheus' own
|
||||
# promql/promqltest testdata by scripts/promqltestcorpus (upstream load scripts
|
||||
# + the vendored reference engine). Unlike live-vs-live parity suites, the
|
||||
# oracle is a committed file, so the suite keeps working when the serving path
|
||||
# itself is the thing being changed — the one situation where comparing two
|
||||
# live paths against each other is blind.
|
||||
|
||||
# One ledger per leg, enforced exactly in both directions. The default leg's
|
||||
# ledger is empty and pinned there; the clickhousev2 ledger is the rollout
|
||||
@@ -40,9 +40,6 @@ LEGS: list[tuple[str, dict | None]] = [
|
||||
("clickhousev2", {"X-SigNoz-PromQL-Provider": "clickhousev2"}),
|
||||
]
|
||||
|
||||
# Datasets sit on disjoint time windows (2h gaps, far beyond the 5m lookback) so
|
||||
# one bulk ingest serves every case without cross-talk.
|
||||
ISOLATION_GAP_MS = 2 * 3600 * 1000
|
||||
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "-Inf": -math.inf}
|
||||
|
||||
|
||||
@@ -52,51 +49,7 @@ def test_upstream_promqltest_corpus(
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
with open(CORPUS_FILE, encoding="utf-8") as f:
|
||||
corpus = json.load(f)
|
||||
|
||||
cases_by_dataset: dict[int, list[dict]] = {}
|
||||
for case in corpus["cases"]:
|
||||
cases_by_dataset.setdefault(case["dataset"], []).append(case)
|
||||
|
||||
# Lay datasets end to end on the timeline, newest last, ending safely in
|
||||
# the past; spans are per-dataset so the whole corpus stays within days.
|
||||
spans = {}
|
||||
for ds in corpus["datasets"]:
|
||||
sample_max = max((s["samples"][-1][0] for s in ds["series"] if s["samples"]), default=0)
|
||||
case_max = max((c["end_ms"] for c in cases_by_dataset.get(ds["id"], [])), default=0)
|
||||
spans[ds["id"]] = max(sample_max, case_max) + corpus["meta"]["lookback_ms"]
|
||||
|
||||
# Hour-aligned dataset bases: registration rows are hour-bucketed, so
|
||||
# behavior depends on where samples fall relative to hour boundaries —
|
||||
# the exact known-divergences enforcement needs that identical every run.
|
||||
hour_ms = 3_600_000
|
||||
advances = {ds["id"]: -(-(spans[ds["id"]] + ISOLATION_GAP_MS) // hour_ms) * hour_ms for ds in corpus["datasets"]}
|
||||
total = sum(advances.values())
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
cursor = (int((now - timedelta(hours=1)).timestamp() * 1000) - total) // hour_ms * hour_ms
|
||||
|
||||
bases: dict[int, int] = {}
|
||||
metrics: list[Metrics] = []
|
||||
for ds in corpus["datasets"]:
|
||||
bases[ds["id"]] = cursor
|
||||
for series in ds["series"]:
|
||||
labels = dict(series["labels"])
|
||||
metric_name = labels.pop("__name__")
|
||||
for off_ms, raw in series["samples"]:
|
||||
stale = raw == "stale"
|
||||
metrics.append(
|
||||
Metrics(
|
||||
metric_name=metric_name,
|
||||
labels=labels,
|
||||
timestamp=datetime.fromtimestamp((cursor + off_ms) / 1000, tz=UTC),
|
||||
value=0.0 if stale else (SPECIALS[raw] if isinstance(raw, str) else float(raw)),
|
||||
flags=1 if stale else 0,
|
||||
)
|
||||
)
|
||||
cursor += advances[ds["id"]]
|
||||
|
||||
insert_metrics(metrics)
|
||||
corpus, bases = ingest_promqltest_corpus(insert_metrics)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
failures: dict[str, list[str]] = {leg: [] for leg, _ in LEGS}
|
||||
|
||||
Reference in New Issue
Block a user