mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-11-09 23:24:05 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Merge pull request #12350 from prometheus/beorn7/histogram

Browse source 
textparse/scrape: Add option to scrape both classic and native histograms
This commit is contained in:
Björn Rabenstein 2023-05-13 02:16:11 +02:00 committed by GitHub
parent 5c5fa5c319 9e500345f3
commit b727e69b76
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 1148 additions and 486 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
config/config.go
View file

@ -146,13 +146,14 @@ var (
// DefaultScrapeConfig is the default scrape configuration.
DefaultScrapeConfig = ScrapeConfig{
// ScrapeTimeout and ScrapeInterval default to the
// configured globals.
MetricsPath: "/metrics",
Scheme: "http",
HonorLabels: false,
HonorTimestamps: true,
HTTPClientConfig: config.DefaultHTTPClientConfig,
// ScrapeTimeout and ScrapeInterval default to the configured
// globals.
ScrapeClassicHistograms: false,
MetricsPath: "/metrics",
Scheme: "http",
HonorLabels: false,
HonorTimestamps: true,
HTTPClientConfig: config.DefaultHTTPClientConfig,
}
// DefaultAlertmanagerConfig is the default alertmanager configuration.
@ -467,6 +468,8 @@ type ScrapeConfig struct {
ScrapeInterval model.Duration `yaml:"scrape_interval,omitempty"`
// The timeout for scraping targets of this config.
ScrapeTimeout model.Duration `yaml:"scrape_timeout,omitempty"`
// Whether to scrape a classic histogram that is also exposed as a native histogram.
ScrapeClassicHistograms bool `yaml:"scrape_classic_histograms,omitempty"`
// The HTTP resource path on which to fetch metrics from targets.
MetricsPath string `yaml:"metrics_path,omitempty"`
// The URL scheme with which to fetch metrics from targets.

4
docs/configuration/configuration.md
View file

@ -134,6 +134,10 @@ job_name: <job_name>
# Per-scrape timeout when scraping this job.
[ scrape_timeout: <duration> | default = <global_config.scrape_timeout> ]
# Whether to scrape a classic histogram that is also exposed as a native
# histogram (has no effect without --enable-feature=native-histograms).
[ scrape_classic_histograms: <boolean> | default = false ]
# The HTTP resource path on which to fetch metrics from targets.
[ metrics_path: <path> | default = /metrics ]

6
model/textparse/interface.go
View file

@ -71,7 +71,7 @@ type Parser interface {
//
// This function always returns a valid parser, but might additionally
// return an error if the content type cannot be parsed.
func New(b []byte, contentType string) (Parser, error) {
func New(b []byte, contentType string, parseClassicHistograms bool) (Parser, error) {
if contentType == "" {
return NewPromParser(b), nil
}
@ -84,7 +84,7 @@ func New(b []byte, contentType string) (Parser, error) {
case "application/openmetrics-text":
return NewOpenMetricsParser(b), nil
case "application/vnd.google.protobuf":
return NewProtobufParser(b), nil
return NewProtobufParser(b, parseClassicHistograms), nil
default:
return NewPromParser(b), nil
}
@ -100,7 +100,7 @@ const (
EntrySeries Entry = 2 // A series with a simple float64 as value.
EntryComment Entry = 3
EntryUnit Entry = 4
EntryHistogram Entry = 5 // A series with a sparse histogram as a value.
EntryHistogram Entry = 5 // A series with a native histogram as a value.
)
// MetricType represents metric type values.

2
model/textparse/interface_test.go
View file

@ -91,7 +91,7 @@ func TestNewParser(t *testing.T) {
tt := tt // Copy to local variable before going parallel.
t.Parallel()
p, err := New([]byte{}, tt.contentType)
p, err := New([]byte{}, tt.contentType, false)
tt.validateParser(t, p)
if tt.err == "" {
require.NoError(t, err)

49
model/textparse/protobufparse.go
View file

@ -52,8 +52,10 @@ type ProtobufParser struct {
// fieldPos is the position within a Summary or (legacy) Histogram. -2
// is the count. -1 is the sum. Otherwise it is the index within
// quantiles/buckets.
fieldPos int
fieldsDone bool // true if no more fields of a Summary or (legacy) Histogram to be processed.
fieldPos int
fieldsDone bool // true if no more fields of a Summary or (legacy) Histogram to be processed.
redoClassic bool // true after parsing a native histogram if we need to parse it again as a classit histogram.
// state is marked by the entry we are processing. EntryInvalid implies
// that we have to decode the next MetricFamily.
state Entry
@ -62,17 +64,22 @@ type ProtobufParser struct {
mf *dto.MetricFamily
// Wether to also parse a classic histogram that is also present as a
// native histogram.
parseClassicHistograms bool
// The following are just shenanigans to satisfy the Parser interface.
metricBytes *bytes.Buffer // A somewhat fluid representation of the current metric.
}
// NewProtobufParser returns a parser for the payload in the byte slice.
func NewProtobufParser(b []byte) Parser {
func NewProtobufParser(b []byte, parseClassicHistograms bool) Parser {
return &ProtobufParser{
in: b,
state: EntryInvalid,
mf: &dto.MetricFamily{},
metricBytes: &bytes.Buffer{},
in: b,
state: EntryInvalid,
mf: &dto.MetricFamily{},
metricBytes: &bytes.Buffer{},
parseClassicHistograms: parseClassicHistograms,
}
}
@ -106,19 +113,28 @@ func (p *ProtobufParser) Series() ([]byte, *int64, float64) {
v = s.GetQuantile()[p.fieldPos].GetValue()
}
case dto.MetricType_HISTOGRAM, dto.MetricType_GAUGE_HISTOGRAM:
// This should only happen for a legacy histogram.
// This should only happen for a classic histogram.
h := m.GetHistogram()
switch p.fieldPos {
case -2:
v = float64(h.GetSampleCount())
v = h.GetSampleCountFloat()
if v == 0 {
v = float64(h.GetSampleCount())
}
case -1:
v = h.GetSampleSum()
default:
bb := h.GetBucket()
if p.fieldPos >= len(bb) {
v = float64(h.GetSampleCount())
v = h.GetSampleCountFloat()
if v == 0 {
v = float64(h.GetSampleCount())
}
} else {
v = float64(bb[p.fieldPos].GetCumulativeCount())
v = bb[p.fieldPos].GetCumulativeCountFloat()
if v == 0 {
v = float64(bb[p.fieldPos].GetCumulativeCount())
}
}
}
default:
@ -149,6 +165,9 @@ func (p *ProtobufParser) Histogram() ([]byte, *int64, *histogram.Histogram, *his
ts = m.GetTimestampMs()
h = m.GetHistogram()
)
if p.parseClassicHistograms && len(h.GetBucket()) > 0 {
p.redoClassic = true
}
if h.GetSampleCountFloat() > 0 || h.GetZeroCountFloat() > 0 {
// It is a float histogram.
fh := histogram.FloatHistogram{
@ -376,6 +395,12 @@ func (p *ProtobufParser) Next() (Entry, error) {
return EntryInvalid, err
}
case EntryHistogram, EntrySeries:
if p.redoClassic {
p.redoClassic = false
p.state = EntrySeries
p.fieldPos = -3
p.fieldsDone = false
}
t := p.mf.GetType()
if p.state == EntrySeries && !p.fieldsDone &&
(t == dto.MetricType_SUMMARY ||
@ -432,7 +457,7 @@ func (p *ProtobufParser) updateMetricBytes() error {
// state.
func (p *ProtobufParser) getMagicName() string {
t := p.mf.GetType()
if p.state == EntryHistogram || (t != dto.MetricType_HISTOGRAM && t != dto.MetricType_SUMMARY) {
if p.state == EntryHistogram || (t != dto.MetricType_HISTOGRAM && t != dto.MetricType_GAUGE_HISTOGRAM && t != dto.MetricType_SUMMARY) {
return p.mf.GetName()
}
if p.fieldPos == -2 {

1393
model/textparse/protobufparse_test.go
View file

File diff suppressed because it is too large Load diff

2
promql/fuzz.go
View file

@ -58,7 +58,7 @@ const (
)
func fuzzParseMetricWithContentType(in []byte, contentType string) int {
p, warning := textparse.New(in, contentType)
p, warning := textparse.New(in, contentType, false)
if warning != nil {
// An invalid content type is being passed, which should not happen
// in this context.

123
scrape/scrape.go
View file

@ -260,17 +260,18 @@ type labelLimits struct {
}
type scrapeLoopOptions struct {
target *Target
scraper scraper
sampleLimit int
bucketLimit int
labelLimits *labelLimits
honorLabels bool
honorTimestamps bool
interval time.Duration
timeout time.Duration
mrc []*relabel.Config
cache *scrapeCache
target *Target
scraper scraper
sampleLimit int
bucketLimit int
labelLimits *labelLimits
honorLabels bool
honorTimestamps bool
interval time.Duration
timeout time.Duration
scrapeClassicHistograms bool
mrc []*relabel.Config
cache *scrapeCache
}
const maxAheadTime = 10 * time.Minute
@ -331,6 +332,7 @@ func newScrapePool(cfg *config.ScrapeConfig, app storage.Appendable, jitterSeed
opts.labelLimits,
opts.interval,
opts.timeout,
opts.scrapeClassicHistograms,
options.ExtraMetrics,
options.EnableMetadataStorage,
opts.target,
@ -547,9 +549,10 @@ func (sp *scrapePool) sync(targets []*Target) {
labelNameLengthLimit: int(sp.config.LabelNameLengthLimit),
labelValueLengthLimit: int(sp.config.LabelValueLengthLimit),
}
honorLabels = sp.config.HonorLabels
honorTimestamps = sp.config.HonorTimestamps
mrc = sp.config.MetricRelabelConfigs
honorLabels = sp.config.HonorLabels
honorTimestamps = sp.config.HonorTimestamps
mrc = sp.config.MetricRelabelConfigs
scrapeClassicHistograms = sp.config.ScrapeClassicHistograms
)
sp.targetMtx.Lock()
@ -568,16 +571,17 @@ func (sp *scrapePool) sync(targets []*Target) {
}
s := &targetScraper{Target: t, client: sp.client, timeout: timeout, bodySizeLimit: bodySizeLimit, acceptHeader: acceptHeader}
l := sp.newLoop(scrapeLoopOptions{
target: t,
scraper: s,
sampleLimit: sampleLimit,
bucketLimit: bucketLimit,
labelLimits: labelLimits,
honorLabels: honorLabels,
honorTimestamps: honorTimestamps,
mrc: mrc,
interval: interval,
timeout: timeout,
target: t,
scraper: s,
sampleLimit: sampleLimit,
bucketLimit: bucketLimit,
labelLimits: labelLimits,
honorLabels: honorLabels,
honorTimestamps: honorTimestamps,
mrc: mrc,
interval: interval,
timeout: timeout,
scrapeClassicHistograms: scrapeClassicHistograms,
})
if err != nil {
l.setForcedError(err)
@ -882,20 +886,21 @@ type cacheEntry struct {
}
type scrapeLoop struct {
scraper scraper
l log.Logger
cache *scrapeCache
lastScrapeSize int
buffers *pool.Pool
jitterSeed uint64
honorTimestamps bool
forcedErr error
forcedErrMtx sync.Mutex
sampleLimit int
bucketLimit int
labelLimits *labelLimits
interval time.Duration
timeout time.Duration
scraper scraper
l log.Logger
cache *scrapeCache
lastScrapeSize int
buffers *pool.Pool
jitterSeed uint64
honorTimestamps bool
forcedErr error
forcedErrMtx sync.Mutex
sampleLimit int
bucketLimit int
labelLimits *labelLimits
interval time.Duration
timeout time.Duration
scrapeClassicHistograms bool
appender func(ctx context.Context) storage.Appender
sampleMutator labelsMutator
@ -1177,6 +1182,7 @@ func newScrapeLoop(ctx context.Context,
labelLimits *labelLimits,
interval time.Duration,
timeout time.Duration,
scrapeClassicHistograms bool,
reportExtraMetrics bool,
appendMetadataToWAL bool,
target *Target,
@ -1204,25 +1210,26 @@ func newScrapeLoop(ctx context.Context,
}
sl := &scrapeLoop{
scraper: sc,
buffers: buffers,
cache: cache,
appender: appender,
sampleMutator: sampleMutator,
reportSampleMutator: reportSampleMutator,
stopped: make(chan struct{}),
jitterSeed: jitterSeed,
l: l,
parentCtx: ctx,
appenderCtx: appenderCtx,
honorTimestamps: honorTimestamps,
sampleLimit: sampleLimit,
bucketLimit: bucketLimit,
labelLimits: labelLimits,
interval: interval,
timeout: timeout,
reportExtraMetrics: reportExtraMetrics,
appendMetadataToWAL: appendMetadataToWAL,
scraper: sc,
buffers: buffers,
cache: cache,
appender: appender,
sampleMutator: sampleMutator,
reportSampleMutator: reportSampleMutator,
stopped: make(chan struct{}),
jitterSeed: jitterSeed,
l: l,
parentCtx: ctx,
appenderCtx: appenderCtx,
honorTimestamps: honorTimestamps,
sampleLimit: sampleLimit,
bucketLimit: bucketLimit,
labelLimits: labelLimits,
interval: interval,
timeout: timeout,
scrapeClassicHistograms: scrapeClassicHistograms,
reportExtraMetrics: reportExtraMetrics,
appendMetadataToWAL: appendMetadataToWAL,
}
sl.ctx, sl.cancel = context.WithCancel(ctx)
@ -1492,7 +1499,7 @@ type appendErrors struct {
}
func (sl *scrapeLoop) append(app storage.Appender, b []byte, contentType string, ts time.Time) (total, added, seriesAdded int, err error) {
p, err := textparse.New(b, contentType)
p, err := textparse.New(b, contentType, sl.scrapeClassicHistograms)
if err != nil {
level.Debug(sl.l).Log(
"msg", "Invalid content type on scrape, using prometheus parser as fallback.",

36
scrape/scrape_test.go
View file

@ -633,6 +633,7 @@ func TestScrapeLoopStopBeforeRun(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -705,6 +706,7 @@ func TestScrapeLoopStop(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -781,6 +783,7 @@ func TestScrapeLoopRun(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -836,6 +839,7 @@ func TestScrapeLoopRun(t *testing.T) {
100*time.Millisecond,
false,
false,
false,
nil,
false,
)
@ -895,6 +899,7 @@ func TestScrapeLoopForcedErr(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -953,6 +958,7 @@ func TestScrapeLoopMetadata(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1010,6 +1016,7 @@ func simpleTestScrapeLoop(t testing.TB) (context.Context, *scrapeLoop) {
0,
false,
false,
false,
nil,
false,
)
@ -1070,6 +1077,7 @@ func TestScrapeLoopFailWithInvalidLabelsAfterRelabel(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1148,6 +1156,7 @@ func TestScrapeLoopRunCreatesStaleMarkersOnFailedScrape(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -1211,6 +1220,7 @@ func TestScrapeLoopRunCreatesStaleMarkersOnParseFailure(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -1277,6 +1287,7 @@ func TestScrapeLoopCache(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -1360,6 +1371,7 @@ func TestScrapeLoopCacheMemoryExhaustionProtection(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -1474,6 +1486,7 @@ func TestScrapeLoopAppend(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1563,7 +1576,7 @@ func TestScrapeLoopAppendForConflictingPrefixedLabels(t *testing.T) {
return mutateSampleLabels(l, &Target{labels: labels.FromStrings(tc.targetLabels...)}, false, nil)
},
nil,
func(ctx context.Context) storage.Appender { return app }, nil, 0, true, 0, 0, nil, 0, 0, false, false, nil, false,
func(ctx context.Context) storage.Appender { return app }, nil, 0, true, 0, 0, nil, 0, 0, false, false, false, nil, false,
)
slApp := sl.appender(context.Background())
_, _, _, err := sl.append(slApp, []byte(tc.exposedLabels), "", time.Date(2000, 1, 1, 1, 0, 0, 0, time.UTC))
@ -1600,6 +1613,7 @@ func TestScrapeLoopAppendCacheEntryButErrNotFound(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1607,7 +1621,7 @@ func TestScrapeLoopAppendCacheEntryButErrNotFound(t *testing.T) {
fakeRef := storage.SeriesRef(1)
expValue := float64(1)
metric := []byte(`metric{n="1"} 1`)
p, warning := textparse.New(metric, "")
p, warning := textparse.New(metric, "", false)
require.NoError(t, warning)
var lset labels.Labels
@ -1658,6 +1672,7 @@ func TestScrapeLoopAppendSampleLimit(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1735,6 +1750,7 @@ func TestScrapeLoop_HistogramBucketLimit(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1833,6 +1849,7 @@ func TestScrapeLoop_ChangingMetricString(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1881,6 +1898,7 @@ func TestScrapeLoopAppendStaleness(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -1932,6 +1950,7 @@ func TestScrapeLoopAppendNoStalenessIfTimestamp(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2043,6 +2062,7 @@ metric_total{n="2"} 2 # {t="2"} 2.0 20000
0,
false,
false,
false,
nil,
false,
)
@ -2108,6 +2128,7 @@ func TestScrapeLoopAppendExemplarSeries(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2160,6 +2181,7 @@ func TestScrapeLoopRunReportsTargetDownOnScrapeError(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -2196,6 +2218,7 @@ func TestScrapeLoopRunReportsTargetDownOnInvalidUTF8(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -2245,6 +2268,7 @@ func TestScrapeLoopAppendGracefullyIfAmendOrOutOfOrderOrOutOfBounds(t *testing.T
0,
false,
false,
false,
nil,
false,
)
@ -2290,6 +2314,7 @@ func TestScrapeLoopOutOfBoundsTimeError(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2562,6 +2587,7 @@ func TestScrapeLoop_RespectTimestamps(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2603,6 +2629,7 @@ func TestScrapeLoop_DiscardTimestamps(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2643,6 +2670,7 @@ func TestScrapeLoopDiscardDuplicateLabels(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2701,6 +2729,7 @@ func TestScrapeLoopDiscardUnnamedMetrics(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -2964,6 +2993,7 @@ func TestScrapeAddFast(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)
@ -3050,6 +3080,7 @@ func TestScrapeReportSingleAppender(t *testing.T) {
time.Hour,
false,
false,
false,
nil,
false,
)
@ -3252,6 +3283,7 @@ func TestScrapeLoopLabelLimit(t *testing.T) {
0,
false,
false,
false,
nil,
false,
)

2
web/federate_test.go
View file

@ -378,7 +378,7 @@ func TestFederationWithNativeHistograms(t *testing.T) {
body, err := io.ReadAll(res.Body)
require.NoError(t, err)
p := textparse.NewProtobufParser(body)
p := textparse.NewProtobufParser(body, false)
var actVec promql.Vector
metricFamilies := 0
l := labels.Labels{}