// Decode implements the Decoder interface.
func (d *protoDecoder) Decode(v *dto.MetricFamily) error {
	_, err := pbutil.ReadDelimited(d.r, v)
	if err != nil {
		return err
	}
	if !model.IsValidMetricName(model.LabelValue(v.GetName())) {
		return fmt.Errorf("invalid metric name %q", v.GetName())
	}
	for _, m := range v.GetMetric() {
		if m == nil {
			continue
		}
		for _, l := range m.GetLabel() {
			if l == nil {
				continue
			}
			if !model.LabelValue(l.GetValue()).IsValid() {
				return fmt.Errorf("invalid label value %q", l.GetValue())
			}
			if !model.LabelName(l.GetName()).IsValid() {
				return fmt.Errorf("invalid label name %q", l.GetName())
			}
		}
	}
	return nil
}

func newMetricFamily(dtoMF *dto.MetricFamily) *metricFamily {
	mf := &metricFamily{
		Name:    dtoMF.GetName(),
		Help:    dtoMF.GetHelp(),
		Type:    dtoMF.GetType().String(),
		Metrics: make([]interface{}, len(dtoMF.Metric)),
	}
	for i, m := range dtoMF.Metric {
		if dtoMF.GetType() == dto.MetricType_SUMMARY {
			mf.Metrics[i] = summary{
				Labels:    makeLabels(m),
				Quantiles: makeQuantiles(m),
				Count:     fmt.Sprint(m.GetSummary().GetSampleCount()),
				Sum:       fmt.Sprint(m.GetSummary().GetSampleSum()),
			}
		} else if dtoMF.GetType() == dto.MetricType_HISTOGRAM {
			mf.Metrics[i] = histogram{
				Labels:  makeLabels(m),
				Buckets: makeBuckets(m),
				Count:   fmt.Sprint(m.GetHistogram().GetSampleCount()),
				Sum:     fmt.Sprint(m.GetSummary().GetSampleSum()),
			}
		} else {
			mf.Metrics[i] = metric{
				Labels: makeLabels(m),
				Value:  fmt.Sprint(getValue(m)),
			}
		}
	}
	return mf
}

func (r *registry) giveMetricFamily(mf *dto.MetricFamily) {
	mf.Reset()
	select {
	case r.metricFamilyPool <- mf:
	default:
	}
}

func extractCounter(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
	samples := make(model.Samples, 0, len(f.Metric))

	for _, m := range f.Metric {
		if m.Counter == nil {
			continue
		}

		sample := new(model.Sample)
		samples = append(samples, sample)

		if m.TimestampMs != nil {
			sample.Timestamp = model.TimestampFromUnix(*m.TimestampMs / 1000)
		} else {
			sample.Timestamp = o.Timestamp
		}
		sample.Metric = model.Metric{}
		metric := sample.Metric

		for _, p := range m.Label {
			metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
		}

		metric[model.MetricNameLabel] = model.LabelValue(f.GetName())

		sample.Value = model.SampleValue(m.Counter.GetValue())
	}

	return out.Ingest(&Result{Samples: samples})
}

// PrintAsText outputs all metrics in text format.
func (r *Registry) PrintAsText(w io.Writer) error {
	var metricFamily prometheusgo.MetricFamily
	var ret error
	labels := r.getLabels()
	for _, metric := range r.tracked {
		metric.Inspect(func(v interface{}) {
			if ret != nil {
				return
			}
			if prom, ok := v.(PrometheusExportable); ok {
				metricFamily.Reset()
				metricFamily.Name = proto.String(exportedName(metric.GetName()))
				metricFamily.Help = proto.String(metric.GetHelp())
				prom.FillPrometheusMetric(&metricFamily)
				if len(labels) != 0 {
					// Set labels from registry. We only set one metric in the slice, but loop anyway.
					for _, m := range metricFamily.Metric {
						m.Label = labels
					}
				}
				if l := prom.GetLabels(); len(l) != 0 {
					// Append per-metric labels.
					for _, m := range metricFamily.Metric {
						m.Label = append(m.Label, l...)
					}
				}
				if _, err := expfmt.MetricFamilyToText(w, &metricFamily); err != nil {
					ret = err
				}
			}
		})
	}
	return ret
}

