refactor(go/pkg/prometheus): extract the scrape transport into a fetcher (#22664)
Ilya Mashchenko committed
Jun 8, 2026 at 23:34 UTC
36455a7475f52544b304e46326831697892a1331
7 files changed
+457
-416
src/go/pkg/prometheus/assemble.go
new
+305
@@ -0,0 +1,305 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package prometheus
4
+
5
+import (
6
+ "strconv"
7
+ "strings"
8
+
9
+ "github.com/prometheus/common/model"
10
+ "github.com/prometheus/prometheus/model/labels"
11
+)
12
+
13
+// assembler folds the driver's classified sample stream into typed
14
+// MetricFamilies. Grouping of summary quantiles / histogram buckets with their
15
+// _sum/_count is keyed by (family name, hash of the base labels). Buffers are
16
+// reused across cycles via reset().
17
+type assembler struct {
18
+ metrics MetricFamilies
19
+ summaries map[assemblyKey]*Summary
20
+ histograms map[assemblyKey]*Histogram
21
+ scratch labels.Labels
22
+
23
+ // currName/currFamily cache the most recent family to skip the metrics map
24
+ // lookup for the common case of consecutive samples in the same family.
25
+ currName string
26
+ currFamily *MetricFamily
27
+}
28
+
29
+type assemblyKey struct {
30
+ name string
31
+ hash uint64
32
+}
33
+
34
+func (a *assembler) reset() {
35
+ a.currName = ""
36
+ a.currFamily = nil
37
+
38
+ if a.metrics == nil {
39
+ a.metrics = make(MetricFamilies)
40
+ }
41
+ for _, mf := range a.metrics {
42
+ mf.help = ""
43
+ mf.typ = ""
44
+ mf.metrics = mf.metrics[:0]
45
+ }
46
+
47
+ if a.summaries == nil {
48
+ a.summaries = make(map[assemblyKey]*Summary)
49
+ }
50
+ for k := range a.summaries {
51
+ delete(a.summaries, k)
52
+ }
53
+
54
+ if a.histograms == nil {
55
+ a.histograms = make(map[assemblyKey]*Histogram)
56
+ }
57
+ for k := range a.histograms {
58
+ delete(a.histograms, k)
59
+ }
60
+}
61
+
62
+func (a *assembler) applyHelp(name, help string) {
63
+ mf := a.ensureFamily(name)
64
+ mf.help = help
65
+}
66
+
67
+func (a *assembler) applySample(sample Sample) error {
68
+ switch sample.Kind {
69
+ case SampleKindSummaryQuantile:
70
+ a.addSummaryQuantile(sample)
71
+ case SampleKindSummarySum:
72
+ a.addSummarySum(sample)
73
+ case SampleKindSummaryCount:
74
+ a.addSummaryCount(sample)
75
+ case SampleKindHistogramBucket:
76
+ a.addHistogramBucket(sample)
77
+ case SampleKindHistogramSum:
78
+ a.addHistogramSum(sample)
79
+ case SampleKindHistogramCount:
80
+ a.addHistogramCount(sample)
81
+ default:
82
+ switch sample.FamilyType {
83
+ case model.MetricTypeSummary:
84
+ mf := a.summaryFamily(sample.Name)
85
+ a.summaryFor(mf, assemblyKey{name: sample.Name, hash: sample.Labels.Hash()}, sample.Labels)
86
+ case model.MetricTypeHistogram:
87
+ mf := a.histogramFamily(sample.Name)
88
+ a.histogramFor(mf, assemblyKey{name: sample.Name, hash: sample.Labels.Hash()}, sample.Labels)
89
+ default:
90
+ a.addScalar(sample)
91
+ }
92
+ }
93
+ return nil
94
+}
95
+
96
+func (a *assembler) families() MetricFamilies {
97
+ for name, mf := range a.metrics {
98
+ if len(mf.metrics) == 0 {
99
+ delete(a.metrics, name)
100
+ }
101
+ }
102
+ return a.metrics
103
+}
104
+
105
+func (a *assembler) addScalar(sample Sample) {
106
+ mf := a.ensureFamily(sample.Name)
107
+
108
+ typ := sample.FamilyType
109
+ if typ == "" {
110
+ typ = model.MetricTypeUnknown
111
+ }
112
+ if mf.typ == "" || mf.typ == model.MetricTypeUnknown {
113
+ mf.typ = typ
114
+ }
115
+
116
+ m := a.appendMetric(mf, sample.Labels)
117
+
118
+ switch typ {
119
+ case model.MetricTypeGauge:
120
+ if m.gauge == nil {
121
+ m.gauge = &Gauge{}
122
+ }
123
+ m.gauge.value = sample.Value
124
+ case model.MetricTypeCounter:
125
+ if m.counter == nil {
126
+ m.counter = &Counter{}
127
+ }
128
+ m.counter.value = sample.Value
129
+ default:
130
+ if m.untyped == nil {
131
+ m.untyped = &Untyped{}
132
+ }
133
+ m.untyped.value = sample.Value
134
+ }
135
+}
136
+
137
+func (a *assembler) addSummaryQuantile(sample Sample) {
138
+ mf := a.summaryFamily(sample.Name)
139
+ base, qv, ok := a.stripLabel(sample.Labels, quantileLabel)
140
+ key := assemblyKey{name: sample.Name}
141
+ if ok {
142
+ key.hash = labels.Labels(base).Hash()
143
+ } else {
144
+ base = sample.Labels
145
+ key.hash = sample.Labels.Hash()
146
+ }
147
+
148
+ s := a.summaryFor(mf, key, base)
149
+ if !ok {
150
+ return
151
+ }
152
+ quantile, _ := strconv.ParseFloat(qv, 64)
153
+ s.quantiles = append(s.quantiles, Quantile{quantile: quantile, value: sample.Value})
154
+}
155
+
156
+func (a *assembler) addSummarySum(sample Sample) {
157
+ name := strings.TrimSuffix(sample.Name, sumSuffix)
158
+ mf := a.summaryFamily(name)
159
+ s := a.summaryFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
160
+ s.sum = sample.Value
161
+}
162
+
163
+func (a *assembler) addSummaryCount(sample Sample) {
164
+ name := strings.TrimSuffix(sample.Name, countSuffix)
165
+ mf := a.summaryFamily(name)
166
+ s := a.summaryFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
167
+ s.count = sample.Value
168
+}
169
+
170
+func (a *assembler) addHistogramBucket(sample Sample) {
171
+ name := strings.TrimSuffix(sample.Name, bucketSuffix)
172
+ mf := a.histogramFamily(name)
173
+ base, lev, ok := a.stripLabel(sample.Labels, bucketLabel)
174
+ key := assemblyKey{name: name}
175
+ if ok {
176
+ key.hash = labels.Labels(base).Hash()
177
+ } else {
178
+ base = sample.Labels
179
+ key.hash = sample.Labels.Hash()
180
+ }
181
+
182
+ h := a.histogramFor(mf, key, base)
183
+ if !ok {
184
+ return
185
+ }
186
+ bound, _ := strconv.ParseFloat(lev, 64)
187
+ h.buckets = append(h.buckets, Bucket{upperBound: bound, cumulativeCount: sample.Value})
188
+}
189
+
190
+func (a *assembler) addHistogramSum(sample Sample) {
191
+ name := strings.TrimSuffix(sample.Name, sumSuffix)
192
+ mf := a.histogramFamily(name)
193
+ h := a.histogramFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
194
+ h.sum = sample.Value
195
+}
196
+
197
+func (a *assembler) addHistogramCount(sample Sample) {
198
+ name := strings.TrimSuffix(sample.Name, countSuffix)
199
+ mf := a.histogramFamily(name)
200
+ h := a.histogramFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
201
+ h.count = sample.Value
202
+}
203
+
204
+func (a *assembler) summaryFamily(name string) *MetricFamily {
205
+ mf := a.ensureFamily(name)
206
+ mf.typ = model.MetricTypeSummary
207
+ return mf
208
+}
209
+
210
+func (a *assembler) histogramFamily(name string) *MetricFamily {
211
+ mf := a.ensureFamily(name)
212
+ mf.typ = model.MetricTypeHistogram
213
+ return mf
214
+}
215
+
216
+func (a *assembler) summaryFor(mf *MetricFamily, key assemblyKey, lbs labels.Labels) *Summary {
217
+ if s, ok := a.summaries[key]; ok {
218
+ return s
219
+ }
220
+
221
+ m := a.appendMetric(mf, lbs)
222
+ if m.summary == nil {
223
+ m.summary = &Summary{}
224
+ } else {
225
+ m.summary.sum = 0
226
+ m.summary.count = 0
227
+ m.summary.quantiles = m.summary.quantiles[:0]
228
+ }
229
+
230
+ a.summaries[key] = m.summary
231
+ return m.summary
232
+}
233
+
234
+func (a *assembler) histogramFor(mf *MetricFamily, key assemblyKey, lbs labels.Labels) *Histogram {
235
+ if h, ok := a.histograms[key]; ok {
236
+ return h
237
+ }
238
+
239
+ m := a.appendMetric(mf, lbs)
240
+ if m.histogram == nil {
241
+ m.histogram = &Histogram{}
242
+ } else {
243
+ m.histogram.sum = 0
244
+ m.histogram.count = 0
245
+ m.histogram.buckets = m.histogram.buckets[:0]
246
+ }
247
+
248
+ a.histograms[key] = m.histogram
249
+ return m.histogram
250
+}
251
+
252
+// appendMetric grows mf.metrics by one, reusing the backing array across cycles
253
+// and storing a copy of lbs. Instrument pointers on a reused slot are left in
254
+// place: a family keeps a stable type across scrapes, so the caller reuses the
255
+// existing instrument and allocates only when its pointer is nil. This preserves
256
+// the legacy allocation profile (no per-scrape instrument churn).
257
+func (a *assembler) appendMetric(mf *MetricFamily, lbs labels.Labels) *Metric {
258
+ idx := len(mf.metrics)
259
+ if idx == cap(mf.metrics) {
260
+ mf.metrics = append(mf.metrics, Metric{})
261
+ } else {
262
+ mf.metrics = mf.metrics[:idx+1]
263
+ }
264
+
265
+ m := &mf.metrics[idx]
266
+ m.labels = m.labels[:0]
267
+ m.labels = append(m.labels, lbs...)
268
+ return m
269
+}
270
+
271
+func (a *assembler) ensureFamily(name string) *MetricFamily {
272
+ if a.currFamily != nil && a.currName == name {
273
+ return a.currFamily
274
+ }
275
+ mf, ok := a.metrics[name]
276
+ if !ok {
277
+ mf = &MetricFamily{name: name, typ: model.MetricTypeUnknown}
278
+ a.metrics[name] = mf
279
+ }
280
+ a.currName = name
281
+ a.currFamily = mf
282
+ return mf
283
+}
284
+
285
+// stripLabel returns the label set without name and the removed value, using a
286
+// reusable scratch buffer. The result is valid only until the next stripLabel.
287
+func (a *assembler) stripLabel(lbs labels.Labels, name string) (labels.Labels, string, bool) {
288
+ a.scratch = a.scratch[:0]
289
+ var (
290
+ value string
291
+ found bool
292
+ )
293
+ for _, lb := range lbs {
294
+ if lb.Name == name {
295
+ value = lb.Value
296
+ found = true
297
+ continue
298
+ }
299
+ a.scratch = append(a.scratch, lb)
300
+ }
301
+ if !found {
302
+ return nil, "", false
303
+ }
304
+ return a.scratch, value, true
305
+}
src/go/pkg/prometheus/fetch.go
new
+96
@@ -0,0 +1,96 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package prometheus
4
+
5
+import (
6
+ "bufio"
7
+ "compress/gzip"
8
+ "context"
9
+ "fmt"
10
+ "io"
11
+ "net/http"
12
+ "os"
13
+ "strings"
14
+
15
+ "github.com/netdata/netdata/go/plugins/pkg/web"
16
+)
17
+
18
+const acceptHeader = `text/plain;version=0.0.4;q=1,*/*;q=0.1`
19
+
20
+// fetcher writes the raw exposition text for one scrape into w. A single
21
+// prometheus instance owns one fetcher and reuses it across scrapes.
22
+type fetcher interface {
23
+ fetch(ctx context.Context, w io.Writer) error
24
+}
25
+
26
+// fileFetcher reads the exposition text from a local file (file:// URLs).
27
+type fileFetcher struct {
28
+ path string
29
+}
30
+
31
+func (f *fileFetcher) fetch(_ context.Context, w io.Writer) error {
32
+ file, err := os.Open(f.path)
33
+ if err != nil {
34
+ return err
35
+ }
36
+ defer func() { _ = file.Close() }()
37
+
38
+ _, err = io.Copy(w, file)
39
+
40
+ return err
41
+}
42
+
43
+// httpFetcher scrapes the exposition text over HTTP, transparently decompressing
44
+// gzip responses. The gzip reader and its buffered source are reused across scrapes.
45
+type httpFetcher struct {
46
+ client *http.Client
47
+ request web.RequestConfig
48
+
49
+ gzipr *gzip.Reader
50
+ bodyBuf *bufio.Reader
51
+}
52
+
53
+func (f *httpFetcher) fetch(ctx context.Context, w io.Writer) error {
54
+ req, err := web.NewHTTPRequest(f.request)
55
+ if err != nil {
56
+ return err
57
+ }
58
+ req = req.WithContext(ctx)
59
+
60
+ req.Header.Add("Accept", acceptHeader)
61
+ req.Header.Add("Accept-Encoding", "gzip")
62
+
63
+ resp, err := f.client.Do(req)
64
+ if err != nil {
65
+ return err
66
+ }
67
+
68
+ defer web.CloseBody(resp)
69
+
70
+ if resp.StatusCode != http.StatusOK {
71
+ return fmt.Errorf("server '%s' returned HTTP status code %d (%s)", req.URL, resp.StatusCode, resp.Status)
72
+ }
73
+
74
+ if !strings.EqualFold(resp.Header.Get("Content-Encoding"), "gzip") {
75
+ _, err = io.Copy(w, resp.Body)
76
+ return err
77
+ }
78
+
79
+ if f.gzipr == nil {
80
+ f.bodyBuf = bufio.NewReader(resp.Body)
81
+ f.gzipr, err = gzip.NewReader(f.bodyBuf)
82
+ if err != nil {
83
+ return err
84
+ }
85
+ } else {
86
+ f.bodyBuf.Reset(resp.Body)
87
+ if err := f.gzipr.Reset(f.bodyBuf); err != nil {
88
+ return err
89
+ }
90
+ }
91
+
92
+ _, err = io.Copy(w, f.gzipr)
93
+ _ = f.gzipr.Close()
94
+
95
+ return err
96
+}
src/go/pkg/prometheus/labels.go
new
+45
@@ -0,0 +1,45 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package prometheus
4
+
5
+import "github.com/prometheus/prometheus/model/labels"
6
+
7
+func copyLabels(lbs []labels.Label) []labels.Label {
8
+ return append([]labels.Label(nil), lbs...)
9
+}
10
+
11
+// copyLabelsWithoutName returns a fresh copy of lbs with __name__ removed. In the
12
+// common case __name__ sorts first (it precedes lowercase label names), so the
13
+// remainder is contiguous and copied directly; otherwise a rare label that sorts
14
+// before __name__ (e.g. "UUID") is skipped element by element.
15
+func copyLabelsWithoutName(lbs labels.Labels) labels.Labels {
16
+ if len(lbs) > 0 && lbs[0].Name == labels.MetricName {
17
+ return copyLabels(lbs[1:])
18
+ }
19
+ out := make([]labels.Label, 0, len(lbs))
20
+ for _, lb := range lbs {
21
+ if lb.Name == labels.MetricName {
22
+ continue
23
+ }
24
+ out = append(out, lb)
25
+ }
26
+ return out
27
+}
28
+
29
+func removeLabel(lbs labels.Labels, name string) (labels.Labels, string, bool) {
30
+ for i, v := range lbs {
31
+ if v.Name == name {
32
+ return append(lbs[:i], lbs[i+1:]...), v.Value, true
33
+ }
34
+ }
35
+ return lbs, "", false
36
+}
37
+
38
+func metricNameValue(lbs labels.Labels) (string, bool) {
39
+ for _, v := range lbs {
40
+ if v.Name == labels.MetricName {
41
+ return v.Value, true
42
+ }
43
+ }
44
+ return "", false
45
+}
src/go/pkg/prometheus/parse.go
-335
@@ -6,7 +6,6 @@ import (
6
"errors"
7
"fmt"
8
"io"
9
- "strconv"
9
"strings"
10
11
"github.com/prometheus/common/model"
@@ -398,340 +397,6 @@ func emitResolvedPending(pending []pendingSample, baseName string, typ model.Met
397
return out, nil
398
}
399
401
-// assembler folds the driver's classified sample stream into typed
402
-// MetricFamilies. Grouping of summary quantiles / histogram buckets with their
403
-// _sum/_count is keyed by (family name, hash of the base labels). Buffers are
404
-// reused across cycles via reset().
405
-type assembler struct {
406
- metrics MetricFamilies
407
- summaries map[assemblyKey]*Summary
408
- histograms map[assemblyKey]*Histogram
409
- scratch labels.Labels
410
-
411
- // currName/currFamily cache the most recent family to skip the metrics map
412
- // lookup for the common case of consecutive samples in the same family.
413
- currName string
414
- currFamily *MetricFamily
415
-}
416
-
417
-type assemblyKey struct {
418
- name string
419
- hash uint64
420
-}
421
-
422
-func (a *assembler) reset() {
423
- a.currName = ""
424
- a.currFamily = nil
425
-
426
- if a.metrics == nil {
427
- a.metrics = make(MetricFamilies)
428
- }
429
- for _, mf := range a.metrics {
430
- mf.help = ""
431
- mf.typ = ""
432
- mf.metrics = mf.metrics[:0]
433
- }
434
-
435
- if a.summaries == nil {
436
- a.summaries = make(map[assemblyKey]*Summary)
437
- }
438
- for k := range a.summaries {
439
- delete(a.summaries, k)
440
- }
441
-
442
- if a.histograms == nil {
443
- a.histograms = make(map[assemblyKey]*Histogram)
444
- }
445
- for k := range a.histograms {
446
- delete(a.histograms, k)
447
- }
448
-}
449
-
450
-func (a *assembler) applyHelp(name, help string) {
451
- mf := a.ensureFamily(name)
452
- mf.help = help
453
-}
454
-
455
-func (a *assembler) applySample(sample Sample) error {
456
- switch sample.Kind {
457
- case SampleKindSummaryQuantile:
458
- a.addSummaryQuantile(sample)
459
- case SampleKindSummarySum:
460
- a.addSummarySum(sample)
461
- case SampleKindSummaryCount:
462
- a.addSummaryCount(sample)
463
- case SampleKindHistogramBucket:
464
- a.addHistogramBucket(sample)
465
- case SampleKindHistogramSum:
466
- a.addHistogramSum(sample)
467
- case SampleKindHistogramCount:
468
- a.addHistogramCount(sample)
469
- default:
470
- switch sample.FamilyType {
471
- case model.MetricTypeSummary:
472
- mf := a.summaryFamily(sample.Name)
473
- a.summaryFor(mf, assemblyKey{name: sample.Name, hash: sample.Labels.Hash()}, sample.Labels)
474
- case model.MetricTypeHistogram:
475
- mf := a.histogramFamily(sample.Name)
476
- a.histogramFor(mf, assemblyKey{name: sample.Name, hash: sample.Labels.Hash()}, sample.Labels)
477
- default:
478
- a.addScalar(sample)
479
- }
480
- }
481
- return nil
482
-}
483
-
484
-func (a *assembler) families() MetricFamilies {
485
- for name, mf := range a.metrics {
486
- if len(mf.metrics) == 0 {
487
- delete(a.metrics, name)
488
- }
489
- }
490
- return a.metrics
491
-}
492
-
493
-func (a *assembler) addScalar(sample Sample) {
494
- mf := a.ensureFamily(sample.Name)
495
-
496
- typ := sample.FamilyType
497
- if typ == "" {
498
- typ = model.MetricTypeUnknown
499
- }
500
- if mf.typ == "" || mf.typ == model.MetricTypeUnknown {
501
- mf.typ = typ
502
- }
503
-
504
- m := a.appendMetric(mf, sample.Labels)
505
-
506
- switch typ {
507
- case model.MetricTypeGauge:
508
- if m.gauge == nil {
509
- m.gauge = &Gauge{}
510
- }
511
- m.gauge.value = sample.Value
512
- case model.MetricTypeCounter:
513
- if m.counter == nil {
514
- m.counter = &Counter{}
515
- }
516
- m.counter.value = sample.Value
517
- default:
518
- if m.untyped == nil {
519
- m.untyped = &Untyped{}
520
- }
521
- m.untyped.value = sample.Value
522
- }
523
-}
524
-
525
-func (a *assembler) addSummaryQuantile(sample Sample) {
526
- mf := a.summaryFamily(sample.Name)
527
- base, qv, ok := a.stripLabel(sample.Labels, quantileLabel)
528
- key := assemblyKey{name: sample.Name}
529
- if ok {
530
- key.hash = labels.Labels(base).Hash()
531
- } else {
532
- base = sample.Labels
533
- key.hash = sample.Labels.Hash()
534
- }
535
-
536
- s := a.summaryFor(mf, key, base)
537
- if !ok {
538
- return
539
- }
540
- quantile, _ := strconv.ParseFloat(qv, 64)
541
- s.quantiles = append(s.quantiles, Quantile{quantile: quantile, value: sample.Value})
542
-}
543
-
544
-func (a *assembler) addSummarySum(sample Sample) {
545
- name := strings.TrimSuffix(sample.Name, sumSuffix)
546
- mf := a.summaryFamily(name)
547
- s := a.summaryFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
548
- s.sum = sample.Value
549
-}
550
-
551
-func (a *assembler) addSummaryCount(sample Sample) {
552
- name := strings.TrimSuffix(sample.Name, countSuffix)
553
- mf := a.summaryFamily(name)
554
- s := a.summaryFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
555
- s.count = sample.Value
556
-}
557
-
558
-func (a *assembler) addHistogramBucket(sample Sample) {
559
- name := strings.TrimSuffix(sample.Name, bucketSuffix)
560
- mf := a.histogramFamily(name)
561
- base, lev, ok := a.stripLabel(sample.Labels, bucketLabel)
562
- key := assemblyKey{name: name}
563
- if ok {
564
- key.hash = labels.Labels(base).Hash()
565
- } else {
566
- base = sample.Labels
567
- key.hash = sample.Labels.Hash()
568
- }
569
-
570
- h := a.histogramFor(mf, key, base)
571
- if !ok {
572
- return
573
- }
574
- bound, _ := strconv.ParseFloat(lev, 64)
575
- h.buckets = append(h.buckets, Bucket{upperBound: bound, cumulativeCount: sample.Value})
576
-}
577
-
578
-func (a *assembler) addHistogramSum(sample Sample) {
579
- name := strings.TrimSuffix(sample.Name, sumSuffix)
580
- mf := a.histogramFamily(name)
581
- h := a.histogramFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
582
- h.sum = sample.Value
583
-}
584
-
585
-func (a *assembler) addHistogramCount(sample Sample) {
586
- name := strings.TrimSuffix(sample.Name, countSuffix)
587
- mf := a.histogramFamily(name)
588
- h := a.histogramFor(mf, assemblyKey{name: name, hash: sample.Labels.Hash()}, sample.Labels)
589
- h.count = sample.Value
590
-}
591
-
592
-func (a *assembler) summaryFamily(name string) *MetricFamily {
593
- mf := a.ensureFamily(name)
594
- mf.typ = model.MetricTypeSummary
595
- return mf
596
-}
597
-
598
-func (a *assembler) histogramFamily(name string) *MetricFamily {
599
- mf := a.ensureFamily(name)
600
- mf.typ = model.MetricTypeHistogram
601
- return mf
602
-}
603
-
604
-func (a *assembler) summaryFor(mf *MetricFamily, key assemblyKey, lbs labels.Labels) *Summary {
605
- if s, ok := a.summaries[key]; ok {
606
- return s
607
- }
608
-
609
- m := a.appendMetric(mf, lbs)
610
- if m.summary == nil {
611
- m.summary = &Summary{}
612
- } else {
613
- m.summary.sum = 0
614
- m.summary.count = 0
615
- m.summary.quantiles = m.summary.quantiles[:0]
616
- }
617
-
618
- a.summaries[key] = m.summary
619
- return m.summary
620
-}
621
-
622
-func (a *assembler) histogramFor(mf *MetricFamily, key assemblyKey, lbs labels.Labels) *Histogram {
623
- if h, ok := a.histograms[key]; ok {
624
- return h
625
- }
626
-
627
- m := a.appendMetric(mf, lbs)
628
- if m.histogram == nil {
629
- m.histogram = &Histogram{}
630
- } else {
631
- m.histogram.sum = 0
632
- m.histogram.count = 0
633
- m.histogram.buckets = m.histogram.buckets[:0]
634
- }
635
-
636
- a.histograms[key] = m.histogram
637
- return m.histogram
638
-}
639
-
640
-// appendMetric grows mf.metrics by one, reusing the backing array across cycles
641
-// and storing a copy of lbs. Instrument pointers on a reused slot are left in
642
-// place: a family keeps a stable type across scrapes, so the caller reuses the
643
-// existing instrument and allocates only when its pointer is nil. This preserves
644
-// the legacy allocation profile (no per-scrape instrument churn).
645
-func (a *assembler) appendMetric(mf *MetricFamily, lbs labels.Labels) *Metric {
646
- idx := len(mf.metrics)
647
- if idx == cap(mf.metrics) {
648
- mf.metrics = append(mf.metrics, Metric{})
649
- } else {
650
- mf.metrics = mf.metrics[:idx+1]
651
- }
652
-
653
- m := &mf.metrics[idx]
654
- m.labels = m.labels[:0]
655
- m.labels = append(m.labels, lbs...)
656
- return m
657
-}
658
-
659
-func (a *assembler) ensureFamily(name string) *MetricFamily {
660
- if a.currFamily != nil && a.currName == name {
661
- return a.currFamily
662
- }
663
- mf, ok := a.metrics[name]
664
- if !ok {
665
- mf = &MetricFamily{name: name, typ: model.MetricTypeUnknown}
666
- a.metrics[name] = mf
667
- }
668
- a.currName = name
669
- a.currFamily = mf
670
- return mf
671
-}
672
-
673
-// stripLabel returns the label set without name and the removed value, using a
674
-// reusable scratch buffer. The result is valid only until the next stripLabel.
675
-func (a *assembler) stripLabel(lbs labels.Labels, name string) (labels.Labels, string, bool) {
676
- a.scratch = a.scratch[:0]
677
- var (
678
- value string
679
- found bool
680
- )
681
- for _, lb := range lbs {
682
- if lb.Name == name {
683
- value = lb.Value
684
- found = true
685
- continue
686
- }
687
- a.scratch = append(a.scratch, lb)
688
- }
689
- if !found {
690
- return nil, "", false
691
- }
692
- return a.scratch, value, true
693
-}
694
-
695
-func copyLabels(lbs []labels.Label) []labels.Label {
696
- return append([]labels.Label(nil), lbs...)
697
-}
698
-
699
-// copyLabelsWithoutName returns a fresh copy of lbs with __name__ removed. In the
700
-// common case __name__ sorts first (it precedes lowercase label names), so the
701
-// remainder is contiguous and copied directly; otherwise a rare label that sorts
702
-// before __name__ (e.g. "UUID") is skipped element by element.
703
-func copyLabelsWithoutName(lbs labels.Labels) labels.Labels {
704
- if len(lbs) > 0 && lbs[0].Name == labels.MetricName {
705
- return copyLabels(lbs[1:])
706
- }
707
- out := make([]labels.Label, 0, len(lbs))
708
- for _, lb := range lbs {
709
- if lb.Name == labels.MetricName {
710
- continue
711
- }
712
- out = append(out, lb)
713
- }
714
- return out
715
-}
716
-
717
-func removeLabel(lbs labels.Labels, name string) (labels.Labels, string, bool) {
718
- for i, v := range lbs {
719
- if v.Name == name {
720
- return append(lbs[:i], lbs[i+1:]...), v.Value, true
721
- }
722
- }
723
- return lbs, "", false
724
-}
725
-
726
-func metricNameValue(lbs labels.Labels) (string, bool) {
727
- for _, v := range lbs {
728
- if v.Name == labels.MetricName {
729
- return v.Value, true
730
- }
731
- }
732
- return "", false
733
-}
734
-
400
func sanitizeHelp(help string) string {
401
if strings.IndexByte(help, '\n') == -1 {
402
return help
src/go/pkg/prometheus/sample.go
renamed
src/go/pkg/prometheus/scrape.go
renamed
+11
-81
@@ -3,15 +3,10 @@
3
package prometheus
4
5
import (
6
- "bufio"
6
"bytes"
8
- "compress/gzip"
7
"context"
10
- "fmt"
11
- "io"
8
"net/http"
9
"net/url"
14
- "os"
10
"path/filepath"
11
12
"github.com/netdata/netdata/go/plugins/pkg/prometheus/selector"
@@ -40,24 +35,15 @@ type (
35
}
36
37
prometheus struct {
43
- client *http.Client
44
- request web.RequestConfig
45
- filepath string
46
-
47
- sr selector.Selector
38
+ client *http.Client
39
+ src fetcher
40
41
parser promTextParser
42
51
- buf *bytes.Buffer
52
- gzipr *gzip.Reader
53
- bodyBuf *bufio.Reader
43
+ buf *bytes.Buffer
44
}
45
)
46
57
-const (
58
- acceptHeader = `text/plain;version=0.0.4;q=1,*/*;q=0.1`
59
-)
60
-
47
// New creates a Prometheus instance.
48
func New(client *http.Client, request web.RequestConfig) Prometheus {
49
return NewWithSelector(client, request, nil)
@@ -66,15 +52,15 @@ func New(client *http.Client, request web.RequestConfig) Prometheus {
52
// NewWithSelector creates a Prometheus instance with the selector.
53
func NewWithSelector(client *http.Client, request web.RequestConfig, sr selector.Selector) Prometheus {
54
p := &prometheus{
69
- client: client,
70
- request: request,
71
- sr: sr,
72
- buf: bytes.NewBuffer(make([]byte, 0, 16000)),
73
- parser: promTextParser{sr: sr},
55
+ client: client,
56
+ buf: bytes.NewBuffer(make([]byte, 0, 16000)),
57
+ parser: promTextParser{sr: sr},
58
}
59
60
if v, err := url.Parse(request.URL); err == nil && v.Scheme == "file" {
77
- p.filepath = filepath.Join(v.Host, v.Path)
61
+ p.src = &fileFetcher{path: filepath.Join(v.Host, v.Path)}
62
+ } else {
63
+ p.src = &httpFetcher{client: client, request: request}
64
}
65
66
return p
@@ -88,7 +74,7 @@ func (p *prometheus) HTTPClient() *http.Client {
74
func (p *prometheus) ScrapeSeries() (Series, error) {
75
p.buf.Reset()
76
91
- if err := p.fetch(context.Background(), p.buf); err != nil {
77
+ if err := p.src.fetch(context.Background(), p.buf); err != nil {
78
return nil, err
79
}
80
@@ -102,65 +88,9 @@ func (p *prometheus) Scrape() (MetricFamilies, error) {
88
func (p *prometheus) ScrapeWithTransform(ctx context.Context, transform SampleTransform) (MetricFamilies, error) {
89
p.buf.Reset()
90
105
- if err := p.fetch(ctx, p.buf); err != nil {
91
+ if err := p.src.fetch(ctx, p.buf); err != nil {
92
return nil, err
93
}
94
95
return p.parser.parseToMetricFamilies(p.buf.Bytes(), transform)
96
}
111
-
112
-func (p *prometheus) fetch(ctx context.Context, w io.Writer) error {
113
- // TODO: should be a separate text file prom client
114
- if p.filepath != "" {
115
- f, err := os.Open(p.filepath)
116
- if err != nil {
117
- return err
118
- }
119
- defer func() { _ = f.Close() }()
120
-
121
- _, err = io.Copy(w, f)
122
-
123
- return err
124
- }
125
-
126
- req, err := web.NewHTTPRequest(p.request)
127
- if err != nil {
128
- return err
129
- }
130
- req = req.WithContext(ctx)
131
-
132
- req.Header.Add("Accept", acceptHeader)
133
- req.Header.Add("Accept-Encoding", "gzip")
134
-
135
- resp, err := p.client.Do(req)
136
- if err != nil {
137
- return err
138
- }
139
-
140
- defer web.CloseBody(resp)
141
-
142
- if resp.StatusCode != http.StatusOK {
143
- return fmt.Errorf("server '%s' returned HTTP status code %d (%s)", req.URL, resp.StatusCode, resp.Status)
144
- }
145
-
146
- if resp.Header.Get("Content-Encoding") != "gzip" {
147
- _, err = io.Copy(w, resp.Body)
148
- return err
149
- }
150
-
151
- if p.gzipr == nil {
152
- p.bodyBuf = bufio.NewReader(resp.Body)
153
- p.gzipr, err = gzip.NewReader(p.bodyBuf)
154
- if err != nil {
155
- return err
156
- }
157
- } else {
158
- p.bodyBuf.Reset(resp.Body)
159
- _ = p.gzipr.Reset(p.bodyBuf)
160
- }
161
-
162
- _, err = io.Copy(w, p.gzipr)
163
- _ = p.gzipr.Close()
164
-
165
- return err
166
-}