cubefs

1
// Copyright 2015 The Prometheus Authors
2
// Licensed under the Apache License, Version 2.0 (the "License");
3
// you may not use this file except in compliance with the License.
4
// You may obtain a copy of the License at
5
//
6
// http://www.apache.org/licenses/LICENSE-2.0
7
//
8
// Unless required by applicable law or agreed to in writing, software
9
// distributed under the License is distributed on an "AS IS" BASIS,
10
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11
// See the License for the specific language governing permissions and
12
// limitations under the License.
13

14
package expfmt
15

16
import (
17
	"fmt"
18
	"io"
19
	"math"
20
	"mime"
21
	"net/http"
22

23
	dto "github.com/prometheus/client_model/go"
24

25
	"github.com/matttproud/golang_protobuf_extensions/pbutil"
26
	"github.com/prometheus/common/model"
27
)
28

29
// Decoder types decode an input stream into metric families.
30
type Decoder interface {
31
	Decode(*dto.MetricFamily) error
32
}
33

34
// DecodeOptions contains options used by the Decoder and in sample extraction.
35
type DecodeOptions struct {
36
	// Timestamp is added to each value from the stream that has no explicit timestamp set.
37
	Timestamp model.Time
38
}
39

40
// ResponseFormat extracts the correct format from a HTTP response header.
41
// If no matching format can be found FormatUnknown is returned.
42
func ResponseFormat(h http.Header) Format {
43
	ct := h.Get(hdrContentType)
44

45
	mediatype, params, err := mime.ParseMediaType(ct)
46
	if err != nil {
47
		return FmtUnknown
48
	}
49

50
	const textType = "text/plain"
51

52
	switch mediatype {
53
	case ProtoType:
54
		if p, ok := params["proto"]; ok && p != ProtoProtocol {
55
			return FmtUnknown
56
		}
57
		if e, ok := params["encoding"]; ok && e != "delimited" {
58
			return FmtUnknown
59
		}
60
		return FmtProtoDelim
61

62
	case textType:
63
		if v, ok := params["version"]; ok && v != TextVersion {
64
			return FmtUnknown
65
		}
66
		return FmtText
67
	}
68

69
	return FmtUnknown
70
}
71

72
// NewDecoder returns a new decoder based on the given input format.
73
// If the input format does not imply otherwise, a text format decoder is returned.
74
func NewDecoder(r io.Reader, format Format) Decoder {
75
	switch format {
76
	case FmtProtoDelim:
77
		return &protoDecoder{r: r}
78
	}
79
	return &textDecoder{r: r}
80
}
81

82
// protoDecoder implements the Decoder interface for protocol buffers.
83
type protoDecoder struct {
84
	r io.Reader
85
}
86

87
// Decode implements the Decoder interface.
88
func (d *protoDecoder) Decode(v *dto.MetricFamily) error {
89
	_, err := pbutil.ReadDelimited(d.r, v)
90
	if err != nil {
91
		return err
92
	}
93
	if !model.IsValidMetricName(model.LabelValue(v.GetName())) {
94
		return fmt.Errorf("invalid metric name %q", v.GetName())
95
	}
96
	for _, m := range v.GetMetric() {
97
		if m == nil {
98
			continue
99
		}
100
		for _, l := range m.GetLabel() {
101
			if l == nil {
102
				continue
103
			}
104
			if !model.LabelValue(l.GetValue()).IsValid() {
105
				return fmt.Errorf("invalid label value %q", l.GetValue())
106
			}
107
			if !model.LabelName(l.GetName()).IsValid() {
108
				return fmt.Errorf("invalid label name %q", l.GetName())
109
			}
110
		}
111
	}
112
	return nil
113
}
114

115
// textDecoder implements the Decoder interface for the text protocol.
116
type textDecoder struct {
117
	r    io.Reader
118
	p    TextParser
119
	fams []*dto.MetricFamily
120
}
121

122
// Decode implements the Decoder interface.
123
func (d *textDecoder) Decode(v *dto.MetricFamily) error {
124
	// TODO(fabxc): Wrap this as a line reader to make streaming safer.
125
	if len(d.fams) == 0 {
126
		// No cached metric families, read everything and parse metrics.
127
		fams, err := d.p.TextToMetricFamilies(d.r)
128
		if err != nil {
129
			return err
130
		}
131
		if len(fams) == 0 {
132
			return io.EOF
133
		}
134
		d.fams = make([]*dto.MetricFamily, 0, len(fams))
135
		for _, f := range fams {
136
			d.fams = append(d.fams, f)
137
		}
138
	}
139

140
	*v = *d.fams[0]
141
	d.fams = d.fams[1:]
142

143
	return nil
144
}
145

