web/federate_test.go
Rama main · réplica de lectura del repositorio original.
// Copyright The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package web
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"sort"
"strconv"
"strings"
"testing"
"time"
"github.com/prometheus/common/model"
"github.com/stretchr/testify/require"
"github.com/prometheus/prometheus/config"
"github.com/prometheus/prometheus/model/histogram"
"github.com/prometheus/prometheus/model/labels"
"github.com/prometheus/prometheus/model/textparse"
"github.com/prometheus/prometheus/promql"
"github.com/prometheus/prometheus/promql/parser"
"github.com/prometheus/prometheus/promql/promqltest"
"github.com/prometheus/prometheus/storage"
"github.com/prometheus/prometheus/tsdb"
"github.com/prometheus/prometheus/util/teststorage"
"github.com/prometheus/prometheus/util/testutil"
)
var testParser = parser.NewParser(parser.Options{})
var scenarios = map[string]struct {
params string
externalLabels labels.Labels
code int
body string
}{
"empty": {
params: "",
code: http.StatusOK,
body: ``,
},
"match nothing": {
params: "match[]=does_not_match_anything",
code: http.StatusOK,
body: ``,
},
"invalid params from the beginning": {
params: "match[]=-not-a-valid-metric-name",
code: http.StatusBadRequest,
body: `1:1: parse error: unexpected <op:->
`,
},
"invalid params somewhere in the middle": {
params: "match[]=not-a-valid-metric-name",
code: http.StatusBadRequest,
body: `1:4: parse error: unexpected <op:->
`,
},
"test_metric1": {
params: "match[]=test_metric1",
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i"} 10000 6000000
test_metric1{foo="boo",instance="i"} 1 6000000
`,
},
"test_metric2": {
params: "match[]=test_metric2",
code: http.StatusOK,
body: `# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i"} 1 6000000
`,
},
"test_metric_without_labels": {
params: "match[]=test_metric_without_labels",
code: http.StatusOK,
body: `# TYPE test_metric_without_labels untyped
test_metric_without_labels{instance=""} 1001 6000000
`,
},
"test_stale_metric": {
params: "match[]=test_metric_stale",
code: http.StatusOK,
body: ``,
},
"test_old_metric": {
params: "match[]=test_metric_old",
code: http.StatusOK,
body: `# TYPE test_metric_old untyped
test_metric_old{instance=""} 981 5880000
`,
},
"{foo='boo'}": {
params: "match[]={foo='boo'}",
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="boo",instance="i"} 1 6000000
# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i"} 1 6000000
`,
},
"two matchers": {
params: "match[]=test_metric1&match[]=test_metric2",
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i"} 10000 6000000
test_metric1{foo="boo",instance="i"} 1 6000000
# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i"} 1 6000000
`,
},
"two matchers with overlap": {
params: "match[]={__name__=~'test_metric1'}&match[]={foo='bar'}",
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i"} 10000 6000000
test_metric1{foo="boo",instance="i"} 1 6000000
`,
},
"everything": {
params: "match[]={__name__=~'.%2b'}", // '%2b' is an URL-encoded '+'.
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i"} 10000 6000000
test_metric1{foo="boo",instance="i"} 1 6000000
# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i"} 1 6000000
# TYPE test_metric_old untyped
test_metric_old{instance=""} 981 5880000
# TYPE test_metric_without_labels untyped
test_metric_without_labels{instance=""} 1001 6000000
`,
},
"empty label value matches everything that doesn't have that label": {
params: "match[]={foo='',__name__=~'.%2b'}",
code: http.StatusOK,
body: `# TYPE test_metric_old untyped
test_metric_old{instance=""} 981 5880000
# TYPE test_metric_without_labels untyped
test_metric_without_labels{instance=""} 1001 6000000
`,
},
"empty label value for a label that doesn't exist at all, matches everything": {
params: "match[]={bar='',__name__=~'.%2b'}",
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i"} 10000 6000000
test_metric1{foo="boo",instance="i"} 1 6000000
# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i"} 1 6000000
# TYPE test_metric_old untyped
test_metric_old{instance=""} 981 5880000
# TYPE test_metric_without_labels untyped
test_metric_without_labels{instance=""} 1001 6000000
`,
},
"external labels are added if not already present": {
params: "match[]={__name__=~'.%2b'}", // '%2b' is an URL-encoded '+'.
externalLabels: labels.FromStrings("foo", "baz", "zone", "ie"),
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i",zone="ie"} 10000 6000000
test_metric1{foo="boo",instance="i",zone="ie"} 1 6000000
# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i",zone="ie"} 1 6000000
# TYPE test_metric_old untyped
test_metric_old{foo="baz",instance="",zone="ie"} 981 5880000
# TYPE test_metric_without_labels untyped
test_metric_without_labels{foo="baz",instance="",zone="ie"} 1001 6000000
`,
},
"instance is an external label": {
// This makes no sense as a configuration, but we should
// know what it does anyway.
params: "match[]={__name__=~'.%2b'}", // '%2b' is an URL-encoded '+'.
externalLabels: labels.FromStrings("instance", "baz"),
code: http.StatusOK,
body: `# TYPE test_metric1 untyped
test_metric1{foo="bar",instance="i"} 10000 6000000
test_metric1{foo="boo",instance="i"} 1 6000000
# TYPE test_metric2 untyped
test_metric2{foo="boo",instance="i"} 1 6000000
# TYPE test_metric_old untyped
test_metric_old{instance="baz"} 981 5880000
# TYPE test_metric_without_labels untyped
test_metric_without_labels{instance="baz"} 1001 6000000
`,
},
}
func TestFederation(t *testing.T) {
storage := promqltest.LoadedStorage(t, `
load 1m
test_metric1{foo="bar",instance="i"} 0+100x100
test_metric1{foo="boo",instance="i"} 1+0x100
test_metric2{foo="boo",instance="i"} 1+0x100
test_metric_without_labels 1+10x100
test_metric_stale 1+10x99 stale
test_metric_old 1+10x98
`)
h := &Handler{
localStorage: &dbAdapter{storage.DB},
lookbackDelta: 5 * time.Minute,
now: func() model.Time { return 101 * 60 * 1000 }, // 101min after epoch.
config: &config.Config{
GlobalConfig: config.GlobalConfig{},
},
options: &Options{Parser: testParser},
}
for name, scenario := range scenarios {
t.Run(name, func(t *testing.T) {
h.config.GlobalConfig.ExternalLabels = scenario.externalLabels
req := httptest.NewRequest(http.MethodGet, "http://example.org/federate?"+scenario.params, http.NoBody)
res := httptest.NewRecorder()
h.federation(res, req)
require.Equal(t, scenario.code, res.Code)
require.Equal(t, scenario.body, normalizeBody(res.Body))
})
}
}
type notReadyReadStorage struct {
LocalStorage
}
func (notReadyReadStorage) Querier(int64, int64) (storage.Querier, error) {
return nil, fmt.Errorf("wrap: %w", tsdb.ErrNotReady)
}
func (notReadyReadStorage) StartTime() (int64, error) {
return 0, fmt.Errorf("wrap: %w", tsdb.ErrNotReady)
}
func (notReadyReadStorage) Stats(string, int) (*tsdb.Stats, error) {
return nil, fmt.Errorf("wrap: %w", tsdb.ErrNotReady)
}
// Regression test for https://github.com/prometheus/prometheus/issues/7181.
func TestFederation_NotReady(t *testing.T) {
for name, scenario := range scenarios {
t.Run(name, func(t *testing.T) {
h := &Handler{
localStorage: notReadyReadStorage{},
lookbackDelta: 5 * time.Minute,
now: func() model.Time { return 101 * 60 * 1000 }, // 101min after epoch.
config: &config.Config{
GlobalConfig: config.GlobalConfig{
ExternalLabels: scenario.externalLabels,
},
},
options: &Options{Parser: testParser},
}
req := httptest.NewRequest(http.MethodGet, "http://example.org/federate?"+scenario.params, http.NoBody)
res := httptest.NewRecorder()
h.federation(res, req)
if scenario.code == http.StatusBadRequest {
// Request are expected to be checked before DB readiness.
require.Equal(t, http.StatusBadRequest, res.Code)
return
}
require.Equal(t, http.StatusServiceUnavailable, res.Code)
})
}
}
// normalizeBody sorts the lines within a metric to make it easy to verify the body.
// (Federation is not taking care of sorting within a metric family.)
func normalizeBody(body *bytes.Buffer) string {
var (
lines []string
lastHash int
)
for line, err := body.ReadString('\n'); err == nil; line, err = body.ReadString('\n') {
if line[0] == '#' && len(lines) > 0 {
sort.Strings(lines[lastHash+1:])
lastHash = len(lines)
}
lines = append(lines, line)
}
if len(lines) > 0 {
sort.Strings(lines[lastHash+1:])
}
return strings.Join(lines, "")
}
func TestFederationWithNativeHistograms(t *testing.T) {
var expVec promql.Vector
db := teststorage.New(t).DB
hist := &histogram.Histogram{
Count: 12,
ZeroCount: 2,
ZeroThreshold: 0.001,
Sum: 39.4,
Schema: 1,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 1, Length: 2},
},
PositiveBuckets: []int64{1, 1, -1, 0},
NegativeSpans: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 1, Length: 2},
},
NegativeBuckets: []int64{1, 1, -1, 0},
}
histWithoutZeroBucket := &histogram.Histogram{
Count: 20,
Sum: 99.23,
Schema: 1,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 1, Length: 2},
},
PositiveBuckets: []int64{2, 2, -2, 0},
NegativeSpans: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 1, Length: 2},
},
NegativeBuckets: []int64{2, 2, -2, 0},
}
nhcb := &histogram.Histogram{
Count: 6,
Sum: 1.234,
Schema: -53,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 2, Length: 1},
},
PositiveBuckets: []int64{3, -1, -1},
CustomValues: []float64{0.1, 0.2, 0.5, 1, 2},
}
app := db.AppenderV2(context.Background())
for i := range 7 {
l := labels.FromStrings("__name__", "test_metric", "foo", strconv.Itoa(i))
expL := labels.FromStrings("__name__", "test_metric", "instance", "", "foo", strconv.Itoa(i))
var err error
switch i {
case 0, 3:
_, err = app.Append(0, l, 0, 100*60*1000, float64(i*100), nil, nil, storage.AOptions{})
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: float64(i * 100),
Metric: expL,
})
case 1, 2:
// Use two float histograms to ensure buffer reuse does not overwrite the first.
hist.ZeroCount++
hist.Count++
fh := hist.ToFloat(nil)
_, err = app.Append(0, l, 0, 100*60*1000, 0, nil, fh.Copy(), storage.AOptions{})
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
H: fh,
Metric: expL,
})
case 4:
_, err = app.Append(0, l, 0, 100*60*1000, 0, histWithoutZeroBucket.Copy(), nil, storage.AOptions{})
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
H: histWithoutZeroBucket.ToFloat(nil),
Metric: expL,
})
case 5:
hist.ZeroCount++
hist.Count++
_, err = app.Append(0, l, 0, 100*60*1000, 0, hist.Copy(), nil, storage.AOptions{})
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
H: hist.ToFloat(nil),
Metric: expL,
})
case 6:
_, err = app.Append(0, l, 0, 100*60*1000, 0, nhcb.Copy(), nil, storage.AOptions{})
expL = labels.FromStrings("__name__", "test_metric_count", "instance", "", "foo", strconv.Itoa(i))
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 6,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_sum", "instance", "", "foo", strconv.Itoa(i))
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 1.234,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_bucket", "instance", "", "foo", strconv.Itoa(i), "le", "0.1")
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 3,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_bucket", "instance", "", "foo", strconv.Itoa(i), "le", "0.2")
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 5,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_bucket", "instance", "", "foo", strconv.Itoa(i), "le", "0.5")
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 5,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_bucket", "instance", "", "foo", strconv.Itoa(i), "le", "1.0")
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 5,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_bucket", "instance", "", "foo", strconv.Itoa(i), "le", "2.0")
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 6,
Metric: expL,
})
expL = labels.FromStrings("__name__", "test_metric_bucket", "instance", "", "foo", strconv.Itoa(i), "le", "+Inf")
expVec = append(expVec, promql.Sample{
T: 100 * 60 * 1000,
F: 6,
Metric: expL,
})
}
require.NoError(t, err)
}
require.NoError(t, app.Commit())
h := &Handler{
localStorage: &dbAdapter{db},
lookbackDelta: 5 * time.Minute,
now: func() model.Time { return 101 * 60 * 1000 }, // 101min after epoch.
config: &config.Config{
GlobalConfig: config.GlobalConfig{},
},
options: &Options{Parser: testParser},
}
req := httptest.NewRequest(http.MethodGet, "http://example.org/federate?match[]=test_metric", http.NoBody)
req.Header.Add("Accept", `application/vnd.google.protobuf;proto=io.prometheus.client.MetricFamily;encoding=delimited,application/openmetrics-text;version=1.0.0;q=0.8,application/openmetrics-text;version=0.0.1;q=0.75,text/plain;version=0.0.4;q=0.5,*/*;q=0.1`)
res := httptest.NewRecorder()
h.federation(res, req)
require.Equal(t, http.StatusOK, res.Code)
body, err := io.ReadAll(res.Body)
require.NoError(t, err)
p := textparse.NewProtobufParser(body, false, false, false, false, labels.NewSymbolTable())
var actVec promql.Vector
metricFamilies := 0
l := labels.Labels{}
for {
et, err := p.Next()
if err != nil && errors.Is(err, io.EOF) {
break
}
require.NoError(t, err)
if et == textparse.EntryHistogram || et == textparse.EntrySeries {
p.Labels(&l)
}
switch et {
case textparse.EntryHelp:
metricFamilies++
case textparse.EntryHistogram:
_, parsedTimestamp, h, fh := p.Histogram()
require.Nil(t, h)
actVec = append(actVec, promql.Sample{
T: *parsedTimestamp,
H: fh,
Metric: l,
})
case textparse.EntrySeries:
_, parsedTimestamp, f := p.Series()
actVec = append(actVec, promql.Sample{
T: *parsedTimestamp,
F: f,
Metric: l,
})
}
}
// TODO(codesome): Once PromQL is able to set the CounterResetHint on histograms,
// test it with switching histogram types for metric families.
require.Equal(t, 4, metricFamilies)
testutil.RequireEqual(t, expVec, actVec)
}
// TestFederationNHCBCountBelowBucketSum checks that a float NHCB whose count is
// marginally below the sum of its buckets, as can result from floating point
// rounding in PromQL (e.g. sum(rate(...))), is federated as a valid classic
// histogram that converts back into an NHCB without negative buckets.
func TestFederationNHCBCountBelowBucketSum(t *testing.T) {
db := teststorage.New(t).DB
// 0.1 + 0.2 == 0.30000000000000004 > 0.3.
fh := &histogram.FloatHistogram{
CounterResetHint: histogram.GaugeType,
Count: 0.3,
Sum: 1,
Schema: histogram.CustomBucketsSchema,
PositiveSpans: []histogram.Span{{Offset: 0, Length: 2}},
PositiveBuckets: []float64{0.1, 0.2},
CustomValues: []float64{1, 2},
}
require.NoError(t, fh.Validate())
app := db.AppenderV2(context.Background())
_, err := app.Append(0, labels.FromStrings("__name__", "test_metric"), 0, 100*60*1000, 0, nil, fh, storage.AOptions{})
require.NoError(t, err)
require.NoError(t, app.Commit())
h := &Handler{
localStorage: &dbAdapter{db},
lookbackDelta: 5 * time.Minute,
now: func() model.Time { return 101 * 60 * 1000 },
config: &config.Config{GlobalConfig: config.GlobalConfig{}},
options: &Options{Parser: testParser},
}
req := httptest.NewRequest(http.MethodGet, "http://example.org/federate?match[]=test_metric", http.NoBody)
req.Header.Add("Accept", `application/vnd.google.protobuf;proto=io.prometheus.client.MetricFamily;encoding=delimited`)
res := httptest.NewRecorder()
h.federation(res, req)
require.Equal(t, http.StatusOK, res.Code)
body, err := io.ReadAll(res.Body)
require.NoError(t, err)
p := textparse.NewProtobufParser(body, false, false, true, false, labels.NewSymbolTable())
var histograms int
for {
et, err := p.Next()
if errors.Is(err, io.EOF) {
break
}
require.NoError(t, err)
if et != textparse.EntryHistogram {
continue
}
histograms++
_, _, _, got := p.Histogram()
require.NotNil(t, got)
require.NoError(t, got.Validate())
require.InDelta(t, 0.3, got.Count, 1e-15)
}
require.Equal(t, 1, histograms)
}
func TestMakeClassicHistogramCount(t *testing.T) {
for name, tc := range map[string]struct {
count float64
buckets []float64
spans []histogram.Span
wantCount float64 // Zero means the bucket sum, added at runtime.
}{
"count equals bucket sum": {
count: 6,
buckets: []float64{3, 2, 1},
spans: []histogram.Span{{Offset: 0, Length: 3}},
wantCount: 6,
},
"count above bucket sum is kept": {
count: 7,
buckets: []float64{3, 2, 1},
spans: []histogram.Span{{Offset: 0, Length: 3}},
wantCount: 7,
},
"count below bucket sum without +Inf bucket": {
count: 0.3,
buckets: []float64{0.1, 0.2},
spans: []histogram.Span{{Offset: 0, Length: 2}},
},
"count below bucket sum with +Inf bucket": {
count: 0.6,
buckets: []float64{0.1, 0.2, 0.3},
spans: []histogram.Span{{Offset: 0, Length: 1}, {Offset: 1, Length: 2}},
},
} {
t.Run(name, func(t *testing.T) {
fh := &histogram.FloatHistogram{
Count: tc.count,
Schema: histogram.CustomBucketsSchema,
PositiveSpans: tc.spans,
PositiveBuckets: tc.buckets,
CustomValues: []float64{1, 2, 5},
}
require.NoError(t, fh.Validate())
if tc.wantCount == 0 {
for _, b := range tc.buckets {
tc.wantCount += b
}
require.Greater(t, tc.wantCount, tc.count)
}
got := makeClassicHistogram(fh)
require.Equal(t, tc.wantCount, got.GetSampleCountFloat())
last := got.Bucket[len(got.Bucket)-1].GetCumulativeCountFloat()
require.GreaterOrEqual(t, got.GetSampleCountFloat(), last)
})
}
}