decode.go

Documentation: github.com/prometheus/common/expfmt

     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  	"bufio"
    18  	"fmt"
    19  	"io"
    20  	"math"
    21  	"mime"
    22  	"net/http"
    23  
    24  	dto "github.com/prometheus/client_model/go"
    25  	"google.golang.org/protobuf/encoding/protodelim"
    26  
    27  	"github.com/prometheus/common/model"
    28  )
    29  
    30  // Decoder types decode an input stream into metric families.
    31  type Decoder interface {
    32  	Decode(*dto.MetricFamily) error
    33  }
    34  
    35  // DecodeOptions contains options used by the Decoder and in sample extraction.
    36  type DecodeOptions struct {
    37  	// Timestamp is added to each value from the stream that has no explicit timestamp set.
    38  	Timestamp model.Time
    39  }
    40  
    41  // ResponseFormat extracts the correct format from a HTTP response header.
    42  // If no matching format can be found FormatUnknown is returned.
    43  func ResponseFormat(h http.Header) Format {
    44  	ct := h.Get(hdrContentType)
    45  
    46  	mediatype, params, err := mime.ParseMediaType(ct)
    47  	if err != nil {
    48  		return fmtUnknown
    49  	}
    50  
    51  	const textType = "text/plain"
    52  
    53  	switch mediatype {
    54  	case ProtoType:
    55  		if p, ok := params["proto"]; ok && p != ProtoProtocol {
    56  			return fmtUnknown
    57  		}
    58  		if e, ok := params["encoding"]; ok && e != "delimited" {
    59  			return fmtUnknown
    60  		}
    61  		return fmtProtoDelim
    62  
    63  	case textType:
    64  		if v, ok := params["version"]; ok && v != TextVersion {
    65  			return fmtUnknown
    66  		}
    67  		return fmtText
    68  	}
    69  
    70  	return fmtUnknown
    71  }
    72  
    73  // NewDecoder returns a new decoder based on the given input format.
    74  // If the input format does not imply otherwise, a text format decoder is returned.
    75  func NewDecoder(r io.Reader, format Format) Decoder {
    76  	switch format.FormatType() {
    77  	case TypeProtoDelim:
    78  		return &protoDecoder{r: bufio.NewReader(r)}
    79  	}
    80  	return &textDecoder{r: r}
    81  }
    82  
    83  // protoDecoder implements the Decoder interface for protocol buffers.
    84  type protoDecoder struct {
    85  	r protodelim.Reader
    86  }
    87  
    88  // Decode implements the Decoder interface.
    89  func (d *protoDecoder) Decode(v *dto.MetricFamily) error {
    90  	opts := protodelim.UnmarshalOptions{
    91  		MaxSize: -1,
    92  	}
    93  	if err := opts.UnmarshalFrom(d.r, v); err != nil {
    94  		return err
    95  	}
    96  	if !model.IsValidMetricName(model.LabelValue(v.GetName())) {
    97  		return fmt.Errorf("invalid metric name %q", v.GetName())
    98  	}
    99  	for _, m := range v.GetMetric() {
   100  		if m == nil {
   101  			continue
   102  		}
   103  		for _, l := range m.GetLabel() {
   104  			if l == nil {
   105  				continue
   106  			}
   107  			if !model.LabelValue(l.GetValue()).IsValid() {
   108  				return fmt.Errorf("invalid label value %q", l.GetValue())
   109  			}
   110  			if !model.LabelName(l.GetName()).IsValid() {
   111  				return fmt.Errorf("invalid label name %q", l.GetName())
   112  			}
   113  		}
   114  	}
   115  	return nil
   116  }
   117  
   118  // textDecoder implements the Decoder interface for the text protocol.
   119  type textDecoder struct {
   120  	r    io.Reader
   121  	fams map[string]*dto.MetricFamily
   122  	err  error
   123  }
   124  
   125  // Decode implements the Decoder interface.
   126  func (d *textDecoder) Decode(v *dto.MetricFamily) error {
   127  	if d.err == nil {
   128  		// Read all metrics in one shot.
   129  		var p TextParser
   130  		d.fams, d.err = p.TextToMetricFamilies(d.r)
   131  		// If we don't get an error, store io.EOF for the end.
   132  		if d.err == nil {
   133  			d.err = io.EOF
   134  		}
   135  	}
   136  	// Pick off one MetricFamily per Decode until there's nothing left.
   137  	for key, fam := range d.fams {
   138  		v.Name = fam.Name
   139  		v.Help = fam.Help
   140  		v.Type = fam.Type
   141  		v.Metric = fam.Metric
   142  		delete(d.fams, key)
   143  		return nil
   144  	}
   145  	return d.err
   146  }
   147  
   148  // SampleDecoder wraps a Decoder to extract samples from the metric families
   149  // decoded by the wrapped Decoder.
   150  type SampleDecoder struct {
   151  	Dec  Decoder
   152  	Opts *DecodeOptions
   153  
   154  	f dto.MetricFamily
   155  }
   156  
   157  // Decode calls the Decode method of the wrapped Decoder and then extracts the
   158  // samples from the decoded MetricFamily into the provided model.Vector.
   159  func (sd *SampleDecoder) Decode(s *model.Vector) error {
   160  	err := sd.Dec.Decode(&sd.f)
   161  	if err != nil {
   162  		return err
   163  	}
   164  	*s, err = extractSamples(&sd.f, sd.Opts)
   165  	return err
   166  }
   167  
   168  // ExtractSamples builds a slice of samples from the provided metric
   169  // families. If an error occurs during sample extraction, it continues to
   170  // extract from the remaining metric families. The returned error is the last
   171  // error that has occurred.
   172  func ExtractSamples(o *DecodeOptions, fams ...*dto.MetricFamily) (model.Vector, error) {
   173  	var (
   174  		all     model.Vector
   175  		lastErr error
   176  	)
   177  	for _, f := range fams {
   178  		some, err := extractSamples(f, o)
   179  		if err != nil {
   180  			lastErr = err
   181  			continue
   182  		}
   183  		all = append(all, some...)
   184  	}
   185  	return all, lastErr
   186  }
   187  
   188  func extractSamples(f *dto.MetricFamily, o *DecodeOptions) (model.Vector, error) {
   189  	switch f.GetType() {
   190  	case dto.MetricType_COUNTER:
   191  		return extractCounter(o, f), nil
   192  	case dto.MetricType_GAUGE:
   193  		return extractGauge(o, f), nil
   194  	case dto.MetricType_SUMMARY:
   195  		return extractSummary(o, f), nil
   196  	case dto.MetricType_UNTYPED:
   197  		return extractUntyped(o, f), nil
   198  	case dto.MetricType_HISTOGRAM:
   199  		return extractHistogram(o, f), nil
   200  	}
   201  	return nil, fmt.Errorf("expfmt.extractSamples: unknown metric family type %v", f.GetType())
   202  }
   203  
   204  func extractCounter(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
   205  	samples := make(model.Vector, 0, len(f.Metric))
   206  
   207  	for _, m := range f.Metric {
   208  		if m.Counter == nil {
   209  			continue
   210  		}
   211  
   212  		lset := make(model.LabelSet, len(m.Label)+1)
   213  		for _, p := range m.Label {
   214  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   215  		}
   216  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
   217  
   218  		smpl := &model.Sample{
   219  			Metric: model.Metric(lset),
   220  			Value:  model.SampleValue(m.Counter.GetValue()),
   221  		}
   222  
   223  		if m.TimestampMs != nil {
   224  			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
   225  		} else {
   226  			smpl.Timestamp = o.Timestamp
   227  		}
   228  
   229  		samples = append(samples, smpl)
   230  	}
   231  
   232  	return samples
   233  }
   234  
   235  func extractGauge(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
   236  	samples := make(model.Vector, 0, len(f.Metric))
   237  
   238  	for _, m := range f.Metric {
   239  		if m.Gauge == nil {
   240  			continue
   241  		}
   242  
   243  		lset := make(model.LabelSet, len(m.Label)+1)
   244  		for _, p := range m.Label {
   245  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   246  		}
   247  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
   248  
   249  		smpl := &model.Sample{
   250  			Metric: model.Metric(lset),
   251  			Value:  model.SampleValue(m.Gauge.GetValue()),
   252  		}
   253  
   254  		if m.TimestampMs != nil {
   255  			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
   256  		} else {
   257  			smpl.Timestamp = o.Timestamp
   258  		}
   259  
   260  		samples = append(samples, smpl)
   261  	}
   262  
   263  	return samples
   264  }
   265  
   266  func extractUntyped(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
   267  	samples := make(model.Vector, 0, len(f.Metric))
   268  
   269  	for _, m := range f.Metric {
   270  		if m.Untyped == nil {
   271  			continue
   272  		}
   273  
   274  		lset := make(model.LabelSet, len(m.Label)+1)
   275  		for _, p := range m.Label {
   276  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   277  		}
   278  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
   279  
   280  		smpl := &model.Sample{
   281  			Metric: model.Metric(lset),
   282  			Value:  model.SampleValue(m.Untyped.GetValue()),
   283  		}
   284  
   285  		if m.TimestampMs != nil {
   286  			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
   287  		} else {
   288  			smpl.Timestamp = o.Timestamp
   289  		}
   290  
   291  		samples = append(samples, smpl)
   292  	}
   293  
   294  	return samples
   295  }
   296  
   297  func extractSummary(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
   298  	samples := make(model.Vector, 0, len(f.Metric))
   299  
   300  	for _, m := range f.Metric {
   301  		if m.Summary == nil {
   302  			continue
   303  		}
   304  
   305  		timestamp := o.Timestamp
   306  		if m.TimestampMs != nil {
   307  			timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
   308  		}
   309  
   310  		for _, q := range m.Summary.Quantile {
   311  			lset := make(model.LabelSet, len(m.Label)+2)
   312  			for _, p := range m.Label {
   313  				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   314  			}
   315  			// BUG(matt): Update other names to "quantile".
   316  			lset[model.LabelName(model.QuantileLabel)] = model.LabelValue(fmt.Sprint(q.GetQuantile()))
   317  			lset[model.MetricNameLabel] = model.LabelValue(f.GetName())
   318  
   319  			samples = append(samples, &model.Sample{
   320  				Metric:    model.Metric(lset),
   321  				Value:     model.SampleValue(q.GetValue()),
   322  				Timestamp: timestamp,
   323  			})
   324  		}
   325  
   326  		lset := make(model.LabelSet, len(m.Label)+1)
   327  		for _, p := range m.Label {
   328  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   329  		}
   330  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
   331  
   332  		samples = append(samples, &model.Sample{
   333  			Metric:    model.Metric(lset),
   334  			Value:     model.SampleValue(m.Summary.GetSampleSum()),
   335  			Timestamp: timestamp,
   336  		})
   337  
   338  		lset = make(model.LabelSet, len(m.Label)+1)
   339  		for _, p := range m.Label {
   340  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   341  		}
   342  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
   343  
   344  		samples = append(samples, &model.Sample{
   345  			Metric:    model.Metric(lset),
   346  			Value:     model.SampleValue(m.Summary.GetSampleCount()),
   347  			Timestamp: timestamp,
   348  		})
   349  	}
   350  
   351  	return samples
   352  }
   353  
   354  func extractHistogram(o *DecodeOptions, f *dto.MetricFamily) model.Vector {
   355  	samples := make(model.Vector, 0, len(f.Metric))
   356  
   357  	for _, m := range f.Metric {
   358  		if m.Histogram == nil {
   359  			continue
   360  		}
   361  
   362  		timestamp := o.Timestamp
   363  		if m.TimestampMs != nil {
   364  			timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
   365  		}
   366  
   367  		infSeen := false
   368  
   369  		for _, q := range m.Histogram.Bucket {
   370  			lset := make(model.LabelSet, len(m.Label)+2)
   371  			for _, p := range m.Label {
   372  				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   373  			}
   374  			lset[model.LabelName(model.BucketLabel)] = model.LabelValue(fmt.Sprint(q.GetUpperBound()))
   375  			lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
   376  
   377  			if math.IsInf(q.GetUpperBound(), +1) {
   378  				infSeen = true
   379  			}
   380  
   381  			samples = append(samples, &model.Sample{
   382  				Metric:    model.Metric(lset),
   383  				Value:     model.SampleValue(q.GetCumulativeCount()),
   384  				Timestamp: timestamp,
   385  			})
   386  		}
   387  
   388  		lset := make(model.LabelSet, len(m.Label)+1)
   389  		for _, p := range m.Label {
   390  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   391  		}
   392  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
   393  
   394  		samples = append(samples, &model.Sample{
   395  			Metric:    model.Metric(lset),
   396  			Value:     model.SampleValue(m.Histogram.GetSampleSum()),
   397  			Timestamp: timestamp,
   398  		})
   399  
   400  		lset = make(model.LabelSet, len(m.Label)+1)
   401  		for _, p := range m.Label {
   402  			lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   403  		}
   404  		lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
   405  
   406  		count := &model.Sample{
   407  			Metric:    model.Metric(lset),
   408  			Value:     model.SampleValue(m.Histogram.GetSampleCount()),
   409  			Timestamp: timestamp,
   410  		}
   411  		samples = append(samples, count)
   412  
   413  		if !infSeen {
   414  			// Append an infinity bucket sample.
   415  			lset := make(model.LabelSet, len(m.Label)+2)
   416  			for _, p := range m.Label {
   417  				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
   418  			}
   419  			lset[model.LabelName(model.BucketLabel)] = model.LabelValue("+Inf")
   420  			lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
   421  
   422  			samples = append(samples, &model.Sample{
   423  				Metric:    model.Metric(lset),
   424  				Value:     count.Value,
   425  				Timestamp: timestamp,
   426  			})
   427  		}
   428  	}
   429  
   430  	return samples
   431  }
   432
View as plain text