@cryptotaxi247 / netdata-1 / commits / 415f0fcf8

chore(go/pkg/prometheus): unified single-pass stream parser (#22640)

Ilya Mashchenko committed Jun 6, 2026 at 00:39 UTC 415f0fcf8d4bf35956182b158140b60ab25e5ede
8 files changed +1083 -267
src/go/pkg/prometheus/client.go
+11
@@ -23,6 +23,17 @@ type (
23 // ScrapeSeries and parse prometheus format metrics
24 ScrapeSeries() (Series, error)
25 Scrape() (MetricFamilies, error)
26 + // ScrapeStream scrapes and invokes onSample for every sample — a flat
27 + // [Sample] stream before typed-family assembly — with the selector (if
28 + // any) already applied; onHelp, if non-nil, receives per-family HELP. It
29 + // stops and returns the first error from a callback.
30 + //
31 + // Order is exposition order, with one exception: a _sum/_count sample
32 + // whose family type is not yet known (its # TYPE, first bucket, or first
33 + // quantile has not appeared) is deferred and emitted once the type
34 + // resolves, or at end of stream — so it may arrive after a later,
35 + // unrelated sample. Each sample's Kind and FamilyType are always correct.
36 + ScrapeStream(onHelp func(name, help string), onSample func(Sample) error) error
37 HTTPClient() *http.Client
38 }
39
src/go/pkg/prometheus/client_test.go
+37
@@ -5,6 +5,7 @@ package prometheus
5 import (
6 "bytes"
7 "compress/gzip"
8 + "errors"
9 "net/http"
10 "net/http/httptest"
11 "os"
@@ -131,6 +132,42 @@ func TestPrometheusReadFromFile(t *testing.T) {
132 }
133 }
134
135 +func TestPrometheusScrapeStream(t *testing.T) {
136 + errBoom := errors.New("boom")
137 +
138 + tests := map[string]struct {
139 + onSampleErr error // returned by onSample (nil = stream everything)
140 + wantErr error
141 + }{
142 + "streams all samples and help": {},
143 + "onSample error propagates": {onSampleErr: errBoom, wantErr: errBoom},
144 + }
145 +
146 + for name, tc := range tests {
147 + t.Run(name, func(t *testing.T) {
148 + prom := New(http.DefaultClient, web.RequestConfig{URL: "file://testdata/testdata.txt"})
149 +
150 + var samples int
151 + var help []string
152 + err := prom.ScrapeStream(
153 + func(name, _ string) { help = append(help, name) },
154 + func(Sample) error {
155 + samples++
156 + return tc.onSampleErr
157 + },
158 + )
159 +
160 + if tc.wantErr != nil {
161 + assert.ErrorIs(t, err, tc.wantErr)
162 + return
163 + }
164 + require.NoError(t, err)
165 + assert.Positive(t, samples)
166 + assert.Contains(t, help, "go_gc_duration_seconds")
167 + })
168 + }
169 +}
170 +
171 func verifyTestData(t *testing.T, ms Series) {
172 assert.Equal(t, 410, len(ms))
173 assert.Equal(t, "go_gc_duration_seconds", ms[0].Labels.Get("__name__"))
src/go/pkg/prometheus/doc.go new
+19
@@ -0,0 +1,19 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +// Package prometheus scrapes and parses the Prometheus text exposition format.
4 +//
5 +// One parse pass drives three outputs from a single [New] / [NewWithSelector]
6 +// instance:
7 +//
8 +// - [Prometheus.Scrape] assembles typed [MetricFamilies] — gauges, counters,
9 +// summaries, and histograms — folding _sum/_count/_bucket and quantile series
10 +// into their families.
11 +// - [Prometheus.ScrapeSeries] returns the raw [Series]: one [SeriesSample] per
12 +// scraped series, with labels in textparse-sorted order.
13 +// - [Prometheus.ScrapeStream] exposes a flat [Sample] stream before typed
14 +// assembly — the form a Prometheus metric-relabeling step operates on.
15 +//
16 +// Results are valid only until the next scrape on the same instance; buffers are
17 +// reused across scrapes. An optional selector (see the selector subpackage)
18 +// filters series during the parse.
19 +package prometheus
src/go/pkg/prometheus/model.go new
+44
@@ -0,0 +1,44 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package prometheus
4 +
5 +import (
6 + "github.com/prometheus/common/model"
7 + "github.com/prometheus/prometheus/model/labels"
8 +)
9 +
10 +// SampleKind classifies a streamed [Sample] by the role it plays in its typed
11 +// family. The driver assigns it as it parses, so a consumer can re-assemble
12 +// typed families (or relabel) without re-deriving the role from the name.
13 +type SampleKind uint8
14 +
15 +const (
16 + // SampleKindScalar is a plain sample (gauge, counter, untyped, or the base
17 + // series of a summary/histogram). Interpret it together with FamilyType.
18 + SampleKindScalar SampleKind = iota
19 + SampleKindHistogramBucket
20 + SampleKindHistogramSum
21 + SampleKindHistogramCount
22 + SampleKindSummaryQuantile
23 + SampleKindSummarySum
24 + SampleKindSummaryCount
25 +)
26 +
27 +// Sample is a single scraped series exposed before typed-family assembly.
28 +//
29 +// Name is the __name__ label value (found by lookup, not by position — do not
30 +// assume label index 0). Labels holds every other label, including structural
31 +// labels such as "le" (histogram buckets) and "quantile" (summary quantiles) —
32 +// textparse canonicalizes these to floats (e.g. "1" -> "1.0") only when the family
33 +// type is known from # TYPE, otherwise leaving them raw, so do not assume a fixed
34 +// form when matching. Labels never contains __name__. Value is the sample value.
35 +// Kind and FamilyType carry the classification the driver derived for this sample.
36 +//
37 +// Sample is the unit a Prometheus metric-relabeling step operates on.
38 +type Sample struct {
39 + Name string
40 + Labels labels.Labels
41 + Value float64
42 + Kind SampleKind
43 + FamilyType model.MetricType
44 +}
src/go/pkg/prometheus/parse.go
+564 -267
@@ -1,10 +1,11 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 package prometheus
4
5 import (
6 "errors"
7 "fmt"
8 "io"
7 - "regexp"
9 "strconv"
10 "strings"
11
@@ -26,68 +27,102 @@ const (
27 bucketSuffix = "_bucket"
28 )
29
30 +// promTextParser orchestrates a single parse pass. The driver parses the
31 +// exposition once into a flat sample stream; the assembler folds that stream
32 +// into typed MetricFamilies. Scrape() (families), ScrapeSeries() (Series), and
33 +// the exported sample stream are all produced from this one model.
34 type promTextParser struct {
30 - metrics MetricFamilies
31 - series Series
32 -
35 sr selector.Selector
36
35 - currMF *MetricFamily
36 - currSeries labels.Labels
37 + driver parseDriver
38 + asm assembler
39 + series Series
40 +}
41
38 - summaries map[uint64]*Summary
39 - histograms map[uint64]*Histogram
42 +func (p *promTextParser) parseToMetricFamilies(text []byte) (MetricFamilies, error) {
43 + p.driver.sr = p.sr
44 + p.asm.reset()
45
41 - isCount bool
42 - isSum bool
43 - isQuantile bool
44 - isBucket bool
46 + // ownLabels=false: the assembler copies labels into its own buffers, so the
47 + // driver may lend the scratch label set (the no-allocation fast path).
48 + if err := p.driver.parseSamples(text, false, p.asm.applyHelp, p.asm.applySample); err != nil {
49 + return nil, err
50 + }
51
46 - currQuantile float64
47 - currBucket float64
52 + return p.asm.families(), nil
53 }
54
55 func (p *promTextParser) parseToSeries(text []byte) (Series, error) {
56 + p.driver.sr = p.sr
57 p.series.Reset()
58
53 - parser := textparse.NewPromParser(text, labels.NewSymbolTable())
54 - for {
55 - entry, err := parser.Next()
56 - if err != nil {
57 - if errors.Is(err, io.EOF) {
58 - break
59 - }
60 - if entry == textparse.EntryInvalid && strings.HasPrefix(err.Error(), "invalid metric type") {
61 - continue
62 - }
63 - return nil, fmt.Errorf("failed to parse prometheus metrics: %v", err)
64 - }
65 -
66 - switch entry {
67 - case textparse.EntrySeries:
68 - p.currSeries = p.currSeries[:0]
59 + // Series keeps the raw label set straight from textparse (sorted, with __name__
60 + // in its sorted position) — identical to the legacy parser. It does NOT go
61 + // through the Sample model (which separates __name__), so there is no deferral,
62 + // no reordering, and the sorted-label invariant is preserved.
63 + err := p.driver.iterate(text, nil, nil, func(series labels.Labels, value float64) error {
64 + p.series.Add(SeriesSample{Labels: copyLabels(series), Value: value})
65 + return nil
66 + })
67 + if err != nil {
68 + return nil, err
69 + }
70
70 - parser.Metric(&p.currSeries)
71 + p.series.Sort()
72
72 - if p.sr != nil && !p.sr.Matches(p.currSeries) {
73 - continue
74 - }
73 + return p.series, nil
74 +}
75
76 - _, _, val := parser.Series()
77 - p.series.Add(SeriesSample{Labels: copyLabels(p.currSeries), Value: val})
78 - }
76 +func (p *promTextParser) parseToStream(text []byte, onHelp func(name, help string), onSample func(Sample) error) error {
77 + if onSample == nil {
78 + return nil
79 }
80 + p.driver.sr = p.sr
81
81 - p.series.Sort()
82 + // ownLabels=true: each Sample must own its labels because the consumer may
83 + // retain or mutate them (e.g. relabeling) past the next sample.
84 + return p.driver.parseSamples(text, true, onHelp, onSample)
85 +}
86
83 - return p.series, nil
87 +// parseDriver runs the single exposition parse pass (iterate). On top of it,
88 +// parseSamples emits a flat, classified sample stream, deferring a _sum/_count
89 +// whose family type is not yet known and back-resolving it once the type appears
90 +// (a later # TYPE, _bucket, or quantile series) or at EOF. familyTypes/pending
91 +// hold that deferral state.
92 +type parseDriver struct {
93 + sr selector.Selector
94 +
95 + familyTypes map[string]model.MetricType
96 + pending []pendingSample
97 + currSeries labels.Labels
98 }
99
86 -var reSpace = regexp.MustCompile(`\s+`)
100 +type pendingSample struct {
101 + baseName string
102 + sample Sample
103 + role pendingRole
104 +}
105
88 -func (p *promTextParser) parseToMetricFamilies(text []byte) (MetricFamilies, error) {
89 - p.reset()
106 +type pendingRole uint8
107
108 +const (
109 + pendingNone pendingRole = iota
110 + pendingSum
111 + pendingCount
112 +)
113 +
114 +// iterate runs the shared single-pass exposition loop. For every series it calls
115 +// onSeries with the raw label set (textparse order — sorted, __name__ in its
116 +// sorted position) and value, after applying the selector; onHelp/onType deliver
117 +// per-family metadata. This is the one parse loop: ScrapeSeries consumes it
118 +// directly (raw labels, byte-identical to the legacy parser), while the flat
119 +// sample stream is layered on top by parseSamples.
120 +func (d *parseDriver) iterate(
121 + text []byte,
122 + onHelp func(name, help string),
123 + onType func(name string, typ model.MetricType) error,
124 + onSeries func(series labels.Labels, value float64) error,
125 +) error {
126 parser := textparse.NewPromParser(text, labels.NewSymbolTable())
127 for {
128 entry, err := parser.Next()
@@ -98,312 +133,580 @@ func (p *promTextParser) parseToMetricFamilies(text []byte) (MetricFamilies, err
133 if entry == textparse.EntryInvalid && strings.HasPrefix(err.Error(), "invalid metric type") {
134 continue
135 }
101 - return nil, fmt.Errorf("failed to parse prometheus metrics: %v", err)
136 + return fmt.Errorf("failed to parse prometheus metrics: %v", err)
137 }
138
139 switch entry {
140 case textparse.EntryHelp:
106 - name, help := parser.Help()
107 - p.setMetricFamilyByName(string(name))
108 - p.currMF.help = string(help)
109 - if strings.IndexByte(p.currMF.help, '\n') != -1 {
110 - // convert multiline to one line because HELP is used as the chart title.
111 - p.currMF.help = reSpace.ReplaceAllString(strings.TrimSpace(p.currMF.help), " ")
141 + if onHelp != nil {
142 + name, help := parser.Help()
143 + onHelp(string(name), sanitizeHelp(string(help)))
144 }
145 case textparse.EntryType:
114 - name, typ := parser.Type()
115 - p.setMetricFamilyByName(string(name))
116 - p.currMF.typ = typ
146 + if onType != nil {
147 + name, typ := parser.Type()
148 + if err := onType(string(name), typ); err != nil {
149 + return err
150 + }
151 + }
152 case textparse.EntrySeries:
118 - p.currSeries = p.currSeries[:0]
153 + d.currSeries = d.currSeries[:0]
154 + parser.Metric(&d.currSeries)
155
120 - parser.Metric(&p.currSeries)
121 -
122 - if p.sr != nil && !p.sr.Matches(p.currSeries) {
156 + if d.sr != nil && !d.sr.Matches(d.currSeries) {
157 continue
158 }
159
126 - p.setMetricFamilyBySeries()
127 -
160 _, _, value := parser.Series()
161
130 - switch p.currMF.typ {
131 - case model.MetricTypeGauge:
132 - p.addGauge(value)
133 - case model.MetricTypeCounter:
134 - p.addCounter(value)
135 - case model.MetricTypeSummary:
136 - p.addSummary(value)
137 - case model.MetricTypeHistogram:
138 - p.addHistogram(value)
139 - case model.MetricTypeUnknown:
140 - p.addUnknown(value)
162 + if onSeries != nil {
163 + if err := onSeries(d.currSeries, value); err != nil {
164 + return err
165 + }
166 }
167 }
168 }
169
145 - for k, v := range p.metrics {
146 - if len(v.Metrics()) == 0 {
147 - delete(p.metrics, k)
170 + return nil
171 +}
172 +
173 +// parseSamples layers the flat, classified sample stream on top of iterate. It
174 +// turns each series into a Sample (Kind + FamilyType) and defers a _sum/_count
175 +// whose family type is not yet known, back-resolving it once the type appears (a
176 +// later # TYPE, _bucket, or quantile series) or flushing it at EOF. Deferral can
177 +// emit a _sum/_count after a later, unrelated series — see ScrapeStream's doc.
178 +func (d *parseDriver) parseSamples(text []byte, ownLabels bool, onHelp func(name, help string), onSample func(Sample) error) error {
179 + d.reset()
180 +
181 + err := d.iterate(text, onHelp,
182 + func(name string, typ model.MetricType) error {
183 + d.familyTypes[name] = typ
184 + var err error
185 + d.pending, err = emitResolvedPending(d.pending, name, typ, onSample)
186 + return err
187 + },
188 + func(series labels.Labels, value float64) error {
189 + sample, baseName, role, ok := d.makeSample(series, value, ownLabels)
190 + if !ok {
191 + return nil
192 + }
193 +
194 + // A quantile/bucket series reveals the family type; back-resolve any
195 + // _sum/_count buffered before it.
196 + switch sample.Kind {
197 + case SampleKindSummaryQuantile:
198 + var err error
199 + d.pending, err = emitResolvedPending(d.pending, sample.Name, model.MetricTypeSummary, onSample)
200 + if err != nil {
201 + return err
202 + }
203 + case SampleKindHistogramBucket:
204 + var err error
205 + d.pending, err = emitResolvedPending(d.pending, strings.TrimSuffix(sample.Name, bucketSuffix), model.MetricTypeHistogram, onSample)
206 + if err != nil {
207 + return err
208 + }
209 + }
210 +
211 + if role != pendingNone {
212 + if !ownLabels {
213 + sample.Labels = copyLabels(sample.Labels)
214 + }
215 + d.pending = append(d.pending, pendingSample{
216 + baseName: baseName,
217 + sample: sample,
218 + role: role,
219 + })
220 + return nil
221 + }
222 +
223 + return onSample(sample)
224 + },
225 + )
226 + if err != nil {
227 + return err
228 + }
229 +
230 + // Flush still-unresolved _sum/_count as plain scalars (matches the legacy
231 + // behavior for a _sum/_count whose family type never appears).
232 + for _, ps := range d.pending {
233 + if err := onSample(ps.sample); err != nil {
234 + return err
235 }
236 }
237 + d.pending = d.pending[:0]
238
151 - return p.metrics, nil
239 + return nil
240 }
241
154 -func (p *promTextParser) setMetricFamilyByName(name string) {
155 - mf, ok := p.metrics[name]
242 +func (d *parseDriver) makeSample(series labels.Labels, value float64, ownLabels bool) (Sample, string, pendingRole, bool) {
243 + name, ok := metricNameValue(series)
244 if !ok {
157 - mf = &MetricFamily{name: name, typ: model.MetricTypeUnknown}
158 - p.metrics[name] = mf
245 + return Sample{}, "", pendingNone, false
246 }
160 - p.currMF = mf
161 -}
247
163 -func (p *promTextParser) setMetricFamilyBySeries() {
164 - p.isSum, p.isCount, p.isQuantile, p.isBucket = false, false, false, false
165 - p.currQuantile, p.currBucket = 0, 0
248 + var lbs labels.Labels
249 + if ownLabels {
250 + lbs = copyLabelsWithoutName(series)
251 + } else {
252 + lbs, _, _ = removeLabel(series, labels.MetricName)
253 + }
254
167 - name, ok := metricNameValue(p.currSeries)
168 - if !ok {
169 - p.currMF = nil
170 - return
255 + sample := Sample{
256 + Name: name,
257 + Labels: lbs,
258 + Value: value,
259 + Kind: SampleKindScalar,
260 + FamilyType: d.familyTypes[name],
261 + }
262 + if sample.FamilyType == "" {
263 + sample.FamilyType = model.MetricTypeUnknown
264 }
265
173 - if p.currMF != nil && p.currMF.name == name {
174 - if p.currMF.typ == model.MetricTypeSummary {
175 - p.setQuantile()
266 + if sample.Labels.Has(quantileLabel) {
267 + if sample.FamilyType != model.MetricTypeUnknown && sample.FamilyType != model.MetricTypeSummary {
268 + return sample, "", pendingNone, true
269 }
177 - return
270 + sample.Kind = SampleKindSummaryQuantile
271 + sample.FamilyType = model.MetricTypeSummary
272 + d.familyTypes[name] = model.MetricTypeSummary
273 + return sample, "", pendingNone, true
274 }
275
180 - typ := model.MetricTypeUnknown
276 + // A histogram bucket requires an "le" label. A _bucket-named series without le
277 + // is malformed: it is NOT treated as a bucket but falls through to a plain
278 + // metric, preserving its value. (The legacy parser folded such a series into
279 + // the histogram family and dropped its value; valid buckets always carry le,
280 + // so real input is unaffected.)
281 + if strings.HasSuffix(name, bucketSuffix) && sample.Labels.Has(bucketLabel) {
282 + if sample.FamilyType != model.MetricTypeUnknown && sample.FamilyType != model.MetricTypeHistogram {
283 + return sample, "", pendingNone, true
284 + }
285 + baseName := strings.TrimSuffix(name, bucketSuffix)
286 + sample.Kind = SampleKindHistogramBucket
287 + sample.FamilyType = model.MetricTypeHistogram
288 + d.familyTypes[baseName] = model.MetricTypeHistogram
289 + return sample, "", pendingNone, true
290 + }
291
182 - switch {
183 - case strings.HasSuffix(name, sumSuffix):
184 - n := strings.TrimSuffix(name, sumSuffix)
185 - if mf, ok := p.metrics[n]; ok && isSummaryOrHistogram(mf.typ) {
186 - p.isSum = true
187 - _ = setLabelValue(p.currSeries, labels.MetricName, n)
188 - p.currMF = mf
189 - return
292 + if strings.HasSuffix(name, sumSuffix) {
293 + if sample.FamilyType != model.MetricTypeUnknown &&
294 + sample.FamilyType != model.MetricTypeSummary &&
295 + sample.FamilyType != model.MetricTypeHistogram {
296 + return sample, "", pendingNone, true
297 }
191 - case strings.HasSuffix(name, countSuffix):
192 - n := strings.TrimSuffix(name, countSuffix)
193 - if mf, ok := p.metrics[n]; ok && isSummaryOrHistogram(mf.typ) {
194 - p.isCount = true
195 - _ = setLabelValue(p.currSeries, labels.MetricName, n)
196 - p.currMF = mf
197 - return
298 +
299 + baseName := strings.TrimSuffix(name, sumSuffix)
300 + switch d.familyTypes[baseName] {
301 + case model.MetricTypeSummary:
302 + sample.Kind = SampleKindSummarySum
303 + sample.FamilyType = model.MetricTypeSummary
304 + return sample, "", pendingNone, true
305 + case model.MetricTypeHistogram:
306 + sample.Kind = SampleKindHistogramSum
307 + sample.FamilyType = model.MetricTypeHistogram
308 + return sample, "", pendingNone, true
309 + default:
310 + return sample, baseName, pendingSum, true
311 }
199 - case strings.HasSuffix(name, bucketSuffix):
200 - n := strings.TrimSuffix(name, bucketSuffix)
201 - if mf, ok := p.metrics[n]; ok && isSummaryOrHistogram(mf.typ) {
202 - _ = setLabelValue(p.currSeries, labels.MetricName, n)
203 - p.setBucket()
204 - p.currMF = mf
205 - return
312 + }
313 +
314 + if strings.HasSuffix(name, countSuffix) {
315 + if sample.FamilyType != model.MetricTypeUnknown &&
316 + sample.FamilyType != model.MetricTypeSummary &&
317 + sample.FamilyType != model.MetricTypeHistogram {
318 + return sample, "", pendingNone, true
319 }
207 - if p.currSeries.Has(bucketLabel) {
208 - _ = setLabelValue(p.currSeries, labels.MetricName, n)
209 - p.setBucket()
210 - name = n
211 - typ = model.MetricTypeHistogram
320 +
321 + baseName := strings.TrimSuffix(name, countSuffix)
322 + switch d.familyTypes[baseName] {
323 + case model.MetricTypeSummary:
324 + sample.Kind = SampleKindSummaryCount
325 + sample.FamilyType = model.MetricTypeSummary
326 + return sample, "", pendingNone, true
327 + case model.MetricTypeHistogram:
328 + sample.Kind = SampleKindHistogramCount
329 + sample.FamilyType = model.MetricTypeHistogram
330 + return sample, "", pendingNone, true
331 + default:
332 + return sample, baseName, pendingCount, true
333 }
213 - case p.currSeries.Has(quantileLabel):
214 - typ = model.MetricTypeSummary
215 - p.setQuantile()
334 }
335
218 - p.setMetricFamilyByName(name)
219 - if p.currMF.typ == "" || p.currMF.typ == model.MetricTypeUnknown {
220 - p.currMF.typ = typ
221 - }
336 + return sample, "", pendingNone, true
337 }
338
224 -func (p *promTextParser) setQuantile() {
225 - if lbs, v, ok := removeLabel(p.currSeries, quantileLabel); ok {
226 - p.isQuantile = true
227 - p.currSeries = lbs
228 - p.currQuantile, _ = strconv.ParseFloat(v, 64)
339 +func (d *parseDriver) reset() {
340 + d.currSeries = d.currSeries[:0]
341 +
342 + if d.familyTypes == nil {
343 + d.familyTypes = make(map[string]model.MetricType)
344 + }
345 + for k := range d.familyTypes {
346 + delete(d.familyTypes, k)
347 }
348 +
349 + d.pending = d.pending[:0]
350 }
351
232 -func (p *promTextParser) setBucket() {
233 - if lbs, v, ok := removeLabel(p.currSeries, bucketLabel); ok {
234 - p.isBucket = true
235 - p.currSeries = lbs
236 - p.currBucket, _ = strconv.ParseFloat(v, 64)
352 +// emitResolvedPending flushes buffered _sum/_count samples for baseName now that
353 +// its family type is known, emitting them in buffered (exposition) order.
354 +func emitResolvedPending(pending []pendingSample, baseName string, typ model.MetricType, onSample func(Sample) error) ([]pendingSample, error) {
355 + if len(pending) == 0 {
356 + return pending, nil
357 }
238 -}
358
240 -func (p *promTextParser) addGauge(value float64) {
241 - p.currSeries, _, _ = removeLabel(p.currSeries, labels.MetricName)
359 + out := pending[:0]
360 + for _, ps := range pending {
361 + if ps.baseName != baseName {
362 + out = append(out, ps)
363 + continue
364 + }
365
243 - if v := len(p.currMF.metrics); v == cap(p.currMF.metrics) {
244 - p.currMF.metrics = append(p.currMF.metrics, Metric{
245 - labels: copyLabels(p.currSeries),
246 - gauge: &Gauge{value: value},
247 - })
248 - } else {
249 - p.currMF.metrics = p.currMF.metrics[:v+1]
250 - if p.currMF.metrics[v].gauge == nil {
251 - p.currMF.metrics[v].gauge = &Gauge{}
366 + sample := ps.sample
367 + sample.FamilyType = typ
368 + switch typ {
369 + case model.MetricTypeSummary:
370 + if ps.role == pendingSum {
371 + sample.Kind = SampleKindSummarySum
372 + } else {
373 + sample.Kind = SampleKindSummaryCount
374 + }
375 + case model.MetricTypeHistogram:
376 + if ps.role == pendingSum {
377 + sample.Kind = SampleKindHistogramSum
378 + } else {
379 + sample.Kind = SampleKindHistogramCount
380 + }
381 + default:
382 + sample.Kind = SampleKindScalar
383 + sample.FamilyType = model.MetricTypeUnknown
384 + }
385 +
386 + if err := onSample(sample); err != nil {
387 + return nil, err
388 }
253 - p.currMF.metrics[v].gauge.value = value
254 - p.currMF.metrics[v].labels = p.currMF.metrics[v].labels[:0]
255 - p.currMF.metrics[v].labels = append(p.currMF.metrics[v].labels, p.currSeries...)
389 }
390 +
391 + return out, nil
392 }
393
259 -func (p *promTextParser) addCounter(value float64) {
260 - p.currSeries, _, _ = removeLabel(p.currSeries, labels.MetricName)
394 +// assembler folds the driver's classified sample stream into typed
395 +// MetricFamilies. Grouping of summary quantiles / histogram buckets with their
396 +// _sum/_count is keyed by (family name, hash of the base labels). Buffers are
397 +// reused across cycles via reset().
398 +type assembler struct {
399 + metrics MetricFamilies
400 + summaries map[assemblyKey]*Summary
401 + histograms map[assemblyKey]*Histogram
402 + scratch labels.Labels
403 +
404 + // currName/currFamily cache the most recent family to skip the metrics map
405 + // lookup for the common case of consecutive samples in the same family.
406 + currName string
407 + currFamily *MetricFamily
408 +}
409
262 - if v := len(p.currMF.metrics); v == cap(p.currMF.metrics) {
263 - p.currMF.metrics = append(p.currMF.metrics, Metric{
264 - labels: copyLabels(p.currSeries),
265 - counter: &Counter{value: value},
266 - })
267 - } else {
268 - p.currMF.metrics = p.currMF.metrics[:v+1]
269 - if p.currMF.metrics[v].counter == nil {
270 - p.currMF.metrics[v].counter = &Counter{}
271 - }
272 - p.currMF.metrics[v].counter.value = value
273 - p.currMF.metrics[v].labels = p.currMF.metrics[v].labels[:0]
274 - p.currMF.metrics[v].labels = append(p.currMF.metrics[v].labels, p.currSeries...)
410 +type assemblyKey struct {
411 + name string
412 + hash uint64
413 +}
414 +
415 +func (a *assembler) reset() {
416 + a.currName = ""
417 + a.currFamily = nil
418 +
419 + if a.metrics == nil {
420 + a.metrics = make(MetricFamilies)
421 + }
422 + for _, mf := range a.metrics {
423 + mf.help = ""
424 + mf.typ = ""
425 + mf.metrics = mf.metrics[:0]
426 + }
427 +
428 + if a.summaries == nil {
429 + a.summaries = make(map[assemblyKey]*Summary)
430 + }
431 + for k := range a.summaries {
432 + delete(a.summaries, k)
433 + }
434 +
435 + if a.histograms == nil {
436 + a.histograms = make(map[assemblyKey]*Histogram)
437 + }
438 + for k := range a.histograms {
439 + delete(a.histograms, k)
440 }
441 }
442
278 -func (p *promTextParser) addUnknown(value float64) {
279 - p.currSeries, _, _ = removeLabel(p.currSeries, labels.MetricName)
443 +func (a *assembler) applyHelp(name, help string) {
444 + mf := a.ensureFamily(name)
445 + mf.help = help
446 +}
447
281 - if v := len(p.currMF.metrics); v == cap(p.currMF.metrics) {
282 - p.currMF.metrics = append(p.currMF.metrics, Metric{
283 - labels: copyLabels(p.currSeries),
284 - untyped: &Untyped{value: value},
285 - })
286 - } else {
287 - p.currMF.metrics = p.currMF.metrics[:v+1]
288 - if p.currMF.metrics[v].untyped == nil {
289 - p.currMF.metrics[v].untyped = &Untyped{}
448 +func (a *assembler) applySample(sample Sample) error {
449 + switch sample.Kind {
450 + case SampleKindSummaryQuantile:
451 + a.addSummaryQuantile(sample)
452 + case SampleKindSummarySum:
453 + a.addSummarySum(sample)
454 + case SampleKindSummaryCount:
455 + a.addSummaryCount(sample)
456 + case SampleKindHistogramBucket:
457 + a.addHistogramBucket(sample)
458 + case SampleKindHistogramSum:
459 + a.addHistogramSum(sample)
460 + case SampleKindHistogramCount:
461 + a.addHistogramCount(sample)
462 + default:
463 + switch sample.FamilyType {
464 + case model.MetricTypeSummary:
465 + mf := a.summaryFamily(sample.Name)
466 + a.summaryFor(mf, assemblyKey{name: sample.Name, hash: sample.Labels.Hash()}, sample.Labels)
467 + case model.MetricTypeHistogram:
468 + mf := a.histogramFamily(sample.Name)
469 + a.histogramFor(mf, assemblyKey{name: sample.Name, hash: sample.Labels.Hash()}, sample.Labels)
470 + default:
471 + a.addScalar(sample)
472 }
291 - p.currMF.metrics[v].untyped.value = value
292 - p.currMF.metrics[v].labels = p.currMF.metrics[v].labels[:0]
293 - p.currMF.metrics[v].labels = append(p.currMF.metrics[v].labels, p.currSeries...)
473 }
474 + return nil
475 }
476
297 -func (p *promTextParser) addSummary(value float64) {
298 - hash := p.currSeries.Hash()
477 +func (a *assembler) families() MetricFamilies {
478 + for name, mf := range a.metrics {
479 + if len(mf.metrics) == 0 {
480 + delete(a.metrics, name)
481 + }
482 + }
483 + return a.metrics
484 +}
485
300 - p.currSeries, _, _ = removeLabel(p.currSeries, labels.MetricName)
486 +func (a *assembler) addScalar(sample Sample) {
487 + mf := a.ensureFamily(sample.Name)
488
302 - s, ok := p.summaries[hash]
303 - if !ok {
304 - if v := len(p.currMF.metrics); v == cap(p.currMF.metrics) {
305 - s = &Summary{}
306 - p.currMF.metrics = append(p.currMF.metrics, Metric{
307 - labels: copyLabels(p.currSeries),
308 - summary: s,
309 - })
310 - } else {
311 - p.currMF.metrics = p.currMF.metrics[:v+1]
312 - if p.currMF.metrics[v].summary == nil {
313 - p.currMF.metrics[v].summary = &Summary{}
314 - }
315 - p.currMF.metrics[v].summary.sum = 0
316 - p.currMF.metrics[v].summary.count = 0
317 - p.currMF.metrics[v].summary.quantiles = p.currMF.metrics[v].summary.quantiles[:0]
318 - p.currMF.metrics[v].labels = p.currMF.metrics[v].labels[:0]
319 - p.currMF.metrics[v].labels = append(p.currMF.metrics[v].labels, p.currSeries...)
320 - s = p.currMF.metrics[v].summary
489 + typ := sample.FamilyType
490 + if typ == "" {
491 + typ = model.MetricTypeUnknown
492 + }
493 + if mf.typ == "" || mf.typ == model.MetricTypeUnknown {
494 + mf.typ = typ
495 + }
496 +
497 + m := a.appendMetric(mf, sample.Labels)
498 +
499 + switch typ {
500 + case model.MetricTypeGauge:
501 + if m.gauge == nil {
502 + m.gauge = &Gauge{}
503 }
504 + m.gauge.value = sample.Value
505 + case model.MetricTypeCounter:
506 + if m.counter == nil {
507 + m.counter = &Counter{}
508 + }
509 + m.counter.value = sample.Value
510 + default:
511 + if m.untyped == nil {
512 + m.untyped = &Untyped{}
513 + }
514 + m.untyped.value = sample.Value
515 + }
516 +}
517
323 - p.summaries[hash] = s
518 +func (a *assembler) addSummaryQuantile(sample Sample) {
519 + mf := a.summaryFamily(sample.Name)
520 + base, qv, ok := a.stripLabel(sample.Labels, quantileLabel)
521 + key := assemblyKey{name: sample.Name}
522 + if ok {
523 + key.hash = labels.Labels(base).Hash()
524 + } else {
525 + base = sample.Labels
526 + key.hash = sample.Labels.Hash()
527 }
528
326 - switch {
327 - case p.isQuantile:
328 - s.quantiles = append(s.quantiles, Quantile{quantile: p.currQuantile, value: value})
329 - case p.isSum:
330 - s.sum = value
331 - case p.isCount:
332 - s.count = value
529 + s := a.summaryFor(mf, key, base)
530 + if !ok {
531 + return
532 }
533 + quantile, _ := strconv.ParseFloat(qv, 64)
534 + s.quantiles = append(s.quantiles, Quantile{quantile: quantile, value: sample.Value})
535 }
536
336 -func (p *promTextParser) addHistogram(value float64) {
337 - hash := p.currSeries.Hash()
537 +func (a *assembler) addSummarySum(sample Sample) {
538 + name := strings.TrimSuffix(sample.Name, sumSuffix)
539 + mf := a.summaryFamily(name)
540 + s := a.summaryFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
541 + s.sum = sample.Value
542 +}
543
339 - p.currSeries, _, _ = removeLabel(p.currSeries, labels.MetricName)
544 +func (a *assembler) addSummaryCount(sample Sample) {
545 + name := strings.TrimSuffix(sample.Name, countSuffix)
546 + mf := a.summaryFamily(name)
547 + s := a.summaryFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
548 + s.count = sample.Value
549 +}
550
341 - h, ok := p.histograms[hash]
551 +func (a *assembler) addHistogramBucket(sample Sample) {
552 + name := strings.TrimSuffix(sample.Name, bucketSuffix)
553 + mf := a.histogramFamily(name)
554 + base, lev, ok := a.stripLabel(sample.Labels, bucketLabel)
555 + key := assemblyKey{name: name}
556 + if ok {
557 + key.hash = labels.Labels(base).Hash()
558 + } else {
559 + base = sample.Labels
560 + key.hash = sample.Labels.Hash()
561 + }
562 +
563 + h := a.histogramFor(mf, key, base)
564 if !ok {
343 - if v := len(p.currMF.metrics); v == cap(p.currMF.metrics) {
344 - h = &Histogram{}
345 - p.currMF.metrics = append(p.currMF.metrics, Metric{
346 - labels: copyLabels(p.currSeries),
347 - histogram: h,
348 - })
349 - } else {
350 - p.currMF.metrics = p.currMF.metrics[:v+1]
351 - if p.currMF.metrics[v].histogram == nil {
352 - p.currMF.metrics[v].histogram = &Histogram{}
353 - }
354 - p.currMF.metrics[v].histogram.sum = 0
355 - p.currMF.metrics[v].histogram.count = 0
356 - p.currMF.metrics[v].histogram.buckets = p.currMF.metrics[v].histogram.buckets[:0]
357 - p.currMF.metrics[v].labels = p.currMF.metrics[v].labels[:0]
358 - p.currMF.metrics[v].labels = append(p.currMF.metrics[v].labels, p.currSeries...)
359 - h = p.currMF.metrics[v].histogram
360 - }
565 + return
566 + }
567 + bound, _ := strconv.ParseFloat(lev, 64)
568 + h.buckets = append(h.buckets, Bucket{upperBound: bound, cumulativeCount: sample.Value})
569 +}
570 +
571 +func (a *assembler) addHistogramSum(sample Sample) {
572 + name := strings.TrimSuffix(sample.Name, sumSuffix)
573 + mf := a.histogramFamily(name)
574 + h := a.histogramFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
575 + h.sum = sample.Value
576 +}
577
362 - p.histograms[hash] = h
578 +func (a *assembler) addHistogramCount(sample Sample) {
579 + name := strings.TrimSuffix(sample.Name, countSuffix)
580 + mf := a.histogramFamily(name)
581 + h := a.histogramFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
582 + h.count = sample.Value
583 +}
584 +
585 +func (a *assembler) summaryFamily(name string) *MetricFamily {
586 + mf := a.ensureFamily(name)
587 + mf.typ = model.MetricTypeSummary
588 + return mf
589 +}
590 +
591 +func (a *assembler) histogramFamily(name string) *MetricFamily {
592 + mf := a.ensureFamily(name)
593 + mf.typ = model.MetricTypeHistogram
594 + return mf
595 +}
596 +
597 +func (a *assembler) summaryFor(mf *MetricFamily, key assemblyKey, lbs labels.Labels) *Summary {
598 + if s, ok := a.summaries[key]; ok {
599 + return s
600 }
601
365 - switch {
366 - case p.isBucket:
367 - h.buckets = append(h.buckets, Bucket{upperBound: p.currBucket, cumulativeCount: value})
368 - case p.isSum:
369 - h.sum = value
370 - case p.isCount:
371 - h.count = value
602 + m := a.appendMetric(mf, lbs)
603 + if m.summary == nil {
604 + m.summary = &Summary{}
605 + } else {
606 + m.summary.sum = 0
607 + m.summary.count = 0
608 + m.summary.quantiles = m.summary.quantiles[:0]
609 }
610 +
611 + a.summaries[key] = m.summary
612 + return m.summary
613 }
614
375 -func (p *promTextParser) reset() {
376 - p.currMF = nil
377 - p.currSeries = p.currSeries[:0]
615 +func (a *assembler) histogramFor(mf *MetricFamily, key assemblyKey, lbs labels.Labels) *Histogram {
616 + if h, ok := a.histograms[key]; ok {
617 + return h
618 + }
619
379 - if p.metrics == nil {
380 - p.metrics = make(MetricFamilies)
620 + m := a.appendMetric(mf, lbs)
621 + if m.histogram == nil {
622 + m.histogram = &Histogram{}
623 + } else {
624 + m.histogram.sum = 0
625 + m.histogram.count = 0
626 + m.histogram.buckets = m.histogram.buckets[:0]
627 }
382 - for _, mf := range p.metrics {
383 - mf.help = ""
384 - mf.typ = ""
385 - mf.metrics = mf.metrics[:0]
628 +
629 + a.histograms[key] = m.histogram
630 + return m.histogram
631 +}
632 +
633 +// appendMetric grows mf.metrics by one, reusing the backing array across cycles
634 +// and storing a copy of lbs. Instrument pointers on a reused slot are left in
635 +// place: a family keeps a stable type across scrapes, so the caller reuses the
636 +// existing instrument and allocates only when its pointer is nil. This preserves
637 +// the legacy allocation profile (no per-scrape instrument churn).
638 +func (a *assembler) appendMetric(mf *MetricFamily, lbs labels.Labels) *Metric {
639 + idx := len(mf.metrics)
640 + if idx == cap(mf.metrics) {
641 + mf.metrics = append(mf.metrics, Metric{})
642 + } else {
643 + mf.metrics = mf.metrics[:idx+1]
644 }
645
388 - if p.summaries == nil {
389 - p.summaries = make(map[uint64]*Summary)
646 + m := &mf.metrics[idx]
647 + m.labels = m.labels[:0]
648 + m.labels = append(m.labels, lbs...)
649 + return m
650 +}
651 +
652 +func (a *assembler) ensureFamily(name string) *MetricFamily {
653 + if a.currFamily != nil && a.currName == name {
654 + return a.currFamily
655 }
391 - for k := range p.summaries {
392 - delete(p.summaries, k)
656 + mf, ok := a.metrics[name]
657 + if !ok {
658 + mf = &MetricFamily{name: name, typ: model.MetricTypeUnknown}
659 + a.metrics[name] = mf
660 }
661 + a.currName = name
662 + a.currFamily = mf
663 + return mf
664 +}
665
395 - if p.histograms == nil {
396 - p.histograms = make(map[uint64]*Histogram)
666 +// stripLabel returns the label set without name and the removed value, using a
667 +// reusable scratch buffer. The result is valid only until the next stripLabel.
668 +func (a *assembler) stripLabel(lbs labels.Labels, name string) (labels.Labels, string, bool) {
669 + a.scratch = a.scratch[:0]
670 + var (
671 + value string
672 + found bool
673 + )
674 + for _, lb := range lbs {
675 + if lb.Name == name {
676 + value = lb.Value
677 + found = true
678 + continue
679 + }
680 + a.scratch = append(a.scratch, lb)
681 }
398 - for k := range p.histograms {
399 - delete(p.histograms, k)
682 + if !found {
683 + return nil, "", false
684 }
685 + return a.scratch, value, true
686 }
687
688 func copyLabels(lbs []labels.Label) []labels.Label {
689 return append([]labels.Label(nil), lbs...)
690 }
691
692 +// copyLabelsWithoutName returns a fresh copy of lbs with __name__ removed. In the
693 +// common case __name__ sorts first (it precedes lowercase label names), so the
694 +// remainder is contiguous and copied directly; otherwise a rare label that sorts
695 +// before __name__ (e.g. "UUID") is skipped element by element.
696 +func copyLabelsWithoutName(lbs labels.Labels) labels.Labels {
697 + if len(lbs) > 0 && lbs[0].Name == labels.MetricName {
698 + return copyLabels(lbs[1:])
699 + }
700 + out := make([]labels.Label, 0, len(lbs))
701 + for _, lb := range lbs {
702 + if lb.Name == labels.MetricName {
703 + continue
704 + }
705 + out = append(out, lb)
706 + }
707 + return out
708 +}
709 +
710 func removeLabel(lbs labels.Labels, name string) (labels.Labels, string, bool) {
711 for i, v := range lbs {
712 if v.Name == name {
@@ -422,16 +725,10 @@ func metricNameValue(lbs labels.Labels) (string, bool) {
725 return "", false
726 }
727
425 -func setLabelValue(lbs labels.Labels, name, value string) bool {
426 - for i, v := range lbs {
427 - if v.Name == name {
428 - lbs[i].Value = value
429 - return true
430 - }
728 +func sanitizeHelp(help string) string {
729 + if strings.IndexByte(help, '\n') == -1 {
730 + return help
731 }
432 - return false
433 -}
434 -
435 -func isSummaryOrHistogram(typ model.MetricType) bool {
436 - return typ == model.MetricTypeSummary || typ == model.MetricTypeHistogram
732 + // HELP is used as a chart title; collapse multiline help to one line.
733 + return strings.Join(strings.Fields(help), " ")
734 }
src/go/pkg/prometheus/parse_bench_test.go new
+97
@@ -0,0 +1,97 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package prometheus
4 +
5 +import (
6 + "os"
7 + "testing"
8 +)
9 +
10 +// pkg/prometheus parses an exposition on every scrape across ~19 importing
11 +// collectors, so it is a hot path. Keep these results updated before/after parser
12 +// changes so regressions stay visible.
13 +//
14 +// The payload is the real exposition fixtures the parser tests use:
15 +// testdata/testdata.txt (a go-process scrape of gauges/counters/summaries) plus
16 +// testdata/histogram-meta.txt (the histogram path), so the benchmark exercises
17 +// every assembler branch on representative data.
18 +//
19 +// Command:
20 +//
21 +// go test ./pkg/prometheus -run '^$' -bench parseTo -benchmem
22 +//
23 +// Measured on a developer laptop (Apple M-series, 14 logical CPUs), not CI, so the
24 +// absolute numbers are machine-specific — compare relative before/after deltas.
25 +//
26 +// Legacy fused parser (master, pre-rewrite — captured 2026-06-05):
27 +//
28 +// parseToMetricFamilies 127868 ns/op 73404 B/op 1314 allocs/op
29 +// parseToSeries 111382 ns/op 100444 B/op 1584 allocs/op
30 +//
31 +// Unified driver+assembler stream parser (this rewrite — captured 2026-06-05):
32 +//
33 +// parseToMetricFamilies 133911 ns/op 73408 B/op 1314 allocs/op
34 +// parseToSeries 112803 ns/op 100448 B/op 1584 allocs/op
35 +// parseToStream 113248 ns/op 102807 B/op 1644 allocs/op
36 +//
37 +// No allocation win and no allocation regression: the legacy parser already
38 +// parsed once per call and reused buffers, and both Scrape and ScrapeSeries match
39 +// it exactly (flat allocs). ScrapeSeries uses the raw-label path (identical to
40 +// legacy), so it is essentially free. Scrape/stream are a few percent slower on
41 +// CPU — the per-sample Sample struct + the shared iterate indirection + assembler
42 +// dispatch on the families/stream path (profiled: makeSample + applySample,
43 +// inherent to the single-driver model, not a fixable hotspot). The win is the
44 +// sample stream that metric relabeling consumes, not raw speed.
45 +
46 +func readBenchData(tb testing.TB) []byte {
47 + tb.Helper()
48 + var data []byte
49 + for _, f := range []string{"testdata/testdata.txt", "testdata/histogram-meta.txt"} {
50 + b, err := os.ReadFile(f)
51 + if err != nil {
52 + tb.Fatal(err)
53 + }
54 + data = append(data, b...)
55 + data = append(data, '\n') // keep files separated when concatenated
56 + }
57 + return data
58 +}
59 +
60 +func BenchmarkPromTextParser_parseToMetricFamilies(b *testing.B) {
61 + data := readBenchData(b)
62 + var p promTextParser
63 + b.ReportAllocs()
64 + b.SetBytes(int64(len(data)))
65 + b.ResetTimer()
66 + for range b.N {
67 + if _, err := p.parseToMetricFamilies(data); err != nil {
68 + b.Fatal(err)
69 + }
70 + }
71 +}
72 +
73 +func BenchmarkPromTextParser_parseToSeries(b *testing.B) {
74 + data := readBenchData(b)
75 + var p promTextParser
76 + b.ReportAllocs()
77 + b.SetBytes(int64(len(data)))
78 + b.ResetTimer()
79 + for range b.N {
80 + if _, err := p.parseToSeries(data); err != nil {
81 + b.Fatal(err)
82 + }
83 + }
84 +}
85 +
86 +func BenchmarkPromTextParser_parseToStream(b *testing.B) {
87 + data := readBenchData(b)
88 + var p promTextParser
89 + b.ReportAllocs()
90 + b.SetBytes(int64(len(data)))
91 + b.ResetTimer()
92 + for range b.N {
93 + if err := p.parseToStream(data, nil, func(Sample) error { return nil }); err != nil {
94 + b.Fatal(err)
95 + }
96 + }
97 +}
src/go/pkg/prometheus/stream.go new
+15
@@ -0,0 +1,15 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package prometheus
4 +
5 +// ScrapeStream implements [Prometheus]. See the interface for the ordering and
6 +// callback contract. onHelp may be nil when per-family HELP is not needed.
7 +func (p *prometheus) ScrapeStream(onHelp func(name, help string), onSample func(Sample) error) error {
8 + p.buf.Reset()
9 +
10 + if err := p.fetch(p.buf); err != nil {
11 + return err
12 + }
13 +
14 + return p.parser.parseToStream(p.buf.Bytes(), onHelp, onSample)
15 +}
src/go/pkg/prometheus/stream_test.go new
+296
@@ -0,0 +1,296 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package prometheus
4 +
5 +import (
6 + "fmt"
7 + "math"
8 + "testing"
9 +
10 + "github.com/prometheus/common/model"
11 + "github.com/prometheus/prometheus/model/labels"
12 + "github.com/stretchr/testify/assert"
13 + "github.com/stretchr/testify/require"
14 +)
15 +
16 +func TestPromTextParser_parseToStream(t *testing.T) {
17 + type wantSample struct {
18 + name string
19 + labels string
20 + value float64
21 + kind SampleKind
22 + familyType model.MetricType
23 + }
24 +
25 + tests := map[string]struct {
26 + input []byte
27 + wantHelp []string
28 + want []wantSample
29 + }{
30 + "all metric types are classified in exposition order": {
31 + input: []byte(`# HELP test_gauge A gauge metric.
32 +# TYPE test_gauge gauge
33 +test_gauge{label="a"} 1
34 +# TYPE test_counter_total counter
35 +test_counter_total 5
36 +# TYPE test_hist histogram
37 +test_hist_bucket{le="0.1"} 1
38 +test_hist_bucket{le="+Inf"} 2
39 +test_hist_sum 3
40 +test_hist_count 2
41 +# TYPE test_summary summary
42 +test_summary{quantile="0.5"} 0.2
43 +test_summary_sum 1
44 +test_summary_count 10
45 +`),
46 + wantHelp: []string{"test_gauge=A gauge metric."},
47 + want: []wantSample{
48 + {"test_gauge", `{label="a"}`, 1, SampleKindScalar, model.MetricTypeGauge},
49 + {"test_counter_total", `{}`, 5, SampleKindScalar, model.MetricTypeCounter},
50 + {"test_hist_bucket", `{le="0.1"}`, 1, SampleKindHistogramBucket, model.MetricTypeHistogram},
51 + {"test_hist_bucket", `{le="+Inf"}`, 2, SampleKindHistogramBucket, model.MetricTypeHistogram},
52 + {"test_hist_sum", `{}`, 3, SampleKindHistogramSum, model.MetricTypeHistogram},
53 + {"test_hist_count", `{}`, 2, SampleKindHistogramCount, model.MetricTypeHistogram},
54 + {"test_summary", `{quantile="0.5"}`, 0.2, SampleKindSummaryQuantile, model.MetricTypeSummary},
55 + {"test_summary_sum", `{}`, 1, SampleKindSummarySum, model.MetricTypeSummary},
56 + {"test_summary_count", `{}`, 10, SampleKindSummaryCount, model.MetricTypeSummary},
57 + },
58 + },
59 + "__name__ is delivered via Name and excluded from Labels": {
60 + input: []byte("# TYPE m gauge\nm{a=\"1\",b=\"2\"} 7\n"),
61 + want: []wantSample{
62 + {"m", `{a="1", b="2"}`, 7, SampleKindScalar, model.MetricTypeGauge},
63 + },
64 + },
65 + "deferred sum/count before # TYPE are buffered and back-resolved": {
66 + input: []byte(`my_summary_sum 10
67 +my_summary_count 2
68 +# TYPE my_summary summary
69 +my_summary{quantile="0.5"} 0.5
70 +`),
71 + want: []wantSample{
72 + {"my_summary_sum", `{}`, 10, SampleKindSummarySum, model.MetricTypeSummary},
73 + {"my_summary_count", `{}`, 2, SampleKindSummaryCount, model.MetricTypeSummary},
74 + {"my_summary", `{quantile="0.5"}`, 0.5, SampleKindSummaryQuantile, model.MetricTypeSummary},
75 + },
76 + },
77 + }
78 +
79 + for name, test := range tests {
80 + t.Run(name, func(t *testing.T) {
81 + var p promTextParser
82 +
83 + for i := range 10 {
84 + t.Run(fmt.Sprintf("parse num %d", i+1), func(t *testing.T) {
85 + var got []wantSample
86 + var help []string
87 +
88 + err := p.parseToStream(test.input,
89 + func(name, h string) { help = append(help, name+"="+h) },
90 + func(s Sample) error {
91 + assert.Falsef(t, s.Labels.Has(labels.MetricName),
92 + "sample %q must not carry __name__ in Labels", s.Name)
93 + got = append(got, wantSample{s.Name, s.Labels.String(), s.Value, s.Kind, s.FamilyType})
94 + return nil
95 + },
96 + )
97 + require.NoError(t, err)
98 + assert.Equal(t, test.want, got)
99 + for _, h := range test.wantHelp {
100 + assert.Contains(t, help, h)
101 + }
102 + })
103 + }
104 + })
105 + }
106 +}
107 +
108 +// Deferred _sum/_count (emitted before the family type is known) fold into the
109 +// typed family once it is resolved — criterion #5 at the assembled level.
110 +func TestPromTextParser_parseToMetricFamilies_deferredClassification(t *testing.T) {
111 + tests := map[string]struct {
112 + input []byte
113 + want MetricFamilies
114 + }{
115 + "summary _sum/_count before # TYPE fold into one summary": {
116 + input: []byte(`my_summary_sum{label1="value1"} 10
117 +my_summary_count{label1="value1"} 2
118 +# TYPE my_summary summary
119 +my_summary{label1="value1",quantile="0.5"} 0.5
120 +`),
121 + want: MetricFamilies{
122 + "my_summary": {
123 + name: "my_summary",
124 + typ: model.MetricTypeSummary,
125 + metrics: []Metric{
126 + {
127 + labels: labels.Labels{{Name: "label1", Value: "value1"}},
128 + summary: &Summary{
129 + sum: 10,
130 + count: 2,
131 + quantiles: []Quantile{{quantile: 0.5, value: 0.5}},
132 + },
133 + },
134 + },
135 + },
136 + },
137 + },
138 + "histogram _sum/_count before _bucket (no # TYPE) fold into one histogram": {
139 + input: []byte(`my_hist_sum{label1="value1"} 5
140 +my_hist_count{label1="value1"} 3
141 +my_hist_bucket{label1="value1",le="0.1"} 1
142 +my_hist_bucket{label1="value1",le="+Inf"} 3
143 +`),
144 + want: MetricFamilies{
145 + "my_hist": {
146 + name: "my_hist",
147 + typ: model.MetricTypeHistogram,
148 + metrics: []Metric{
149 + {
150 + labels: labels.Labels{{Name: "label1", Value: "value1"}},
151 + histogram: &Histogram{
152 + sum: 5,
153 + count: 3,
154 + buckets: []Bucket{
155 + {upperBound: 0.1, cumulativeCount: 1},
156 + {upperBound: math.Inf(1), cumulativeCount: 3},
157 + },
158 + },
159 + },
160 + },
161 + },
162 + },
163 + },
164 + }
165 +
166 + for name, test := range tests {
167 + t.Run(name, func(t *testing.T) {
168 + var p promTextParser
169 +
170 + for i := range 10 {
171 + t.Run(fmt.Sprintf("parse num %d", i+1), func(t *testing.T) {
172 + mfs, err := p.parseToMetricFamilies(test.input)
173 + require.NoError(t, err)
174 + assert.Equal(t, test.want, mfs)
175 + })
176 + }
177 + })
178 + }
179 +}
180 +
181 +// The stream supports Prometheus-style relabeling on __name__/le/quantile.
182 +// ownLabels=true isolates each sample's labels, so a transform can rename via
183 +// Name and mutate Labels in place without affecting later samples.
184 +func TestPromTextParser_parseToStream_relabelStyle(t *testing.T) {
185 + data := []byte(`# TYPE req_seconds histogram
186 +req_seconds_bucket{le="0.1",path="/a"} 1
187 +req_seconds_bucket{le="+Inf",path="/a"} 3
188 +# TYPE rpc summary
189 +rpc{quantile="0.99",path="/a"} 0.5
190 +`)
191 +
192 + type out struct {
193 + name string
194 + le string
195 + quantile string
196 + labels string
197 + }
198 +
199 + var p promTextParser
200 + var got []out
201 + err := p.parseToStream(data, nil, func(s Sample) error {
202 + o := out{
203 + name: s.Name + ":relabeled", // __name__ is mutable via Name
204 + le: s.Labels.Get(bucketLabel),
205 + quantile: s.Labels.Get(quantileLabel),
206 + }
207 + // Drop the "path" target label in place (this sample owns its labels).
208 + kept := s.Labels[:0]
209 + for _, l := range s.Labels {
210 + if l.Name == "path" {
211 + continue
212 + }
213 + kept = append(kept, l)
214 + }
215 + o.labels = labels.Labels(kept).String()
216 + got = append(got, o)
217 + return nil
218 + })
219 + require.NoError(t, err)
220 +
221 + assert.Equal(t, []out{
222 + {name: "req_seconds_bucket:relabeled", le: "0.1", quantile: "", labels: `{le="0.1"}`},
223 + {name: "req_seconds_bucket:relabeled", le: "+Inf", quantile: "", labels: `{le="+Inf"}`},
224 + {name: "rpc:relabeled", le: "", quantile: "0.99", labels: `{quantile="0.99"}`},
225 + }, got)
226 +}
227 +
228 +func TestPromTextParser_parseToStream_nilCallbackIsNoop(t *testing.T) {
229 + var p promTextParser
230 + require.NoError(t, p.parseToStream([]byte("metric 1\n"), nil, nil))
231 +}
232 +
233 +// Byte-identical series ordering: textparse sorts labels, so __name__ is NOT
234 +// always first — a label like "UUID" (0x55) sorts before "__name__" (0x5f).
235 +// ScrapeSeries must preserve that raw sorted order (same as the legacy parser),
236 +// not force __name__ first, which would also break the labels.Labels sorted
237 +// invariant.
238 +func TestPromTextParser_parseToSeries_labelOrderMatchesTextparse(t *testing.T) {
239 + var p promTextParser
240 + series, err := p.parseToSeries([]byte("m{UUID=\"x\",gpu=\"0\"} 5\n"))
241 + require.NoError(t, err)
242 + require.Len(t, series, 1)
243 + assert.Equal(t, `{UUID="x", __name__="m", gpu="0"}`, series[0].Labels.String())
244 +}
245 +
246 +// The deferred _sum/_count buffer (criterion #5) can emit a _sum after a later,
247 +// unrelated sample: here a_sum is buffered (type unknown), b is emitted, then
248 +// a_sum resolves on "# TYPE a summary". Documents the ScrapeStream order caveat.
249 +func TestPromTextParser_parseToStream_deferralReordersAcrossMetrics(t *testing.T) {
250 + data := []byte("a_sum 1\nb 2\n# TYPE a summary\na{quantile=\"0.5\"} 3\n")
251 +
252 + var p promTextParser
253 + var names []string
254 + err := p.parseToStream(data, nil, func(s Sample) error {
255 + names = append(names, s.Name)
256 + return nil
257 + })
258 + require.NoError(t, err)
259 + assert.Equal(t, []string{"b", "a_sum", "a"}, names)
260 +}
261 +
262 +// A _bucket-named series is a histogram bucket only if it carries an "le" label.
263 +// Without le it is malformed and is kept as a plain metric (value preserved) — not
264 +// folded into the histogram family. (Legacy folded it and dropped the value;
265 +// valid buckets always have le, so real histograms are unaffected.)
266 +func TestPromTextParser_parseToMetricFamilies_bucketRequiresLe(t *testing.T) {
267 + tests := map[string]struct {
268 + input []byte
269 + wantFam string
270 + wantType model.MetricType
271 + }{
272 + "valid _bucket (has le) folds into the histogram family": {
273 + input: []byte("# TYPE h histogram\nh_bucket{le=\"1\",label=\"x\"} 1\n"),
274 + wantFam: "h",
275 + wantType: model.MetricTypeHistogram,
276 + },
277 + "_bucket without le is a plain metric, not a histogram bucket": {
278 + input: []byte("# TYPE h histogram\nh_bucket{label=\"x\"} 1\n"),
279 + wantFam: "h_bucket",
280 + wantType: model.MetricTypeUnknown,
281 + },
282 + }
283 +
284 + for name, tc := range tests {
285 + t.Run(name, func(t *testing.T) {
286 + var p promTextParser
287 + mfs, err := p.parseToMetricFamilies(tc.input)
288 + require.NoError(t, err)
289 +
290 + mf := mfs.Get(tc.wantFam)
291 + require.NotNilf(t, mf, "expected family %q", tc.wantFam)
292 + assert.Equal(t, tc.wantType, mf.Type())
293 + assert.Len(t, mf.Metrics(), 1)
294 + })
295 + }
296 +}