mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-11 14:00:42 +01:00
Compare commits
1 Commits
test/expli
...
promql-rem
| Author | SHA1 | Date | |
|---|---|---|---|
|
18bb0805e1 |
@@ -17,7 +17,7 @@ For example, the [prometheus](/pkg/prometheus) provider delivers a prometheus en
|
||||
|
||||
- `pkg/prometheus/prometheus.go` - Interface definition
|
||||
- `pkg/prometheus/config.go` - Configuration
|
||||
- `pkg/prometheus/clickhouseprometheus/provider.go` - Clickhouse-powered implementation
|
||||
- `pkg/prometheus/clickhouseprometheusv2/provider.go` - Clickhouse-powered implementation
|
||||
- `pkg/prometheus/prometheustest/provider.go` - Mock implementation
|
||||
|
||||
## How to wire it up?
|
||||
|
||||
@@ -9,15 +9,16 @@ change breaks an invariant, flag it and discuss it first.
|
||||
|
||||
---
|
||||
|
||||
## Why a second provider
|
||||
## Why the provider looks like this
|
||||
|
||||
The v1 provider (`pkg/prometheus/clickhouseprometheus`) serves the promql
|
||||
engine through the remote-read protobuf adapter. It fetches every raw sample
|
||||
of a query's union window. It serializes all of them and gives them to the
|
||||
engine. The cost follows the ingested data, not the question. This is how a
|
||||
dashboard of PromQL panels can take an instance down.
|
||||
The removed v1 provider served the promql engine through the remote-read
|
||||
protobuf adapter. It fetched every raw sample of a query's union window,
|
||||
serialized all of them, and gave them to the engine. The cost followed the
|
||||
ingested data, not the question. This is how a dashboard of PromQL panels
|
||||
could take an instance down. v2 replaced it after a byte-level parity
|
||||
rollout, and v1 was then deleted.
|
||||
|
||||
In v2, each query runs in one of two ways. The classifier decides per query:
|
||||
Each query runs in one of two ways. The classifier decides per query:
|
||||
|
||||
- **Transpiled**: ClickHouse evaluates the query. Only final (or near-final)
|
||||
per-group grid arrays come back. The statements use the
|
||||
@@ -30,7 +31,7 @@ In v2, each query runs in one of two ways. The classifier decides per query:
|
||||
lost user. A construct that cannot reproduce engine semantics exactly falls
|
||||
back. It does not approximate.** The conformance suite
|
||||
(`tests/integration/tests/promqlconformance/`) replays Prometheus' own test
|
||||
corpus against both providers. It is the arbiter. The classification golden
|
||||
corpus against the provider. It is the arbiter. The classification golden
|
||||
(`testdata/classification_golden.json`) freezes the route of each corpus
|
||||
expression. The rest of this document is the PromQL-to-SQL story. That
|
||||
mapping is where correctness is won or lost.
|
||||
@@ -263,7 +264,8 @@ per-thread partials scaled memory with the thread count. The slide then
|
||||
combines each slot's at-most-W bucket partials by direct aggregation
|
||||
(`arraySum(arraySlice(...))`). Window sums are added the way the engine adds
|
||||
them. There is no prefix-sum differencing: its large-minus-large
|
||||
cancellation would drift past the shadow tolerance on counter-sized values.
|
||||
cancellation would drift past the conformance tolerance on counter-sized
|
||||
values.
|
||||
This is correct per slot because the bucket union is the exact window
|
||||
multiset, and avg/min/max/sum/count are order-insensitive on a multiset
|
||||
(sum/avg up to summation order; see the float caveat above). A slot with
|
||||
@@ -333,7 +335,7 @@ can carry them.
|
||||
## The engine path
|
||||
|
||||
Queries that do not transpile run in the stock engine over this package's
|
||||
`storage.Querier`. This is still not the v1 path. Samples are fetched per
|
||||
`storage.Querier`. Samples are fetched per
|
||||
selector with the engine's per-selector hints, not the query-wide union
|
||||
window. So `foo / foo offset 1d` reads two narrow windows, not the widest
|
||||
one twice. Instant selectors of subquery-free queries fetch only the last
|
||||
@@ -367,9 +369,9 @@ same predicates as a shard-local semi-join, not a GLOBAL broadcast of the
|
||||
matched set. The temporality filter on every samples statement is a
|
||||
semantic no-op: the matched fingerprints already come from those
|
||||
temporalities. It engages the leading samples primary-key column.
|
||||
Delta-temporality series stay invisible to PromQL here, exactly as in v1.
|
||||
The rollout gate is parity with v1. To make Delta visible is its own change
|
||||
with its own semantics to design. A Delta stream fed to `rate()`
|
||||
Delta-temporality series stay invisible to PromQL here, as they were before
|
||||
v2. To make Delta visible is its own change with its own semantics to
|
||||
design. A Delta stream fed to `rate()`
|
||||
as-if-cumulative would be wrong, not just new.
|
||||
|
||||
## Observability
|
||||
|
||||
@@ -160,7 +160,7 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
triggeredTestAlerts := []map[*alertmanagertypes.PostableAlert][]string{}
|
||||
|
||||
// Variable to store promProvider for cleanup
|
||||
var promProvider *prometheustest.Provider
|
||||
var promProvider prometheus.Prometheus
|
||||
|
||||
// Create manager using test factory with hooks
|
||||
mgr := rules.NewTestManager(t, &rules.TestManagerOptions{
|
||||
@@ -185,76 +185,29 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
TelemetryStoreHook: func(store telemetrystore.TelemetryStore) {
|
||||
mockStore := store.(*telemetrystoretest.Provider)
|
||||
|
||||
// Set up Prometheus-specific mock data
|
||||
// Fingerprint columns for Prometheus queries
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
// Samples columns for Prometheus queries
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
// Calculate query time range similar to Prometheus rule tests
|
||||
// TestNotification uses time.Now().UTC() for evaluation
|
||||
// We calculate the query window based on current time to match what the actual evaluation will use
|
||||
// Grid the TestNotification eval computes over (see
|
||||
// Timestamps on base_rule); nil args match any window.
|
||||
evalTime := baseTime
|
||||
evalWindowMs := int64(5 * 60 * 1000) // 5 minutes in ms
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-2*evalWindowMs)/60000)*60000 + 1 // truncate to minute + 1ms
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
gridEnd := (evalTime.UnixMilli() / 60000) * 60000
|
||||
gridStart := gridEnd - evalWindowMs
|
||||
|
||||
// Create fingerprint data
|
||||
fingerprint := uint64(12345)
|
||||
labelsJSON := `{"__name__":"test_metric"}`
|
||||
fingerprintData := [][]interface{}{
|
||||
{fingerprint, labelsJSON},
|
||||
}
|
||||
fingerprintRows := cmock.NewRows(fingerprintCols, fingerprintData)
|
||||
|
||||
// Create samples data from test case values, calculating timestamps relative to baseTime
|
||||
validSamplesData := make([][]interface{}, 0)
|
||||
tsList := make([]int64, 0, len(tc.Values))
|
||||
vList := make([]float64, 0, len(tc.Values))
|
||||
for _, v := range tc.Values {
|
||||
// Skip NaN and Inf values in the samples data
|
||||
if math.IsNaN(v.Value) || math.IsInf(v.Value, 0) {
|
||||
continue
|
||||
}
|
||||
// Calculate timestamp relative to baseTime
|
||||
sampleTimestamp := baseTime.Add(v.Offset).UnixMilli()
|
||||
validSamplesData = append(validSamplesData, []interface{}{
|
||||
"test_metric",
|
||||
fingerprint,
|
||||
sampleTimestamp,
|
||||
v.Value,
|
||||
uint32(0), // flags - 0 means normal value
|
||||
})
|
||||
tsList = append(tsList, baseTime.Add(v.Offset).UnixMilli())
|
||||
vList = append(vList, v.Value)
|
||||
}
|
||||
samplesRows := cmock.NewRows(samplesCols, validSamplesData)
|
||||
grid := prometheustest.LastSampleGrid(tsList, vList, gridStart, gridEnd, 60_000, 300_000)
|
||||
|
||||
mock := mockStore.Mock()
|
||||
|
||||
// Mock the fingerprint query (for Prometheus label matching)
|
||||
// args: $1=metric_name (the __name__ matcher maps onto the column)
|
||||
mock.ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(fingerprintRows)
|
||||
|
||||
// Mock the samples query (for Prometheus metric data)
|
||||
// args: metric_name IN (discovered names), subquery metric_name, start, end
|
||||
mock.ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs(
|
||||
"test_metric",
|
||||
"test_metric",
|
||||
queryStart,
|
||||
queryEnd,
|
||||
).
|
||||
WillReturnRows(samplesRows)
|
||||
mock.ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{{`[["__name__","test_metric"]]`, grid}}))
|
||||
|
||||
// Create Prometheus provider for this test
|
||||
promProvider = prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, store)
|
||||
@@ -289,7 +242,6 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
assert.Empty(t, triggeredTestAlerts)
|
||||
}
|
||||
|
||||
promProvider.Close()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
15
go.mod
15
go.mod
@@ -131,7 +131,6 @@ require (
|
||||
github.com/go-playground/locales v0.14.1 // indirect
|
||||
github.com/go-playground/universal-translator v0.18.1 // indirect
|
||||
github.com/goccy/go-yaml v1.19.2 // indirect
|
||||
github.com/google/go-cmp v0.7.0 // indirect
|
||||
github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
|
||||
github.com/hashicorp/go-metrics v0.5.4 // indirect
|
||||
github.com/huandu/go-clone v1.7.3 // indirect
|
||||
@@ -139,8 +138,7 @@ require (
|
||||
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||
github.com/ncruces/go-strftime v1.0.0 // indirect
|
||||
github.com/nxadm/tail v1.4.11 // indirect
|
||||
github.com/prometheus/client_golang/exp v0.0.0-20260325093428-d8591d0db856 // indirect
|
||||
github.com/puzpuzpuz/xsync/v4 v4.4.0 // indirect
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.148.0 // indirect
|
||||
github.com/redis/go-redis/extra/rediscmd/v9 v9.15.1 // indirect
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
||||
github.com/swaggest/refl v1.4.0 // indirect
|
||||
@@ -151,7 +149,6 @@ require (
|
||||
github.com/x448/float16 v0.8.4 // indirect
|
||||
go.opentelemetry.io/collector/internal/componentalias v0.148.0 // indirect
|
||||
go.opentelemetry.io/collector/pdata/xpdata v0.148.0 // indirect
|
||||
go.uber.org/goleak v1.3.0 // indirect
|
||||
go.yaml.in/yaml/v2 v2.4.4 // indirect
|
||||
golang.org/x/arch v0.20.0 // indirect
|
||||
golang.org/x/term v0.43.0 // indirect
|
||||
@@ -171,10 +168,6 @@ require (
|
||||
cloud.google.com/go/auth v0.18.2 // indirect
|
||||
cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
|
||||
cloud.google.com/go/compute/metadata v0.9.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.13.1 // indirect
|
||||
github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 // indirect
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 // indirect
|
||||
github.com/ClickHouse/ch-go v0.71.0
|
||||
github.com/Masterminds/squirrel v1.5.4 // indirect
|
||||
github.com/Yiling-J/theine-go v0.6.2 // indirect
|
||||
@@ -220,7 +213,6 @@ require (
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/gojek/valkyrie v0.0.0-20180215180059-6aee720afcdf // indirect
|
||||
github.com/golang/protobuf v1.5.4 // indirect
|
||||
github.com/golang/snappy v1.0.0 // indirect
|
||||
github.com/google/btree v1.1.3 // indirect
|
||||
github.com/google/cel-go v0.29.0 // indirect
|
||||
github.com/google/s2a-go v0.1.9 // indirect
|
||||
@@ -253,7 +245,6 @@ require (
|
||||
github.com/jtolds/gls v4.20.0+incompatible // indirect
|
||||
github.com/klauspost/compress v1.18.5 // indirect
|
||||
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
|
||||
github.com/kylelemons/godebug v1.1.0 // indirect
|
||||
github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
|
||||
github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
|
||||
github.com/leodido/go-syslog/v4 v4.3.0 // indirect
|
||||
@@ -278,14 +269,10 @@ require (
|
||||
github.com/oklog/ulid/v2 v2.1.1
|
||||
github.com/open-feature/go-sdk v1.17.0
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.144.0 // indirect
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.148.0 // indirect
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.148.0 // indirect
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.148.0 // indirect
|
||||
github.com/openfga/openfga v1.14.1
|
||||
github.com/paulmach/orb v0.12.0 // indirect
|
||||
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
|
||||
github.com/pierrec/lz4/v4 v4.1.25 // indirect
|
||||
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
|
||||
github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
|
||||
github.com/pressly/goose/v3 v3.27.0 // indirect
|
||||
|
||||
83
go.sum
83
go.sum
@@ -73,16 +73,8 @@ github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0 h1:fou+2+WFTib47nS+nz/ozhEB
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0/go.mod h1:t76Ruy8AHvUAC8GfMWJMa0ElSbuIcO03NLpynfbgsPA=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.13.1 h1:Hk5QBxZQC1jb2Fwj6mpzme37xbCDdNTxU7O9eb5+LB4=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.13.1/go.mod h1:IYus9qsFobWIc2YVwe/WPjcnyCkPKtnHAqUYeebc8z0=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2 h1:yz1bePFlP5Vws5+8ez6T3HWXPmwOK7Yvq8QxDBD3SKY=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2/go.mod h1:Pa9ZNPuoNu/GztvBSKk9J1cDJW6vk/n0zLtV4mgd8N8=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 h1:9iefClla7iYpfYWdzPCRDozdmndjTm8DXdpCzPajMgA=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2/go.mod h1:XtLgD3ZD34DAaVIIAyG3objl5DynM3CQ/vMcbBNJZGI=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute/v5 v5.7.0 h1:LkHbJbgF3YyvC53aqYGR+wWQDn2Rdp9AQdGndf9QvY4=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute/v5 v5.7.0/go.mod h1:QyiQdW4f4/BIfB8ZutZ2s+28RAgfa/pT+zS++ZHyM1I=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork/v4 v4.3.0 h1:bXwSugBiSbgtz7rOtbfGf+woewp4f06orW9OP5BjHLA=
|
||||
github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork/v4 v4.3.0/go.mod h1:Y/HgrePTmGy9HjdSGTqZNa+apUpTVIEVKXJyARP2lrk=
|
||||
github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM=
|
||||
github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE=
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgvJqCH0sFfrBUTnUJSBrBf7++ypk+twtRs=
|
||||
github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0/go.mod h1:HKpQxkWaGLJ+D/5H8QRpyQXA1eKjxkFlOMwck5+33Jk=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
@@ -91,8 +83,6 @@ github.com/ClickHouse/ch-go v0.71.0 h1:bUdZ/EZj/LcVHsMqaRUP2holqygrPWQKeMjc6nZoy
|
||||
github.com/ClickHouse/ch-go v0.71.0/go.mod h1:NwbNc+7jaqfY58dmdDUbG4Jl22vThgx1cYjBw0vtgXw=
|
||||
github.com/ClickHouse/clickhouse-go/v2 v2.44.0 h1:9pxs5pRwIvhni5BDRPn/n5A8DeUod5TnBaeulFBX8EQ=
|
||||
github.com/ClickHouse/clickhouse-go/v2 v2.44.0/go.mod h1:giJfUVlMkcfUEPVfRpt51zZaGEx9i17gCos8gBl392c=
|
||||
github.com/Code-Hex/go-generics-cache v1.5.1 h1:6vhZGc5M7Y/YD8cIUcY8kcuQLB4cHR7U+0KMqAA0KcU=
|
||||
github.com/Code-Hex/go-generics-cache v1.5.1/go.mod h1:qxcC9kRVrct9rHeiYpFWSoW1vxyillCVzX13KZG8dl4=
|
||||
github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
|
||||
github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
|
||||
github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
|
||||
@@ -162,23 +152,11 @@ github.com/aws/aws-sdk-go-v2/internal/ini v1.2.4/go.mod h1:ZcBrrI3zBKlhGFNYWvju0
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 h1:qYQ4pzQ2Oz6WpQ8T3HvGHnZydA72MnLuFK9tJwmrbHw=
|
||||
github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6/go.mod h1:O3h0IK87yXci+kg6flUKzJnWeziQUKciKrLjcatSNcY=
|
||||
github.com/aws/aws-sdk-go-v2/service/appconfig v1.4.2/go.mod h1:FZ3HkCe+b10uFZZkFdvf98LHW21k49W8o8J366lqVKY=
|
||||
github.com/aws/aws-sdk-go-v2/service/ec2 v1.296.0 h1:98Miqj16un1WLNyM1RjVDhXYumhqZrQfAeG8i4jPG6o=
|
||||
github.com/aws/aws-sdk-go-v2/service/ec2 v1.296.0/go.mod h1:T6ndRfdhnXLIY5oKBHjYZDVj706los2zGdpThppquvA=
|
||||
github.com/aws/aws-sdk-go-v2/service/ecs v1.74.0 h1:YS5TXaEvzDb+sV+wdQFUtuCAk0GeFR9Ai6HFdxpz6q8=
|
||||
github.com/aws/aws-sdk-go-v2/service/ecs v1.74.0/go.mod h1:10kBgdaNJz0FO/+JWDUH+0rtSjkn5yafgavDDmmhFzs=
|
||||
github.com/aws/aws-sdk-go-v2/service/elasticache v1.51.12 h1:S066ajzfPRCSW4lsSHOYglne6SNi2CHt1u5omzW1RBg=
|
||||
github.com/aws/aws-sdk-go-v2/service/elasticache v1.51.12/go.mod h1:86SE4NcXxbxr8KTG3yOyDmd4HyiFmKl8TexXnhYJ+Bw=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 h1:5EniKhLZe4xzL7a+fU3C2tfUN4nWIqlLesfrjkuPFTY=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7/go.mod h1:x0nZssQ3qZSnIcePWLvcoFisRXJzcTVvYpAAdYX8+GI=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.3.2/go.mod h1:72HRZDLMtmVQiLG2tLfQcaWLCssELvGl+Zf2WVxMmR8=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3xgIJMSC8S6hEVq+38DcvUlgFY0FM6mSI5oto=
|
||||
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21/go.mod h1:r6+pf23ouCB718FUxaqzZdbpYFyDtehyZcmP5KL9FkA=
|
||||
github.com/aws/aws-sdk-go-v2/service/kafka v1.49.1 h1:BgBatWcQIFqF1l6KGHjv66V0d/ISnWrTwxDx/Jf6EJM=
|
||||
github.com/aws/aws-sdk-go-v2/service/kafka v1.49.1/go.mod h1:pMpys+PlrN//vj8j5s0oOAMJjauj81VkHzIZxPVWOro=
|
||||
github.com/aws/aws-sdk-go-v2/service/lightsail v1.51.0 h1:cg6PxzoIide2wiEyLfikOFN+XwHafwR8p5+L9U1E8dQ=
|
||||
github.com/aws/aws-sdk-go-v2/service/lightsail v1.51.0/go.mod h1:YvX7hjUWecrKX8fBkbEncyddEW85xjNH+u5JHioITOw=
|
||||
github.com/aws/aws-sdk-go-v2/service/rds v1.117.0 h1:T1Xe9sYxSUUQOvd1RsFeVk/IXFPdqSiN0atXu/Hy/8A=
|
||||
github.com/aws/aws-sdk-go-v2/service/rds v1.117.0/go.mod h1:QbXW4coAMakHQhf1qhE0eVVCen9gwB/Kvn+HHHKhpGY=
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.8 h1:0GFOLzEbOyZABS3PhYfBIx2rNBACYcKty+XGkTgw1ow=
|
||||
github.com/aws/aws-sdk-go-v2/service/signin v1.0.8/go.mod h1:LXypKvk85AROkKhOG6/YEcHFPoX+prKTowKnVdcaIxE=
|
||||
github.com/aws/aws-sdk-go-v2/service/sns v1.39.11 h1:Ke7RS0NuP9Xwk31prXYcFGA1Qfn8QmNWcxyjKPcXZdc=
|
||||
@@ -247,8 +225,6 @@ github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWH
|
||||
github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os8IaYg++6uMOdKK83QtkkvJik=
|
||||
github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4=
|
||||
github.com/coder/quartz v0.3.0 h1:bUoSEJ77NBfKtUqv6CPSC0AS8dsjqAqqAv7bN02m1mg=
|
||||
github.com/coder/quartz v0.3.0/go.mod h1:BgE7DOj/8NfvRgvKw0jPLDQH/2Lya2kxcTaNJ8X0rZk=
|
||||
github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
|
||||
@@ -275,12 +251,8 @@ github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa5
|
||||
github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/digitalocean/godo v1.178.0 h1:+B4xGOaoFwwwpM7TKhoyGHdmFg5eF9zDB1YfOLvNJ2E=
|
||||
github.com/digitalocean/godo v1.178.0/go.mod h1:xQsWpVCCbkDrWisHA72hPzPlnC+4W5w/McZY5ij9uvU=
|
||||
github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
|
||||
github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
|
||||
github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM=
|
||||
github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
|
||||
github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94=
|
||||
github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE=
|
||||
github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
|
||||
@@ -313,9 +285,6 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.m
|
||||
github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
|
||||
github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
|
||||
github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
|
||||
github.com/envoyproxy/go-control-plane v0.14.0 h1:hbG2kr4RuFj222B6+7T83thSPqLjwBIfQawTkC++2HA=
|
||||
github.com/envoyproxy/go-control-plane/envoy v1.37.0 h1:u3riX6BoYRfF4Dr7dwSOroNfdSbEPe9Yyl09/B6wBrQ=
|
||||
github.com/envoyproxy/go-control-plane/envoy v1.37.0/go.mod h1:DReE9MMrmecPy+YvQOAOHNYMALuowAnbjjEMkkWOi6A=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/envoyproxy/protoc-gen-validate v1.3.3 h1:MVQghNeW+LZcmXe7SY1V36Z+WFMDjpqGAGacLe2T0ds=
|
||||
github.com/envoyproxy/protoc-gen-validate v1.3.3/go.mod h1:TsndJ/ngyIdQRhMcVVGDDHINPLWB7C82oDArY51KfB0=
|
||||
@@ -327,8 +296,6 @@ github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5Kwzbycv
|
||||
github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
|
||||
github.com/fatih/color v1.10.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
|
||||
github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
|
||||
github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
|
||||
github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
|
||||
github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
|
||||
github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
|
||||
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
|
||||
@@ -438,16 +405,12 @@ github.com/go-playground/validator/v10 v10.27.0 h1:w8+XrWVMhGkxOaaowyKH35gFydVHO
|
||||
github.com/go-playground/validator/v10 v10.27.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo=
|
||||
github.com/go-redis/redismock/v9 v9.2.0 h1:ZrMYQeKPECZPjOj5u9eyOjg8Nnb0BS9lkVIZ6IpsKLw=
|
||||
github.com/go-redis/redismock/v9 v9.2.0/go.mod h1:18KHfGDK4Y6c2R0H38EUGWAdc7ZQS9gfYxc94k7rWT0=
|
||||
github.com/go-resty/resty/v2 v2.17.2 h1:FQW5oHYcIlkCNrMD2lloGScxcHJ0gkjshV3qcQAyHQk=
|
||||
github.com/go-resty/resty/v2 v2.17.2/go.mod h1:kCKZ3wWmwJaNc7S29BRtUhJwy7iqmn+2mLtQrOyQlVA=
|
||||
github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo=
|
||||
github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU=
|
||||
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
|
||||
github.com/go-test/deep v1.0.2-0.20181118220953-042da051cf31/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
|
||||
github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro=
|
||||
github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
|
||||
github.com/go-zookeeper/zk v1.0.4 h1:DPzxraQx7OrPyXq2phlGlNSIyWEsAox0RJmjTseMV6I=
|
||||
github.com/go-zookeeper/zk v1.0.4/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
|
||||
github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
|
||||
github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
|
||||
github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
|
||||
@@ -527,8 +490,6 @@ github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8
|
||||
github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/google/go-querystring v1.2.0 h1:yhqkPbu2/OH+V9BfpCVPZkNmUXhb2gBxJArfhIxNtP0=
|
||||
github.com/google/go-querystring v1.2.0/go.mod h1:8IFJqpSRITyJ8QhQ13bmbeMBDfmeEJZD5A0egEOmkqU=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
|
||||
@@ -571,8 +532,6 @@ github.com/googleapis/gax-go/v2 v2.4.0/go.mod h1:XOTVJ59hdnfJLIP/dh8n5CGryZR2LxK
|
||||
github.com/googleapis/gax-go/v2 v2.18.0 h1:jxP5Uuo3bxm3M6gGtV94P4lliVetoCB4Wk2x8QA86LI=
|
||||
github.com/googleapis/gax-go/v2 v2.18.0/go.mod h1:uSzZN4a356eRG985CzJ3WfbFSpqkLTjsnhWGJR6EwrE=
|
||||
github.com/googleapis/google-cloud-go-testing v0.0.0-20200911160855-bcd43fbb19e8/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g=
|
||||
github.com/gophercloud/gophercloud/v2 v2.11.1 h1:jCs4vLH8sJgRqrPzqVfWgl7uI6JnIIlsgeIRM0uHjxY=
|
||||
github.com/gophercloud/gophercloud/v2 v2.11.1/go.mod h1:Rm0YvKQ4QYX2rY9XaDKnjRzSGwlG5ge4h6ABYnmkKQM=
|
||||
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
|
||||
github.com/gopherjs/gopherjs v1.17.2 h1:fQnZVsXk8uxXIStYb0N4bGk7jeyTalG/wsZjQ25dO0g=
|
||||
github.com/gopherjs/gopherjs v1.17.2/go.mod h1:pRRIvn/QzFLrKfvEz3qUuEhtE/zLCWfreZ6J5gM2i+k=
|
||||
@@ -594,11 +553,7 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz
|
||||
github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c=
|
||||
github.com/hashicorp/consul/api v1.12.0/go.mod h1:6pVBMo0ebnYdt2S3H87XhekM/HHrUoTD2XXb/VrZVy0=
|
||||
github.com/hashicorp/consul/api v1.13.0/go.mod h1:ZlVrynguJKcYr54zGaDbaL3fOvKC9m72FhPvA8T35KQ=
|
||||
github.com/hashicorp/consul/api v1.32.1 h1:0+osr/3t/aZNAdJX558crU3PEjVrG4x6715aZHRgceE=
|
||||
github.com/hashicorp/consul/api v1.32.1/go.mod h1:mXUWLnxftwTmDv4W3lzxYCPD199iNLLUyLfLGFJbtl4=
|
||||
github.com/hashicorp/consul/sdk v0.8.0/go.mod h1:GBvyrGALthsZObzUGsfgHZQDXjg4lOjagTIwIR1vPms=
|
||||
github.com/hashicorp/cronexpr v1.1.3 h1:rl5IkxXN2m681EfivTlccqIryzYJSXRGRNa0xeG7NA4=
|
||||
github.com/hashicorp/cronexpr v1.1.3/go.mod h1:P4wA0KBl9C5q2hABiMO7cp6jcIg96CDh1Efb3g1PWA4=
|
||||
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
|
||||
github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
|
||||
github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
|
||||
@@ -610,8 +565,6 @@ github.com/hashicorp/go-hclog v0.0.0-20180709165350-ff2cf002a8dd/go.mod h1:9bjs9
|
||||
github.com/hashicorp/go-hclog v0.8.0/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
|
||||
github.com/hashicorp/go-hclog v0.12.0/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ=
|
||||
github.com/hashicorp/go-hclog v1.2.0/go.mod h1:whpDNt7SSdeAju8AWKIWsul05p54N/39EeqMAyrmvFQ=
|
||||
github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB11/k=
|
||||
github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M=
|
||||
github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1 h1:DKHmCUm2hRBK510BaiZlwvpD40f8bJFeZnpfm2KLowc=
|
||||
github.com/hashicorp/go-immutable-radix v1.3.1/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
@@ -630,7 +583,6 @@ github.com/hashicorp/go-retryablehttp v0.5.4/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es
|
||||
github.com/hashicorp/go-retryablehttp v0.7.8 h1:ylXZWnqa7Lhqpk0L1P1LzDtGcCR0rPVUrx/c8Unxc48=
|
||||
github.com/hashicorp/go-retryablehttp v0.7.8/go.mod h1:rjiScheydd+CxvumBsIrFKlx3iS0jrZ7LvzFGFmuKbw=
|
||||
github.com/hashicorp/go-rootcerts v1.0.1/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
|
||||
github.com/hashicorp/go-rootcerts v1.0.2 h1:jzhAVGtqPKbwpyCPELlgNWhE1znq+qwJtW5Oi2viEzc=
|
||||
github.com/hashicorp/go-rootcerts v1.0.2/go.mod h1:pqUvnprVnM5bf7AOirdbb01K4ccR319Vf4pU3K5EGc8=
|
||||
github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
|
||||
github.com/hashicorp/go-sockaddr v1.0.2/go.mod h1:rB4wwRAUzs07qva3c5SdrY/NEtAUjGlgmH/UkBUC97A=
|
||||
@@ -658,18 +610,12 @@ github.com/hashicorp/mdns v1.0.4/go.mod h1:mtBihi+LeNXGtG8L9dX59gAEa12BDtBQSp4v/
|
||||
github.com/hashicorp/memberlist v0.3.0/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
|
||||
github.com/hashicorp/memberlist v0.5.4 h1:40YY+3qq2tAUhZIMEK8kqusKZBBjdwJ3NUjvYkcxh74=
|
||||
github.com/hashicorp/memberlist v0.5.4/go.mod h1:OgN6xiIo6RlHUWk+ALjP9e32xWCoQrsOCmHrWCm2MWA=
|
||||
github.com/hashicorp/nomad/api v0.0.0-20260324203407-b27b0c2e019a h1:HGwfgBNl90YBiHdbzZ/+8aMxO1UL9B/yNTAXa8iB8z8=
|
||||
github.com/hashicorp/nomad/api v0.0.0-20260324203407-b27b0c2e019a/go.mod h1:KkLNLU0Nyfh5jWsFoF/PsmMbKpRIAoIV4lmQoJWgKCk=
|
||||
github.com/hashicorp/serf v0.9.6/go.mod h1:TXZNMjZQijwlDvp+r0b63xZ45H7JmCmgg4gpTwn9UV4=
|
||||
github.com/hashicorp/serf v0.9.7/go.mod h1:TXZNMjZQijwlDvp+r0b63xZ45H7JmCmgg4gpTwn9UV4=
|
||||
github.com/hashicorp/serf v0.10.1 h1:Z1H2J60yRKvfDYAOZLd2MU0ND4AH/WDz7xYHDWQsIPY=
|
||||
github.com/hashicorp/serf v0.10.1/go.mod h1:yL2t6BqATOLGc5HF7qbFkTfXoPIY0WZdWHfEvMqbG+4=
|
||||
github.com/hashicorp/vault/api v1.0.4/go.mod h1:gDcqh3WGcR1cpF5AJz/B1UFheUEneMoIospckxBxk6Q=
|
||||
github.com/hashicorp/vault/sdk v0.1.13/go.mod h1:B+hVj7TpuQY1Y/GPbCpffmgd+tSEwvhkWnjtSYCaS2M=
|
||||
github.com/hashicorp/yamux v0.0.0-20180604194846-3520598351bb/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
|
||||
github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d/go.mod h1:+NfK9FKeTrX5uv1uIXGdwYDTeHna2qgaIlx54MXqjAM=
|
||||
github.com/hetznercloud/hcloud-go/v2 v2.36.0 h1:HlLL/aaVXUulqe+rsjoJmrxKhPi1MflL5O9iq5QEtvo=
|
||||
github.com/hetznercloud/hcloud-go/v2 v2.36.0/go.mod h1:MnN/QJEa/RYNQiiVoJjNHPntM7Z1wlYPgJ2HA40/cDE=
|
||||
github.com/hjson/hjson-go/v4 v4.0.0 h1:wlm6IYYqHjOdXH1gHev4VoXCaW20HdQAGCxdOEEg2cs=
|
||||
github.com/hjson/hjson-go/v4 v4.0.0/go.mod h1:KaYt3bTw3zhBjYqnXkYywcYctk0A2nxeEFTse3rH13E=
|
||||
github.com/huandu/go-assert v1.1.5/go.mod h1:yOLvuqZwmcHIC5rIzrBhT7D3Q9c3GFnd0JrPVhn/06U=
|
||||
@@ -689,8 +635,6 @@ github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:
|
||||
github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
|
||||
github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
|
||||
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
|
||||
github.com/ionos-cloud/sdk-go/v6 v6.3.6 h1:l/TtKgdQ1wUH3DDe2SfFD78AW+TJWdEbDpQhHkWd6CM=
|
||||
github.com/ionos-cloud/sdk-go/v6 v6.3.6/go.mod h1:nUGHP4kZHAZngCVr4v6C8nuargFrtvt7GrzH/hqn7c4=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
|
||||
@@ -727,8 +671,6 @@ github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7
|
||||
github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
|
||||
github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
|
||||
github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
|
||||
github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRtuthU=
|
||||
github.com/keybase/go-keychain v0.0.1/go.mod h1:PdEILRW3i9D8JcdM+FmY6RwkHGnhHxXwkPPMeUgOK1k=
|
||||
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=
|
||||
@@ -741,8 +683,6 @@ github.com/knadh/koanf v1.5.0 h1:q2TSd/3Pyc/5yP9ldIrSdIz26MCcyNQzW0pEAugLPNs=
|
||||
github.com/knadh/koanf v1.5.0/go.mod h1:Hgyjp4y8v44hpZtPzs7JZfRAW5AhN7KfZcwv1RYggDs=
|
||||
github.com/knadh/koanf/v2 v2.3.3 h1:jLJC8XCRfLC7n4F+ZKKdBsbq1bfXTpuFhf4L7t94D94=
|
||||
github.com/knadh/koanf/v2 v2.3.3/go.mod h1:gRb40VRAbd4iJMYYD5IxZ6hfuopFcXBpc9bbQpZwo28=
|
||||
github.com/kolo/xmlrpc v0.0.0-20220921171641-a4b6fa1dd06b h1:udzkj9S/zlT5X367kqJis0QP7YMxobob6zhzq6Yre00=
|
||||
github.com/kolo/xmlrpc v0.0.0-20220921171641-a4b6fa1dd06b/go.mod h1:pcaDhQK0/NJZEvtCO0qQPPropqV0sJOJ6YW7X+9kRwM=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
|
||||
github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
|
||||
@@ -770,8 +710,6 @@ github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
|
||||
github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI=
|
||||
github.com/leodido/ragel-machinery v0.0.0-20190525184631-5f46317e436b h1:11UHH39z1RhZ5dc4y4r/4koJo6IYFgTRMe/LlwRTEw0=
|
||||
github.com/leodido/ragel-machinery v0.0.0-20190525184631-5f46317e436b/go.mod h1:WZxr2/6a/Ar9bMDc2rN/LJrE/hF6bXE4LPyDSIxwAfg=
|
||||
github.com/linode/linodego v1.66.0 h1:rK8QJFaV53LWOEJvb/evhTg/dP5ElvtuZmx4iv4RJds=
|
||||
github.com/linode/linodego v1.66.0/go.mod h1:12ykGs9qsvxE+OU3SXuW2w+DTruWF35FPlXC7gGk2tU=
|
||||
github.com/lufia/plan9stats v0.0.0-20251013123823-9fd1530e3ec3 h1:PwQumkgq4/acIiZhtifTV5OUqqiP82UAl0h87xj/l9k=
|
||||
github.com/lufia/plan9stats v0.0.0-20251013123823-9fd1530e3ec3/go.mod h1:autxFIvghDt3jPTLoqZ9OZ7s9qTGNAWmYCjVFWPX/zg=
|
||||
github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
|
||||
@@ -787,8 +725,6 @@ github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope
|
||||
github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
|
||||
github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
|
||||
github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4=
|
||||
github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
|
||||
github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
|
||||
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
|
||||
github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
|
||||
github.com/mattn/go-isatty v0.0.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84=
|
||||
@@ -815,7 +751,6 @@ github.com/mitchellh/cli v1.1.0/go.mod h1:xcISNoH86gajksDmfB23e/pu+B+GeFRMYmoHXx
|
||||
github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw=
|
||||
github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
|
||||
github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
|
||||
github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
|
||||
github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
|
||||
github.com/mitchellh/go-testing-interface v0.0.0-20171004221916-a61a99592b77/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
|
||||
github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
|
||||
@@ -875,8 +810,6 @@ github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.144
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.144.0/go.mod h1:R0go5FMmUe51VpKl8YCk/rUxibA+U3lfPYMoihQ/nhw=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.144.0 h1:Qv3nLVGKJ9LQCGwxteJxjSNyQ5CP99QRvYPFn6d8Y60=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.144.0/go.mod h1:O2rZKRXk1WeYhzfJBVXES/g7+PlIds/TzPZW/4NfTNA=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.148.0 h1:CiTjQE/Hh5xK2t56ogrDK4nl0+tJPNmASCs4zEYZ/xU=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.148.0/go.mod h1:WUFkzTiOpt7EYyL67gv1GOf3RD8qKWGtin3lY9LYzW4=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.144.0 h1:Ywu5mU4K5TMJigiXdyZloCRs/cq3/2OnoK3WjxNHWJo=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.144.0/go.mod h1:iebqlu6UvpiV1hO37r1sXA9fXaCaA8sQXilG0///xss=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.144.0 h1:TMRTvQSAeeLtkKwSrqcbectxDRPiqB6yYM3IvjC75es=
|
||||
@@ -889,8 +822,6 @@ github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza v0.144.0 h1
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza v0.144.0/go.mod h1:3Y6ctEEwRg19B0jqsrQH6Hiquqte+zC0ZxpXLLSa5sA=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.144.0 h1:gRk73SsIJv3q/HI0kxMRN5TiIiJj+MRxrz0GEYx3jZw=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.144.0/go.mod h1:7b8ZcPdN6JSm6+LUq3rienoqHHiX60kyoOHDPCMM+L8=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.148.0 h1:xgD/kNGp/wWY+bwY599Pc01OamYN17phRiTP934bM5Y=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.148.0/go.mod h1:ZK7wvaefla9lB3bAW0rNKt7IzRPcTRQoOFqr4sZy/XM=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor v0.144.0 h1:HbmpzTixpQG/xGhQuQoiJTXQPrixe+yivAsF6tl2o4g=
|
||||
github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor v0.144.0/go.mod h1:32jR9iqxVozOJ/Lg5RkCNoW18uCNwpKSbs6h5c28Ep4=
|
||||
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
|
||||
@@ -906,8 +837,6 @@ github.com/openfga/openfga v1.14.1/go.mod h1:AqMyFFi3y24Hko1mIME6ctOdCsCru2HA3uH
|
||||
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
|
||||
github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
|
||||
github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
|
||||
github.com/ovh/go-ovh v1.9.0 h1:6K8VoL3BYjVV3In9tPJUdT7qMx9h0GExN9EXx1r2kKE=
|
||||
github.com/ovh/go-ovh v1.9.0/go.mod h1:cTVDnl94z4tl8pP1uZ/8jlVxntjSIf09bNcQ5TJSC7c=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
|
||||
github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
@@ -934,8 +863,6 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
|
||||
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
|
||||
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
@@ -985,8 +912,6 @@ github.com/prometheus/sigv4 v0.4.1 h1:EIc3j+8NBea9u1iV6O5ZAN8uvPq2xOIUPcqCTivHuX
|
||||
github.com/prometheus/sigv4 v0.4.1/go.mod h1:eu+ZbRvsc5TPiHwqh77OWuCnWK73IdkETYY46P4dXOU=
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg=
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
|
||||
github.com/puzpuzpuz/xsync/v4 v4.4.0 h1:vlSN6/CkEY0pY8KaB0yqo/pCLZvp9nhdbBdjipT4gWo=
|
||||
github.com/puzpuzpuz/xsync/v4 v4.4.0/go.mod h1:VJDmTCJMBt8igNxnkQd86r+8KUeN1quSfNKu5bLYFQo=
|
||||
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
|
||||
github.com/redis/go-redis/extra/rediscmd/v9 v9.15.1 h1:G3pzZlMvMX9VX9TBB8zr03CAkeyMtbyW2D59PdyaGkM=
|
||||
github.com/redis/go-redis/extra/rediscmd/v9 v9.15.1/go.mod h1:JiJ4f0bngycE8LQqzY/4TB23witBbFnlUS6hPvHn6Zc=
|
||||
@@ -1022,8 +947,6 @@ github.com/samber/lo v1.47.0 h1:z7RynLwP5nbyRscyvcD043DWYoOcYRv3mV8lBeqOCLc=
|
||||
github.com/samber/lo v1.47.0/go.mod h1:RmDH9Ct32Qy3gduHQuKJ3gW1fMHAnE/fAzQuf6He5cU=
|
||||
github.com/santhosh-tekuri/jsonschema/v3 v3.1.0 h1:levPcBfnazlA1CyCMC3asL/QLZkq9pa8tQZOH513zQw=
|
||||
github.com/santhosh-tekuri/jsonschema/v3 v3.1.0/go.mod h1:8kzK2TC0k0YjOForaAHdNEa7ik0fokNa2k30BKJ/W7Y=
|
||||
github.com/scaleway/scaleway-sdk-go v1.0.0-beta.36 h1:ObX9hZmK+VmijreZO/8x9pQ8/P/ToHD/bdSb4Eg4tUo=
|
||||
github.com/scaleway/scaleway-sdk-go v1.0.0-beta.36/go.mod h1:LEsDu4BubxK7/cWhtlQWfuxwL4rf/2UEpxXz1o1EMtM=
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
|
||||
github.com/sebdah/goldie/v2 v2.5.3 h1:9ES/mNN+HNUbNWpVAlrzuZ7jE+Nrczbj8uFRjM7624Y=
|
||||
@@ -1078,8 +1001,6 @@ github.com/spf13/viper v1.20.1 h1:ZMi+z/lvLyPSCoNtFCpqjy0S4kPbirhpTMwl8BkW9X4=
|
||||
github.com/spf13/viper v1.20.1/go.mod h1:P9Mdzt1zoHIG8m2eZQinpiBjo6kCmZSKBClNNqjJvu4=
|
||||
github.com/srikanthccv/ClickHouse-go-mock v0.12.0 h1:KUzaWTwuqMc2uf5FylM/oAcTFdE2DdZjvISm9V0/NAA=
|
||||
github.com/srikanthccv/ClickHouse-go-mock v0.12.0/go.mod h1:1oUmLtXEXOyS0EEWVKlKEfLfv9y02agCMAvD3tVnhlo=
|
||||
github.com/stackitcloud/stackit-sdk-go/core v0.23.0 h1:zPrOhf3Xe47rKRs1fg/AqKYUiJJRYjdcv+3qsS50mEs=
|
||||
github.com/stackitcloud/stackit-sdk-go/core v0.23.0/go.mod h1:osMglDby4csGZ5sIfhNyYq1bS1TxIdPY88+skE/kkmI=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.3.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
|
||||
@@ -1158,8 +1079,6 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
|
||||
github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
|
||||
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
|
||||
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
|
||||
github.com/vultr/govultr/v3 v3.28.1 h1:KR3LhppYARlBujY7+dcrE7YKL0Yo9qXL+msxykKQrLI=
|
||||
github.com/vultr/govultr/v3 v3.28.1/go.mod h1:2zyUw9yADQaGwKnwDesmIOlBNLrm7edsCfWHFJpWKf8=
|
||||
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
|
||||
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
|
||||
github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
|
||||
@@ -1921,8 +1840,6 @@ gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWM
|
||||
gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
|
||||
gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
|
||||
gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
|
||||
gopkg.in/ini.v1 v1.67.1 h1:tVBILHy0R6e4wkYOn3XmiITt/hEVH4TFMYvAX2Ytz6k=
|
||||
gopkg.in/ini.v1 v1.67.1/go.mod h1:x/cyOwCgZqOkJoDIJ3c1KNHMo10+nLGAhh+kn3Zizss=
|
||||
gopkg.in/square/go-jose.v2 v2.3.1/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=
|
||||
gopkg.in/telebot.v3 v3.3.8 h1:uVDGjak9l824FN9YARWUHMsiNZnlohAVwUycw21k6t8=
|
||||
gopkg.in/telebot.v3 v3.3.8/go.mod h1:1mlbqcLTVSfK9dx7fdp+Nb5HZsy4LLPtpZTKmwhwtzM=
|
||||
|
||||
@@ -3,16 +3,15 @@ package flagger
|
||||
import "github.com/SigNoz/signoz/pkg/types/featuretypes"
|
||||
|
||||
var (
|
||||
FeatureUseSpanMetrics = featuretypes.MustNewName("use_span_metrics")
|
||||
FeatureKafkaSpanEval = featuretypes.MustNewName("kafka_span_eval")
|
||||
FeatureHideRootUser = featuretypes.MustNewName("hide_root_user")
|
||||
FeaturePutMetersInZeus = featuretypes.MustNewName("put_meters_in_zeus")
|
||||
FeatureUseMeterReporter = featuretypes.MustNewName("use_meter_reporter")
|
||||
FeatureUseJSONBody = featuretypes.MustNewName("use_json_body")
|
||||
FeatureEnableAIObservability = featuretypes.MustNewName("enable_ai_observability")
|
||||
FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
|
||||
FeatureUsePrometheusClickhouseV2 = featuretypes.MustNewName("use_prometheus_clickhouse_v2")
|
||||
FeatureResolveSemconvFamilies = featuretypes.MustNewName("resolve_semconv_families")
|
||||
FeatureUseSpanMetrics = featuretypes.MustNewName("use_span_metrics")
|
||||
FeatureKafkaSpanEval = featuretypes.MustNewName("kafka_span_eval")
|
||||
FeatureHideRootUser = featuretypes.MustNewName("hide_root_user")
|
||||
FeaturePutMetersInZeus = featuretypes.MustNewName("put_meters_in_zeus")
|
||||
FeatureUseMeterReporter = featuretypes.MustNewName("use_meter_reporter")
|
||||
FeatureUseJSONBody = featuretypes.MustNewName("use_json_body")
|
||||
FeatureEnableAIObservability = featuretypes.MustNewName("enable_ai_observability")
|
||||
FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
|
||||
FeatureResolveSemconvFamilies = featuretypes.MustNewName("resolve_semconv_families")
|
||||
)
|
||||
|
||||
func MustNewRegistry() featuretypes.Registry {
|
||||
@@ -81,14 +80,6 @@ func MustNewRegistry() featuretypes.Registry {
|
||||
DefaultVariant: featuretypes.MustNewName("disabled"),
|
||||
Variants: featuretypes.NewBooleanVariants(),
|
||||
},
|
||||
&featuretypes.Feature{
|
||||
Name: FeatureUsePrometheusClickhouseV2,
|
||||
Kind: featuretypes.KindBoolean,
|
||||
Stage: featuretypes.StageExperimental,
|
||||
Description: "Runs PromQL queries on the clickhousev2 provider alongside the served engine result and logs any difference; serving is unaffected.",
|
||||
DefaultVariant: featuretypes.MustNewName("disabled"),
|
||||
Variants: featuretypes.NewBooleanVariants(),
|
||||
},
|
||||
&featuretypes.Feature{
|
||||
Name: FeatureResolveSemconvFamilies,
|
||||
Kind: featuretypes.KindBoolean,
|
||||
|
||||
@@ -1,90 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/prometheus/prometheus/prompb"
|
||||
"github.com/prometheus/prometheus/storage"
|
||||
)
|
||||
|
||||
// statementRecorder collects the statements a PromQL evaluation would run.
|
||||
// Safe for concurrent use: the engine may Select selectors concurrently.
|
||||
type statementRecorder struct {
|
||||
mu sync.Mutex
|
||||
statements []prometheus.CapturedStatement
|
||||
}
|
||||
|
||||
func (r *statementRecorder) record(query string, args []any) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.statements = append(r.statements, prometheus.CapturedStatement{Query: query, Args: args})
|
||||
}
|
||||
|
||||
func (r *statementRecorder) Statements() []prometheus.CapturedStatement {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
out := make([]prometheus.CapturedStatement, len(r.statements))
|
||||
copy(out, r.statements)
|
||||
return out
|
||||
}
|
||||
|
||||
// captureClient builds the same SQL as the real client but records it and
|
||||
// returns an empty result instead of executing.
|
||||
type captureClient struct {
|
||||
*client
|
||||
recorder *statementRecorder
|
||||
}
|
||||
|
||||
func (c *captureClient) Read(ctx context.Context, query *prompb.Query, _ bool) (storage.SeriesSet, error) {
|
||||
// Raw-SQL passthrough ({job="rawsql", query="..."}): record the raw query.
|
||||
if len(query.Matchers) == 2 {
|
||||
var hasJob bool
|
||||
var queryString string
|
||||
for _, m := range query.Matchers {
|
||||
if m.Type == prompb.LabelMatcher_EQ && m.Name == "job" && m.Value == "rawsql" {
|
||||
hasJob = true
|
||||
}
|
||||
if m.Type == prompb.LabelMatcher_EQ && m.Name == "query" {
|
||||
queryString = m.Value
|
||||
}
|
||||
}
|
||||
if hasJob && queryString != "" {
|
||||
c.recorder.record(queryString, nil)
|
||||
return storage.EmptySeriesSet(), nil
|
||||
}
|
||||
}
|
||||
|
||||
// Without executing the series lookup, only an exact-name selector's
|
||||
// metric name is known.
|
||||
var metricNames []string
|
||||
for _, matcher := range query.Matchers {
|
||||
if matcher.Name == "__name__" && matcher.Type == prompb.LabelMatcher_EQ {
|
||||
metricNames = []string{matcher.Value}
|
||||
}
|
||||
}
|
||||
|
||||
// Build the executing path's queries, but only record them.
|
||||
sub, err := seriesLookupQuery(query, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
samplesQuery, samplesArgs := buildSamplesQuery(int64(query.StartTimestampMs), int64(query.EndTimestampMs), metricNames, sub)
|
||||
c.recorder.record(samplesQuery, samplesArgs)
|
||||
|
||||
return storage.EmptySeriesSet(), nil
|
||||
}
|
||||
|
||||
// captureQueryable adapts the capturing read client to storage.Queryable.
|
||||
type captureQueryable struct {
|
||||
inner storage.SampleAndChunkQueryable
|
||||
}
|
||||
|
||||
func (c captureQueryable) Querier(mint, maxt int64) (storage.Querier, error) {
|
||||
querier, err := c.inner.Querier(mint, maxt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return storage.NewMergeQuerier(nil, []storage.Querier{querier}, storage.ChainedSeriesMerge), nil
|
||||
}
|
||||
@@ -1,517 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"sort"
|
||||
"sync"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/cespare/xxhash/v2"
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
promValue "github.com/prometheus/prometheus/model/value"
|
||||
"github.com/prometheus/prometheus/prompb"
|
||||
"github.com/prometheus/prometheus/storage"
|
||||
"github.com/prometheus/prometheus/storage/remote"
|
||||
)
|
||||
|
||||
type client struct {
|
||||
settings factory.ScopedProviderSettings
|
||||
telemetryStore telemetrystore.TelemetryStore
|
||||
}
|
||||
|
||||
func NewReadClient(settings factory.ScopedProviderSettings, telemetryStore telemetrystore.TelemetryStore) remote.ReadClient {
|
||||
return &client{
|
||||
settings: settings,
|
||||
telemetryStore: telemetryStore,
|
||||
}
|
||||
}
|
||||
|
||||
func (client *client) Read(ctx context.Context, query *prompb.Query, sortSeries bool) (storage.SeriesSet, error) {
|
||||
if len(query.Matchers) == 2 {
|
||||
var hasJob bool
|
||||
var queryString string
|
||||
for _, m := range query.Matchers {
|
||||
if m.Type == prompb.LabelMatcher_EQ && m.Name == "job" && m.Value == "rawsql" {
|
||||
hasJob = true
|
||||
}
|
||||
if m.Type == prompb.LabelMatcher_EQ && m.Name == "query" {
|
||||
queryString = m.Value
|
||||
}
|
||||
}
|
||||
|
||||
if hasJob && queryString != "" {
|
||||
res, err := client.queryRaw(ctx, queryString, int64(query.EndTimestampMs))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return remote.FromQueryResult(sortSeries, res), nil
|
||||
}
|
||||
}
|
||||
|
||||
lookup, err := seriesLookupQuery(query, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
lookupSQL, lookupArgs := lookup.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
fingerprints, metricNames, err := client.getFingerprintsFromClickhouseQuery(ctx, lookupSQL, lookupArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(fingerprints) == 0 {
|
||||
return remote.FromQueryResult(sortSeries, new(prompb.QueryResult)), nil
|
||||
}
|
||||
|
||||
sub, err := seriesLookupQuery(query, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
samplesSQL, samplesArgs := buildSamplesQuery(int64(query.StartTimestampMs), int64(query.EndTimestampMs), metricNames, sub)
|
||||
|
||||
res := new(prompb.QueryResult)
|
||||
timeseries, err := client.querySamples(ctx, samplesSQL, samplesArgs, fingerprints)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
res.Timeseries = timeseries
|
||||
|
||||
return remote.FromQueryResult(sortSeries, res), nil
|
||||
}
|
||||
|
||||
func (c *client) ReadMultiple(ctx context.Context, queries []*prompb.Query, sortSeries bool) (storage.SeriesSet, error) {
|
||||
if len(queries) == 0 {
|
||||
return storage.EmptySeriesSet(), nil
|
||||
}
|
||||
if len(queries) == 1 {
|
||||
return c.Read(ctx, queries[0], sortSeries)
|
||||
}
|
||||
|
||||
type result struct {
|
||||
ss storage.SeriesSet
|
||||
err error
|
||||
}
|
||||
|
||||
results := make([]result, len(queries))
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(len(queries))
|
||||
for i, q := range queries {
|
||||
go func(i int, q *prompb.Query) {
|
||||
defer wg.Done()
|
||||
ss, err := c.Read(ctx, q, sortSeries)
|
||||
results[i] = result{ss, err}
|
||||
}(i, q)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
sets := make([]storage.SeriesSet, 0, len(queries))
|
||||
for _, r := range results {
|
||||
if r.err != nil {
|
||||
return nil, r.err
|
||||
}
|
||||
sets = append(sets, r.ss)
|
||||
}
|
||||
return storage.NewMergeSeriesSet(sets, 0, storage.ChainedSeriesMerge), nil
|
||||
}
|
||||
|
||||
// anchorRegex makes a pattern fully anchored, the way Prometheus compiles
|
||||
// matcher regexes; ClickHouse's match() would otherwise substring-match.
|
||||
func anchorRegex(pattern string) string {
|
||||
return "^(?:" + pattern + ")$"
|
||||
}
|
||||
|
||||
// seriesLookupQuery builds the time-series lookup. It returns a builder so
|
||||
// the samples query can embed it as a subquery with the args merged in
|
||||
// render order by the builder instead of hand-numbered placeholders.
|
||||
func seriesLookupQuery(query *prompb.Query, subQuery bool) (*sqlbuilder.SelectBuilder, error) {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
if subQuery {
|
||||
sb.Select("fingerprint")
|
||||
} else {
|
||||
sb.Select("fingerprint", "any(labels)")
|
||||
}
|
||||
|
||||
start, end, tableName := getStartAndEndAndTableName(query.StartTimestampMs, query.EndTimestampMs)
|
||||
sb.From(databaseName + "." + tableName)
|
||||
|
||||
sb.Where("temporality IN ['Cumulative', 'Unspecified']")
|
||||
// Inclusive upper bound: registration rows are hour-floored by the
|
||||
// exporter, so a series first registered in the hour starting exactly at
|
||||
// `end` would otherwise be invisible while its samples (<= end) are in
|
||||
// range.
|
||||
sb.Where(fmt.Sprintf("unix_milli >= %d AND unix_milli <= %d", start, end))
|
||||
|
||||
for _, m := range query.Matchers {
|
||||
if m.Name == "__name__" {
|
||||
// __name__ maps onto the metric_name column per matcher type;
|
||||
// reducing regex/negated/absent name matchers to one equality
|
||||
// made such selectors silently return empty.
|
||||
switch m.Type {
|
||||
case prompb.LabelMatcher_EQ:
|
||||
sb.Where(sb.E("metric_name", m.Value))
|
||||
case prompb.LabelMatcher_NEQ:
|
||||
sb.Where(sb.NE("metric_name", m.Value))
|
||||
case prompb.LabelMatcher_RE:
|
||||
sb.Where(fmt.Sprintf("match(metric_name, %s)", sb.Var(anchorRegex(m.Value))))
|
||||
case prompb.LabelMatcher_NRE:
|
||||
sb.Where(fmt.Sprintf("not match(metric_name, %s)", sb.Var(anchorRegex(m.Value))))
|
||||
default:
|
||||
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported or invalid matcher type: %s", m.Type.String())
|
||||
}
|
||||
continue
|
||||
}
|
||||
switch m.Type {
|
||||
case prompb.LabelMatcher_EQ:
|
||||
sb.Where(fmt.Sprintf("JSONExtractString(labels, %s) = %s", sb.Var(m.Name), sb.Var(m.Value)))
|
||||
case prompb.LabelMatcher_NEQ:
|
||||
sb.Where(fmt.Sprintf("JSONExtractString(labels, %s) != %s", sb.Var(m.Name), sb.Var(m.Value)))
|
||||
case prompb.LabelMatcher_RE:
|
||||
sb.Where(fmt.Sprintf("match(JSONExtractString(labels, %s), %s)", sb.Var(m.Name), sb.Var(anchorRegex(m.Value))))
|
||||
case prompb.LabelMatcher_NRE:
|
||||
sb.Where(fmt.Sprintf("not match(JSONExtractString(labels, %s), %s)", sb.Var(m.Name), sb.Var(anchorRegex(m.Value))))
|
||||
default:
|
||||
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported or invalid matcher type: %s", m.Type.String())
|
||||
}
|
||||
}
|
||||
|
||||
sb.GroupBy("fingerprint")
|
||||
return sb, nil
|
||||
}
|
||||
|
||||
func (client *client) getFingerprintsFromClickhouseQuery(ctx context.Context, query string, args []any) (map[uint64][]prompb.Label, []string, error) {
|
||||
ctx = client.withClickhousePrometheusContext(ctx, "getFingerprintsFromClickhouseQuery")
|
||||
rows, err := client.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
fingerprints := make(map[uint64][]prompb.Label)
|
||||
nameSet := make(map[string]struct{})
|
||||
|
||||
var fingerprint uint64
|
||||
var labelString string
|
||||
for rows.Next() {
|
||||
if err = rows.Scan(&fingerprint, &labelString); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
labels, metricName, err := unmarshalLabels(labelString)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
fingerprints[fingerprint] = labels
|
||||
if metricName != "" {
|
||||
nameSet[metricName] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
metricNames := make([]string, 0, len(nameSet))
|
||||
for name := range nameSet {
|
||||
metricNames = append(metricNames, name)
|
||||
}
|
||||
sort.Strings(metricNames)
|
||||
|
||||
return fingerprints, metricNames, nil
|
||||
}
|
||||
|
||||
// buildSamplesQuery renders the samples SQL for the series selected by
|
||||
// subQuery. The metric_name condition exists only for primary-key pruning;
|
||||
// the fingerprint filter already selects the right rows.
|
||||
//
|
||||
// Time bounds are inclusive on both ends because that is Prometheus's
|
||||
// storage contract: Select(mint, maxt) returns [start, end] and the engine
|
||||
// itself trims each evaluation window to left-open (T-window, T], so the
|
||||
// sample at exactly `end` belongs to the last point. This deliberately
|
||||
// differs from the query builder's `unix_milli < end`, which is correct for
|
||||
// its own model — toStartOfInterval buckets covering [t, t+step), where a
|
||||
// sample at `end` falls in an unrendered bucket and end-exclusive ranges
|
||||
// tile exactly across cached time slices.
|
||||
func buildSamplesQuery(start int64, end int64, metricNames []string, sub *sqlbuilder.SelectBuilder) (string, []any) {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
sb.Select("metric_name", "fingerprint", "unix_milli", "value", "flags")
|
||||
sb.From(databaseName + "." + distributedSamplesV4)
|
||||
|
||||
if len(metricNames) > 0 {
|
||||
names := make([]any, len(metricNames))
|
||||
for i, name := range metricNames {
|
||||
names[i] = name
|
||||
}
|
||||
sb.Where(sb.In("metric_name", names...))
|
||||
}
|
||||
sb.Where(fmt.Sprintf("fingerprint GLOBAL IN (%s)", sb.Var(sub)))
|
||||
sb.Where(sb.GTE("unix_milli", start), sb.LTE("unix_milli", end))
|
||||
sb.OrderBy("fingerprint", "unix_milli")
|
||||
|
||||
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
}
|
||||
|
||||
func (client *client) querySamples(ctx context.Context, query string, args []any, fingerprints map[uint64][]prompb.Label) ([]*prompb.TimeSeries, error) {
|
||||
ctx = client.withClickhousePrometheusContext(ctx, "querySamples")
|
||||
|
||||
rows, err := client.telemetryStore.ClickhouseDB().Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var res []*prompb.TimeSeries
|
||||
var ts *prompb.TimeSeries
|
||||
var metricName string
|
||||
var fingerprint, prevFingerprint uint64
|
||||
var timestampMs, prevTimestamp int64
|
||||
var value float64
|
||||
var flags uint32
|
||||
|
||||
prevTimestamp = math.MinInt64
|
||||
|
||||
for rows.Next() {
|
||||
if err := rows.Scan(&metricName, &fingerprint, ×tampMs, &value, &flags); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// collect samples in time series
|
||||
if fingerprint != prevFingerprint {
|
||||
// add collected time series to result
|
||||
prevFingerprint = fingerprint
|
||||
if ts != nil {
|
||||
res = append(res, ts)
|
||||
}
|
||||
|
||||
labels := fingerprints[fingerprint]
|
||||
ts = &prompb.TimeSeries{
|
||||
Labels: labels,
|
||||
}
|
||||
prevTimestamp = math.MinInt64
|
||||
}
|
||||
|
||||
if flags&1 == 1 {
|
||||
value = math.Float64frombits(promValue.StaleNaN)
|
||||
}
|
||||
|
||||
if timestampMs == prevTimestamp {
|
||||
continue
|
||||
}
|
||||
prevTimestamp = timestampMs
|
||||
|
||||
// add samples to current time series
|
||||
ts.Samples = append(ts.Samples, prompb.Sample{
|
||||
Timestamp: timestampMs,
|
||||
Value: value,
|
||||
})
|
||||
}
|
||||
|
||||
// add last time series
|
||||
if ts != nil {
|
||||
res = append(res, ts)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return mergeSeriesWithIdenticalLabels(res), nil
|
||||
}
|
||||
|
||||
// mergeSeriesWithIdenticalLabels collapses series sharing one labelset into
|
||||
// one series each. Distinct fingerprints can map to one labelset: a label
|
||||
// value goes empty over a series' lifetime (#8563), the fingerprint
|
||||
// algorithm changes across exporter versions, or the env changes. The
|
||||
// engine treats the labelset as series identity — duplicates raise
|
||||
// "duplicate series", and #8563's workaround of injecting a synthetic
|
||||
// fingerprint label silently broke without() and vector matching. Merging
|
||||
// at the last point before hand-off keeps any future input-side label
|
||||
// normalization collision-safe. Grouping is by an order-insensitive 64-bit
|
||||
// hash so the common no-collision case costs one hash and one map insert
|
||||
// per series; hash-equal groups are confirmed by exact labelset equality
|
||||
// before any merge. Regression:
|
||||
// TestClient_QuerySamplesMergesIdenticalLabelSets and
|
||||
// tests/integration/tests/promqlconformance/02_fingerprint_probe.py.
|
||||
func mergeSeriesWithIdenticalLabels(series []*prompb.TimeSeries) []*prompb.TimeSeries {
|
||||
if len(series) < 2 {
|
||||
return series
|
||||
}
|
||||
|
||||
groups := make(map[uint64][]*prompb.TimeSeries, len(series))
|
||||
order := make([]uint64, 0, len(series))
|
||||
for _, ts := range series {
|
||||
key := labelsHash(ts.Labels)
|
||||
if _, ok := groups[key]; !ok {
|
||||
order = append(order, key)
|
||||
}
|
||||
groups[key] = append(groups[key], ts)
|
||||
}
|
||||
if len(order) == len(series) {
|
||||
return series
|
||||
}
|
||||
|
||||
res := make([]*prompb.TimeSeries, 0, len(order))
|
||||
for _, key := range order {
|
||||
group := groups[key]
|
||||
if len(group) == 1 {
|
||||
res = append(res, group[0])
|
||||
continue
|
||||
}
|
||||
for _, sub := range splitByLabelSet(group) {
|
||||
if len(sub) == 1 {
|
||||
res = append(res, sub[0])
|
||||
continue
|
||||
}
|
||||
res = append(res, mergeSamples(sub))
|
||||
}
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
var labelHashSep = []byte{0xff}
|
||||
|
||||
// labelsHash combines per-label hashes commutatively, so the stored JSON's
|
||||
// key order (not canonical across fingerprints) needs no sort.
|
||||
func labelsHash(lbls []prompb.Label) uint64 {
|
||||
var h uint64
|
||||
var d xxhash.Digest
|
||||
for _, l := range lbls {
|
||||
d.Reset()
|
||||
_, _ = d.WriteString(l.Name)
|
||||
_, _ = d.Write(labelHashSep)
|
||||
_, _ = d.WriteString(l.Value)
|
||||
h += d.Sum64()
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// splitByLabelSet partitions a hash-equal group into sub-groups of exactly
|
||||
// equal labelsets, preserving input order; series that merely collide on the
|
||||
// 64-bit hash must not be merged.
|
||||
func splitByLabelSet(group []*prompb.TimeSeries) [][]*prompb.TimeSeries {
|
||||
var out [][]*prompb.TimeSeries
|
||||
outer:
|
||||
for _, ts := range group {
|
||||
for i, sub := range out {
|
||||
if labelSetsEqual(sub[0].Labels, ts.Labels) {
|
||||
out[i] = append(out[i], ts)
|
||||
continue outer
|
||||
}
|
||||
}
|
||||
out = append(out, []*prompb.TimeSeries{ts})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func labelSetsEqual(a, b []prompb.Label) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for _, la := range a {
|
||||
found := false
|
||||
for _, lb := range b {
|
||||
if la.Name == lb.Name {
|
||||
found = la.Value == lb.Value
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// mergeSamples k-way merges sample streams that share one labelset. On
|
||||
// equal timestamps the highest fingerprint wins: the input is in ascending
|
||||
// fingerprint order (samples SQL), keeping the choice deterministic.
|
||||
func mergeSamples(group []*prompb.TimeSeries) *prompb.TimeSeries {
|
||||
merged := &prompb.TimeSeries{Labels: group[0].Labels}
|
||||
idx := make([]int, len(group))
|
||||
for {
|
||||
minTs := int64(math.MaxInt64)
|
||||
for i, ts := range group {
|
||||
if idx[i] < len(ts.Samples) && ts.Samples[idx[i]].Timestamp < minTs {
|
||||
minTs = ts.Samples[idx[i]].Timestamp
|
||||
}
|
||||
}
|
||||
if minTs == math.MaxInt64 {
|
||||
return merged
|
||||
}
|
||||
var chosen prompb.Sample
|
||||
for i, ts := range group {
|
||||
if idx[i] < len(ts.Samples) && ts.Samples[idx[i]].Timestamp == minTs {
|
||||
chosen = ts.Samples[idx[i]]
|
||||
idx[i]++
|
||||
}
|
||||
}
|
||||
merged.Samples = append(merged.Samples, chosen)
|
||||
}
|
||||
}
|
||||
|
||||
func (client *client) queryRaw(ctx context.Context, query string, ts int64) (*prompb.QueryResult, error) {
|
||||
ctx = client.withClickhousePrometheusContext(ctx, "queryRaw")
|
||||
|
||||
rows, err := client.telemetryStore.ClickhouseDB().Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
columns := rows.Columns()
|
||||
var res prompb.QueryResult
|
||||
targets := make([]any, len(columns))
|
||||
for i := range targets {
|
||||
targets[i] = new(scanner)
|
||||
}
|
||||
|
||||
for rows.Next() {
|
||||
if err = rows.Scan(targets...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
labels := make([]prompb.Label, 0, len(columns))
|
||||
var value float64
|
||||
for i, c := range columns {
|
||||
v := targets[i].(*scanner)
|
||||
switch c {
|
||||
case "value":
|
||||
value = v.f
|
||||
default:
|
||||
labels = append(labels, prompb.Label{
|
||||
Name: c,
|
||||
Value: v.s,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
res.Timeseries = append(res.Timeseries, &prompb.TimeSeries{
|
||||
Labels: labels,
|
||||
Samples: []prompb.Sample{{
|
||||
Value: value,
|
||||
Timestamp: ts,
|
||||
}},
|
||||
})
|
||||
}
|
||||
if err = rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &res, nil
|
||||
}
|
||||
|
||||
func (client *client) withClickhousePrometheusContext(ctx context.Context, functionName string) context.Context {
|
||||
comments := map[string]string{
|
||||
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
|
||||
instrumentationtypes.CodeNamespace: "clickhouse-prometheus",
|
||||
instrumentationtypes.CodeFunctionName: functionName,
|
||||
}
|
||||
return ctxtypes.NewContextWithCommentVals(ctx, comments)
|
||||
}
|
||||
@@ -1,382 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
"testing"
|
||||
|
||||
cmock "github.com/SigNoz/clickhouse-go-mock"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
"github.com/prometheus/prometheus/prompb"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// Test for querySamples method.
|
||||
func TestClient_QuerySamples(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cols := make([]cmock.ColumnType, 0)
|
||||
cols = append(cols, cmock.ColumnType{Name: "metric_name", Type: "String"})
|
||||
cols = append(cols, cmock.ColumnType{Name: "fingerprint", Type: "UInt64"})
|
||||
cols = append(cols, cmock.ColumnType{Name: "unix_milli", Type: "Int64"})
|
||||
cols = append(cols, cmock.ColumnType{Name: "value", Type: "Float64"})
|
||||
cols = append(cols, cmock.ColumnType{Name: "flags", Type: "UInt32"})
|
||||
tests := []struct {
|
||||
name string
|
||||
start int64
|
||||
end int64
|
||||
fingerprints map[uint64][]prompb.Label
|
||||
metricNames []string
|
||||
subQuery string
|
||||
args []any
|
||||
setupMock func(mock cmock.ClickConnMockCommon, args ...any)
|
||||
expectedTimeSeries int
|
||||
expectError bool
|
||||
description string
|
||||
result []*prompb.TimeSeries
|
||||
}{
|
||||
{
|
||||
name: "successful samples retrieval",
|
||||
start: int64(1000),
|
||||
end: int64(2000),
|
||||
fingerprints: map[uint64][]prompb.Label{
|
||||
123: {
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "instance", Value: "localhost:9090"},
|
||||
},
|
||||
456: {
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "instance", Value: "localhost:9091"},
|
||||
},
|
||||
},
|
||||
metricNames: []string{"cpu_usage"},
|
||||
subQuery: "SELECT metric_name, fingerprint, unix_milli, value, flags",
|
||||
expectedTimeSeries: 2,
|
||||
expectError: false,
|
||||
description: "Should successfully retrieve samples for multiple time series",
|
||||
setupMock: func(mock cmock.ClickConnMockCommon, args ...any) {
|
||||
values := [][]interface{}{
|
||||
{"cpu_usage", uint64(123), int64(1001), float64(1.1), uint32(0)},
|
||||
{"cpu_usage", uint64(123), int64(1001), float64(1.1), uint32(0)},
|
||||
{"cpu_usage", uint64(456), int64(1001), float64(1.2), uint32(0)},
|
||||
{"cpu_usage", uint64(456), int64(1001), float64(1.2), uint32(0)},
|
||||
{"cpu_usage", uint64(456), int64(1001), float64(1.2), uint32(0)},
|
||||
}
|
||||
mock.ExpectQuery("SELECT metric_name, fingerprint, unix_milli, value, flags").WithArgs(args...).WillReturnRows(
|
||||
cmock.NewRows(cols, values),
|
||||
)
|
||||
},
|
||||
result: []*prompb.TimeSeries{
|
||||
{
|
||||
Labels: []prompb.Label{
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "instance", Value: "localhost:9090"},
|
||||
},
|
||||
Samples: []prompb.Sample{
|
||||
{Timestamp: 1001, Value: 1.1},
|
||||
},
|
||||
},
|
||||
{
|
||||
Labels: []prompb.Label{
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "instance", Value: "localhost:9091"},
|
||||
},
|
||||
Samples: []prompb.Sample{
|
||||
{Timestamp: 1001, Value: 1.2},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{Provider: "clickhouse"}, sqlmock.QueryMatcherRegexp)
|
||||
readClient := client{telemetryStore: telemetryStore}
|
||||
if tt.setupMock != nil {
|
||||
tt.setupMock(telemetryStore.Mock(), "cpu_usage", tt.start, tt.end)
|
||||
|
||||
}
|
||||
result, err := readClient.querySamples(ctx, tt.subQuery, []any{"cpu_usage", tt.start, tt.end}, tt.fingerprints)
|
||||
|
||||
if tt.expectError {
|
||||
assert.Error(t, err)
|
||||
assert.Nil(t, result)
|
||||
} else {
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, tt.expectedTimeSeries, len(result))
|
||||
assert.Equal(t, result, tt.result)
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Regression for the duplicate-series class behind #8563: fingerprints
|
||||
// sharing one labelset must come back as one merged series, the higher
|
||||
// fingerprint winning equal timestamps.
|
||||
func TestClient_QuerySamplesMergesIdenticalLabelSets(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
canary := []prompb.Label{
|
||||
{Name: "__name__", Value: "requests"},
|
||||
{Name: "group", Value: "canary"},
|
||||
}
|
||||
production := []prompb.Label{
|
||||
{Name: "__name__", Value: "requests"},
|
||||
{Name: "group", Value: "production"},
|
||||
}
|
||||
fingerprints := map[uint64][]prompb.Label{
|
||||
100: canary,
|
||||
200: canary,
|
||||
300: production,
|
||||
}
|
||||
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{Provider: "clickhouse"}, sqlmock.QueryMatcherRegexp)
|
||||
// Rows arrive ordered by (fingerprint, unix_milli), matching the SQL.
|
||||
values := [][]any{
|
||||
{"requests", uint64(100), int64(1000), float64(1.0), uint32(0)},
|
||||
{"requests", uint64(100), int64(2000), float64(2.0), uint32(0)},
|
||||
{"requests", uint64(200), int64(2000), float64(20.0), uint32(0)},
|
||||
{"requests", uint64(200), int64(3000), float64(30.0), uint32(0)},
|
||||
{"requests", uint64(300), int64(1500), float64(5.0), uint32(0)},
|
||||
}
|
||||
telemetryStore.Mock().ExpectQuery("SELECT metric_name, fingerprint, unix_milli, value, flags").
|
||||
WithArgs("requests", int64(1000), int64(3000)).
|
||||
WillReturnRows(cmock.NewRows(cols, values))
|
||||
|
||||
readClient := client{telemetryStore: telemetryStore}
|
||||
result, err := readClient.querySamples(ctx, "SELECT metric_name, fingerprint, unix_milli, value, flags", []any{"requests", int64(1000), int64(3000)}, fingerprints)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, []*prompb.TimeSeries{
|
||||
{
|
||||
Labels: canary,
|
||||
Samples: []prompb.Sample{
|
||||
{Timestamp: 1000, Value: 1.0},
|
||||
{Timestamp: 2000, Value: 20.0},
|
||||
{Timestamp: 3000, Value: 30.0},
|
||||
},
|
||||
},
|
||||
{
|
||||
Labels: production,
|
||||
Samples: []prompb.Sample{
|
||||
{Timestamp: 1500, Value: 5.0},
|
||||
},
|
||||
},
|
||||
}, result)
|
||||
}
|
||||
|
||||
func TestClient_getFingerprintsFromClickhouseQuery(t *testing.T) {
|
||||
cols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "labels", Type: "String"},
|
||||
}
|
||||
|
||||
sortLabels := func(ls []prompb.Label) {
|
||||
sort.Slice(ls, func(i, j int) bool {
|
||||
if ls[i].Name == ls[j].Name {
|
||||
return ls[i].Value < ls[j].Value
|
||||
}
|
||||
return ls[i].Name < ls[j].Name
|
||||
})
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
start, end int64
|
||||
metricName string
|
||||
subQuery string
|
||||
args []any
|
||||
setupMock func(m cmock.ClickConnMockCommon, args ...any)
|
||||
want map[uint64][]prompb.Label
|
||||
wantNames []string
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "happy-path - two fingerprints",
|
||||
start: 1000,
|
||||
end: 2000,
|
||||
metricName: "cpu_usage",
|
||||
subQuery: `SELECT fingerprint,labels`,
|
||||
// args slice is empty here, but test‑case still owns it
|
||||
args: []any{},
|
||||
|
||||
setupMock: func(m cmock.ClickConnMockCommon, args ...any) {
|
||||
rows := [][]any{
|
||||
{uint64(123), `{"__name__":"cpu_usage","t1":"s1","t2":"s2"}`},
|
||||
{uint64(234), `{"__name__":"cpu_usage","t1":"s1","t2":"s2","empty":""}`},
|
||||
}
|
||||
m.ExpectQuery(`SELECT fingerprint,labels`).WithArgs(args...).WillReturnRows(
|
||||
cmock.NewRows(cols, rows),
|
||||
)
|
||||
},
|
||||
|
||||
// No synthetic fingerprint label (#8563), empty-valued labels
|
||||
// dropped: both fingerprints present one labelset for
|
||||
// querySamples to merge.
|
||||
want: map[uint64][]prompb.Label{
|
||||
123: {
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "t1", Value: "s1"},
|
||||
{Name: "t2", Value: "s2"},
|
||||
},
|
||||
234: {
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "t1", Value: "s1"},
|
||||
{Name: "t2", Value: "s2"},
|
||||
},
|
||||
},
|
||||
wantNames: []string{"cpu_usage"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := telemetrystoretest.New(
|
||||
telemetrystore.Config{Provider: "clickhouse"},
|
||||
sqlmock.QueryMatcherRegexp,
|
||||
)
|
||||
|
||||
if tc.setupMock != nil {
|
||||
tc.setupMock(store.Mock(), tc.args...)
|
||||
}
|
||||
|
||||
c := client{telemetryStore: store}
|
||||
|
||||
got, gotNames, err := c.getFingerprintsFromClickhouseQuery(ctx, tc.subQuery, tc.args)
|
||||
if tc.wantErr {
|
||||
require.Error(t, err)
|
||||
require.Nil(t, got)
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.wantNames, gotNames, "discovered metric names mismatch")
|
||||
require.Equal(t, len(tc.want), len(got), "fingerprint map length mismatch")
|
||||
for fp, expLabels := range tc.want {
|
||||
gotLabels, ok := got[fp]
|
||||
require.Truef(t, ok, "missing fingerprint %d", fp)
|
||||
|
||||
sortLabels(expLabels)
|
||||
sortLabels(gotLabels)
|
||||
|
||||
assert.Equalf(t, expLabels, gotLabels, "labels mismatch for fingerprint %d", fp)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Regression for nameless/regex-name selectors silently returning empty:
|
||||
// the old code reduced every __name__ matcher to `metric_name = <value>`
|
||||
// (empty string when absent). Regexes must come out anchored — Prometheus
|
||||
// matcher semantics, while ClickHouse match() substring-matches.
|
||||
func TestQueryToClickhouseQueryNameMatchers(t *testing.T) {
|
||||
query := func(matchers ...*prompb.LabelMatcher) *prompb.Query {
|
||||
return &prompb.Query{StartTimestampMs: 0, EndTimestampMs: 1000, Matchers: matchers}
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
query *prompb.Query
|
||||
contains []string
|
||||
absent []string
|
||||
args []any
|
||||
}{
|
||||
{
|
||||
name: "exact name",
|
||||
query: query(&prompb.LabelMatcher{Type: prompb.LabelMatcher_EQ, Name: "__name__", Value: "cpu_usage"}),
|
||||
contains: []string{"metric_name = ?"},
|
||||
args: []any{"cpu_usage"},
|
||||
},
|
||||
{
|
||||
name: "regex name is anchored",
|
||||
query: query(&prompb.LabelMatcher{Type: prompb.LabelMatcher_RE, Name: "__name__", Value: ".+"}),
|
||||
contains: []string{"match(metric_name, ?)"},
|
||||
args: []any{"^(?:.+)$"},
|
||||
},
|
||||
{
|
||||
name: "nameless selector has no metric_name condition",
|
||||
query: query(
|
||||
&prompb.LabelMatcher{Type: prompb.LabelMatcher_EQ, Name: "job", Value: "api"},
|
||||
&prompb.LabelMatcher{Type: prompb.LabelMatcher_NRE, Name: "group", Value: "can.*"},
|
||||
),
|
||||
contains: []string{
|
||||
"JSONExtractString(labels, ?) = ?",
|
||||
"not match(JSONExtractString(labels, ?), ?)",
|
||||
},
|
||||
absent: []string{"metric_name"},
|
||||
args: []any{"job", "api", "group", "^(?:can.*)$"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
lookup, err := seriesLookupQuery(tt.query, false)
|
||||
require.NoError(t, err)
|
||||
sql, args := lookup.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
for _, want := range tt.contains {
|
||||
assert.Contains(t, sql, want)
|
||||
}
|
||||
for _, notWant := range tt.absent {
|
||||
assert.NotContains(t, sql, notWant)
|
||||
}
|
||||
assert.Equal(t, tt.args, args)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The samples query narrows by the metric names the lookup discovered and
|
||||
// embeds the series lookup as a subquery, the builder merging its args in
|
||||
// render order.
|
||||
func TestBuildSamplesQueryMetricNames(t *testing.T) {
|
||||
sub := sqlbuilder.NewSelectBuilder()
|
||||
sub.Select("fingerprint")
|
||||
sub.From("t")
|
||||
sub.Where(sub.E("k", "v"))
|
||||
|
||||
sql, args := buildSamplesQuery(5, 9, []string{"a_total", "b_total"}, sub)
|
||||
assert.Contains(t, sql, "metric_name IN (?, ?)")
|
||||
assert.Contains(t, sql, "fingerprint GLOBAL IN (SELECT fingerprint FROM t WHERE k = ?)")
|
||||
assert.Contains(t, sql, "unix_milli >= ? AND unix_milli <= ?")
|
||||
assert.Equal(t, []any{"a_total", "b_total", "v", int64(5), int64(9)}, args)
|
||||
|
||||
sub2 := sqlbuilder.NewSelectBuilder()
|
||||
sub2.Select("fingerprint")
|
||||
sub2.From("t")
|
||||
sql, args = buildSamplesQuery(5, 9, nil, sub2)
|
||||
assert.NotContains(t, sql, "metric_name IN")
|
||||
assert.Equal(t, []any{int64(5), int64(9)}, args)
|
||||
}
|
||||
|
||||
// Hash grouping must stay order-insensitive (stored JSON key order is not
|
||||
// canonical across fingerprints), and a 64-bit hash collision between
|
||||
// distinct labelsets must not merge them — splitByLabelSet is that guard.
|
||||
func TestLabelsHashAndCollisionSplit(t *testing.T) {
|
||||
lbls := []prompb.Label{
|
||||
{Name: "__name__", Value: "requests"},
|
||||
{Name: "job", Value: "api"},
|
||||
{Name: "instance", Value: "0"},
|
||||
}
|
||||
reversed := []prompb.Label{lbls[2], lbls[1], lbls[0]}
|
||||
assert.Equal(t, labelsHash(lbls), labelsHash(reversed))
|
||||
|
||||
a := &prompb.TimeSeries{Labels: []prompb.Label{{Name: "job", Value: "x"}}}
|
||||
b := &prompb.TimeSeries{Labels: []prompb.Label{{Name: "job", Value: "y"}}}
|
||||
c := &prompb.TimeSeries{Labels: []prompb.Label{{Name: "job", Value: "x"}}}
|
||||
got := splitByLabelSet([]*prompb.TimeSeries{a, b, c})
|
||||
require.Len(t, got, 2)
|
||||
assert.Equal(t, []*prompb.TimeSeries{a, c}, got[0])
|
||||
assert.Equal(t, []*prompb.TimeSeries{b}, got[1])
|
||||
}
|
||||
@@ -1,34 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"github.com/prometheus/prometheus/prompb"
|
||||
)
|
||||
|
||||
// Unmarshals JSON into Prometheus labels. It does not preserve order.
|
||||
// Empty-valued labels are dropped: Prometheus treats them as absent, and
|
||||
// keeping them lets two fingerprints present duplicate labelsets to the
|
||||
// engine (the incident behind #8563).
|
||||
func unmarshalLabels(s string) ([]prompb.Label, string, error) {
|
||||
var metricName string
|
||||
m := make(map[string]string)
|
||||
if err := json.Unmarshal([]byte(s), &m); err != nil {
|
||||
return nil, metricName, err
|
||||
}
|
||||
res := make([]prompb.Label, 0, len(m))
|
||||
for n, v := range m {
|
||||
if v == "" {
|
||||
continue
|
||||
}
|
||||
if n == "__name__" {
|
||||
metricName = v
|
||||
}
|
||||
|
||||
res = append(res, prompb.Label{
|
||||
Name: n,
|
||||
Value: v,
|
||||
})
|
||||
}
|
||||
return res, metricName, nil
|
||||
}
|
||||
@@ -1,82 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/prometheus/prompb"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func mkSeries(value string, samples ...prompb.Sample) *prompb.TimeSeries {
|
||||
return &prompb.TimeSeries{
|
||||
Labels: []prompb.Label{{Name: "job", Value: value}},
|
||||
Samples: samples,
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeSeriesWithIdenticalLabels(t *testing.T) {
|
||||
s := func(ts int64, v float64) prompb.Sample { return prompb.Sample{Timestamp: ts, Value: v} }
|
||||
|
||||
t.Run("empty and single series pass through untouched", func(t *testing.T) {
|
||||
assert.Nil(t, mergeSeriesWithIdenticalLabels(nil))
|
||||
one := []*prompb.TimeSeries{mkSeries("a", s(1, 1))}
|
||||
got := mergeSeriesWithIdenticalLabels(one)
|
||||
assert.Equal(t, one, got)
|
||||
})
|
||||
|
||||
t.Run("no collisions returns the input slice as is", func(t *testing.T) {
|
||||
in := []*prompb.TimeSeries{mkSeries("a", s(1, 1)), mkSeries("b", s(1, 2))}
|
||||
got := mergeSeriesWithIdenticalLabels(in)
|
||||
// same backing slice: the fast path must not rebuild anything
|
||||
assert.Equal(t, &in[0], &got[0])
|
||||
})
|
||||
|
||||
t.Run("disjoint streams concatenate in timestamp order", func(t *testing.T) {
|
||||
// the #8563 shape: the series transitioned fingerprints at a point
|
||||
// in time, so the streams do not overlap at all
|
||||
got := mergeSeriesWithIdenticalLabels([]*prompb.TimeSeries{
|
||||
mkSeries("a", s(1, 1), s(2, 2)),
|
||||
mkSeries("a", s(3, 3), s(4, 4)),
|
||||
})
|
||||
assert.Equal(t, []prompb.Sample{s(1, 1), s(2, 2), s(3, 3), s(4, 4)}, got[0].Samples)
|
||||
})
|
||||
|
||||
t.Run("three fingerprints one labelset", func(t *testing.T) {
|
||||
got := mergeSeriesWithIdenticalLabels([]*prompb.TimeSeries{
|
||||
mkSeries("a", s(1, 1), s(4, 4)),
|
||||
mkSeries("a", s(2, 2)),
|
||||
mkSeries("a", s(3, 3)),
|
||||
})
|
||||
assert.Len(t, got, 1)
|
||||
assert.Equal(t, []prompb.Sample{s(1, 1), s(2, 2), s(3, 3), s(4, 4)}, got[0].Samples)
|
||||
})
|
||||
|
||||
t.Run("equal timestamps everywhere keep the last stream's value", func(t *testing.T) {
|
||||
// input is in ascending fingerprint order; the highest wins each tie
|
||||
got := mergeSeriesWithIdenticalLabels([]*prompb.TimeSeries{
|
||||
mkSeries("a", s(1, 1), s(2, 1)),
|
||||
mkSeries("a", s(1, 9), s(2, 9)),
|
||||
})
|
||||
assert.Equal(t, []prompb.Sample{s(1, 9), s(2, 9)}, got[0].Samples)
|
||||
})
|
||||
|
||||
t.Run("zero-sample series in a group is harmless", func(t *testing.T) {
|
||||
got := mergeSeriesWithIdenticalLabels([]*prompb.TimeSeries{
|
||||
mkSeries("a"),
|
||||
mkSeries("a", s(1, 1)),
|
||||
})
|
||||
assert.Len(t, got, 1)
|
||||
assert.Equal(t, []prompb.Sample{s(1, 1)}, got[0].Samples)
|
||||
})
|
||||
|
||||
t.Run("colliding and distinct series interleave without cross-talk", func(t *testing.T) {
|
||||
got := mergeSeriesWithIdenticalLabels([]*prompb.TimeSeries{
|
||||
mkSeries("a", s(1, 1)),
|
||||
mkSeries("b", s(1, 2)),
|
||||
mkSeries("a", s(2, 3)),
|
||||
})
|
||||
assert.Len(t, got, 2)
|
||||
assert.Equal(t, []prompb.Sample{s(1, 1), s(2, 3)}, got[0].Samples)
|
||||
assert.Equal(t, []prompb.Sample{s(1, 2)}, got[1].Samples)
|
||||
})
|
||||
}
|
||||
@@ -1,78 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/prometheus/common/model"
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
"github.com/prometheus/prometheus/storage"
|
||||
"github.com/prometheus/prometheus/storage/remote"
|
||||
)
|
||||
|
||||
var stCallback = func() (int64, error) {
|
||||
return int64(model.Latest), nil
|
||||
}
|
||||
|
||||
type provider struct {
|
||||
settings factory.ScopedProviderSettings
|
||||
telemetryStore telemetrystore.TelemetryStore
|
||||
engine *prometheus.Engine
|
||||
parser prometheus.Parser
|
||||
queryable storage.SampleAndChunkQueryable
|
||||
}
|
||||
|
||||
func NewFactory(telemetryStore telemetrystore.TelemetryStore) factory.ProviderFactory[prometheus.Prometheus, prometheus.Config] {
|
||||
return factory.NewProviderFactory(factory.MustNewName("clickhouse"), func(ctx context.Context, providerSettings factory.ProviderSettings, config prometheus.Config) (prometheus.Prometheus, error) {
|
||||
return New(ctx, providerSettings, config, telemetryStore)
|
||||
})
|
||||
}
|
||||
|
||||
func New(ctx context.Context, providerSettings factory.ProviderSettings, config prometheus.Config, telemetryStore telemetrystore.TelemetryStore) (prometheus.Prometheus, error) {
|
||||
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheus")
|
||||
|
||||
readClient := NewReadClient(settings, telemetryStore)
|
||||
|
||||
return &provider{
|
||||
settings: settings,
|
||||
telemetryStore: telemetryStore,
|
||||
engine: prometheus.NewEngine(settings.Logger(), config),
|
||||
parser: prometheus.NewParser(),
|
||||
queryable: remote.NewSampleAndChunkQueryableClient(readClient, labels.EmptyLabels(), []*labels.Matcher{}, false, stCallback),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (provider *provider) Engine() *prometheus.Engine {
|
||||
return provider.engine
|
||||
}
|
||||
|
||||
func (provider *provider) Parser() prometheus.Parser {
|
||||
return provider.parser
|
||||
}
|
||||
|
||||
func (provider *provider) Storage() storage.Queryable {
|
||||
return provider
|
||||
}
|
||||
|
||||
func (provider *provider) Querier(mint, maxt int64) (storage.Querier, error) {
|
||||
querier, err := provider.queryable.Querier(mint, maxt)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return storage.NewMergeQuerier(nil, []storage.Querier{querier}, storage.ChainedSeriesMerge), nil
|
||||
}
|
||||
|
||||
// CapturingStorage implements prometheus.StatementCapturer. Uses a fresh
|
||||
// recorder per call so concurrent dry-runs don't share state.
|
||||
func (provider *provider) CapturingStorage() (storage.Queryable, prometheus.StatementRecorder) {
|
||||
recorder := &statementRecorder{}
|
||||
capture := &captureClient{
|
||||
client: &client{settings: provider.settings, telemetryStore: provider.telemetryStore},
|
||||
recorder: recorder,
|
||||
}
|
||||
queryable := remote.NewSampleAndChunkQueryableClient(capture, labels.EmptyLabels(), []*labels.Matcher{}, false, stCallback)
|
||||
return captureQueryable{inner: queryable}, recorder
|
||||
}
|
||||
@@ -1,31 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
var _ sql.Scanner = (*scanner)(nil)
|
||||
|
||||
type scanner struct {
|
||||
f float64
|
||||
s string
|
||||
}
|
||||
|
||||
func (s *scanner) Scan(val any) error {
|
||||
s.f = 0
|
||||
s.s = ""
|
||||
|
||||
s.s = fmt.Sprintf("%v", val)
|
||||
switch val := val.(type) {
|
||||
case int64:
|
||||
s.f = float64(val)
|
||||
case uint64:
|
||||
s.f = float64(val)
|
||||
case float64:
|
||||
s.f = val
|
||||
case []byte:
|
||||
s.s = string(val)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
package clickhouseprometheus
|
||||
|
||||
import "time"
|
||||
|
||||
const (
|
||||
databaseName string = "signoz_metrics"
|
||||
distributedTimeSeriesV4 string = "distributed_time_series_v4"
|
||||
distributedTimeSeriesV46hrs string = "distributed_time_series_v4_6hrs"
|
||||
distributedTimeSeriesV41day string = "distributed_time_series_v4_1day"
|
||||
distributedSamplesV4 string = "distributed_samples_v4"
|
||||
)
|
||||
|
||||
var (
|
||||
sixHoursInMilliseconds = time.Hour.Milliseconds() * 6
|
||||
oneDayInMilliseconds = time.Hour.Milliseconds() * 24
|
||||
)
|
||||
|
||||
// Returns the start time, end time and the table name to use for the query.
|
||||
//
|
||||
// If time range is less than 6 hours, we need to use the `time_series_v4` table
|
||||
// else if time range is less than 1 day and greater than 6 hours, we need to use the `time_series_v4_6hrs` table
|
||||
// else we need to use the `time_series_v4_1day` table
|
||||
func getStartAndEndAndTableName(start, end int64) (int64, int64, string) {
|
||||
var tableName string
|
||||
|
||||
if end-start <= sixHoursInMilliseconds {
|
||||
// adjust the start time to nearest 1 hour
|
||||
start = start - (start % (time.Hour.Milliseconds() * 1))
|
||||
tableName = distributedTimeSeriesV4
|
||||
} else if end-start <= oneDayInMilliseconds {
|
||||
// adjust the start time to nearest 6 hours
|
||||
start = start - (start % (time.Hour.Milliseconds() * 6))
|
||||
tableName = distributedTimeSeriesV46hrs
|
||||
} else {
|
||||
// adjust the start time to nearest 1 day
|
||||
start = start - (start % (time.Hour.Milliseconds() * 24))
|
||||
tableName = distributedTimeSeriesV41day
|
||||
}
|
||||
|
||||
return start, end, tableName
|
||||
}
|
||||
@@ -19,11 +19,7 @@ type provider struct {
|
||||
executor *executor
|
||||
}
|
||||
|
||||
var (
|
||||
_ prometheus.Prometheus = (*provider)(nil)
|
||||
_ prometheus.StatementCapturer = (*provider)(nil)
|
||||
_ prometheus.RangeExecutor = (*provider)(nil)
|
||||
)
|
||||
var _ prometheus.Prometheus = (*provider)(nil)
|
||||
|
||||
func NewFactory(telemetryStore telemetrystore.TelemetryStore) factory.ProviderFactory[prometheus.Prometheus, prometheus.Config] {
|
||||
return factory.NewProviderFactory(factory.MustNewName("clickhousev2"), func(ctx context.Context, providerSettings factory.ProviderSettings, config prometheus.Config) (prometheus.Prometheus, error) {
|
||||
@@ -47,30 +43,69 @@ func New(_ context.Context, providerSettings factory.ProviderSettings, config pr
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *provider) TryExecuteRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
|
||||
return p.executor.TryExecuteRange(ctx, query, start, end, step)
|
||||
func (p *provider) QueryRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (*prometheus.Result, error) {
|
||||
matrix, served, err := p.executor.TryExecuteRange(ctx, query, start, end, step)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if served {
|
||||
return &prometheus.Result{Value: matrix}, nil
|
||||
}
|
||||
|
||||
qry, err := p.engine.NewRangeQuery(p.traitsContext(ctx, query), p, nil, query, start, end, step)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return finishQuery(ctx, qry)
|
||||
}
|
||||
|
||||
func (p *provider) Engine() *prometheus.Engine {
|
||||
return p.engine
|
||||
func (p *provider) Query(ctx context.Context, query string, ts time.Time) (*prometheus.Result, error) {
|
||||
qry, err := p.engine.NewInstantQuery(p.traitsContext(ctx, query), p, nil, query, ts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return finishQuery(ctx, qry)
|
||||
}
|
||||
|
||||
func (p *provider) Parser() prometheus.Parser {
|
||||
return p.parser
|
||||
// A fresh recorder per call keeps concurrent dry-runs isolated. Exec drives
|
||||
// a Select per selector (recording SQL) but reads no data.
|
||||
func (p *provider) Statements(ctx context.Context, query string, start, end time.Time, step time.Duration) ([]prometheus.CapturedStatement, error) {
|
||||
recorder := &statementRecorder{}
|
||||
capture := &captureQueryable{client: p.client, recorder: recorder}
|
||||
qry, err := p.engine.NewRangeQuery(p.traitsContext(ctx, query), capture, nil, query, start, end, step)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer qry.Close()
|
||||
if res := qry.Exec(ctx); res.Err != nil {
|
||||
return nil, res.Err
|
||||
}
|
||||
return recorder.Statements(), nil
|
||||
}
|
||||
|
||||
func (p *provider) Storage() storage.Queryable {
|
||||
return p
|
||||
// traitsContext attaches the query's traits so the storage can prove
|
||||
// step-aligned optimizations safe (see prometheus.QueryTraits). A parse
|
||||
// failure surfaces from the engine with its own error.
|
||||
func (p *provider) traitsContext(ctx context.Context, query string) context.Context {
|
||||
expr, err := p.parser.ParseExpr(query)
|
||||
if err != nil {
|
||||
return ctx
|
||||
}
|
||||
return prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
|
||||
}
|
||||
|
||||
// finishQuery packages an engine evaluation. The query is closed only on
|
||||
// error: Close returns the result's sample slices to the engine's pool, and
|
||||
// the returned Value must stay valid for the caller.
|
||||
func finishQuery(ctx context.Context, qry promql.Query) (*prometheus.Result, error) {
|
||||
res := qry.Exec(ctx)
|
||||
if res.Err != nil {
|
||||
qry.Close()
|
||||
return nil, res.Err
|
||||
}
|
||||
return &prometheus.Result{Value: res.Value, Warnings: res.Warnings, Stats: qry.Stats()}, nil
|
||||
}
|
||||
|
||||
func (p *provider) Querier(mint, maxt int64) (storage.Querier, error) {
|
||||
return &querier{mint: mint, maxt: maxt, client: p.client}, nil
|
||||
}
|
||||
|
||||
// CapturingStorage implements prometheus.StatementCapturer: a storage that
|
||||
// records each selector's SQL without executing it, for the preview path.
|
||||
// A fresh recorder per call keeps concurrent dry-runs isolated.
|
||||
func (p *provider) CapturingStorage() (storage.Queryable, prometheus.StatementRecorder) {
|
||||
recorder := &statementRecorder{}
|
||||
return &captureQueryable{client: p.client, recorder: recorder}, recorder
|
||||
}
|
||||
|
||||
@@ -24,10 +24,6 @@ type Config struct {
|
||||
|
||||
// Timeout is the maximum time a query is allowed to run before being aborted.
|
||||
Timeout time.Duration `mapstructure:"timeout"`
|
||||
|
||||
// ProviderName selects the storage provider: "clickhouse" (default) or
|
||||
// "clickhousev2".
|
||||
ProviderName string `mapstructure:"provider"`
|
||||
}
|
||||
|
||||
func NewConfigFactory() factory.ConfigFactory {
|
||||
@@ -41,8 +37,7 @@ func newConfig() factory.Config {
|
||||
Path: "",
|
||||
MaxConcurrent: 20,
|
||||
},
|
||||
Timeout: 2 * time.Minute,
|
||||
ProviderName: "clickhouse",
|
||||
Timeout: 2 * time.Minute,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,15 +45,9 @@ func (c Config) Validate() error {
|
||||
if c.Timeout <= 0 {
|
||||
return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "prometheus::timeout must be greater than 0")
|
||||
}
|
||||
if c.ProviderName != "" && c.ProviderName != "clickhouse" && c.ProviderName != "clickhousev2" {
|
||||
return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "prometheus::provider must be one of [clickhouse, clickhousev2], got %q", c.ProviderName)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c Config) Provider() string {
|
||||
if c.ProviderName == "" {
|
||||
return "clickhouse"
|
||||
}
|
||||
return c.ProviderName
|
||||
return "clickhousev2"
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
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"
|
||||
@@ -81,36 +82,8 @@ func (h *handler) QueryRange(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
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.(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
|
||||
res, err := h.prom.QueryRange(ctx, r.FormValue("query"), start, end, step)
|
||||
h.respondResult(ctx, w, r, res, err)
|
||||
}
|
||||
|
||||
// Query evaluates an expression at a single instant: query and optional
|
||||
@@ -134,35 +107,34 @@ func (h *handler) Query(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
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)
|
||||
res, err := h.prom.Query(ctx, r.FormValue("query"), ts)
|
||||
h.respondResult(ctx, w, r, res, err)
|
||||
}
|
||||
|
||||
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) {
|
||||
func (h *handler) respondResult(ctx context.Context, w http.ResponseWriter, r *http.Request, res *Result, err error) {
|
||||
if err != nil {
|
||||
h.logger.ErrorContext(ctx, "error evaluating promql query", errors.Attr(err))
|
||||
var parseErrs parser.ParseErrors
|
||||
if errors.As(err, &parseErrs) {
|
||||
h.respondError(ctx, w, errBadData, err)
|
||||
return
|
||||
}
|
||||
switch err.(type) {
|
||||
case promql.ErrQueryCanceled:
|
||||
h.respondError(ctx, w, errCanceled, res.Err)
|
||||
h.respondError(ctx, w, errCanceled, err)
|
||||
case promql.ErrQueryTimeout:
|
||||
h.respondError(ctx, w, errTimeout, res.Err)
|
||||
h.respondError(ctx, w, errTimeout, err)
|
||||
case promql.ErrStorage:
|
||||
h.respondError(ctx, w, errInternal, res.Err)
|
||||
h.respondError(ctx, w, errInternal, err)
|
||||
default:
|
||||
h.respondError(ctx, w, errExec, res.Err)
|
||||
h.respondError(ctx, w, errExec, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
data := &queryData{ResultType: res.Value.Type(), Result: res.Value}
|
||||
if r.FormValue("stats") != "" {
|
||||
data.Stats = stats.NewQueryStats(qry.Stats())
|
||||
if r.FormValue("stats") != "" && res.Stats != nil {
|
||||
data.Stats = stats.NewQueryStats(res.Stats)
|
||||
}
|
||||
warnings, infos := res.Warnings.AsStrings(r.FormValue("query"), 10, 10)
|
||||
h.respond(ctx, w, data, warnings, infos)
|
||||
|
||||
@@ -6,7 +6,8 @@ import (
|
||||
|
||||
"github.com/prometheus/prometheus/promql"
|
||||
"github.com/prometheus/prometheus/promql/parser"
|
||||
"github.com/prometheus/prometheus/storage"
|
||||
"github.com/prometheus/prometheus/util/annotations"
|
||||
"github.com/prometheus/prometheus/util/stats"
|
||||
)
|
||||
|
||||
type Engine = promql.Engine
|
||||
@@ -14,9 +15,25 @@ type Engine = promql.Engine
|
||||
type Parser = parser.Parser
|
||||
|
||||
type Prometheus interface {
|
||||
Engine() *Engine
|
||||
Storage() storage.Queryable
|
||||
Parser() Parser
|
||||
// QueryRange evaluates a range query: inside the datastore when the
|
||||
// query's shape allows it, else in the engine over the provider's
|
||||
// storage, which is always exact.
|
||||
QueryRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (*Result, error)
|
||||
|
||||
// Query evaluates an instant query in the engine.
|
||||
Query(ctx context.Context, query string, ts time.Time) (*Result, error)
|
||||
|
||||
// Statements returns the datastore statements the engine path of a
|
||||
// range query would run, captured without executing them.
|
||||
Statements(ctx context.Context, query string, start, end time.Time, step time.Duration) ([]CapturedStatement, error)
|
||||
}
|
||||
|
||||
// Result is one evaluation's outcome. The caller owns Value: the provider
|
||||
// never returns its sample slices to the engine's pools.
|
||||
type Result struct {
|
||||
Value parser.Value
|
||||
Warnings annotations.Annotations
|
||||
Stats *stats.Statistics
|
||||
}
|
||||
|
||||
// CapturedStatement is one datastore statement a PromQL query would run,
|
||||
@@ -25,33 +42,3 @@ type CapturedStatement struct {
|
||||
Query string
|
||||
Args []any
|
||||
}
|
||||
|
||||
// StatementRecorder reads back the statements captured against a capturing
|
||||
// Storage (see StatementCapturer).
|
||||
type StatementRecorder interface {
|
||||
Statements() []CapturedStatement
|
||||
}
|
||||
|
||||
// StatementCapturer is an optional Prometheus-provider capability, discovered
|
||||
// via type assertion: it returns a Storage that records each Select's statement
|
||||
// without executing it, plus a recorder to read them back.
|
||||
type StatementCapturer interface {
|
||||
CapturingStorage() (storage.Queryable, StatementRecorder)
|
||||
}
|
||||
|
||||
// ProviderClickhouseV2 is the clickhousev2 provider name: the factory
|
||||
// registration, the prometheus::provider config value and the
|
||||
// X-SigNoz-PromQL-Provider request header all use it, so they cannot drift
|
||||
// apart.
|
||||
const ProviderClickhouseV2 = "clickhousev2"
|
||||
|
||||
// RangeExecutor is the optional capability of a provider that can evaluate
|
||||
// some range queries entirely inside the datastore. ok=false means the
|
||||
// query is not evaluable that way. The caller then runs the engine over the
|
||||
// provider's Storage, which is always exact. Only the clickhousev2 provider
|
||||
// implements this capability. When that provider is the only one, the
|
||||
// capability folds into Prometheus itself, and the engine-vs-datastore
|
||||
// decision becomes internal.
|
||||
type RangeExecutor interface {
|
||||
TryExecuteRange(ctx context.Context, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error)
|
||||
}
|
||||
|
||||
31
pkg/prometheus/prometheustest/grid.go
Normal file
31
pkg/prometheus/prometheustest/grid.go
Normal file
@@ -0,0 +1,31 @@
|
||||
package prometheustest
|
||||
|
||||
import (
|
||||
cmock "github.com/SigNoz/clickhouse-go-mock"
|
||||
)
|
||||
|
||||
// GridCols is the result shape of a transpiled unit statement.
|
||||
var GridCols = []cmock.ColumnType{
|
||||
{Name: "gkey", Type: "String"},
|
||||
{Name: "grid", Type: "Array(Nullable(Float64))"},
|
||||
}
|
||||
|
||||
// LastSampleGrid builds the grid a transpiled instant selector returns: per
|
||||
// slot t, the latest sample in the left-open lookback window (t-lookback, t].
|
||||
func LastSampleGrid(tsMs []int64, values []float64, startMs, endMs, stepMs, lookbackMs int64) []*float64 {
|
||||
grid := make([]*float64, (endMs-startMs)/stepMs+1)
|
||||
for i := range grid {
|
||||
slot := startMs + int64(i)*stepMs
|
||||
best := -1
|
||||
for j, ts := range tsMs {
|
||||
if ts > slot-lookbackMs && ts <= slot && (best == -1 || ts >= tsMs[best]) {
|
||||
best = j
|
||||
}
|
||||
}
|
||||
if best >= 0 {
|
||||
v := values[best]
|
||||
grid[i] = &v
|
||||
}
|
||||
}
|
||||
return grid
|
||||
}
|
||||
@@ -5,58 +5,16 @@ import (
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/prometheus/common/model"
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
"github.com/prometheus/prometheus/storage"
|
||||
"github.com/prometheus/prometheus/storage/remote"
|
||||
)
|
||||
|
||||
var _ prometheus.Prometheus = (*Provider)(nil)
|
||||
|
||||
type Provider struct {
|
||||
queryable storage.SampleAndChunkQueryable
|
||||
engine *prometheus.Engine
|
||||
parser prometheus.Parser
|
||||
}
|
||||
|
||||
var stCallback = func() (int64, error) {
|
||||
return int64(model.Latest), nil
|
||||
}
|
||||
|
||||
func New(ctx context.Context, providerSettings factory.ProviderSettings, config prometheus.Config, telemetryStore telemetrystore.TelemetryStore) *Provider {
|
||||
|
||||
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/prometheus/prometheustest")
|
||||
|
||||
engine := prometheus.NewEngine(settings.Logger(), config)
|
||||
|
||||
readClient := clickhouseprometheus.NewReadClient(settings, telemetryStore)
|
||||
|
||||
queryable := remote.NewSampleAndChunkQueryableClient(readClient, labels.EmptyLabels(), []*labels.Matcher{}, false, stCallback)
|
||||
|
||||
return &Provider{
|
||||
engine: engine,
|
||||
parser: prometheus.NewParser(),
|
||||
queryable: queryable,
|
||||
// New returns the clickhousev2 provider over the given telemetry store, so
|
||||
// tests exercise the production read path against a mock store.
|
||||
func New(ctx context.Context, providerSettings factory.ProviderSettings, config prometheus.Config, telemetryStore telemetrystore.TelemetryStore) prometheus.Prometheus {
|
||||
provider, err := clickhouseprometheusv2.New(ctx, providerSettings, config, telemetryStore)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (provider *Provider) Engine() *prometheus.Engine {
|
||||
return provider.engine
|
||||
}
|
||||
|
||||
func (provider *Provider) Storage() storage.Queryable {
|
||||
return provider.queryable
|
||||
}
|
||||
|
||||
func (provider *Provider) Parser() prometheus.Parser {
|
||||
return provider.parser
|
||||
}
|
||||
|
||||
func (provider *Provider) Close() error {
|
||||
if provider.engine != nil {
|
||||
provider.engine.Close()
|
||||
}
|
||||
return nil
|
||||
return provider
|
||||
}
|
||||
|
||||
@@ -57,7 +57,6 @@ func (handler *handler) QueryRange(rw http.ResponseWriter, req *http.Request) {
|
||||
render.Error(rw, err)
|
||||
return
|
||||
}
|
||||
queryRangeRequest.PromQLProvider = req.Header.Get("X-SigNoz-PromQL-Provider")
|
||||
|
||||
// Validate the query request
|
||||
if err := queryRangeRequest.Validate(); err != nil {
|
||||
|
||||
@@ -231,7 +231,7 @@ func (q *querier) buildPreviewProviders(
|
||||
sub.CompositeQuery = qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{query}}
|
||||
}
|
||||
|
||||
built, _, bErr := q.buildQueries(orgID, &sub, deps, missingMetricQuerySet, event, promqlOptions{})
|
||||
built, _, bErr := q.buildQueries(orgID, &sub, deps, missingMetricQuerySet, event)
|
||||
if bErr != nil {
|
||||
errs[name] = bErr
|
||||
continue
|
||||
|
||||
@@ -102,24 +102,6 @@ type promqlQuery struct {
|
||||
tr qbv5.TimeRange
|
||||
requestType qbv5.RequestType
|
||||
vars map[string]qbv5.VariableItem
|
||||
opts promqlOptions
|
||||
}
|
||||
|
||||
// promqlOptions is how a PromQL query relates to the clickhousev2 provider
|
||||
// (see querier.promqlOptions for where the fields come from and why they are
|
||||
// flag-gated). Both providers are nil for a plain request, so a plain
|
||||
// request costs nothing extra.
|
||||
type promqlOptions struct {
|
||||
// shadow, when set, runs the query on this provider after serving and
|
||||
// logs any result difference; the response is never affected.
|
||||
shadow prometheus.Prometheus
|
||||
// shadowSlots is the querier-wide admission for shadow runs, shared by
|
||||
// every query so the bound holds per process.
|
||||
shadowSlots chan struct{}
|
||||
// serve, when set, serves the response from this provider instead of the
|
||||
// default path. Comparison callers fetch the default and the pinned
|
||||
// result as two API calls and diff them.
|
||||
serve prometheus.Prometheus
|
||||
}
|
||||
|
||||
var _ qbv5.Query = (*promqlQuery)(nil)
|
||||
@@ -132,29 +114,19 @@ func newPromqlQuery(
|
||||
tr qbv5.TimeRange,
|
||||
requestType qbv5.RequestType,
|
||||
variables map[string]qbv5.VariableItem,
|
||||
opts promqlOptions,
|
||||
) *promqlQuery {
|
||||
return &promqlQuery{
|
||||
logger: logger,
|
||||
promEngine: promEngine,
|
||||
parser: promEngine.Parser(),
|
||||
parser: prometheus.NewParser(),
|
||||
query: query,
|
||||
tr: tr,
|
||||
requestType: requestType,
|
||||
vars: variables,
|
||||
opts: opts,
|
||||
}
|
||||
}
|
||||
|
||||
func (q *promqlQuery) Fingerprint() string {
|
||||
// A pinned request must not share cache entries with default serving: a
|
||||
// cached default result would satisfy the pin without running the pinned
|
||||
// provider, and a pinned result would poison normal serving. No
|
||||
// fingerprint means no caching at all — the pin exists to observe a
|
||||
// provider, so a cache in front of it defeats the point.
|
||||
if q.opts.serve != nil {
|
||||
return ""
|
||||
}
|
||||
if q.requestType != qbv5.RequestTypeTimeSeries {
|
||||
return ""
|
||||
}
|
||||
@@ -267,15 +239,9 @@ func (q *promqlQuery) Statement(_ context.Context) (*qbv5.Statement, error) {
|
||||
return &qbv5.Statement{Query: rendered}, nil
|
||||
}
|
||||
|
||||
// PreviewStatements returns the ClickHouse statement(s) this PromQL query would
|
||||
// run, captured by driving the engine with a Storage that records each selector's
|
||||
// SQL and returns no data. Returns nil if capture is unsupported.
|
||||
// PreviewStatements returns the ClickHouse statement(s) this PromQL query
|
||||
// would run on the engine path, captured without executing them.
|
||||
func (q *promqlQuery) PreviewStatements(ctx context.Context) ([]prometheus.CapturedStatement, error) {
|
||||
storer, ok := q.promEngine.(prometheus.StatementCapturer)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
rendered, err := q.renderVars(q.query.Query, q.vars, q.tr.From, q.tr.To)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -284,42 +250,11 @@ func (q *promqlQuery) PreviewStatements(ctx context.Context) ([]prometheus.Captu
|
||||
start := int64(querybuilder.ToNanoSecs(q.tr.From))
|
||||
end := int64(querybuilder.ToNanoSecs(q.tr.To))
|
||||
|
||||
// Attach the same query traits as Execute so the captured statements
|
||||
// match what the live path would run.
|
||||
if expr, parseErr := q.parser.ParseExpr(rendered); parseErr == nil {
|
||||
ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
|
||||
}
|
||||
|
||||
capStorage, recorder := storer.CapturingStorage()
|
||||
if capStorage == nil {
|
||||
return nil, nil
|
||||
}
|
||||
qry, err := q.promEngine.Engine().NewRangeQuery(
|
||||
ctx,
|
||||
capStorage,
|
||||
nil,
|
||||
rendered,
|
||||
time.Unix(0, start),
|
||||
time.Unix(0, end),
|
||||
q.query.Step.Duration,
|
||||
)
|
||||
statements, err := q.promEngine.Statements(ctx, rendered, time.Unix(0, start), time.Unix(0, end), q.query.Step.Duration)
|
||||
if err != nil {
|
||||
if e := tryEnhancePromQLExecError(err); e != nil {
|
||||
return nil, e
|
||||
}
|
||||
return nil, enhancePromQLError(rendered, err)
|
||||
return nil, q.evalError(rendered, err)
|
||||
}
|
||||
defer qry.Close()
|
||||
|
||||
// Exec drives a Select per selector (recording SQL) but reads no data.
|
||||
if res := qry.Exec(ctx); res.Err != nil {
|
||||
if e := tryEnhancePromQLExecError(res.Err); e != nil {
|
||||
return nil, e
|
||||
}
|
||||
return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "query execution error: %v", res.Err)
|
||||
}
|
||||
|
||||
return recorder.Statements(), nil
|
||||
return statements, nil
|
||||
}
|
||||
|
||||
func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
|
||||
@@ -337,13 +272,6 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Attach query traits so the storage can prove step-aligned optimizations
|
||||
// safe (see prometheus.QueryTraits). A parse failure surfaces below via
|
||||
// the engine with the enhanced error message.
|
||||
if expr, parseErr := q.parser.ParseExpr(query); parseErr == nil {
|
||||
ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
|
||||
}
|
||||
|
||||
// Accumulate ClickHouse-side scan stats across every storage query this
|
||||
// evaluation issues (engine selectors or the compiled executor): progress
|
||||
// options propagate to each ClickHouse query through the context.
|
||||
@@ -358,97 +286,37 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
|
||||
|
||||
began := time.Now()
|
||||
|
||||
// A pinned provider serves directly from it: comparison callers fetch
|
||||
// the default result and the pinned result as two API calls and diff
|
||||
// them.
|
||||
if q.opts.serve != nil {
|
||||
matrix, err := q.serveFromProvider(ctx, query, start, end)
|
||||
if err != nil {
|
||||
if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
|
||||
return nil, enhanced
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return q.toResult(matrix, nil, began, &statsMu, &rowsScanned, &bytesScanned), nil
|
||||
}
|
||||
|
||||
// When the serving provider has the RangeExecutor capability
|
||||
// (prometheus::provider: clickhousev2), serve the way the provider is
|
||||
// designed to serve: transpiled when the shape allows. Without this the
|
||||
// override would silently run the engine path only.
|
||||
if re, ok := q.promEngine.(prometheus.RangeExecutor); ok {
|
||||
matrix, served, err := re.TryExecuteRange(ctx, query, time.Unix(0, start), time.Unix(0, end), q.query.Step.Duration)
|
||||
if err != nil {
|
||||
if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
|
||||
return nil, enhanced
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if served {
|
||||
return q.toResult(matrix, nil, began, &statsMu, &rowsScanned, &bytesScanned), nil
|
||||
}
|
||||
}
|
||||
|
||||
qry, err := q.promEngine.Engine().NewRangeQuery(
|
||||
ctx,
|
||||
q.promEngine.Storage(),
|
||||
nil,
|
||||
query,
|
||||
time.Unix(0, start),
|
||||
time.Unix(0, end),
|
||||
q.query.Step.Duration,
|
||||
)
|
||||
res, err := q.promEngine.QueryRange(ctx, query, time.Unix(0, start), time.Unix(0, end), q.query.Step.Duration)
|
||||
if err != nil {
|
||||
// NewRangeQuery can fail with execution errors (e.g. context deadline exceeded)
|
||||
// during the query queue/scheduling stage, not just parse errors.
|
||||
if err := tryEnhancePromQLExecError(err); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return nil, enhancePromQLError(query, err)
|
||||
return nil, q.evalError(query, err)
|
||||
}
|
||||
|
||||
res := qry.Exec(ctx)
|
||||
if res.Err != nil {
|
||||
if err := tryEnhancePromQLExecError(res.Err); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "query execution error: %v", res.Err)
|
||||
}
|
||||
|
||||
defer qry.Close()
|
||||
|
||||
matrix, promErr := res.Matrix()
|
||||
if promErr != nil {
|
||||
return nil, errors.WrapInternalf(promErr, errors.CodeInternal, "error getting matrix from promql query %q", query)
|
||||
}
|
||||
|
||||
if q.opts.shadow != nil {
|
||||
// Shadows detach from the request, so without admission a dashboard
|
||||
// burst would stack unbounded ClickHouse work for up to the shadow
|
||||
// timeout — the concurrency pattern behind the original outages.
|
||||
// Non-blocking: at the cap the comparison is skipped, not queued;
|
||||
// a sampled shadow stream is exactly as useful for rollout evidence.
|
||||
select {
|
||||
case q.opts.shadowSlots <- struct{}{}:
|
||||
// The engine pools the result's sample slices on Close; the
|
||||
// shadow comparison needs a stable copy of what was served.
|
||||
served := copyMatrix(matrix)
|
||||
servedIn := time.Since(began)
|
||||
go func() {
|
||||
defer func() { <-q.opts.shadowSlots }()
|
||||
q.runShadowCompare(context.WithoutCancel(ctx), query, start, end, served, servedIn)
|
||||
}()
|
||||
default:
|
||||
q.logger.DebugContext(ctx, "promql shadow skipped: at concurrency cap", slog.String("query", query))
|
||||
}
|
||||
matrix, ok := res.Value.(promql.Matrix)
|
||||
if !ok {
|
||||
return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "promql query %q returned %T, expected a matrix", query, res.Value)
|
||||
}
|
||||
|
||||
warnings, _ := res.Warnings.AsStrings(query, 10, 0)
|
||||
return q.toResult(matrix, warnings, began, &statsMu, &rowsScanned, &bytesScanned), nil
|
||||
}
|
||||
|
||||
// evalError types an evaluation error: engine execution classes first, then
|
||||
// parse errors with the migration hints, everything else internal.
|
||||
func (q *promqlQuery) evalError(query string, err error) error {
|
||||
if enhanced := tryEnhancePromQLExecError(err); enhanced != nil {
|
||||
return enhanced
|
||||
}
|
||||
var parseErrs parser.ParseErrors
|
||||
if errors.As(err, &parseErrs) {
|
||||
return enhancePromQLError(query, err)
|
||||
}
|
||||
// The transpiled path raises typed user errors of its own; keep them.
|
||||
if errors.Ast(err, errors.TypeInvalidInput) {
|
||||
return err
|
||||
}
|
||||
return errors.Newf(errors.TypeInternal, errors.CodeInternal, "query execution error: %v", err)
|
||||
}
|
||||
|
||||
// toResult converts an evaluated matrix into the v5 result shape, attaching
|
||||
// the ClickHouse scan stats accumulated during evaluation.
|
||||
func (q *promqlQuery) toResult(matrix promql.Matrix, warnings []string, began time.Time, statsMu *sync.Mutex, rowsScanned, bytesScanned *uint64) *qbv5.Result {
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/prometheustest"
|
||||
qbv5 "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -449,18 +448,6 @@ func TestQuotedMetricOutsideBracesPattern(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A pinned request must not share cache entries with default serving: a
|
||||
// cached default result would satisfy the pin without running the pinned
|
||||
// provider.
|
||||
func TestFingerprint_PinnedProviderBypassesCache(t *testing.T) {
|
||||
q := &promqlQuery{
|
||||
logger: slog.Default(),
|
||||
query: qbv5.PromQuery{Query: "up"},
|
||||
opts: promqlOptions{serve: &prometheustest.Provider{}},
|
||||
}
|
||||
assert.Empty(t, q.Fingerprint())
|
||||
}
|
||||
|
||||
func TestToResultDropsNonFiniteValues(t *testing.T) {
|
||||
tests := []struct {
|
||||
description string
|
||||
|
||||
@@ -1,183 +0,0 @@
|
||||
package querier
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
"github.com/prometheus/prometheus/promql"
|
||||
)
|
||||
|
||||
// shadowTimeout bounds a shadow evaluation; a shadow run must never outlive
|
||||
// the request by much or pile up.
|
||||
const shadowTimeout = 2 * time.Minute
|
||||
|
||||
// runShadowCompare executes the query on the clickhousev2 provider exactly
|
||||
// as it would serve (transpiled when the shape allows, engine over the v2
|
||||
// querier otherwise), compares against the served result and logs the
|
||||
// outcome. Serving is never affected: this runs after the response, off the
|
||||
// request context, and only logs. The mismatch and failure logs are the
|
||||
// rollout evidence — serving cuts over to v2 only after they stay clean.
|
||||
func (q *promqlQuery) runShadowCompare(ctx context.Context, query string, startNs, endNs int64, served promql.Matrix, servedIn time.Duration) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
q.logger.ErrorContext(ctx, "promql shadow comparison panicked", slog.Any("panic", r), slog.String("query", query))
|
||||
}
|
||||
}()
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, shadowTimeout)
|
||||
defer cancel()
|
||||
|
||||
// The request context carries the served response's scan-stats progress
|
||||
// callback; without replacing it the shadow's ClickHouse progress would
|
||||
// race into the served stats. The response itself was already sent.
|
||||
ctx = clickhouse.Context(ctx, clickhouse.WithProgress(func(*clickhouse.Progress) {}))
|
||||
|
||||
if expr, parseErr := q.parser.ParseExpr(query); parseErr == nil {
|
||||
ctx = prometheus.NewContextWithQueryTraits(ctx, prometheus.DetectQueryTraits(expr))
|
||||
}
|
||||
|
||||
start, end := time.Unix(0, startNs), time.Unix(0, endNs)
|
||||
began := time.Now()
|
||||
shadow, transpiled, err := executeOnProvider(ctx, q.opts.shadow, query, start, end, q.query.Step.Duration)
|
||||
shadowIn := time.Since(began)
|
||||
|
||||
logAttrs := []any{
|
||||
slog.String("query", query),
|
||||
slog.Int64("start_ms", startNs/int64(time.Millisecond)),
|
||||
slog.Int64("end_ms", endNs/int64(time.Millisecond)),
|
||||
slog.Duration("step", q.query.Step.Duration),
|
||||
slog.Bool("transpiled", transpiled),
|
||||
slog.Duration("served_in", servedIn),
|
||||
slog.Duration("shadow_in", shadowIn),
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
// A shadow failure would be a serving failure after rollout; surface
|
||||
// it at the same level as a result mismatch.
|
||||
q.logger.WarnContext(ctx, "promql shadow execution failed", append(logAttrs, slog.Any("error", err))...)
|
||||
return
|
||||
}
|
||||
|
||||
servedNorm := normalizeShadowMatrix(served)
|
||||
shadowNorm := normalizeShadowMatrix(shadow)
|
||||
if diff := diffShadowMatrices(servedNorm, shadowNorm); diff != "" {
|
||||
q.logger.WarnContext(ctx, "promql shadow comparison mismatch", append(logAttrs,
|
||||
slog.String("diff", diff),
|
||||
slog.Int("served_series", len(servedNorm)),
|
||||
slog.Int("shadow_series", len(shadowNorm)),
|
||||
)...)
|
||||
return
|
||||
}
|
||||
// Matches log the timings: served_in vs shadow_in across the fleet is
|
||||
// the perf evidence for the cutover, gathered for free.
|
||||
q.logger.DebugContext(ctx, "promql shadow comparison matched", logAttrs...)
|
||||
}
|
||||
|
||||
func (q *promqlQuery) serveFromProvider(ctx context.Context, query string, startNs, endNs int64) (promql.Matrix, error) {
|
||||
matrix, _, err := executeOnProvider(ctx, q.opts.serve, query, time.Unix(0, startNs), time.Unix(0, endNs), q.query.Step.Duration)
|
||||
return matrix, err
|
||||
}
|
||||
|
||||
// The returned matrix is an owned copy.
|
||||
func executeOnProvider(ctx context.Context, prov prometheus.Prometheus, query string, start, end time.Time, step time.Duration) (promql.Matrix, bool, error) {
|
||||
if re, ok := prov.(prometheus.RangeExecutor); ok {
|
||||
matrix, served, err := re.TryExecuteRange(ctx, query, start, end, step)
|
||||
if err != nil {
|
||||
return nil, true, err
|
||||
}
|
||||
if served {
|
||||
return matrix, true, nil
|
||||
}
|
||||
}
|
||||
|
||||
qry, err := prov.Engine().NewRangeQuery(ctx, prov.Storage(), nil, query, start, end, step)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
defer qry.Close()
|
||||
|
||||
res := qry.Exec(ctx)
|
||||
if res.Err != nil {
|
||||
return nil, false, res.Err
|
||||
}
|
||||
matrix, err := res.Matrix()
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
// Close returns the result's sample slices to the engine pool.
|
||||
return copyMatrix(matrix), false, nil
|
||||
}
|
||||
|
||||
func copyMatrix(matrix promql.Matrix) promql.Matrix {
|
||||
out := make(promql.Matrix, 0, len(matrix))
|
||||
for _, s := range matrix {
|
||||
floats := make([]promql.FPoint, len(s.Floats))
|
||||
copy(floats, s.Floats)
|
||||
out = append(out, promql.Series{Metric: s.Metric.Copy(), Floats: floats})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// normalizeShadowMatrix sorts by label set for order-independent
|
||||
// comparison. Both providers now resolve series identity the same way
|
||||
// (empty-valued labels dropped at read, no synthetic fingerprint label
|
||||
// since the v1 series-identity fix), so labels need no normalization.
|
||||
func normalizeShadowMatrix(matrix promql.Matrix) promql.Matrix {
|
||||
out := make(promql.Matrix, 0, len(matrix))
|
||||
out = append(out, matrix...)
|
||||
sort.Slice(out, func(i, j int) bool { return labels.Compare(out[i].Metric, out[j].Metric) < 0 })
|
||||
return out
|
||||
}
|
||||
|
||||
// diffShadowMatrices returns a description of the first difference, or "".
|
||||
// Values compare with relative tolerance: spatial aggregations accumulate
|
||||
// floats in storage order, which differs between the providers in the last
|
||||
// ULP.
|
||||
func diffShadowMatrices(served, shadow promql.Matrix) string {
|
||||
const relTol = 1e-9
|
||||
if len(served) != len(shadow) {
|
||||
return fmt.Sprintf("series count: served=%d shadow=%d", len(served), len(shadow))
|
||||
}
|
||||
for i := range served {
|
||||
if labels.Compare(served[i].Metric, shadow[i].Metric) != 0 {
|
||||
return fmt.Sprintf("series %d labels: served=%s shadow=%s", i, served[i].Metric, shadow[i].Metric)
|
||||
}
|
||||
if len(served[i].Floats) != len(shadow[i].Floats) {
|
||||
return fmt.Sprintf("series %s points: served=%d shadow=%d", served[i].Metric, len(served[i].Floats), len(shadow[i].Floats))
|
||||
}
|
||||
for j := range served[i].Floats {
|
||||
a, b := served[i].Floats[j], shadow[i].Floats[j]
|
||||
if a.T != b.T {
|
||||
return fmt.Sprintf("series %s point %d ts: served=%d shadow=%d", served[i].Metric, j, a.T, b.T)
|
||||
}
|
||||
// NaN and infinities first: NaN != NaN and Inf-Inf arithmetic
|
||||
// would otherwise make one-sided NaN and Inf-vs-finite compare
|
||||
// as equal (NaN > x and Inf > Inf are both false).
|
||||
if math.IsNaN(a.F) || math.IsNaN(b.F) {
|
||||
if math.IsNaN(a.F) != math.IsNaN(b.F) {
|
||||
return fmt.Sprintf("series %s @%d value: served=%v shadow=%v", served[i].Metric, a.T, a.F, b.F)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if math.IsInf(a.F, 0) || math.IsInf(b.F, 0) {
|
||||
if a.F != b.F {
|
||||
return fmt.Sprintf("series %s @%d value: served=%v shadow=%v", served[i].Metric, a.T, a.F, b.F)
|
||||
}
|
||||
continue
|
||||
}
|
||||
diff := math.Abs(a.F - b.F)
|
||||
scale := math.Max(math.Abs(a.F), math.Abs(b.F))
|
||||
if diff > relTol*math.Max(scale, 1e-300) && diff > 1e-12 {
|
||||
return fmt.Sprintf("series %s @%d value: served=%v shadow=%v", served[i].Metric, a.T, a.F, b.F)
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -1,67 +0,0 @@
|
||||
package querier
|
||||
|
||||
import (
|
||||
"math"
|
||||
"testing"
|
||||
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
"github.com/prometheus/prometheus/promql"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestNormalizeShadowMatrix(t *testing.T) {
|
||||
matrix := promql.Matrix{
|
||||
{
|
||||
Metric: labels.FromStrings("__name__", "up", "job", "api"),
|
||||
Floats: []promql.FPoint{{T: 1000, F: 1}},
|
||||
},
|
||||
{
|
||||
Metric: labels.FromStrings("a", "1"),
|
||||
Floats: []promql.FPoint{{T: 1000, F: 2}},
|
||||
},
|
||||
}
|
||||
norm := normalizeShadowMatrix(matrix)
|
||||
// sorted by label set; labels pass through untouched — both providers
|
||||
// resolve series identity identically since the v1 series-identity fix
|
||||
assert.Equal(t, labels.FromStrings("__name__", "up", "job", "api"), norm[0].Metric)
|
||||
assert.Equal(t, labels.FromStrings("a", "1"), norm[1].Metric)
|
||||
}
|
||||
|
||||
func TestDiffShadowMatrices(t *testing.T) {
|
||||
series := func(v float64) promql.Matrix {
|
||||
return promql.Matrix{{Metric: labels.FromStrings("a", "1"), Floats: []promql.FPoint{{T: 1000, F: v}}}}
|
||||
}
|
||||
|
||||
assert.Empty(t, diffShadowMatrices(series(1.5), series(1.5)))
|
||||
// last-ULP differences from storage-order float accumulation are expected
|
||||
assert.Empty(t, diffShadowMatrices(series(0.08888888888888889), series(0.08888888888888888)))
|
||||
assert.Empty(t, diffShadowMatrices(series(math.NaN()), series(math.NaN())))
|
||||
|
||||
assert.Contains(t, diffShadowMatrices(series(1.5), series(1.6)), "value")
|
||||
assert.Contains(t, diffShadowMatrices(series(1.5), promql.Matrix{}), "series count")
|
||||
assert.Contains(t, diffShadowMatrices(
|
||||
series(1.5),
|
||||
promql.Matrix{{Metric: labels.FromStrings("a", "2"), Floats: []promql.FPoint{{T: 1000, F: 1.5}}}},
|
||||
), "labels")
|
||||
assert.Contains(t, diffShadowMatrices(
|
||||
series(1.5),
|
||||
promql.Matrix{{Metric: labels.FromStrings("a", "1"), Floats: []promql.FPoint{{T: 2000, F: 1.5}}}},
|
||||
), "ts")
|
||||
}
|
||||
|
||||
// One-sided NaN makes every float comparison false, and Inf-Inf arithmetic
|
||||
// yields Inf > Inf == false; without explicit handling both divergences log
|
||||
// as matched — a shadow comparator that cannot see them would green-light a
|
||||
// broken rollout.
|
||||
func TestDiffShadowMatrices_SpecialFloats(t *testing.T) {
|
||||
point := func(v float64) promql.Matrix {
|
||||
return promql.Matrix{{Metric: labels.FromStrings("a", "1"), Floats: []promql.FPoint{{T: 1000, F: v}}}}
|
||||
}
|
||||
|
||||
assert.NotEmpty(t, diffShadowMatrices(point(math.NaN()), point(1.5)), "one-sided NaN must diff")
|
||||
assert.NotEmpty(t, diffShadowMatrices(point(1.5), point(math.NaN())), "one-sided NaN must diff either way")
|
||||
assert.NotEmpty(t, diffShadowMatrices(point(math.Inf(1)), point(1.5)), "Inf vs finite must diff")
|
||||
assert.NotEmpty(t, diffShadowMatrices(point(math.Inf(1)), point(math.Inf(-1))), "opposite infinities must diff")
|
||||
assert.Empty(t, diffShadowMatrices(point(math.Inf(1)), point(math.Inf(1))), "equal infinities match")
|
||||
assert.Empty(t, diffShadowMatrices(point(math.NaN()), point(math.NaN())), "both NaN match")
|
||||
}
|
||||
@@ -24,7 +24,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/statsreporter"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/featuretypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
@@ -47,19 +46,11 @@ type Querier interface {
|
||||
}
|
||||
|
||||
type querier struct {
|
||||
logger *slog.Logger
|
||||
fl flagger.Flagger
|
||||
telemetryStore telemetrystore.TelemetryStore
|
||||
metadataStore telemetrytypes.MetadataStore
|
||||
promEngine prometheus.Prometheus
|
||||
// promV2 is the clickhousev2 prometheus provider, wired only when the
|
||||
// serving provider is the default one (nil otherwise). It reads the same
|
||||
// ClickHouse data through a different implementation; PromQL queries
|
||||
// shadow-compare against it behind the use_prometheus_clickhouse_v2 flag
|
||||
// and can be pinned to it for a response (see promqlOptions). It never
|
||||
// serves by default — that cutover happens only after the shadow logs
|
||||
// stay clean.
|
||||
promV2 prometheus.Prometheus
|
||||
logger *slog.Logger
|
||||
fl flagger.Flagger
|
||||
telemetryStore telemetrystore.TelemetryStore
|
||||
metadataStore telemetrytypes.MetadataStore
|
||||
promEngine prometheus.Prometheus
|
||||
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
|
||||
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
|
||||
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
|
||||
@@ -71,16 +62,8 @@ type querier struct {
|
||||
liveDataRefresh time.Duration
|
||||
builderConfig builderConfig
|
||||
maxConcurrentQueries int
|
||||
// shadowSlots bounds concurrent shadow comparisons per process; shadows
|
||||
// detach from their requests, so nothing else limits how many pile up.
|
||||
shadowSlots chan struct{}
|
||||
}
|
||||
|
||||
// maxConcurrentShadows is deliberately small: a shadow is a full extra
|
||||
// ClickHouse evaluation, and a sampled stream of comparisons is exactly as
|
||||
// useful for rollout evidence as an exhaustive one under load.
|
||||
const maxConcurrentShadows = 8
|
||||
|
||||
var _ Querier = (*querier)(nil)
|
||||
|
||||
func New(
|
||||
@@ -88,7 +71,6 @@ func New(
|
||||
telemetryStore telemetrystore.TelemetryStore,
|
||||
metadataStore telemetrytypes.MetadataStore,
|
||||
promEngine prometheus.Prometheus,
|
||||
promV2 prometheus.Prometheus,
|
||||
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
@@ -111,7 +93,6 @@ func New(
|
||||
telemetryStore: telemetryStore,
|
||||
metadataStore: metadataStore,
|
||||
promEngine: promEngine,
|
||||
promV2: promV2,
|
||||
traceStmtBuilder: traceStmtBuilder,
|
||||
aiTraceStmtBuilder: aiTraceStmtBuilder,
|
||||
logStmtBuilder: logStmtBuilder,
|
||||
@@ -125,7 +106,6 @@ func New(
|
||||
logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
|
||||
},
|
||||
maxConcurrentQueries: maxConcurrentQueries,
|
||||
shadowSlots: make(chan struct{}, maxConcurrentShadows),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -165,11 +145,7 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
|
||||
missingMetricQuerySet[name] = true
|
||||
}
|
||||
|
||||
promqlOpts, err := q.promqlOptions(ctx, orgID, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
queries, steps, err := q.buildQueries(orgID, req, dependencyQueries, missingMetricQuerySet, event, promqlOpts)
|
||||
queries, steps, err := q.buildQueries(orgID, req, dependencyQueries, missingMetricQuerySet, event)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -212,41 +188,12 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
|
||||
return qbResp, qbErr
|
||||
}
|
||||
|
||||
// promqlOptions derives the PromQL execution options for a request. With the
|
||||
// org's use_prometheus_clickhouse_v2 flag on, queries are shadow-compared
|
||||
// against the clickhousev2 provider (serving unaffected, diffs logged; see
|
||||
// promql_shadow.go). The X-SigNoz-PromQL-Provider header may instead pin the
|
||||
// response to that provider — integration tests and support fetch both
|
||||
// results for comparison — so it is deliberately flag-gated too: without the
|
||||
// gate the header would be an unaudited switch onto a provider still under
|
||||
// validation.
|
||||
func (q *querier) promqlOptions(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (promqlOptions, error) {
|
||||
enabled := q.fl.BooleanOrEmpty(ctx, flagger.FeatureUsePrometheusClickhouseV2, featuretypes.NewFlaggerEvaluationContext(orgID))
|
||||
if req.PromQLProvider == "" {
|
||||
if enabled && q.promV2 != nil {
|
||||
return promqlOptions{shadow: q.promV2, shadowSlots: q.shadowSlots}, nil
|
||||
}
|
||||
return promqlOptions{}, nil
|
||||
}
|
||||
if req.PromQLProvider != prometheus.ProviderClickhouseV2 {
|
||||
return promqlOptions{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown promql provider %q", req.PromQLProvider)
|
||||
}
|
||||
if !enabled {
|
||||
return promqlOptions{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "promql provider %q requires the use_prometheus_clickhouse_v2 flag", req.PromQLProvider)
|
||||
}
|
||||
if q.promV2 == nil {
|
||||
return promqlOptions{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "promql provider %q is not available", req.PromQLProvider)
|
||||
}
|
||||
return promqlOptions{serve: q.promV2}, nil
|
||||
}
|
||||
|
||||
func (q *querier) buildQueries(
|
||||
orgID valuer.UUID,
|
||||
req *qbtypes.QueryRangeRequest,
|
||||
dependencyQueries map[string]bool,
|
||||
missingMetricQuerySet map[string]bool,
|
||||
event *qbtypes.QBEvent,
|
||||
promqlOpts promqlOptions,
|
||||
) (map[string]qbtypes.Query, map[string]qbtypes.Step, error) {
|
||||
|
||||
tmplVars := req.Variables
|
||||
@@ -271,7 +218,7 @@ func (q *querier) buildQueries(
|
||||
if !ok {
|
||||
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid promql query spec %T", query.Spec)
|
||||
}
|
||||
promqlQuery := newPromqlQuery(q.logger, q.promEngine, promQuery, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType, tmplVars, promqlOpts)
|
||||
promqlQuery := newPromqlQuery(q.logger, q.promEngine, promQuery, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType, tmplVars)
|
||||
queries[promQuery.Name] = promqlQuery
|
||||
steps[promQuery.Name] = promQuery.Step
|
||||
case qbtypes.QueryTypeClickHouseSQL:
|
||||
@@ -929,7 +876,7 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
|
||||
switch qt := originalQuery.(type) {
|
||||
case *promqlQuery:
|
||||
queryCopy := qt.query.Copy()
|
||||
return newPromqlQuery(q.logger, qt.promEngine, queryCopy, timeRange, qt.requestType, qt.vars, qt.opts)
|
||||
return newPromqlQuery(q.logger, qt.promEngine, queryCopy, timeRange, qt.requestType, qt.vars)
|
||||
|
||||
case *chSQLQuery:
|
||||
queryCopy := qt.query.Copy()
|
||||
|
||||
@@ -48,7 +48,6 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
|
||||
nil, // telemetryStore
|
||||
metadataStore,
|
||||
nil, // prometheus
|
||||
nil, // promV2
|
||||
nil, // traceStmtBuilder
|
||||
nil, // aiTraceStmtBuilder
|
||||
nil, // logStmtBuilder
|
||||
@@ -122,7 +121,6 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
|
||||
telemetryStore,
|
||||
metadataStore,
|
||||
nil, // prometheus
|
||||
nil, // promV2
|
||||
nil, // traceStmtBuilder
|
||||
nil, // aiTraceStmtBuilder
|
||||
nil, // logStmtBuilder
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
func NewFactory(
|
||||
telemetryStore telemetrystore.TelemetryStore,
|
||||
prometheus prometheus.Prometheus,
|
||||
promV2 prometheus.Prometheus,
|
||||
metadataStore telemetrytypes.MetadataStore,
|
||||
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
@@ -42,7 +41,6 @@ func NewFactory(
|
||||
telemetryStore,
|
||||
metadataStore,
|
||||
prometheus,
|
||||
promV2,
|
||||
traceStmtBuilder,
|
||||
aiTraceStmtBuilder,
|
||||
logStmtBuilder,
|
||||
|
||||
@@ -35,6 +35,7 @@ import (
|
||||
errorsV2 "github.com/SigNoz/signoz/pkg/errors"
|
||||
|
||||
"github.com/prometheus/prometheus/promql"
|
||||
"github.com/prometheus/prometheus/promql/parser"
|
||||
"github.com/prometheus/prometheus/util/stats"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
@@ -230,41 +231,43 @@ func NewReader(
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetInstantQueryMetricsResult(ctx context.Context, queryParams *model.InstantQueryMetricsParams) (*promql.Result, *stats.QueryStats, *model.ApiError) {
|
||||
qry, err := r.prometheus.Engine().NewInstantQuery(ctx, r.prometheus.Storage(), nil, queryParams.Query, queryParams.Time)
|
||||
res, err := r.prometheus.Query(ctx, queryParams.Query, queryParams.Time)
|
||||
var qs stats.QueryStats
|
||||
if err != nil {
|
||||
return nil, nil, &model.ApiError{Typ: model.ErrorBadData, Err: err}
|
||||
var parseErrs parser.ParseErrors
|
||||
if errorsV2.As(err, &parseErrs) {
|
||||
return nil, nil, &model.ApiError{Typ: model.ErrorBadData, Err: err}
|
||||
}
|
||||
// Evaluation errors travel inside the result, as the engine reports
|
||||
// them; the handler maps them from there.
|
||||
return &promql.Result{Err: err}, &qs, nil
|
||||
}
|
||||
|
||||
res := qry.Exec(ctx)
|
||||
|
||||
// Optional stats field in response if parameter "stats" is not empty.
|
||||
var qs stats.QueryStats
|
||||
if queryParams.Stats != "" {
|
||||
qs = stats.NewQueryStats(qry.Stats())
|
||||
if queryParams.Stats != "" && res.Stats != nil {
|
||||
qs = stats.NewQueryStats(res.Stats)
|
||||
}
|
||||
|
||||
qry.Close()
|
||||
return res, &qs, nil
|
||||
|
||||
return &promql.Result{Value: res.Value, Warnings: res.Warnings}, &qs, nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetQueryRangeResult(ctx context.Context, query *model.QueryRangeParams) (*promql.Result, *stats.QueryStats, *model.ApiError) {
|
||||
qry, err := r.prometheus.Engine().NewRangeQuery(ctx, r.prometheus.Storage(), nil, query.Query, query.Start, query.End, query.Step)
|
||||
|
||||
res, err := r.prometheus.QueryRange(ctx, query.Query, query.Start, query.End, query.Step)
|
||||
var qs stats.QueryStats
|
||||
if err != nil {
|
||||
return nil, nil, &model.ApiError{Typ: model.ErrorBadData, Err: err}
|
||||
var parseErrs parser.ParseErrors
|
||||
if errorsV2.As(err, &parseErrs) {
|
||||
return nil, nil, &model.ApiError{Typ: model.ErrorBadData, Err: err}
|
||||
}
|
||||
return &promql.Result{Err: err}, &qs, nil
|
||||
}
|
||||
|
||||
res := qry.Exec(ctx)
|
||||
|
||||
// Optional stats field in response if parameter "stats" is not empty.
|
||||
var qs stats.QueryStats
|
||||
if query.Stats != "" {
|
||||
qs = stats.NewQueryStats(qry.Stats())
|
||||
if query.Stats != "" && res.Stats != nil {
|
||||
qs = stats.NewQueryStats(res.Stats)
|
||||
}
|
||||
|
||||
qry.Close()
|
||||
return res, &qs, nil
|
||||
return &promql.Result{Value: res.Value, Warnings: res.Warnings}, &qs, nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetServicesList(ctx context.Context) (*[]string, error) {
|
||||
|
||||
@@ -156,7 +156,7 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
triggeredTestAlerts := []map[*alertmanagertypes.PostableAlert][]string{}
|
||||
|
||||
// Variable to store promProvider for cleanup
|
||||
var promProvider *prometheustest.Provider
|
||||
var promProvider prometheus.Prometheus
|
||||
|
||||
// Create manager using test factory with hooks
|
||||
mgr := NewTestManager(t, &TestManagerOptions{
|
||||
@@ -181,74 +181,29 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
TelemetryStoreHook: func(store telemetrystore.TelemetryStore) {
|
||||
mockStore := store.(*telemetrystoretest.Provider)
|
||||
|
||||
// Set up Prometheus-specific mock data
|
||||
// Fingerprint columns for Prometheus queries
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
// Samples columns for Prometheus queries
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
// Calculate query time range similar to Prometheus rule tests
|
||||
// TestNotification uses time.Now().UTC() for evaluation
|
||||
// We calculate the query window based on current time to match what the actual evaluation will use
|
||||
// Grid the TestNotification eval computes over (see
|
||||
// Timestamps on base_rule); nil args match any window.
|
||||
evalTime := baseTime
|
||||
evalWindowMs := int64(5 * 60 * 1000) // 5 minutes in ms
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-2*evalWindowMs)/60000)*60000 + 1 // truncate to minute + 1ms
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
gridEnd := (evalTime.UnixMilli() / 60000) * 60000
|
||||
gridStart := gridEnd - evalWindowMs
|
||||
|
||||
// Create fingerprint data
|
||||
fingerprint := uint64(12345)
|
||||
labelsJSON := `{"__name__":"test_metric"}`
|
||||
fingerprintData := [][]any{
|
||||
{fingerprint, labelsJSON},
|
||||
}
|
||||
fingerprintRows := cmock.NewRows(fingerprintCols, fingerprintData)
|
||||
|
||||
// Create samples data from test case values, calculating timestamps relative to baseTime
|
||||
validSamplesData := make([][]any, 0)
|
||||
tsList := make([]int64, 0, len(tc.Values))
|
||||
vList := make([]float64, 0, len(tc.Values))
|
||||
for _, v := range tc.Values {
|
||||
// Skip NaN and Inf values in the samples data
|
||||
if math.IsNaN(v.Value) || math.IsInf(v.Value, 0) {
|
||||
continue
|
||||
}
|
||||
// Calculate timestamp relative to baseTime
|
||||
sampleTimestamp := baseTime.Add(v.Offset).UnixMilli()
|
||||
validSamplesData = append(validSamplesData, []any{
|
||||
"test_metric",
|
||||
fingerprint,
|
||||
sampleTimestamp,
|
||||
v.Value,
|
||||
uint32(0), // flags - 0 means normal value
|
||||
})
|
||||
tsList = append(tsList, baseTime.Add(v.Offset).UnixMilli())
|
||||
vList = append(vList, v.Value)
|
||||
}
|
||||
samplesRows := cmock.NewRows(samplesCols, validSamplesData)
|
||||
grid := prometheustest.LastSampleGrid(tsList, vList, gridStart, gridEnd, 60_000, 300_000)
|
||||
|
||||
mock := mockStore.Mock()
|
||||
|
||||
// Mock the fingerprint query (for Prometheus label matching)
|
||||
mock.ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(fingerprintRows)
|
||||
|
||||
// Mock the samples query (for Prometheus metric data)
|
||||
mock.ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs(
|
||||
"test_metric",
|
||||
"test_metric",
|
||||
queryStart,
|
||||
queryEnd,
|
||||
).
|
||||
WillReturnRows(samplesRows)
|
||||
mock.ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{{`[["__name__","test_metric"]]`, grid}}))
|
||||
|
||||
// Create Prometheus provider for this test
|
||||
promProvider = prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, store)
|
||||
@@ -282,7 +237,6 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
assert.Empty(t, triggeredTestAlerts)
|
||||
}
|
||||
|
||||
promProvider.Close()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -131,7 +131,7 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
|
||||
meterStmtBuilder, err := meterstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
|
||||
require.NoError(t, err)
|
||||
bucketCache := querier.NewBucketCache(providerSettings, cache, 0, 0)
|
||||
providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, nil, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
|
||||
providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
|
||||
mockQuerier, err := providerFactory.New(context.Background(), providerSettings, querier.Config{})
|
||||
require.NoError(t, err)
|
||||
|
||||
|
||||
@@ -376,17 +376,11 @@ func (r *PromRule) String() string {
|
||||
}
|
||||
|
||||
func (r *PromRule) RunAlertQuery(ctx context.Context, qs string, start, end time.Time, interval time.Duration) (promql.Matrix, error) {
|
||||
q, err := r.prometheus.Engine().NewRangeQuery(ctx, r.prometheus.Storage(), nil, qs, start, end, interval)
|
||||
res, err := r.prometheus.QueryRange(ctx, qs, start, end, interval)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
res := q.Exec(ctx)
|
||||
|
||||
if res.Err != nil {
|
||||
return nil, res.Err
|
||||
}
|
||||
|
||||
switch typ := res.Value.(type) {
|
||||
case promql.Vector:
|
||||
series := make([]promql.Series, 0, len(typ))
|
||||
|
||||
@@ -9,8 +9,8 @@ import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
pql "github.com/prometheus/prometheus/promql"
|
||||
cmock "github.com/SigNoz/clickhouse-go-mock"
|
||||
pql "github.com/prometheus/prometheus/promql"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
@@ -768,26 +768,11 @@ func TestPromRuleUnitCombinations(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
// time_series_v4 cols of interest
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
// samples_v4 columns
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
// see Timestamps on base_rule
|
||||
evalWindowMs := int64(5 * 60 * 1000) // 5 minutes in ms
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-2*evalWindowMs)/60000)*60000 + 1 // truncate to minute + 1ms
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
gridStart := queryEnd - evalWindowMs
|
||||
|
||||
cases := []struct {
|
||||
targetUnit string
|
||||
@@ -904,43 +889,19 @@ func TestPromRuleUnitCombinations(t *testing.T) {
|
||||
for idx, c := range cases {
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
|
||||
|
||||
// single fingerprint with labels JSON
|
||||
fingerprint := uint64(12345)
|
||||
labelsJSON := `{"__name__":"test_metric"}`
|
||||
fingerprintData := [][]any{
|
||||
{fingerprint, labelsJSON},
|
||||
}
|
||||
fingerprintRows := cmock.NewRows(fingerprintCols, fingerprintData)
|
||||
|
||||
// create samples data from test case values
|
||||
samplesData := make([][]any, len(c.values))
|
||||
tsList := make([]int64, len(c.values))
|
||||
vList := make([]float64, len(c.values))
|
||||
for i, v := range c.values {
|
||||
samplesData[i] = []any{
|
||||
"test_metric",
|
||||
fingerprint,
|
||||
v.timestamp.UnixMilli(),
|
||||
v.value,
|
||||
uint32(0), // flags - 0 means normal value, 1 means stale, we are not doing staleness tests
|
||||
}
|
||||
tsList[i] = v.timestamp.UnixMilli()
|
||||
vList[i] = v.value
|
||||
}
|
||||
samplesRows := cmock.NewRows(samplesCols, samplesData)
|
||||
grid := prometheustest.LastSampleGrid(tsList, vList, gridStart, queryEnd, 60_000, 300_000)
|
||||
|
||||
// args: $1=metric_name (the __name__ matcher maps onto the column)
|
||||
// args: $1-$3=group-key join conditions, $4-$6=samples conditions
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(fingerprintRows)
|
||||
|
||||
// args: $1=metric_name IN (discovered names), $2=metric_name (subquery), $3=start, $4=end
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs(
|
||||
"test_metric",
|
||||
"test_metric",
|
||||
queryStart,
|
||||
queryEnd,
|
||||
).
|
||||
WillReturnRows(samplesRows)
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{{`[["__name__","test_metric"]]`, grid}}))
|
||||
|
||||
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
|
||||
|
||||
@@ -970,14 +931,12 @@ func TestPromRuleUnitCombinations(t *testing.T) {
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
continue
|
||||
}
|
||||
|
||||
alertsFound, err := rule.Eval(context.Background(), evalTime)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -995,7 +954,6 @@ func TestPromRuleUnitCombinations(t *testing.T) {
|
||||
assert.Equal(t, c.expectAlerts, foundCount, "case %d", idx)
|
||||
}
|
||||
|
||||
promProvider.Close()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1027,12 +985,6 @@ func TestPromRuleNoData(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
// time_series_v4 cols of interest
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
values []struct {
|
||||
timestamp time.Time
|
||||
@@ -1054,15 +1006,11 @@ func TestPromRuleNoData(t *testing.T) {
|
||||
for idx, c := range cases {
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
|
||||
|
||||
// no data
|
||||
fingerprintData := [][]any{}
|
||||
fingerprintRows := cmock.NewRows(fingerprintCols, fingerprintData)
|
||||
|
||||
// no rows == no data
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(fingerprintRows)
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{}))
|
||||
|
||||
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
|
||||
|
||||
@@ -1087,14 +1035,12 @@ func TestPromRuleNoData(t *testing.T) {
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
continue
|
||||
}
|
||||
|
||||
alertsFound, err := rule.Eval(context.Background(), evalTime)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -1107,7 +1053,6 @@ func TestPromRuleNoData(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
promProvider.Close()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1139,24 +1084,11 @@ func TestMultipleThresholdPromRule(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
// see .Timestamps of base rule
|
||||
evalWindowMs := int64(5 * 60 * 1000)
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-2*evalWindowMs)/60000)*60000 + 1
|
||||
queryEnd := (evalTimeMs / 60000) * 60000
|
||||
gridStart := queryEnd - evalWindowMs
|
||||
|
||||
cases := []struct {
|
||||
targetUnit string
|
||||
@@ -1250,39 +1182,19 @@ func TestMultipleThresholdPromRule(t *testing.T) {
|
||||
for idx, c := range cases {
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
|
||||
|
||||
fingerprint := uint64(12345)
|
||||
labelsJSON := `{"__name__":"test_metric"}`
|
||||
fingerprintData := [][]any{
|
||||
{fingerprint, labelsJSON},
|
||||
}
|
||||
fingerprintRows := cmock.NewRows(fingerprintCols, fingerprintData)
|
||||
|
||||
samplesData := make([][]any, len(c.values))
|
||||
tsList := make([]int64, len(c.values))
|
||||
vList := make([]float64, len(c.values))
|
||||
for i, v := range c.values {
|
||||
samplesData[i] = []any{
|
||||
"test_metric",
|
||||
fingerprint,
|
||||
v.timestamp.UnixMilli(),
|
||||
v.value,
|
||||
uint32(0),
|
||||
}
|
||||
tsList[i] = v.timestamp.UnixMilli()
|
||||
vList[i] = v.value
|
||||
}
|
||||
samplesRows := cmock.NewRows(samplesCols, samplesData)
|
||||
grid := prometheustest.LastSampleGrid(tsList, vList, gridStart, queryEnd, 60_000, 300_000)
|
||||
|
||||
// args: $1-$3=group-key join conditions, $4-$6=samples conditions
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(fingerprintRows)
|
||||
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs(
|
||||
"test_metric",
|
||||
"test_metric",
|
||||
queryStart,
|
||||
queryEnd,
|
||||
).
|
||||
WillReturnRows(samplesRows)
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{{`[["__name__","test_metric"]]`, grid}}))
|
||||
|
||||
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
|
||||
|
||||
@@ -1319,14 +1231,12 @@ func TestMultipleThresholdPromRule(t *testing.T) {
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
continue
|
||||
}
|
||||
|
||||
alertsFound, err := rule.Eval(context.Background(), evalTime)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -1344,7 +1254,6 @@ func TestMultipleThresholdPromRule(t *testing.T) {
|
||||
assert.Equal(t, c.expectAlerts, foundCount, "case %d", idx)
|
||||
}
|
||||
|
||||
promProvider.Close()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1378,27 +1287,6 @@ func TestPromRule_NoData(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
// time_series_v4 cols of interest
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
// samples_v4 columns
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
// see Timestamps on base_rule
|
||||
evalWindowMs := int64(5 * 60 * 1000) // 5 minutes in ms
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-2*evalWindowMs)/60000)*60000 + 1 // truncate to minute + 1ms
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
|
||||
cases := []struct {
|
||||
description string
|
||||
alertOnAbsent bool
|
||||
@@ -1430,18 +1318,11 @@ func TestPromRule_NoData(t *testing.T) {
|
||||
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
|
||||
|
||||
// single fingerprint with labels JSON
|
||||
fingerprint := uint64(12345)
|
||||
labelsJSON := `{"__name__":"test_metric"}`
|
||||
// no rows == no data
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(cmock.NewRows(fingerprintCols, [][]any{{fingerprint, labelsJSON}}))
|
||||
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs("test_metric", "test_metric", queryStart, queryEnd).
|
||||
WillReturnRows(cmock.NewRows(samplesCols, [][]any{}))
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{}))
|
||||
|
||||
promProvider := prometheustest.New(
|
||||
context.Background(),
|
||||
@@ -1450,7 +1331,6 @@ func TestPromRule_NoData(t *testing.T) {
|
||||
telemetryStore,
|
||||
)
|
||||
defer func() {
|
||||
_ = promProvider.Close()
|
||||
}()
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
@@ -1510,19 +1390,6 @@ func TestPromRule_NoData_AbsentFor(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
description string
|
||||
absentFor uint64 // grace period in minutes
|
||||
@@ -1556,43 +1423,30 @@ func TestPromRule_NoData_AbsentFor(t *testing.T) {
|
||||
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
|
||||
|
||||
fingerprint := uint64(12345)
|
||||
labelsJSON := `{"__name__":"test_metric"}`
|
||||
|
||||
// Helper to calculate query time range for an eval time
|
||||
calcQueryRange := func(evalTime time.Time) (int64, int64) {
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-2*evalWindow.Milliseconds())/60000)*60000 + 1
|
||||
queryEnd := (evalTimeMs / 60000) * 60000
|
||||
return queryStart, queryEnd
|
||||
// Grid an eval at this time evaluates over (see Timestamps on
|
||||
// base_rule).
|
||||
calcGrid := func(evalTime time.Time) (int64, int64) {
|
||||
gridEnd := (evalTime.UnixMilli() / 60000) * 60000
|
||||
return gridEnd - evalWindow.Milliseconds(), gridEnd
|
||||
}
|
||||
|
||||
// First eval (t1) - with data
|
||||
queryStart1, queryEnd1 := calcQueryRange(t1)
|
||||
// First eval (t1) - with data: points in the past relative to t1
|
||||
gridStart1, gridEnd1 := calcGrid(t1)
|
||||
grid1 := prometheustest.LastSampleGrid(
|
||||
[]int64{baseTime.UnixMilli(), baseTime.Add(1 * time.Minute).UnixMilli(), baseTime.Add(2 * time.Minute).UnixMilli()},
|
||||
[]float64{100, 100, 100},
|
||||
gridStart1, gridEnd1, 60_000, 300_000,
|
||||
)
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(cmock.NewRows(fingerprintCols, [][]any{{fingerprint, labelsJSON}}))
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs("test_metric", "test_metric", queryStart1, queryEnd1).
|
||||
WillReturnRows(cmock.NewRows(samplesCols, [][]any{
|
||||
// Data points in the past relative to t1
|
||||
{"test_metric", fingerprint, baseTime.UnixMilli(), 100.0, uint32(0)},
|
||||
{"test_metric", fingerprint, baseTime.Add(1 * time.Minute).UnixMilli(), 100.0, uint32(0)},
|
||||
{"test_metric", fingerprint, baseTime.Add(2 * time.Minute).UnixMilli(), 100.0, uint32(0)},
|
||||
}))
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{{`[["__name__","test_metric"]]`, grid1}}))
|
||||
|
||||
// Second eval (t2) - no data
|
||||
queryStart2, queryEnd2 := calcQueryRange(t2)
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(cmock.NewRows(fingerprintCols, [][]any{{fingerprint, labelsJSON}}))
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs("test_metric", "test_metric", queryStart2, queryEnd2).
|
||||
WillReturnRows(cmock.NewRows(samplesCols, [][]any{})) // empty - no data
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{}))
|
||||
|
||||
promProvider := prometheustest.New(
|
||||
context.Background(),
|
||||
@@ -1601,7 +1455,6 @@ func TestPromRule_NoData_AbsentFor(t *testing.T) {
|
||||
telemetryStore,
|
||||
)
|
||||
defer func() {
|
||||
_ = promProvider.Close()
|
||||
}()
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
@@ -1652,32 +1505,15 @@ func TestPromRuleEval_RequireMinPoints(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
fingerprintCols := []cmock.ColumnType{
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "any(labels)", Type: "String"},
|
||||
}
|
||||
fingerprint := uint64(12345)
|
||||
fingerprintData := [][]any{{fingerprint, `{"__name__":"test_metric"}`}}
|
||||
|
||||
samplesCols := []cmock.ColumnType{
|
||||
{Name: "metric_name", Type: "String"},
|
||||
{Name: "fingerprint", Type: "UInt64"},
|
||||
{Name: "unix_milli", Type: "Int64"},
|
||||
{Name: "value", Type: "Float64"},
|
||||
{Name: "flags", Type: "UInt32"},
|
||||
}
|
||||
samplesData := [][]any{
|
||||
{"test_metric", fingerprint, baseTime.UnixMilli(), 100.0, 0},
|
||||
{"test_metric", fingerprint, baseTime.Add(time.Minute).UnixMilli(), 150.0, 0},
|
||||
{"test_metric", fingerprint, baseTime.Add(2 * time.Minute).UnixMilli(), 250.0, 0},
|
||||
}
|
||||
sampleTs := []int64{baseTime.UnixMilli(), baseTime.Add(time.Minute).UnixMilli(), baseTime.Add(2 * time.Minute).UnixMilli()}
|
||||
sampleVs := []float64{100, 150, 250}
|
||||
targetForAlert := 200.0
|
||||
targetForNoAlert := 500.0
|
||||
|
||||
// see Timestamps on base_rule
|
||||
evalTimeMs := evalTime.UnixMilli()
|
||||
queryStart := ((evalTimeMs-evalWindow.Milliseconds()-lookBackDelta.Milliseconds())/60000)*60000 + 1 // truncate to minute + 1ms
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
queryEnd := (evalTimeMs / 60000) * 60000 // truncate to minute
|
||||
gridStart := queryEnd - evalWindow.Milliseconds()
|
||||
|
||||
cases := []struct {
|
||||
description string
|
||||
@@ -1746,14 +1582,11 @@ func TestPromRuleEval_RequireMinPoints(t *testing.T) {
|
||||
|
||||
t.Run(c.description, func(t *testing.T) {
|
||||
telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
|
||||
grid := prometheustest.LastSampleGrid(sampleTs, sampleVs, gridStart, queryEnd, 60_000, lookBackDelta.Milliseconds())
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT fingerprint, any").
|
||||
WithArgs("test_metric").
|
||||
WillReturnRows(cmock.NewRows(fingerprintCols, fingerprintData))
|
||||
telemetryStore.Mock().
|
||||
ExpectQuery("SELECT metric_name, fingerprint, unix_milli").
|
||||
WithArgs("test_metric", "test_metric", queryStart, queryEnd).
|
||||
WillReturnRows(cmock.NewRows(samplesCols, samplesData))
|
||||
ExpectQuery("SELECT gkey").
|
||||
WithArgs("test_metric", nil, nil, "test_metric", nil, nil).
|
||||
WillReturnRows(cmock.NewRows(prometheustest.GridCols, [][]any{{`[["__name__","test_metric"]]`, grid}}))
|
||||
promProvider := prometheustest.New(
|
||||
context.Background(),
|
||||
instrumentationtest.New().ToProviderSettings(),
|
||||
@@ -1761,7 +1594,6 @@ func TestPromRuleEval_RequireMinPoints(t *testing.T) {
|
||||
telemetryStore,
|
||||
)
|
||||
defer func() {
|
||||
_ = promProvider.Close()
|
||||
}()
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
|
||||
@@ -40,7 +40,6 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
|
||||
telemetryStore,
|
||||
metadataStore,
|
||||
nil, // prometheus
|
||||
nil, // promV2
|
||||
nil, // traceStmtBuilder
|
||||
nil, // aiTraceStmtBuilder
|
||||
nil, // logStmtBuilder
|
||||
@@ -76,7 +75,6 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
|
||||
telemetryStore,
|
||||
metadataStore,
|
||||
nil, // prometheus
|
||||
nil, // promV2
|
||||
nil, // traceStmtBuilder
|
||||
nil, // aiTraceStmtBuilder
|
||||
logStmtBuilder,
|
||||
@@ -113,7 +111,6 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
|
||||
telemetryStore,
|
||||
metadataStore,
|
||||
nil, // prometheus
|
||||
nil, // promV2
|
||||
traceStmtBuilder,
|
||||
nil, // aiTraceStmtBuilder
|
||||
nil, // logStmtBuilder
|
||||
|
||||
@@ -45,7 +45,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/pprof/httppprof"
|
||||
"github.com/SigNoz/signoz/pkg/pprof/nooppprof"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
|
||||
"github.com/SigNoz/signoz/pkg/querier"
|
||||
"github.com/SigNoz/signoz/pkg/querier/signozquerier"
|
||||
@@ -270,7 +269,6 @@ func NewTelemetryStoreProviderFactories() factory.NamedMap[factory.ProviderFacto
|
||||
|
||||
func NewPrometheusProviderFactories(telemetryStore telemetrystore.TelemetryStore) factory.NamedMap[factory.ProviderFactory[prometheus.Prometheus, prometheus.Config]] {
|
||||
return factory.MustNewNamedMap(
|
||||
clickhouseprometheus.NewFactory(telemetryStore),
|
||||
clickhouseprometheusv2.NewFactory(telemetryStore),
|
||||
)
|
||||
}
|
||||
@@ -313,9 +311,9 @@ func NewStatsReporterProviderFactories(aggregator statsreporter.Aggregator, orgG
|
||||
)
|
||||
}
|
||||
|
||||
func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, promV2 prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
|
||||
func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
|
||||
return factory.MustNewNamedMap(
|
||||
signozquerier.NewFactory(telemetryStore, prometheus, promV2, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -40,7 +40,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/modules/tag/impltag"
|
||||
"github.com/SigNoz/signoz/pkg/modules/user/impluser"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus/clickhouseprometheusv2"
|
||||
"github.com/SigNoz/signoz/pkg/querier"
|
||||
"github.com/SigNoz/signoz/pkg/queryparser"
|
||||
"github.com/SigNoz/signoz/pkg/ruler"
|
||||
@@ -309,11 +308,6 @@ func New(
|
||||
|
||||
retentionGetter := implretention.NewGetter(implretention.NewStore(sqlstore))
|
||||
|
||||
// promV2 is the clickhousev2 provider handed to the querier for shadow
|
||||
// comparison and pinned serving (declared before the serving provider,
|
||||
// whose variable shadows the package name below).
|
||||
var promV2 prometheus.Prometheus
|
||||
|
||||
// Initialize prometheus from the available prometheus provider factories
|
||||
prometheus, err := factory.NewProviderFromNamedMap(
|
||||
ctx,
|
||||
@@ -326,23 +320,6 @@ func New(
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// With the default provider, also stand up the clickhousev2 provider for
|
||||
// the querier: PromQL queries shadow-compare against it behind the
|
||||
// use_prometheus_clickhouse_v2 flag (see pkg/querier/promql_shadow.go).
|
||||
// It never serves by default. An explicit
|
||||
// prometheus::provider: clickhousev2 makes v2 the serving provider
|
||||
// outright, so there is nothing to compare against.
|
||||
if config.Prometheus.Provider() == "clickhouse" {
|
||||
v2Config := config.Prometheus
|
||||
// The v2 engine only evaluates shadow and pinned queries; disable its
|
||||
// active query tracker so two trackers never share a file.
|
||||
v2Config.ActiveQueryTrackerConfig.Enabled = false
|
||||
promV2, err = clickhouseprometheusv2.New(ctx, providerSettings, v2Config, telemetrystore)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// Assemble the query stack (metadata store, statement builders, bucket cache) once,
|
||||
// and reuse the single metadata store everywhere downstream.
|
||||
telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger)
|
||||
@@ -355,7 +332,7 @@ func New(
|
||||
ctx,
|
||||
providerSettings,
|
||||
config.Querier,
|
||||
NewQuerierProviderFactories(telemetrystore, prometheus, promV2, telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
config.Querier.Provider(),
|
||||
)
|
||||
if err != nil {
|
||||
|
||||
@@ -388,14 +388,6 @@ type QueryRangeRequest struct {
|
||||
// NoCache is a flag to disable caching for the request.
|
||||
NoCache bool `json:"noCache,omitempty"`
|
||||
|
||||
// PromQLProvider serves this request's PromQL queries via the named
|
||||
// prometheus provider ("clickhousev2") instead of the default — the same
|
||||
// data read through a different implementation. It is set from the
|
||||
// X-SigNoz-PromQL-Provider header by the API handler, never from the
|
||||
// body: a rollout-scoped comparison hook for integration tests and
|
||||
// support should not become part of the public request schema.
|
||||
PromQLProvider string `json:"-"`
|
||||
|
||||
FormatOptions *FormatOptions `json:"formatOptions,omitempty"`
|
||||
}
|
||||
|
||||
|
||||
3
tests/fixtures/querier.py
vendored
3
tests/fixtures/querier.py
vendored
@@ -175,7 +175,6 @@ def make_query_request(
|
||||
variables: dict | None = None,
|
||||
no_cache: bool = True,
|
||||
timeout: int = QUERY_TIMEOUT,
|
||||
headers: dict | None = None,
|
||||
) -> requests.Response:
|
||||
if format_options is None:
|
||||
format_options = {"formatTableResultForUI": False, "fillGaps": False}
|
||||
@@ -195,7 +194,7 @@ def make_query_request(
|
||||
return requests.post(
|
||||
signoz.self.host_configs["8080"].get("/api/v5/query_range"),
|
||||
timeout=timeout,
|
||||
headers={"authorization": f"Bearer {token}", **(headers or {})},
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
json=payload,
|
||||
)
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"note": "Divergences of the CURRENT promql serving path from the upstream reference engine, enforced exactly by 01_upstream_corpus.py in both directions. These document shipped defects, not test debt: the dominant class is the v1 remote-read fetch injecting a synthetic 'fingerprint' label into every series (pkg/prometheus/clickhouseprometheus/json.go), which breaks without() grouping and default vector matching. Entries must be REMOVED as the serving path is fixed or swapped. Second class, and the bulk of the entries below: promql_query.go drops NaN and +/-Inf from results, mirroring the builder path in consume.go, so every case whose expected output carries a non-finite value diverges on both legs. That class is a product decision rather than a defect, so it is not part of the burn-down.",
|
||||
"note": "Divergences of the clickhousev2 provider from the upstream reference engine, enforced exactly by 01_upstream_corpus.py in both directions: a new divergence is a regression, and an entry that starts passing must be removed. The entries are the non-finite values the API filters and the Kahan-summation class.",
|
||||
"divergences": {
|
||||
"aggregators.test:630[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:630[instant-coarse]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
@@ -15,8 +15,16 @@
|
||||
"aggregators.test:645[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:648[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:648[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:651[base]": "avg over near-max-float64 values: engine's incremental mean never forms the overflowing sum; avgForEach sums then divides, overflowing to +Inf",
|
||||
"aggregators.test:651[instant-coarse]": "same as aggregators.test:651[base] on the coarse-step grid variant",
|
||||
"aggregators.test:654[base]": "avg over near-min-float64 values: engine's incremental mean never forms the overflowing sum; avgForEach overflows to -Inf",
|
||||
"aggregators.test:654[instant-coarse]": "same as aggregators.test:654[base] on the coarse-step grid variant",
|
||||
"aggregators.test:661[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:661[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:687[base]": "sum over {1e100, -1e100, small}: engine uses Kahan compensated summation; sumForEach's naive summation loses the small terms to cancellation and returns 0",
|
||||
"aggregators.test:687[instant-coarse]": "same as aggregators.test:687[base] on the coarse-step grid variant",
|
||||
"aggregators.test:695[base]": "avg over {1e100, -1e100, small}: same Kahan-vs-naive cancellation as aggregators.test:687, divided by count",
|
||||
"aggregators.test:695[instant-coarse]": "same as aggregators.test:695[base] on the coarse-step grid variant",
|
||||
"aggregators.test:698[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:698[instant-coarse]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:702[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
@@ -73,6 +81,10 @@
|
||||
"aggregators.test:963[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:966[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:966[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"functions.test:1084[instant-coarse]": "sum_over_time over a window containing ±1e100: the disjoint coarse-step form's arraySum slide is naive summation, cancelling to 0 (the base variant's W>64 shape falls back to the engine and is exact)",
|
||||
"functions.test:1087[instant-coarse]": "avg_over_time, same window and cancellation as functions.test:1084[instant-coarse]",
|
||||
"functions.test:1149[base]": "avg_over_time over ±2.258e220-magnitude samples: engine's Kahan-compensated incremental mean cancels exactly to 0; the bucketed form's naive slide summation leaves a ~1e202 residue",
|
||||
"functions.test:1149[instant-coarse]": "same as functions.test:1149[base] through the disjoint coarse-step form",
|
||||
"operators.test:533[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"operators.test:533[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"operators.test:539[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
|
||||
@@ -1,128 +0,0 @@
|
||||
{
|
||||
"note": "Divergences of the clickhousev2 provider (pinned via X-SigNoz-PromQL-Provider) from the upstream reference engine, enforced exactly by 01_upstream_corpus.py in both directions. This ledger is the rollout scorecard for the provider swap: the default provider cannot be replaced by clickhousev2 while anything is listed here. Entries must carry the defect's cause and be REMOVED as the provider is fixed. Current class: the engine aggregates floats with Kahan compensated summation (sum, sum_over_time) and an overflow-free incremental mean (avg); ClickHouse's sumForEach/avgForEach/arraySum are naive, so extreme-magnitude corpus data (±1e100 cancellation, ±1.8e308 overflow) diverges on transpiled plans. Burn-down candidates: sumKahanForEach for the cancellation class; the overflow class needs an incremental-mean aggregate ClickHouse does not have. Second class, and the bulk of the entries below: promql_query.go drops NaN and +/-Inf from results, mirroring the builder path in consume.go, so every case whose expected output carries a non-finite value diverges on both legs. That class is a product decision rather than a defect, so it is not part of the burn-down.",
|
||||
"divergences": {
|
||||
"aggregators.test:630[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:630[instant-coarse]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:633[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:633[instant-coarse]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:636[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:636[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:639[base]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:639[instant-coarse]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:642[base]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:642[instant-coarse]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:645[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:645[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:648[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:648[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:651[base]": "avg over near-max-float64 values: engine's incremental mean never forms the overflowing sum; avgForEach sums then divides, overflowing to +Inf",
|
||||
"aggregators.test:651[instant-coarse]": "same as aggregators.test:651[base] on the coarse-step grid variant",
|
||||
"aggregators.test:654[base]": "avg over near-min-float64 values: engine's incremental mean never forms the overflowing sum; avgForEach overflows to -Inf",
|
||||
"aggregators.test:654[instant-coarse]": "same as aggregators.test:654[base] on the coarse-step grid variant",
|
||||
"aggregators.test:661[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:661[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:687[base]": "sum over {1e100, -1e100, small}: engine uses Kahan compensated summation; sumForEach's naive summation loses the small terms to cancellation and returns 0",
|
||||
"aggregators.test:687[instant-coarse]": "same as aggregators.test:687[base] on the coarse-step grid variant",
|
||||
"aggregators.test:695[base]": "avg over {1e100, -1e100, small}: same Kahan-vs-naive cancellation as aggregators.test:687, divided by count",
|
||||
"aggregators.test:695[instant-coarse]": "same as aggregators.test:695[base] on the coarse-step grid variant",
|
||||
"aggregators.test:698[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:698[instant-coarse]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:702[base]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:702[instant-coarse]": "expected output is entirely Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:706[base]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:706[instant-coarse]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:710[base]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:710[instant-coarse]": "expected output is entirely -Inf; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:714[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:714[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:717[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:717[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:720[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:720[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:724[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:724[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:862[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:862[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:865[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:865[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:868[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:868[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:873[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:873[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:885[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:885[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:888[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:888[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:891[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:891[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:896[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:896[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:906[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:906[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:909[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:909[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:919[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:919[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:922[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:922[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:925[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:925[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:930[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:930[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:942[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:942[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:945[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:945[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:948[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:948[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:953[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:953[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"aggregators.test:963[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:963[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:966[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"aggregators.test:966[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"functions.test:1084[instant-coarse]": "sum_over_time over a window containing ±1e100: the disjoint coarse-step form's arraySum slide is naive summation, cancelling to 0 (the base variant's W>64 shape falls back to the engine and is exact)",
|
||||
"functions.test:1087[instant-coarse]": "avg_over_time, same window and cancellation as functions.test:1084[instant-coarse]",
|
||||
"functions.test:1149[base]": "avg_over_time over ±2.258e220-magnitude samples: engine's Kahan-compensated incremental mean cancels exactly to 0; the bucketed form's naive slide summation leaves a ~1e202 residue",
|
||||
"functions.test:1149[instant-coarse]": "same as functions.test:1149[base] through the disjoint coarse-step form",
|
||||
"operators.test:533[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"operators.test:533[instant-coarse]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"operators.test:539[base]": "expected output is entirely NaN; promql_query.go drops non-finite values, so no series is emitted",
|
||||
"trig_functions.test:13[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:13[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:18[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:18[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:23[base]": "2 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:23[instant-coarse]": "2 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:28[base]": "2 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:28[instant-coarse]": "2 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:33[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:33[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:38[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:38[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:43[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:43[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:48[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:48[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:53[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:53[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:58[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:58[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:63[base]": "2 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:63[instant-coarse]": "2 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:68[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:68[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:73[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:73[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:78[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:78[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:83[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:83[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:88[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:88[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:8[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:8[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:93[base]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing",
|
||||
"trig_functions.test:93[instant-coarse]": "1 of 3 expected points are NaN; promql_query.go drops non-finite values, so those timestamps are missing"
|
||||
}
|
||||
}
|
||||
@@ -12,11 +12,9 @@ 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.
|
||||
# — the path nothing else exercises. Range cases go to query_range, which
|
||||
# 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}
|
||||
@@ -38,8 +36,7 @@ def test_prometheus_api_corpus(
|
||||
# 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.
|
||||
# (and its known divergences ledgered) by promqlconformance.
|
||||
if case["variant"] == "instant-coarse":
|
||||
continue
|
||||
|
||||
|
||||
@@ -1,37 +0,0 @@
|
||||
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",
|
||||
},
|
||||
)
|
||||
@@ -18,27 +18,11 @@ TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "testdata")
|
||||
# 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
|
||||
# scorecard — the provider swap is measured by burning it down to empty.
|
||||
LEDGER_FILES = {
|
||||
"default": os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences.json"),
|
||||
"clickhousev2": os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences_v2.json"),
|
||||
}
|
||||
|
||||
# Every case replays on both legs, each asserted against the same frozen
|
||||
# expectations and its own ledger — deliberately never against each other: both
|
||||
# legs can sit within one rounding quantum of the expected value yet differ from
|
||||
# each other by up to two quanta when a true value straddles a rounding boundary,
|
||||
# so a leg-vs-leg equality check would reintroduce exactly the boundary noise the
|
||||
# quantum tolerance absorbs. A case failing on one leg while passing on the other
|
||||
# already localizes the defect to that provider; the printed DIVERGED lines for
|
||||
# both legs are the side-by-side triage view. The clickhousev2 header is
|
||||
# flag-gated (see conftest.py).
|
||||
LEGS: list[tuple[str, dict | None]] = [
|
||||
("default", None),
|
||||
("clickhousev2", {"X-SigNoz-PromQL-Provider": "clickhousev2"}),
|
||||
]
|
||||
# The ledger is enforced exactly in both directions: a new divergence is a
|
||||
# regression, and a known divergence that starts passing must be removed. Its
|
||||
# entries are the frozen defects of the serving path (the non-finite API
|
||||
# filtering, and the clickhousev2 Kahan-summation class).
|
||||
LEDGER_FILE = os.path.join(TESTDATA_DIR, "promqltestcorpus", "known_divergences.json")
|
||||
|
||||
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "-Inf": -math.inf}
|
||||
|
||||
@@ -52,7 +36,7 @@ def test_upstream_promqltest_corpus(
|
||||
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}
|
||||
failures: list[str] = []
|
||||
for case in corpus["cases"]:
|
||||
base = bases[case["dataset"]]
|
||||
start_ms = base + case["start_ms"]
|
||||
@@ -71,104 +55,96 @@ def test_upstream_promqltest_corpus(
|
||||
}
|
||||
case_id = f"{case['source']}[{case['variant']}]"
|
||||
|
||||
for leg, headers in LEGS:
|
||||
response = make_query_request(signoz, token, req_start_ms, end_ms, [query], headers=headers)
|
||||
if response.status_code != HTTPStatus.OK:
|
||||
failures[leg].append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
|
||||
continue
|
||||
response = make_query_request(signoz, token, req_start_ms, end_ms, [query])
|
||||
if response.status_code != HTTPStatus.OK:
|
||||
failures.append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
|
||||
continue
|
||||
|
||||
# A response carrying several series with identical visible labels
|
||||
# is itself a defect signal (e.g. a hidden grouping label stripped
|
||||
# on the way out) and must not be silently collapsed into one entry.
|
||||
actual: dict[tuple, dict[int, float]] = {}
|
||||
duplicates: list[tuple] = []
|
||||
# Empty results serialize with null aggregations/series/values fields.
|
||||
for series in get_all_series(response.json(), "A") or []:
|
||||
lbls = {l["key"]["name"]: str(l["value"]) for l in series.get("labels") or []}
|
||||
points = {int(v["timestamp"]): SPECIALS[v["value"]] if isinstance(v["value"], str) else float(v["value"]) for v in series.get("values") or []}
|
||||
key = tuple(sorted(lbls.items()))
|
||||
if key in actual:
|
||||
duplicates.append(key)
|
||||
actual[key] = points
|
||||
if duplicates:
|
||||
failures[leg].append(f"{case_id}: response carries multiple series with identical labels for {case['expr']!r}: {[dict(d) for d in duplicates[:3]]}")
|
||||
continue
|
||||
# A response carrying several series with identical visible labels
|
||||
# is itself a defect signal (e.g. a hidden grouping label stripped
|
||||
# on the way out) and must not be silently collapsed into one entry.
|
||||
actual: dict[tuple, dict[int, float]] = {}
|
||||
duplicates: list[tuple] = []
|
||||
# Empty results serialize with null aggregations/series/values fields.
|
||||
for series in get_all_series(response.json(), "A") or []:
|
||||
lbls = {l["key"]["name"]: str(l["value"]) for l in series.get("labels") or []}
|
||||
points = {int(v["timestamp"]): SPECIALS[v["value"]] if isinstance(v["value"], str) else float(v["value"]) for v in series.get("values") or []}
|
||||
key = tuple(sorted(lbls.items()))
|
||||
if key in actual:
|
||||
duplicates.append(key)
|
||||
actual[key] = points
|
||||
if duplicates:
|
||||
failures.append(f"{case_id}: response carries multiple series with identical labels for {case['expr']!r}: {[dict(d) for d in duplicates[:3]]}")
|
||||
continue
|
||||
|
||||
if case["instant"]:
|
||||
# Keep only the instant point; the extra grid step is a request
|
||||
# encoding byproduct, not part of the assertion.
|
||||
actual = {lset: {ts: v for ts, v in pts.items() if ts == end_ms} for lset, pts in actual.items()}
|
||||
actual = {lset: pts for lset, pts in actual.items() if pts}
|
||||
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 case["instant"]:
|
||||
# Keep only the instant point; the extra grid step is a request
|
||||
# encoding byproduct, not part of the assertion.
|
||||
actual = {lset: {ts: v for ts, v in pts.items() if ts == end_ms} for lset, pts in actual.items()}
|
||||
actual = {lset: pts for lset, pts in actual.items() if pts}
|
||||
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[leg].append(f"{case_id}: series mismatch for {case['expr']!r} (missing={sorted(missing)[:3]} extra={sorted(extra)[:3]}) actual={[(dict(k), {t - base: v for t, v in pts.items()}) for k, pts in actual.items()]}")
|
||||
continue
|
||||
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]}) actual={[(dict(k), {t - base: v for t, v in pts.items()}) for k, pts in actual.items()]}")
|
||||
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
|
||||
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:
|
||||
# Both sides carry the API's rounding (>=1: three decimal places; <1:
|
||||
# three significant digits). A true value sitting exactly on a rounding
|
||||
# boundary can round either way when the two computations differ at ULP
|
||||
# level (float aggregation order over series is storage-iteration
|
||||
# dependent), so allow one rounding quantum.
|
||||
scale = max(abs(act_v), abs(exp_v))
|
||||
if scale >= 1:
|
||||
# Values too large to round pass through unrounded; give those an
|
||||
# ULP-class relative grace on top of the rounding quantum.
|
||||
quantum = max(1e-3, scale * 1e-9)
|
||||
else:
|
||||
# Both sides carry the API's rounding (>=1: three decimal places; <1:
|
||||
# three significant digits). A true value sitting exactly on a rounding
|
||||
# boundary can round either way when the two computations differ at ULP
|
||||
# level (float aggregation order over series is storage-iteration
|
||||
# dependent), so allow one rounding quantum.
|
||||
scale = max(abs(act_v), abs(exp_v))
|
||||
if scale >= 1:
|
||||
# Values too large to round pass through unrounded; give those an
|
||||
# ULP-class relative grace on top of the rounding quantum.
|
||||
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:
|
||||
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:
|
||||
failures[leg].append(mismatch)
|
||||
break
|
||||
if mismatch:
|
||||
failures.append(mismatch)
|
||||
|
||||
for leg, _ in LEGS:
|
||||
for f_line in failures[leg]:
|
||||
print("DIVERGED", f"[{leg}]", f_line)
|
||||
for f_line in failures:
|
||||
print("DIVERGED", f_line)
|
||||
|
||||
known: dict[str, str] = {}
|
||||
if os.path.exists(LEDGER_FILE):
|
||||
with open(LEDGER_FILE, encoding="utf-8") as f:
|
||||
known = json.load(f)["divergences"]
|
||||
|
||||
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)
|
||||
|
||||
# Known divergences are defects of that leg's serving path, frozen with
|
||||
# reasons. Each set is enforced exactly in both directions: a NEW
|
||||
# divergence is a regression, and a known divergence that starts passing
|
||||
# must be removed from the file. Problems across both legs are collected
|
||||
# before asserting so one leg's failure never hides the other's.
|
||||
problems: list[str] = []
|
||||
for leg, _ in LEGS:
|
||||
known: dict[str, str] = {}
|
||||
if os.path.exists(LEDGER_FILES[leg]):
|
||||
with open(LEDGER_FILES[leg], encoding="utf-8") as f:
|
||||
known = json.load(f)["divergences"]
|
||||
|
||||
failed_ids = {f_line.split(": ", 1)[0] for f_line in failures[leg]}
|
||||
unexpected = [f_line for f_line in failures[leg] if f_line.split(": ", 1)[0] not in known]
|
||||
now_passing = sorted(set(known) - failed_ids)
|
||||
|
||||
if unexpected:
|
||||
problems.append(f"[{leg}] {len(unexpected)} corpus cases diverged beyond the known set:\n" + "\n".join(unexpected[:25]))
|
||||
if now_passing:
|
||||
problems.append(f"[{leg}] {len(now_passing)} known divergences now pass — remove them from {os.path.basename(LEDGER_FILES[leg])}: {now_passing[:25]}")
|
||||
if unexpected:
|
||||
problems.append(f"{len(unexpected)} corpus cases diverged beyond the known set:\n" + "\n".join(unexpected[:25]))
|
||||
if now_passing:
|
||||
problems.append(f"{len(now_passing)} known divergences now pass — remove them from {os.path.basename(LEDGER_FILE)}: {now_passing[:25]}")
|
||||
|
||||
assert not problems, "\n\n".join(problems)
|
||||
|
||||
@@ -10,11 +10,6 @@ from fixtures.querier import get_all_series, make_query_request
|
||||
|
||||
MINUTE_MS = 60_000
|
||||
|
||||
LEGS: list[tuple[str, dict | None]] = [
|
||||
("default", None),
|
||||
("clickhousev2", {"X-SigNoz-PromQL-Provider": "clickhousev2"}),
|
||||
]
|
||||
|
||||
|
||||
def test_promql_subquery_without_step_evaluates(
|
||||
signoz: types.SigNoz,
|
||||
@@ -45,16 +40,15 @@ def test_promql_subquery_without_step_evaluates(
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
for leg, headers in LEGS:
|
||||
query = {"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric}[5m:])"}}
|
||||
response = make_query_request(signoz, token, start_ms, end_ms, [query], headers=headers)
|
||||
assert response.status_code == HTTPStatus.OK, f"{leg}: {response.text[:300]}"
|
||||
series = get_all_series(response.json(), "A")
|
||||
assert series, f"{leg}: the subquery must return the inserted series"
|
||||
values = {point["value"] for entry in series for point in entry.get("values") or []}
|
||||
assert values == {42.0}, f"{leg}: {sorted(values)[:5]}"
|
||||
query = {"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric}[5m:])"}}
|
||||
response = make_query_request(signoz, token, start_ms, end_ms, [query])
|
||||
assert response.status_code == HTTPStatus.OK, response.text[:300]
|
||||
series = get_all_series(response.json(), "A")
|
||||
assert series, "the subquery must return the inserted series"
|
||||
values = {point["value"] for entry in series for point in entry.get("values") or []}
|
||||
assert values == {42.0}, sorted(values)[:5]
|
||||
|
||||
# A plain follow-up query proves the process survived the subquery legs.
|
||||
# A plain follow-up query proves the process survived the subquery.
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
|
||||
@@ -1,39 +0,0 @@
|
||||
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_promql_conformance(
|
||||
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:
|
||||
"""
|
||||
Package-scoped SigNoz with use_prometheus_clickhouse_v2 on, so the corpus
|
||||
can replay every case twice: once against the default provider and once
|
||||
pinned to the clickhousev2 provider via the X-SigNoz-PromQL-Provider
|
||||
header (which the flag gates). Each leg is asserted against the same
|
||||
frozen expectations — see 01_upstream_corpus.py for why the legs are
|
||||
never asserted against each other.
|
||||
"""
|
||||
return create_signoz(
|
||||
network=network,
|
||||
zeus=zeus,
|
||||
gateway=gateway,
|
||||
sqlstore=sqlstore,
|
||||
clickhouse=clickhouse,
|
||||
request=request,
|
||||
pytestconfig=pytestconfig,
|
||||
cache_key="signoz-promql-conformance",
|
||||
env_overrides={
|
||||
"SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__PROMETHEUS__CLICKHOUSE__V2": True,
|
||||
},
|
||||
)
|
||||
Reference in New Issue
Block a user