exporter/prometheusexporter: support Summary

exporter/prometheusexporter/accumulator.go

@@ -94,11 +94,43 @@ func (a *lastValueAccumulator) addMetric(metric pdata.Metric, il pdata.Instrumen
 		return a.accumulateIntHistogram(metric, il, now)
 	case pdata.MetricDataTypeHistogram:
 		return a.accumulateDoubleHistogram(metric, il, now)
+	case pdata.MetricDataTypeSummary:
+		return a.accumulateSummary(metric, il, now)
+	default:
+		a.logger.With(
+			zap.String("data_type", string(metric.DataType())),
+			zap.String("metric_name", metric.Name()),
+		).Error("failed to translate metric")
 	}
 
 	return 0
 }
 
+func (a *lastValueAccumulator) accumulateSummary(metric pdata.Metric, il pdata.InstrumentationLibrary, now time.Time) (n int) {
+	dps := metric.Summary().DataPoints()
+	for i := 0; i < dps.Len(); i++ {
+		ip := dps.At(i)
+
+		signature := timeseriesSignature(il.Name(), metric, ip.LabelsMap())
+
+		v, ok := a.registeredMetrics.Load(signature)
+		stalePoint := ok &&
+			ip.Timestamp().AsTime().Before(v.(*accumulatedValue).value.Summary().DataPoints().At(0).Timestamp().AsTime())
+
+		if stalePoint {
+			// Only keep this datapoint if it has a later timestamp.
+			continue
+		}
+
+		mm := createMetric(metric)
+		mm.Summary().DataPoints().Append(ip)
+		a.registeredMetrics.Store(signature, &accumulatedValue{value: mm, instrumentationLibrary: il, updated: now})
+		n++
+	}
+
+	return n
+}
+
 func (a *lastValueAccumulator) accumulateIntGauge(metric pdata.Metric, il pdata.InstrumentationLibrary, now time.Time) (n int) {
 	dps := metric.IntGauge().DataPoints()
 	for i := 0; i < dps.Len(); i++ {

exporter/prometheusexporter/collector.go

@@ -70,6 +70,8 @@ func (c *collector) convertMetric(metric pdata.Metric) (prometheus.Metric, error
 		return c.convertIntHistogram(metric)
 	case pdata.MetricDataTypeHistogram:
 		return c.convertDoubleHistogram(metric)
+	case pdata.MetricDataTypeSummary:
+		return c.convertSummary(metric)
 	}
 
 	return nil, errUnknownMetricType
@@ -207,6 +209,30 @@ func (c *collector) convertIntHistogram(metric pdata.Metric) (prometheus.Metric,
 	return m, nil
 }
 
+func (c *collector) convertSummary(metric pdata.Metric) (prometheus.Metric, error) {
+	// TODO: In the off chance that we have multiple points
+	// within the same metric, how should we handle them?
+	point := metric.Summary().DataPoints().At(0)
+
+	quantiles := make(map[float64]float64)
+	qv := point.QuantileValues()
+	for j := 0; j < qv.Len(); j++ {
+		qvj := qv.At(j)
+		// There should be EXACTLY one quantile value lest it is an invalid exposition.
+		quantiles[qvj.Quantile()] = qvj.Value()
+	}
+
+	desc, labelValues := c.getMetricMetadata(metric, point.LabelsMap())
+	m, err := prometheus.NewConstSummary(desc, point.Count(), point.Sum(), quantiles, labelValues...)
+	if err != nil {
+		return nil, err
+	}
+	if c.sendTimestamps {
+		return prometheus.NewMetricWithTimestamp(point.Timestamp().AsTime(), m), nil
+	}
+	return m, nil
+}
+
 func (c *collector) convertDoubleHistogram(metric pdata.Metric) (prometheus.Metric, error) {
 	ip := metric.Histogram().DataPoints().At(0)
 	desc, labels := c.getMetricMetadata(metric, ip.LabelsMap())

exporter/prometheusexporter/collector_test.go

@@ -349,10 +349,12 @@ func TestCollectMetrics(t *testing.T) {
 						require.Equal(t, tt.value, *pbMetric.Counter.Value)
 						require.Nil(t, pbMetric.Gauge)
 						require.Nil(t, pbMetric.Histogram)
+						require.Nil(t, pbMetric.Summary)
 					case prometheus.GaugeValue:
 						require.Equal(t, tt.value, *pbMetric.Gauge.Value)
 						require.Nil(t, pbMetric.Counter)
 						require.Nil(t, pbMetric.Histogram)
+						require.Nil(t, pbMetric.Summary)
 					}
 				}
 				require.Equal(t, 1, j)
@@ -488,3 +490,115 @@ func TestAccumulateHistograms(t *testing.T) {
 		}
 	}
 }
