textparse/scrape: Add option to scrape both classic and native histograms

So far, if a target exposes a histogram with both classic and native
buckets, a native-histogram enabled Prometheus would ignore the
classic buckets. With the new scrape config option
`scrape_classic_histograms` set, both buckets will be ingested,
creating all the series of a classic histogram in parallel to the
native histogram series. For example, a histogram `foo` would create a
native histogram series `foo` and classic series called `foo_sum`,
`foo_count`, and `foo_bucket`.

This feature can be used in a migration strategy from classic to
native histograms, where it is desired to have a transition period
during which both native and classic histograms are present.

Note that two bugs in classic histogram parsing were found and fixed
as a byproduct of testing the new feature:

1. Series created from classic _gauge_ histograms didn't get the
   _sum/_count/_bucket prefix set.
2. Values of classic _float_ histograms weren't parsed properly.

Signed-off-by: beorn7 <beorn@grafana.com>

config/config.go

@@ -146,13 +146,14 @@ var (
 
 	// DefaultScrapeConfig is the default scrape configuration.
 	DefaultScrapeConfig = ScrapeConfig{
-		// ScrapeTimeout and ScrapeInterval default to the
+		// ScrapeTimeout and ScrapeInterval default to the configured
-		// configured globals.
+		// globals.
-		MetricsPath:      "/metrics",
+		ScrapeClassicHistograms: false,
-		Scheme:           "http",
+		MetricsPath:             "/metrics",
-		HonorLabels:      false,
+		Scheme:                  "http",
-		HonorTimestamps:  true,
+		HonorLabels:             false,
-		HTTPClientConfig: config.DefaultHTTPClientConfig,
+		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.

docs/configuration/configuration.md

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

model/textparse/interface.go

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

model/textparse/interface_test.go

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

model/textparse/protobufparse.go

@@ -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
+	fieldPos    int
-	fieldsDone bool // true if no more fields of a Summary or (legacy) Histogram to be processed.
+	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,
+		in:                     b,
-		state:       EntryInvalid,
+		state:                  EntryInvalid,
-		mf:          &dto.MetricFamily{},
+		mf:                     &dto.MetricFamily{},
-		metricBytes: &bytes.Buffer{},
+		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 {

promql/fuzz.go

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

scrape/scrape.go

@@ -260,17 +260,18 @@ type labelLimits struct {
 }
 
 type scrapeLoopOptions struct {
-	target          *Target
+	target                  *Target
-	scraper         scraper
+	scraper                 scraper
-	sampleLimit     int
+	sampleLimit             int
-	bucketLimit     int
+	bucketLimit             int
-	labelLimits     *labelLimits
+	labelLimits             *labelLimits
-	honorLabels     bool
+	honorLabels             bool
-	honorTimestamps bool
+	honorTimestamps         bool
-	interval        time.Duration
+	interval                time.Duration
-	timeout         time.Duration
+	timeout                 time.Duration
-	mrc             []*relabel.Config
+	scrapeClassicHistograms bool
-	cache           *scrapeCache
+	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
+		honorLabels             = sp.config.HonorLabels
-		honorTimestamps = sp.config.HonorTimestamps
+		honorTimestamps         = sp.config.HonorTimestamps
-		mrc             = sp.config.MetricRelabelConfigs
+		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,
+				target:                  t,
-				scraper:         s,
+				scraper:                 s,
-				sampleLimit:     sampleLimit,
+				sampleLimit:             sampleLimit,
-				bucketLimit:     bucketLimit,
+				bucketLimit:             bucketLimit,
-				labelLimits:     labelLimits,
+				labelLimits:             labelLimits,
-				honorLabels:     honorLabels,
+				honorLabels:             honorLabels,
-				honorTimestamps: honorTimestamps,
+				honorTimestamps:         honorTimestamps,
-				mrc:             mrc,
+				mrc:                     mrc,
-				interval:        interval,
+				interval:                interval,
-				timeout:         timeout,
+				timeout:                 timeout,
+				scrapeClassicHistograms: scrapeClassicHistograms,
 			})
 			if err != nil {
 				l.setForcedError(err)
@@ -882,20 +886,21 @@ type cacheEntry struct {
 }
 
 type scrapeLoop struct {
-	scraper         scraper
+	scraper                 scraper
-	l               log.Logger
+	l                       log.Logger
-	cache           *scrapeCache
+	cache                   *scrapeCache
-	lastScrapeSize  int
+	lastScrapeSize          int
-	buffers         *pool.Pool
+	buffers                 *pool.Pool
-	jitterSeed      uint64
+	jitterSeed              uint64
-	honorTimestamps bool
+	honorTimestamps         bool
-	forcedErr       error
+	forcedErr               error
-	forcedErrMtx    sync.Mutex
+	forcedErrMtx            sync.Mutex
-	sampleLimit     int
+	sampleLimit             int
-	bucketLimit     int
+	bucketLimit             int
-	labelLimits     *labelLimits
+	labelLimits             *labelLimits
-	interval        time.Duration
+	interval                time.Duration
-	timeout         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,
+		scraper:                 sc,
-		buffers:             buffers,
+		buffers:                 buffers,
-		cache:               cache,
+		cache:                   cache,
-		appender:            appender,
+		appender:                appender,
-		sampleMutator:       sampleMutator,
+		sampleMutator:           sampleMutator,
-		reportSampleMutator: reportSampleMutator,
+		reportSampleMutator:     reportSampleMutator,
-		stopped:             make(chan struct{}),
+		stopped:                 make(chan struct{}),
-		jitterSeed:          jitterSeed,
+		jitterSeed:              jitterSeed,
-		l:                   l,
+		l:                       l,
-		parentCtx:           ctx,
+		parentCtx:               ctx,
-		appenderCtx:         appenderCtx,
+		appenderCtx:             appenderCtx,
-		honorTimestamps:     honorTimestamps,
+		honorTimestamps:         honorTimestamps,
-		sampleLimit:         sampleLimit,
+		sampleLimit:             sampleLimit,
-		bucketLimit:         bucketLimit,
+		bucketLimit:             bucketLimit,
-		labelLimits:         labelLimits,
+		labelLimits:             labelLimits,
-		interval:            interval,
+		interval:                interval,
-		timeout:             timeout,
+		timeout:                 timeout,
-		reportExtraMetrics:  reportExtraMetrics,
+		scrapeClassicHistograms: scrapeClassicHistograms,
-		appendMetadataToWAL: appendMetadataToWAL,
+		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.",

scrape/scrape_test.go

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

web/federate_test.go

@@ -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{}