func extractUntyped(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
	samples := make(model.Samples, 0, len(f.Metric))

	for _, m := range f.Metric {
		if m.Untyped == nil {
			continue
		}

		sample := &model.Sample{
			Metric: model.Metric{},
			Value:  model.SampleValue(m.Untyped.GetValue()),
		}
		samples = append(samples, sample)

		if m.TimestampMs != nil {
			sample.Timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
		} else {
			sample.Timestamp = o.Timestamp
		}

		metric := sample.Metric
		for _, p := range m.Label {
			metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
		}
		metric[model.MetricNameLabel] = model.LabelValue(f.GetName())
	}

	return out.Ingest(samples)
}

func extractMetricFamily(out Ingester, o *ProcessOptions, family *dto.MetricFamily) error {
	switch family.GetType() {
	case dto.MetricType_COUNTER:
		if err := extractCounter(out, o, family); err != nil {
			return err
		}
	case dto.MetricType_GAUGE:
		if err := extractGauge(out, o, family); err != nil {
			return err
		}
	case dto.MetricType_SUMMARY:
		if err := extractSummary(out, o, family); err != nil {
			return err
		}
	case dto.MetricType_UNTYPED:
		if err := extractUntyped(out, o, family); err != nil {
			return err
		}
	case dto.MetricType_HISTOGRAM:
		if err := extractHistogram(out, o, family); err != nil {
			return err
		}
	}
	return nil
}

func scrapeMetrics(s *httptest.Server) ([]*prometheuspb.MetricFamily, error) {
	req, err := http.NewRequest("GET", s.URL+"/metrics", nil)
	if err != nil {
		return nil, fmt.Errorf("Unable to create http request: %v", err)
	}
	// Ask the prometheus exporter for its text protocol buffer format, since it's
	// much easier to parse than its plain-text format. Don't use the serialized
	// proto representation since it uses a non-standard varint delimiter between
	// metric families.
	req.Header.Add("Accept", scrapeRequestHeader)

	client := &http.Client{}
	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("Unable to contact metrics endpoint of master: %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return nil, fmt.Errorf("Non-200 response trying to scrape metrics from master: %v", resp)
	}

	// Each line in the response body should contain all the data for a single metric.
	var metrics []*prometheuspb.MetricFamily
	scanner := bufio.NewScanner(resp.Body)
	for scanner.Scan() {
		var metric prometheuspb.MetricFamily
		if err := proto.UnmarshalText(scanner.Text(), &metric); err != nil {
			return nil, fmt.Errorf("Failed to unmarshal line of metrics response: %v", err)
		}
		glog.Infof("Got metric %q", metric.GetName())
		metrics = append(metrics, &metric)
	}
	return metrics, nil
}

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

	for _, m := range f.Metric {
		if m.Untyped == nil {
			continue
		}

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

		smpl := &model.Sample{
			Metric: model.Metric(lset),
			Value:  model.SampleValue(m.Untyped.GetValue()),
		}

		if m.TimestampMs != nil {
			smpl.Timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
		} else {
			smpl.Timestamp = o.Timestamp
		}

		samples = append(samples, smpl)
	}

	return samples
}

func extractSamples(f *dto.MetricFamily, o *DecodeOptions) model.Vector {
	switch f.GetType() {
	case dto.MetricType_COUNTER:
		return extractCounter(o, f)
	case dto.MetricType_GAUGE:
		return extractGauge(o, f)
	case dto.MetricType_SUMMARY:
		return extractSummary(o, f)
	case dto.MetricType_UNTYPED:
		return extractUntyped(o, f)
	case dto.MetricType_HISTOGRAM:
		return extractHistogram(o, f)
	}
	panic("expfmt.extractSamples: unknown metric family type")
}

// PrintAsText outputs all metrics in text format.
func (r *Registry) PrintAsText(w io.Writer) error {
	var metricFamily prometheusgo.MetricFamily
	var ret error
	r.Each(func(name string, v interface{}) {
		if ret != nil {
			return
		}
		if metric, ok := v.(PrometheusExportable); ok {
			metricFamily.Reset()
			metricFamily.Name = proto.String(exportedName(name))
			metric.FillPrometheusMetric(&metricFamily)
			if _, err := expfmt.MetricFamilyToText(w, &metricFamily); err != nil {
				ret = err
			}
		}
	})
	return ret
}