+
+func TestAccumulateSummary(t *testing.T) {
+	quantileValue := func(pN, value float64) pdata.ValueAtQuantile {
+		vqpN := pdata.NewValueAtQuantile()
+		vqpN.SetQuantile(pN)
+		vqpN.SetValue(value)
+		return vqpN
+	}
+	quantilesFromMap := func(qf map[float64]float64) (qL []*io_prometheus_client.Quantile) {
+		f64Ptr := func(v float64) *float64 { return &v }
+		for quantile, value := range qf {
+			qL = append(qL, &io_prometheus_client.Quantile{
+				Quantile: f64Ptr(quantile), Value: f64Ptr(value),
+			})
+		}
+		return qL
+	}
+	tests := []struct {
+		name          string
+		metric        func(time.Time) pdata.Metric
+		wantSum       float64
+		wantCount     uint64
+		wantQuantiles []*io_prometheus_client.Quantile
+	}{
+		{
+			name:      "Summary with single point",
+			wantSum:   0.012,
+			wantCount: 10,
+			wantQuantiles: quantilesFromMap(map[float64]float64{
+				0.50: 190,
+				0.99: 817,
+			}),
+			metric: func(ts time.Time) (metric pdata.Metric) {
+				sp := pdata.NewSummaryDataPoint()
+				sp.SetCount(10)
+				sp.SetSum(0.012)
+				sp.SetCount(10)
+				sp.LabelsMap().Insert("label_1", "1")
+				sp.LabelsMap().Insert("label_2", "2")
+				sp.SetTimestamp(pdata.TimestampFromTime(ts))
+
+				sp.QuantileValues().Append(quantileValue(0.50, 190))
+				sp.QuantileValues().Append(quantileValue(0.99, 817))
+
+				metric = pdata.NewMetric()
+				metric.SetName("test_metric")
+				metric.SetDataType(pdata.MetricDataTypeSummary)
+				metric.Summary().DataPoints().Append(sp)
+				metric.SetDescription("test description")
+
+				return
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		for _, sendTimestamp := range []bool{true, false} {
+			name := tt.name
+			if sendTimestamp {
+				name += "/WithTimestamp"
+			}
+			t.Run(name, func(t *testing.T) {
+				ts := time.Now()
+				metric := tt.metric(ts)
+				c := collector{
+					accumulator: &mockAccumulator{
+						[]pdata.Metric{metric},
+					},
+					sendTimestamps: sendTimestamp,
+					logger:         zap.NewNop(),
+				}
+
+				ch := make(chan prometheus.Metric, 1)
+				go func() {
+					c.Collect(ch)
+					close(ch)
+				}()
+
+				n := 0
+				for m := range ch {
+					n++
+					require.Contains(t, m.Desc().String(), "fqName: \"test_metric\"")
+					require.Contains(t, m.Desc().String(), "variableLabels: [label_1 label_2]")
+
+					pbMetric := io_prometheus_client.Metric{}
+					m.Write(&pbMetric)
+
+					labelsKeys := map[string]string{"label_1": "1", "label_2": "2"}
+					for _, l := range pbMetric.Label {
+						require.Equal(t, labelsKeys[*l.Name], *l.Value)
+					}
+
+					if sendTimestamp {
+						require.Equal(t, ts.UnixNano()/1e6, *(pbMetric.TimestampMs))
+					} else {
+						require.Nil(t, pbMetric.TimestampMs)
+					}
+
+					require.Nil(t, pbMetric.Gauge)
+					require.Nil(t, pbMetric.Counter)
+					require.Nil(t, pbMetric.Histogram)
+
+					s := *pbMetric.Summary
+					require.Equal(t, tt.wantCount, *s.SampleCount)
+					require.Equal(t, tt.wantSum, *s.SampleSum)
+					require.Equal(t, tt.wantQuantiles, s.Quantile)
+				}
+				require.Equal(t, 1, n)
+			})
+		}
+	}
+}