146
// SampleDecoder wraps a Decoder to extract samples from the metric families
147
// decoded by the wrapped Decoder.
148
type SampleDecoder struct {
149
	Dec  Decoder
150
	Opts *DecodeOptions
151

152
	f dto.MetricFamily
153
}
154

155
// Decode calls the Decode method of the wrapped Decoder and then extracts the
156
// samples from the decoded MetricFamily into the provided model.Vector.
157
func (sd *SampleDecoder) Decode(s *model.Vector) error {
158
	err := sd.Dec.Decode(&sd.f)
159
	if err != nil {
160
		return err
161
	}
162
	*s, err = extractSamples(&sd.f, sd.Opts)
163
	return err
164
}
165

166
// ExtractSamples builds a slice of samples from the provided metric
167
// families. If an error occurs during sample extraction, it continues to
168
// extract from the remaining metric families. The returned error is the last
169
// error that has occurred.
170
func ExtractSamples(o *DecodeOptions, fams ...*dto.MetricFamily) (model.Vector, error) {
171
	var (
172
		all     model.Vector
173
		lastErr error
174
	)
175
	for _, f := range fams {
176
		some, err := extractSamples(f, o)
177
		if err != nil {
178
			lastErr = err
179
			continue
180
		}
181
		all = append(all, some...)
182
	}
183
	return all, lastErr
184
}
185

186
func extractSamples(f *dto.MetricFamily, o *DecodeOptions) (model.Vector, error) {
187
	switch f.GetType() {
188
	case dto.MetricType_COUNTER:
189
		return extractCounter(o, f), nil
190
	case dto.MetricType_GAUGE:
191
		return extractGauge(o, f), nil
192
	case dto.MetricType_SUMMARY:
193
		return extractSummary(o, f), nil
194
	case dto.MetricType_UNTYPED:
195
		return extractUntyped(o, f), nil
196
	case dto.MetricType_HISTOGRAM:
197
		return extractHistogram(o, f), nil
198
	}
199
	return nil, fmt.Errorf("expfmt.extractSamples: unknown metric family type %v", f.GetType())
200
}
201

202
func extractCounter(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
203
	samples := make(model.Vector, 0, len(f.Metric))
204

205
	for _, m := range f.Metric {
206
		if m.Counter == nil {
207
			continue
208
		}
209

210
		lset := make(model.LabelSet, len(m.Label)+1)
211
		for _, p := range m.Label {
212
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
213
		}
214
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
215

216
		smpl := &model.Sample{
217
			Metric: model.Metric(lset),
218
			Value:  model.SampleValue(m.Counter.GetValue()),
219
		}
220

221
		if m.TimestampMs != nil {
222
			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
223
		} else {
224
			smpl.Timestamp = o.Timestamp
225
		}
226

227
		samples = append(samples, smpl)
228
	}
229

230
	return samples
231
}
232

233
func extractGauge(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
234
	samples := make(model.Vector, 0, len(f.Metric))
235

236
	for _, m := range f.Metric {
237
		if m.Gauge == nil {
238
			continue
239
		}
240

241
		lset := make(model.LabelSet, len(m.Label)+1)
242
		for _, p := range m.Label {
243
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
244
		}
245
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
246

247
		smpl := &model.Sample{
248
			Metric: model.Metric(lset),
249
			Value:  model.SampleValue(m.Gauge.GetValue()),
250
		}
251

252
		if m.TimestampMs != nil {
253
			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
254
		} else {
255
			smpl.Timestamp = o.Timestamp
256
		}
257

258
		samples = append(samples, smpl)
259
	}
260

261
	return samples
262
}
263

264
func extractUntyped(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
265
	samples := make(model.Vector, 0, len(f.Metric))
266

267
	for _, m := range f.Metric {
268
		if m.Untyped == nil {
269
			continue
270
		}
271

272
		lset := make(model.LabelSet, len(m.Label)+1)
273
		for _, p := range m.Label {
274
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
275
		}
276
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
277

278
		smpl := &model.Sample{
279
			Metric: model.Metric(lset),
280
			Value:  model.SampleValue(m.Untyped.GetValue()),
281
		}
282

283
		if m.TimestampMs != nil {
284
			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
285
		} else {
286
			smpl.Timestamp = o.Timestamp
287
		}
288

289
		samples = append(samples, smpl)
290
	}
291

292
	return samples
293
}
294

