feat(prometheus): add the Prometheus query API under a /prometheus prefix

New GET|POST /prometheus/api/v1/query_range and /prometheus/api/v1/query
endpoints (pkg/prometheus/promapi), following the Prometheus HTTP API:
float-unix or RFC3339 times, float-seconds or duration-string durations,
the {status, data, errorType, error, warnings, infos} envelope with
Prometheus' status codes, and the 11,000-point cap. Wired through
signoz.Handlers like the other domain handlers. Range queries serve
through the RangeExecutor capability when the provider has it, so a
clickhousev2-serving deployment transpiles through these endpoints too.

The promapiconformance suite replays the frozen promqltest corpus against
these endpoints with clickhousev2 as the serving provider - the two paths
nothing else exercises. Instant cases go through /query with a real time
parameter. Its ledger holds the seven known Kahan-class divergences of
transpiled coarse-step serving.

Purely additive: the existing GET /api/v1/query_range and /api/v1/query
handlers are untouched.

Assisted-by: Claude Fable 5

.github/workflows/integrationci.yaml

@@ -58,6 +58,7 @@ jobs:
           - querierai
           - rawexportdata
           - promqlconformance
+          - promapiconformance
           - querierauthz
           - role
           - rootuser

pkg/prometheus/handler.go

@@ -0,0 +1,9 @@
+package prometheus
+
+import "net/http"
+
+type Handler interface {
+	Query(http.ResponseWriter, *http.Request)
+
+	QueryRange(http.ResponseWriter, *http.Request)
+}

pkg/prometheus/promapi/handler.go

@@ -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)
+}

pkg/query-service/app/http_handler.go

@@ -485,6 +485,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)

pkg/signoz/handler.go

@@ -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),

pkg/signoz/handler_test.go

@@ -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)

pkg/signoz/signoz.go

@@ -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(

tests/fixtures/promapi.py

@@ -0,0 +1,44 @@
+"""Client helpers for the /prometheus/api/v1 endpoints."""
+
+import math
+
+import requests
+
+from fixtures import types
+
+SPECIALS = {"NaN": math.nan, "Inf": math.inf, "+Inf": math.inf, "-Inf": -math.inf}
+QUERY_TIMEOUT = 30
+
+
+def prom_api_get(signoz: types.SigNoz, token: str, path: str, params: dict) -> requests.Response:
+    return requests.get(
+        signoz.self.host_configs["8080"].get(path),
+        params=params,
+        timeout=QUERY_TIMEOUT,
+        headers={"authorization": f"Bearer {token}"},
+    )
+
+
+def prom_api_value(v: str) -> float:
+    """Prometheus API sample values are strings, including "NaN" and "+Inf"."""
+    if v in SPECIALS:
+        return SPECIALS[v]
+    return float(v)
+
+
+def series_from_prom_result(result_type: str, result) -> dict[tuple, dict[int, float]]:
+    """Flattens a matrix/vector/scalar result into
+    {sorted-labels tuple: {unix_ms: value}}."""
+    out: dict[tuple, dict[int, float]] = {}
+    if result_type == "matrix":
+        for series in result:
+            points = {round(float(ts) * 1000): prom_api_value(v) for ts, v in series.get("values") or []}
+            out[tuple(sorted((series.get("metric") or {}).items()))] = points
+    elif result_type == "vector":
+        for series in result:
+            ts, v = series["value"]
+            out[tuple(sorted((series.get("metric") or {}).items()))] = {round(float(ts) * 1000): prom_api_value(v)}
+    elif result_type == "scalar":
+        ts, v = result
+        out[()] = {round(float(ts) * 1000): prom_api_value(v)}
+    return out

tests/fixtures/promqltestcorpus.py

@@ -0,0 +1,112 @@
+"""Shared helpers for suites that replay the frozen promqltest corpus
+(tests/integration/testdata/promqltestcorpus/corpus.json)."""
+
+import json
+import math
+import os
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+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")
+
+ISOLATION_GAP_MS = 2 * 3600 * 1000
+SPECIALS = {"NaN": math.nan, "Inf": math.inf, "+Inf": math.inf, "-Inf": -math.inf}
+
+
+def decode_corpus_value(v: float | str) -> float:
+    if isinstance(v, str):
+        return SPECIALS[v]
+    return float(v)
+
+
+def values_close(a: float, b: float) -> bool:
+    """Expected corpus values carry the v5 API's rounding (>=1: three decimal
+    places; <1: three significant digits). One rounding quantum covers both a
+    raw-vs-rounded comparison and a boundary that rounds either way."""
+    if math.isnan(a) or math.isnan(b):
+        return math.isnan(a) and math.isnan(b)
+    if math.isinf(a) or math.isinf(b):
+        return a == b
+    if a == b:
+        return True
+    scale = max(abs(a), abs(b))
+    if scale >= 1:
+        quantum = max(1e-3, scale * 1e-9)
+    else:
+        quantum = 10 ** (math.floor(math.log10(scale)) - 2)
+    return abs(a - b) <= quantum + 1e-12
+
+
+def labelset(labels: dict[str, str]) -> tuple:
+    return tuple(sorted(labels.items()))
+
+
+def ledger(filename: str) -> dict[str, str]:
+    path = os.path.join(TESTDATA_DIR, filename)
+    if not os.path.exists(path):
+        return {}
+    with open(path, encoding="utf-8") as f:
+        return json.load(f)["divergences"]
+
+
+@pytest.fixture(name="ingest_promqltest_corpus")
+def ingest_promqltest_corpus(insert_metrics: Callable[[list[Metrics]], None]) -> Callable[[], tuple[dict, dict[int, int]]]:
+    """Yields a callable that loads the corpus, lays its datasets end to end
+    on the timeline, 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.
+    Datasets sit on disjoint windows (2h gaps, far beyond the 5m lookback) so
+    one bulk ingest serves every case without cross-talk."""
+
+    def ingest() -> tuple[dict, dict[int, int]]:
+        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 decode_corpus_value(raw),
+                            flags=1 if stale else 0,
+                        )
+                    )
+            cursor += advances[ds["id"]]
+
+        insert_metrics(metrics)
+        return corpus, bases
+
+    return ingest