func (collector *PrometheusCollector) GetSpec() []v1.MetricSpec {

	response, err := collector.httpClient.Get(collector.configFile.Endpoint.URL)
	if err != nil {
		return nil
	}
	defer response.Body.Close()

	if response.StatusCode != http.StatusOK {
		return nil
	}

	dec := expfmt.NewDecoder(response.Body, expfmt.ResponseFormat(response.Header))

	var specs []v1.MetricSpec

	for {
		d := rawmodel.MetricFamily{}
		if err = dec.Decode(&d); err != nil {
			break
		}
		name := d.GetName()
		if len(name) == 0 {
			continue
		}
		// If metrics to collect is specified, skip any metrics not in the list to collect.
		if _, ok := collector.metricsSet[name]; collector.metricsSet != nil && !ok {
			continue
		}

		spec := v1.MetricSpec{
			Name:   name,
			Type:   metricType(d.GetType()),
			Format: v1.FloatType,
		}
		specs = append(specs, spec)
	}

	if err != nil && err != io.EOF {
		return nil
	}

	return specs
}

// FillPrometheusMetric fills the appropriate metric fields.
func (h *Histogram) FillPrometheusMetric(promMetric *prometheusgo.MetricFamily) {
	// TODO(mjibson): change to a Histogram once bucket counts are reasonable
	sum := &prometheusgo.Summary{}

	h.mu.Lock()
	maybeTick(h)
	merged := h.windowed.Merge()
	for _, b := range merged.CumulativeDistribution() {
		sum.Quantile = append(sum.Quantile, &prometheusgo.Quantile{
			Quantile: proto.Float64(b.Quantile),
			Value:    proto.Float64(float64(b.ValueAt)),
		})
	}
	sum.SampleCount = proto.Uint64(uint64(merged.TotalCount()))
	h.mu.Unlock()

	promMetric.Type = prometheusgo.MetricType_SUMMARY.Enum()
	promMetric.Metric = []*prometheusgo.Metric{
		{Summary: sum},
	}
}