295
func extractSummary(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
296
	samples := make(model.Vector, 0, len(f.Metric))
297

298
	for _, m := range f.Metric {
299
		if m.Summary == nil {
300
			continue
301
		}
302

303
		timestamp := o.Timestamp
304
		if m.TimestampMs != nil {
305
			timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
306
		}
307

308
		for _, q := range m.Summary.Quantile {
309
			lset := make(model.LabelSet, len(m.Label)+2)
310
			for _, p := range m.Label {
311
				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
312
			}
313
			// BUG(matt): Update other names to "quantile".
314
			lset[model.LabelName(model.QuantileLabel)] = model.LabelValue(fmt.Sprint(q.GetQuantile()))
315
			lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
316

317
			samples = append(samples, &model.Sample{
318
				Metric:    model.Metric(lset),
319
				Value:     model.SampleValue(q.GetValue()),
320
				Timestamp: timestamp,
321
			})
322
		}
323

324
		lset := make(model.LabelSet, len(m.Label)+1)
325
		for _, p := range m.Label {
326
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
327
		}
328
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
329

330
		samples = append(samples, &model.Sample{
331
			Metric:    model.Metric(lset),
332
			Value:     model.SampleValue(m.Summary.GetSampleSum()),
333
			Timestamp: timestamp,
334
		})
335

336
		lset = make(model.LabelSet, len(m.Label)+1)
337
		for _, p := range m.Label {
338
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
339
		}
340
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
341

342
		samples = append(samples, &model.Sample{
343
			Metric:    model.Metric(lset),
344
			Value:     model.SampleValue(m.Summary.GetSampleCount()),
345
			Timestamp: timestamp,
346
		})
347
	}
348

349
	return samples
350
}
351

352
func extractHistogram(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
353
	samples := make(model.Vector, 0, len(f.Metric))
354

355
	for _, m := range f.Metric {
356
		if m.Histogram == nil {
357
			continue
358
		}
359

360
		timestamp := o.Timestamp
361
		if m.TimestampMs != nil {
362
			timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
363
		}
364

365
		infSeen := false
366

367
		for _, q := range m.Histogram.Bucket {
368
			lset := make(model.LabelSet, len(m.Label)+2)
369
			for _, p := range m.Label {
370
				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
371
			}
372
			lset[model.LabelName(model.BucketLabel)] = model.LabelValue(fmt.Sprint(q.GetUpperBound()))
373
			lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
374

375
			if math.IsInf(q.GetUpperBound(), +1) {
376
				infSeen = true
377
			}
378

379
			samples = append(samples, &model.Sample{
380
				Metric:    model.Metric(lset),
381
				Value:     model.SampleValue(q.GetCumulativeCount()),
382
				Timestamp: timestamp,
383
			})
384
		}
385

386
		lset := make(model.LabelSet, len(m.Label)+1)
387
		for _, p := range m.Label {
388
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
389
		}
390
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
391

392
		samples = append(samples, &model.Sample{
393
			Metric:    model.Metric(lset),
394
			Value:     model.SampleValue(m.Histogram.GetSampleSum()),
395
			Timestamp: timestamp,
396
		})
397

398
		lset = make(model.LabelSet, len(m.Label)+1)
399
		for _, p := range m.Label {
400
			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
401
		}
402
		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
403

404
		count := &model.Sample{
405
			Metric:    model.Metric(lset),
406
			Value:     model.SampleValue(m.Histogram.GetSampleCount()),
407
			Timestamp: timestamp,
408
		}
409
		samples = append(samples, count)
410

411
		if !infSeen {
412
			// Append an infinity bucket sample.
413
			lset := make(model.LabelSet, len(m.Label)+2)
414
			for _, p := range m.Label {
415
				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
416
			}
417
			lset[model.LabelName(model.BucketLabel)] = model.LabelValue("+Inf")
418
			lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
419

420
			samples = append(samples, &model.Sample{
421
				Metric:    model.Metric(lset),
422
				Value:     count.Value,
423
				Timestamp: timestamp,
424
			})
425
		}
426
	}
427

428
	return samples
429
}
430