tests/integration/testdata/promqltestcorpus/known_divergences_promapi.json

@@ -0,0 +1,12 @@
+{
+  "note": "Divergences of the /prometheus/api/v1 endpoints, served by the clickhousev2 provider, from the upstream reference engine. 01_prometheus_api_corpus.py enforces this set exactly in both directions. All current entries are the Kahan class recorded in known_divergences_v2.json: the engine sums with Kahan compensation and an overflow-free incremental mean, ClickHouse's aggregates are naive. Only the [instant-coarse] variants appear here: they are range-encoded, so they serve transpiled; the [base] instant evals go through /prometheus/api/v1/query on the exact engine path.",
+  "divergences": {
+    "aggregators.test:651[instant-coarse]": "avg over near-max-float64 values: avgForEach overflows to +Inf where the engine's incremental mean does not",
+    "aggregators.test:654[instant-coarse]": "avg over near-min-float64 values: avgForEach overflows to -Inf",
+    "aggregators.test:687[instant-coarse]": "sum over {1e100, -1e100, small}: naive summation cancels to 0 where the engine's Kahan sum keeps 10",
+    "aggregators.test:695[instant-coarse]": "avg over {1e100, -1e100, small}: same cancellation divided by count",
+    "functions.test:1084[instant-coarse]": "sum_over_time over a ±1e100 window: the disjoint coarse-step form's arraySum cancels to 0",
+    "functions.test:1087[instant-coarse]": "avg_over_time, same window and cancellation as functions.test:1084",
+    "functions.test:1149[instant-coarse]": "avg_over_time over ±2.258e220 samples: naive slide summation leaves a ~1e202 residue where the engine cancels to 0"
+  }
+}

tests/integration/tests/promapiconformance/01_prometheus_api_corpus.py

@@ -0,0 +1,90 @@
+"""
+Replay the frozen promqltest corpus against the /prometheus/api/v1
+endpoints, with clickhousev2 as the serving provider (see conftest.py).
+
+The oracle is the same committed corpus the promqlconformance package
+replays through /api/v5/query_range. This package differs in two ways.
+Range cases go to /prometheus/api/v1/query_range, where a RangeExecutor
+provider serves transpiled statements when the shape allows. Instant cases
+go to /prometheus/api/v1/query with a real `time` parameter, so they need
+no grid encoding.
+"""
+
+import json
+from collections.abc import Callable
+from http import HTTPStatus
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.promapi import prom_api_get, series_from_prom_result
+from fixtures.promqltestcorpus import decode_corpus_value, labelset, ledger, values_close
+
+
+def test_prometheus_api_corpus(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    ingest_promqltest_corpus: Callable[[], tuple[dict, dict[int, int]]],
+) -> None:
+    corpus, bases = ingest_promqltest_corpus()
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    failures: list[str] = []
+    for case in corpus["cases"]:
+        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"]:
+            response = prom_api_get(signoz, token, "/prometheus/api/v1/query", {"query": case["expr"], "time": end_ms / 1000})
+        else:
+            response = prom_api_get(
+                signoz,
+                token,
+                "/prometheus/api/v1/query_range",
+                {"query": case["expr"], "start": start_ms / 1000, "end": end_ms / 1000, "step": step_s},
+            )
+        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
+
+        actual = series_from_prom_result(body["data"]["resultType"], body["data"]["result"])
+        expected = {
+            labelset(res["labels"]): {base + off_ms: decode_corpus_value(v) for off_ms, v in res["points"]}
+            for res in case["expected"]
+        }
+
+        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
+        for lset, exp_points in expected.items():
+            act_points = actual[lset]
+            if set(act_points) != set(exp_points):
+                failures.append(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():
+                if not values_close(act_points[ts], exp_v):
+                    failures.append(f"{case_id}: value mismatch for {case['expr']!r} series {dict(lset)} at {ts}: expected {exp_v}, got {act_points[ts]}")
+                    break
+            else:
+                continue
+            break
+
+    for f_line in failures:
+        print("DIVERGED", f_line)
+
+    known = ledger("known_divergences_promapi.json")
+    failed_ids = {f_line.split(": ", 1)[0] for f_line in failures}
+    unexpected = [f_line for f_line in failures if f_line.split(": ", 1)[0] not in known]
+    now_passing = sorted(set(known) - failed_ids)
+
+    assert not unexpected, f"{len(unexpected)} corpus cases diverged beyond the known set:\n" + "\n".join(unexpected[:25])
+    assert not now_passing, f"{len(now_passing)} known divergences now pass — remove them from known_divergences_promapi.json: {now_passing[:25]}"

tests/integration/tests/promapiconformance/conftest.py

@@ -0,0 +1,38 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.promqltestcorpus import ingest_promqltest_corpus  # noqa: F401  pylint: disable=unused-import
+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",
+        },
+    )