func checkDescConsistency(
	metricFamily *dto.MetricFamily,
	dtoMetric *dto.Metric,
	desc *Desc,
) error {
	// Desc help consistency with metric family help.
	if metricFamily.GetHelp() != desc.help {
		return fmt.Errorf(
			"collected metric %s %s has help %q but should have %q",
			metricFamily.GetName(), dtoMetric, metricFamily.GetHelp(), desc.help,
		)
	}

	// Is the desc consistent with the content of the metric?
	lpsFromDesc := make([]*dto.LabelPair, 0, len(dtoMetric.Label))
	lpsFromDesc = append(lpsFromDesc, desc.constLabelPairs...)
	for _, l := range desc.variableLabels {
		lpsFromDesc = append(lpsFromDesc, &dto.LabelPair{
			Name: proto.String(l),
		})
	}
	if len(lpsFromDesc) != len(dtoMetric.Label) {
		return fmt.Errorf(
			"labels in collected metric %s %s are inconsistent with descriptor %s",
			metricFamily.GetName(), dtoMetric, desc,
		)
	}
	sort.Sort(LabelPairSorter(lpsFromDesc))
	for i, lpFromDesc := range lpsFromDesc {
		lpFromMetric := dtoMetric.Label[i]
		if lpFromDesc.GetName() != lpFromMetric.GetName() ||
			lpFromDesc.Value != nil && lpFromDesc.GetValue() != lpFromMetric.GetValue() {
			return fmt.Errorf(
				"labels in collected metric %s %s are inconsistent with descriptor %s",
				metricFamily.GetName(), dtoMetric, desc,
			)
		}
	}
	return nil
}

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

	for _, m := range f.Metric {
		if m.Summary == nil {
			continue
		}

		timestamp := o.Timestamp
		if m.TimestampMs != nil {
			timestamp = model.TimeFromUnixNano(*m.TimestampMs * 1000000)
		}

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

			samples = append(samples, &model.Sample{
				Metric:    model.Metric(lset),
				Value:     model.SampleValue(q.GetValue()),
				Timestamp: timestamp,
			})
		}

		if m.Summary.SampleSum != nil {
			lset := make(model.LabelSet, len(m.Label)+1)
			for _, p := range m.Label {
				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")

			samples = append(samples, &model.Sample{
				Metric:    model.Metric(lset),
				Value:     model.SampleValue(m.Summary.GetSampleSum()),
				Timestamp: timestamp,
			})
		}

		if m.Summary.SampleCount != nil {
			lset := make(model.LabelSet, len(m.Label)+1)
			for _, p := range m.Label {
				lset[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			lset[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")

			samples = append(samples, &model.Sample{
				Metric:    model.Metric(lset),
				Value:     model.SampleValue(m.Summary.GetSampleCount()),
				Timestamp: timestamp,
			})
		}
	}

	return samples
}

func extractSummary(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
	samples := make(model.Samples, 0, len(f.Metric))

	for _, m := range f.Metric {
		if m.Summary == nil {
			continue
		}

		timestamp := o.Timestamp
		if m.TimestampMs != nil {
			timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
		}

		for _, q := range m.Summary.Quantile {
			sample := &model.Sample{
				Metric:    model.Metric{},
				Value:     model.SampleValue(q.GetValue()),
				Timestamp: timestamp,
			}
			samples = append(samples, sample)

			metric := sample.Metric
			for _, p := range m.Label {
				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			// BUG(matt): Update other names to "quantile".
			metric[model.LabelName(model.QuantileLabel)] = model.LabelValue(fmt.Sprint(q.GetQuantile()))
			metric[model.MetricNameLabel] = model.LabelValue(f.GetName())
		}

		if m.Summary.SampleSum != nil {
			sum := &model.Sample{
				Metric:    model.Metric{},
				Value:     model.SampleValue(m.Summary.GetSampleSum()),
				Timestamp: timestamp,
			}
			samples = append(samples, sum)

			metric := sum.Metric
			for _, p := range m.Label {
				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
		}

		if m.Summary.SampleCount != nil {
			count := &model.Sample{
				Metric:    model.Metric{},
				Value:     model.SampleValue(m.Summary.GetSampleCount()),
				Timestamp: timestamp,
			}
			samples = append(samples, count)

			metric := count.Metric
			for _, p := range m.Label {
				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")
		}
	}

	return out.Ingest(samples)
}

func (r *registry) checkConsistency(metricFamily *dto.MetricFamily, dtoMetric *dto.Metric, desc *Desc, metricHashes map[uint64]struct{}) error {

	// Type consistency with metric family.
	if metricFamily.GetType() == dto.MetricType_GAUGE && dtoMetric.Gauge == nil ||
		metricFamily.GetType() == dto.MetricType_COUNTER && dtoMetric.Counter == nil ||
		metricFamily.GetType() == dto.MetricType_SUMMARY && dtoMetric.Summary == nil ||
		metricFamily.GetType() == dto.MetricType_HISTOGRAM && dtoMetric.Histogram == nil ||
		metricFamily.GetType() == dto.MetricType_UNTYPED && dtoMetric.Untyped == nil {
		return fmt.Errorf(
			"collected metric %s %s is not a %s",
			metricFamily.GetName(), dtoMetric, metricFamily.GetType(),
		)
	}

	// Is the metric unique (i.e. no other metric with the same name and the same label values)?
	h := fnv.New64a()
	var buf bytes.Buffer
	buf.WriteString(metricFamily.GetName())
	buf.WriteByte(separatorByte)
	h.Write(buf.Bytes())
	// Make sure label pairs are sorted. We depend on it for the consistency
	// check. Label pairs must be sorted by contract. But the point of this
	// method is to check for contract violations. So we better do the sort
	// now.
	sort.Sort(LabelPairSorter(dtoMetric.Label))
	for _, lp := range dtoMetric.Label {
		buf.Reset()
		buf.WriteString(lp.GetValue())
		buf.WriteByte(separatorByte)
		h.Write(buf.Bytes())
	}
	metricHash := h.Sum64()
	if _, exists := metricHashes[metricHash]; exists {
		return fmt.Errorf(
			"collected metric %s %s was collected before with the same name and label values",
			metricFamily.GetName(), dtoMetric,
		)
	}
	metricHashes[metricHash] = struct{}{}

	if desc == nil {
		return nil // Nothing left to check if we have no desc.
	}

	// Desc consistency with metric family.
	if metricFamily.GetName() != desc.fqName {
		return fmt.Errorf(
			"collected metric %s %s has name %q but should have %q",
			metricFamily.GetName(), dtoMetric, metricFamily.GetName(), desc.fqName,
		)
	}
	if metricFamily.GetHelp() != desc.help {
		return fmt.Errorf(
			"collected metric %s %s has help %q but should have %q",
			metricFamily.GetName(), dtoMetric, metricFamily.GetHelp(), desc.help,
		)
	}

	// Is the desc consistent with the content of the metric?
	lpsFromDesc := make([]*dto.LabelPair, 0, len(dtoMetric.Label))
	lpsFromDesc = append(lpsFromDesc, desc.constLabelPairs...)
	for _, l := range desc.variableLabels {
		lpsFromDesc = append(lpsFromDesc, &dto.LabelPair{
			Name: proto.String(l),
		})
	}
	if len(lpsFromDesc) != len(dtoMetric.Label) {
		return fmt.Errorf(
			"labels in collected metric %s %s are inconsistent with descriptor %s",
			metricFamily.GetName(), dtoMetric, desc,
		)
	}
	sort.Sort(LabelPairSorter(lpsFromDesc))
	for i, lpFromDesc := range lpsFromDesc {
		lpFromMetric := dtoMetric.Label[i]
		if lpFromDesc.GetName() != lpFromMetric.GetName() ||
			lpFromDesc.Value != nil && lpFromDesc.GetValue() != lpFromMetric.GetValue() {
			return fmt.Errorf(
				"labels in collected metric %s %s are inconsistent with descriptor %s",
				metricFamily.GetName(), dtoMetric, desc,
			)
		}
	}

	r.mtx.RLock() // Remaining checks need the read lock.
	defer r.mtx.RUnlock()

	// Is the desc registered?
	if _, exist := r.descIDs[desc.id]; !exist {
		return fmt.Errorf(
			"collected metric %s %s with unregistered descriptor %s",
			metricFamily.GetName(), dtoMetric, desc,
		)
	}

	return nil
}

func (c *textFileCollector) parseTextFiles() []*dto.MetricFamily {
	error := 0.0
	metricFamilies := make([]*dto.MetricFamily, 0)
	mtimes := map[string]time.Time{}

	// Iterate over files and accumulate their metrics.
	files, err := ioutil.ReadDir(c.path)
	if err != nil && c.path != "" {
		log.Errorf("Error reading textfile collector directory %s: %s", c.path, err)
		error = 1.0
	}
	for _, f := range files {
		if !strings.HasSuffix(f.Name(), ".prom") {
			continue
		}
		path := filepath.Join(c.path, f.Name())
		file, err := os.Open(path)
		if err != nil {
			log.Errorf("Error opening %s: %v", path, err)
			error = 1.0
			continue
		}
		parsedFamilies, err := (&text.Parser{}).TextToMetricFamilies(file)
		if err != nil {
			log.Errorf("Error parsing %s: %v", path, err)
			error = 1.0
			continue
		}
		// Only set this once it has been parsed, so that
		// a failure does not appear fresh.
		mtimes[f.Name()] = f.ModTime()
		for _, mf := range parsedFamilies {
			if mf.Help == nil {
				help := fmt.Sprintf("Metric read from %s", path)
				mf.Help = &help
			}
			metricFamilies = append(metricFamilies, mf)
		}
	}

	// Export the mtimes of the successful files.
	if len(mtimes) > 0 {
		mtimeMetricFamily := dto.MetricFamily{
			Name:   proto.String("node_textfile_mtime"),
			Help:   proto.String("Unixtime mtime of textfiles successfully read."),
			Type:   dto.MetricType_GAUGE.Enum(),
			Metric: []*dto.Metric{},
		}

		// Sorting is needed for predictable output comparison in tests.
		filenames := make([]string, 0, len(mtimes))
		for filename := range mtimes {
			filenames = append(filenames, filename)
		}
		sort.Strings(filenames)

		for _, filename := range filenames {
			mtimeMetricFamily.Metric = append(mtimeMetricFamily.Metric,
				&dto.Metric{
					Label: []*dto.LabelPair{
						&dto.LabelPair{
							Name:  proto.String("file"),
							Value: proto.String(filename),
						},
					},
					Gauge: &dto.Gauge{Value: proto.Float64(float64(mtimes[filename].UnixNano()) / 1e9)},
				},
			)
		}
		metricFamilies = append(metricFamilies, &mtimeMetricFamily)
	}
	// Export if there were errors.
	metricFamilies = append(metricFamilies, &dto.MetricFamily{
		Name: proto.String("node_textfile_scrape_error"),
		Help: proto.String("1 if there was an error opening or reading a file, 0 otherwise"),
		Type: dto.MetricType_GAUGE.Enum(),
		Metric: []*dto.Metric{
			&dto.Metric{
				Gauge: &dto.Gauge{Value: &error},
			},
		},
	})

	return metricFamilies
}

// checkMetricConsistency checks if the provided Metric is consistent with the
// provided MetricFamily. It also hashed the Metric labels and the MetricFamily
// name. If the resulting hash is alread in the provided metricHashes, an error
// is returned. If not, it is added to metricHashes. The provided dimHashes maps
// MetricFamily names to their dimHash (hashed sorted label names). If dimHashes
// doesn't yet contain a hash for the provided MetricFamily, it is
// added. Otherwise, an error is returned if the existing dimHashes in not equal
// the calculated dimHash.
func checkMetricConsistency(
	metricFamily *dto.MetricFamily,
	dtoMetric *dto.Metric,
	metricHashes map[uint64]struct{},
	dimHashes map[string]uint64,
) error {
	// Type consistency with metric family.
	if metricFamily.GetType() == dto.MetricType_GAUGE && dtoMetric.Gauge == nil ||
		metricFamily.GetType() == dto.MetricType_COUNTER && dtoMetric.Counter == nil ||
		metricFamily.GetType() == dto.MetricType_SUMMARY && dtoMetric.Summary == nil ||
		metricFamily.GetType() == dto.MetricType_HISTOGRAM && dtoMetric.Histogram == nil ||
		metricFamily.GetType() == dto.MetricType_UNTYPED && dtoMetric.Untyped == nil {
		return fmt.Errorf(
			"collected metric %s %s is not a %s",
			metricFamily.GetName(), dtoMetric, metricFamily.GetType(),
		)
	}

	// Is the metric unique (i.e. no other metric with the same name and the same label values)?
	h := hashNew()
	h = hashAdd(h, metricFamily.GetName())
	h = hashAddByte(h, separatorByte)
	dh := hashNew()
	// Make sure label pairs are sorted. We depend on it for the consistency
	// check.
	sort.Sort(LabelPairSorter(dtoMetric.Label))
	for _, lp := range dtoMetric.Label {
		h = hashAdd(h, lp.GetValue())
		h = hashAddByte(h, separatorByte)
		dh = hashAdd(dh, lp.GetName())
		dh = hashAddByte(dh, separatorByte)
	}
	if _, exists := metricHashes[h]; exists {
		return fmt.Errorf(
			"collected metric %s %s was collected before with the same name and label values",
			metricFamily.GetName(), dtoMetric,
		)
	}
	if dimHash, ok := dimHashes[metricFamily.GetName()]; ok {
		if dimHash != dh {
			return fmt.Errorf(
				"collected metric %s %s has label dimensions inconsistent with previously collected metrics in the same metric family",
				metricFamily.GetName(), dtoMetric,
			)
		}
	} else {
		dimHashes[metricFamily.GetName()] = dh
	}
	metricHashes[h] = struct{}{}
	return nil
}

func extractHistogram(out Ingester, o *ProcessOptions, f *dto.MetricFamily) error {
	samples := make(model.Samples, 0, len(f.Metric))

	for _, m := range f.Metric {
		if m.Histogram == nil {
			continue
		}

		timestamp := o.Timestamp
		if m.TimestampMs != nil {
			timestamp = model.TimestampFromUnixNano(*m.TimestampMs * 1000000)
		}

		infSeen := false

		for _, q := range m.Histogram.Bucket {
			sample := &model.Sample{
				Metric:    model.Metric{},
				Value:     model.SampleValue(q.GetCumulativeCount()),
				Timestamp: timestamp,
			}
			samples = append(samples, sample)

			metric := sample.Metric
			for _, p := range m.Label {
				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			metric[model.LabelName(model.BucketLabel)] = model.LabelValue(fmt.Sprint(q.GetUpperBound()))
			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")

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

		if m.Histogram.SampleSum != nil {
			sum := &model.Sample{
				Metric:    model.Metric{},
				Value:     model.SampleValue(m.Histogram.GetSampleSum()),
				Timestamp: timestamp,
			}
			samples = append(samples, sum)

			metric := sum.Metric
			for _, p := range m.Label {
				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_sum")
		}

		if m.Histogram.SampleCount != nil {
			count := &model.Sample{
				Metric:    model.Metric{},
				Value:     model.SampleValue(m.Histogram.GetSampleCount()),
				Timestamp: timestamp,
			}
			samples = append(samples, count)

			metric := count.Metric
			for _, p := range m.Label {
				metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
			}
			metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_count")

			if !infSeen {
				infBucket := &model.Sample{
					Metric:    model.Metric{},
					Value:     count.Value,
					Timestamp: timestamp,
				}
				samples = append(samples, infBucket)

				metric := infBucket.Metric
				for _, p := range m.Label {
					metric[model.LabelName(p.GetName())] = model.LabelValue(p.GetValue())
				}
				metric[model.LabelName(model.BucketLabel)] = model.LabelValue("+Inf")
				metric[model.MetricNameLabel] = model.LabelValue(f.GetName() + "_bucket")
			}
		}
	}

	return out.Ingest(samples)
}