diff --git a/exporter/prometheusremotewriteexporter/exporter.go b/exporter/prometheusremotewriteexporter/exporter.go index 3983801e865..8f9c86193dc 100644 --- a/exporter/prometheusremotewriteexporter/exporter.go +++ b/exporter/prometheusremotewriteexporter/exporter.go @@ -123,7 +123,12 @@ func (prwe *PrwExporter) PushMetrics(ctx context.Context, md pdata.Metrics) (int case *otlp.Metric_DoubleHistogram, *otlp.Metric_IntHistogram: if err := prwe.handleHistogramMetric(tsMap, metric); err != nil { dropped++ - errs = append(errs, err) + errs = append(errs, consumererror.Permanent(err)) + } + case *otlp.Metric_DoubleSummary: + if err := prwe.handleSummaryMetric(tsMap, metric); err != nil { + dropped++ + errs = append(errs, consumererror.Permanent(err)) } default: dropped++ @@ -209,6 +214,19 @@ func (prwe *PrwExporter) handleHistogramMetric(tsMap map[string]*prompb.TimeSeri return nil } +// handleSummaryMetric processes data points in a single OTLP summary metric by mapping the sum, count and each +// quantile of every data point as a Sample, and adding each Sample to its corresponding TimeSeries. +// tsMap and metric cannot be nil. +func (prwe *PrwExporter) handleSummaryMetric(tsMap map[string]*prompb.TimeSeries, metric *otlp.Metric) error { + if metric.GetDoubleSummary().GetDataPoints() == nil { + return fmt.Errorf("nil data point. %s is dropped", metric.GetName()) + } + for _, pt := range metric.GetDoubleSummary().GetDataPoints() { + addSingleDoubleSummaryDataPoint(pt, metric, prwe.namespace, tsMap, prwe.externalLabels) + } + return nil +} + // export sends a Snappy-compressed WriteRequest containing TimeSeries to a remote write endpoint in order func (prwe *PrwExporter) export(ctx context.Context, tsMap map[string]*prompb.TimeSeries) error { // Calls the helper function to convert the TsMap to the desired format diff --git a/exporter/prometheusremotewriteexporter/exporter_test.go b/exporter/prometheusremotewriteexporter/exporter_test.go index 9c31acfcb8b..8a8c567cd5b 100644 --- a/exporter/prometheusremotewriteexporter/exporter_test.go +++ b/exporter/prometheusremotewriteexporter/exporter_test.go @@ -329,6 +329,20 @@ func Test_PushMetrics(t *testing.T) { } doubleHistogramBatch := pdata.MetricsFromOtlp(doubleHistogramMetric) + doubleSummaryMetric := []*otlp.ResourceMetrics{ + { + InstrumentationLibraryMetrics: []*otlp.InstrumentationLibraryMetrics{ + { + Metrics: []*otlp.Metric{ + validMetrics1[validDoubleSummary], + validMetrics2[validDoubleSummary], + }, + }, + }, + }, + } + doubleSummaryBatch := pdata.MetricsFromOtlp(doubleSummaryMetric) + // len(BucketCount) > len(ExplicitBounds) unmatchedBoundBucketIntHistMetric := []*otlp.ResourceMetrics{ { @@ -435,6 +449,19 @@ func Test_PushMetrics(t *testing.T) { } nilDataPointDoubleHistogramBatch := pdata.MetricsFromOtlp(nilDataPointDoubleHistogramMetric) + nilDataPointDoubleSummaryMetric := []*otlp.ResourceMetrics{ + { + InstrumentationLibraryMetrics: []*otlp.InstrumentationLibraryMetrics{ + { + Metrics: []*otlp.Metric{ + errorMetrics[nilDataPointDoubleSummary], + }, + }, + }, + }, + } + nilDataPointDoubleSummaryBatch := pdata.MetricsFromOtlp(nilDataPointDoubleSummaryMetric) + checkFunc := func(t *testing.T, r *http.Request, expected int) { body, err := ioutil.ReadAll(r.Body) if err != nil { @@ -553,6 +580,15 @@ func Test_PushMetrics(t *testing.T) { 0, false, }, + { + "doubleSummary_case", + &doubleSummaryBatch, + checkFunc, + 10, + http.StatusAccepted, + 0, + false, + }, { "unmatchedBoundBucketIntHist_case", &unmatchedBoundBucketIntHistBatch, @@ -634,6 +670,15 @@ func Test_PushMetrics(t *testing.T) { nilDataPointIntHistogramBatch.MetricCount(), true, }, + { + "nilDataPointDoubleSummary_case", + &nilDataPointDoubleSummaryBatch, + checkFunc, + 0, + http.StatusAccepted, + nilDataPointDoubleSummaryBatch.MetricCount(), + true, + }, } for _, tt := range tests { diff --git a/exporter/prometheusremotewriteexporter/helper.go b/exporter/prometheusremotewriteexporter/helper.go index 00559aa9bdf..ab704749bfe 100644 --- a/exporter/prometheusremotewriteexporter/helper.go +++ b/exporter/prometheusremotewriteexporter/helper.go @@ -31,15 +31,16 @@ import ( ) const ( - nameStr = "__name__" - sumStr = "_sum" - countStr = "_count" - bucketStr = "_bucket" - leStr = "le" - pInfStr = "+Inf" - totalStr = "total" - delimeter = "_" - keyStr = "key" + nameStr = "__name__" + sumStr = "_sum" + countStr = "_count" + bucketStr = "_bucket" + leStr = "le" + quantileStr = "quantile" + pInfStr = "+Inf" + totalStr = "total" + delimeter = "_" + keyStr = "key" ) // ByLabelName enables the usage of sort.Sort() with a slice of labels @@ -72,6 +73,8 @@ func validateMetrics(metric *otlp.Metric) bool { case *otlp.Metric_IntHistogram: return metric.GetIntHistogram() != nil && metric.GetIntHistogram().GetAggregationTemporality() == otlp.AggregationTemporality_AGGREGATION_TEMPORALITY_CUMULATIVE + case *otlp.Metric_DoubleSummary: + return metric.GetDoubleSummary() != nil } return false } @@ -425,3 +428,41 @@ func addSingleDoubleHistogramDataPoint(pt *otlp.DoubleHistogramDataPoint, metric infLabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+bucketStr, leStr, pInfStr) addSample(tsMap, infBucket, infLabels, metric) } + +// addSingleDoubleSummaryDataPoint converts pt to len(QuantileValues) + 2 samples. +func addSingleDoubleSummaryDataPoint(pt *otlp.DoubleSummaryDataPoint, metric *otlp.Metric, namespace string, + tsMap map[string]*prompb.TimeSeries, externalLabels map[string]string) { + if pt == nil { + return + } + time := convertTimeStamp(pt.TimeUnixNano) + // sum and count of the summary should append suffix to baseName + baseName := getPromMetricName(metric, namespace) + // treat sum as a sample in an individual TimeSeries + sum := &prompb.Sample{ + Value: pt.GetSum(), + Timestamp: time, + } + + sumlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+sumStr) + addSample(tsMap, sum, sumlabels, metric) + + // treat count as a sample in an individual TimeSeries + count := &prompb.Sample{ + Value: float64(pt.GetCount()), + Timestamp: time, + } + countlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName+countStr) + addSample(tsMap, count, countlabels, metric) + + // process each percentile/quantile + for _, qt := range pt.GetQuantileValues() { + quantile := &prompb.Sample{ + Value: qt.Value, + Timestamp: time, + } + percentileStr := strconv.FormatFloat(qt.GetQuantile(), 'f', -1, 64) + qtlabels := createLabelSet(pt.GetLabels(), externalLabels, nameStr, baseName, quantileStr, percentileStr) + addSample(tsMap, quantile, qtlabels, metric) + } +} diff --git a/exporter/prometheusremotewriteexporter/testutil_test.go b/exporter/prometheusremotewriteexporter/testutil_test.go index dc054aafbdd..1a603f3c223 100644 --- a/exporter/prometheusremotewriteexporter/testutil_test.go +++ b/exporter/prometheusremotewriteexporter/testutil_test.go @@ -79,12 +79,17 @@ var ( bounds = []float64{0.1, 0.5, 0.99} buckets = []uint64{1, 2, 3} + quantileBounds = []float64{0.15, 0.9, 0.99} + quantileValues = []float64{7, 8, 9} + quantiles = getQuantiles(quantileBounds, quantileValues) + validIntGauge = "valid_IntGauge" validDoubleGauge = "valid_DoubleGauge" validIntSum = "valid_IntSum" validDoubleSum = "valid_DoubleSum" validIntHistogram = "valid_IntHistogram" validDoubleHistogram = "valid_DoubleHistogram" + validDoubleSummary = "valid_DoubleSummary" validIntGaugeDirty = "*valid_IntGauge$" @@ -163,6 +168,17 @@ var ( }, }, }, + validDoubleSummary: { + Name: validDoubleSummary, + Data: &otlp.Metric_DoubleSummary{ + DoubleSummary: &otlp.DoubleSummary{ + DataPoints: []*otlp.DoubleSummaryDataPoint{ + getDoubleSummaryDataPoint(lbs1, time1, floatVal1, uint64(intVal1), quantiles), + nil, + }, + }, + }, + }, } validMetrics2 = map[string]*otlp.Metric{ validIntGauge: { @@ -229,6 +245,17 @@ var ( }, }, }, + validDoubleSummary: { + Name: validDoubleSummary, + Data: &otlp.Metric_DoubleSummary{ + DoubleSummary: &otlp.DoubleSummary{ + DataPoints: []*otlp.DoubleSummaryDataPoint{ + getDoubleSummaryDataPoint(lbs2, time2, floatVal2, uint64(intVal2), quantiles), + nil, + }, + }, + }, + }, validIntGaugeDirty: { Name: validIntGaugeDirty, Data: &otlp.Metric_IntGauge{ @@ -280,6 +307,7 @@ var ( notMatchDoubleSum = "notMatchDoubleSum" notMatchIntHistogram = "notMatchIntHistogram" notMatchDoubleHistogram = "notMatchDoubleHistogram" + notMatchDoubleSummary = "notMatchDoubleSummary" // Category 2: invalid type and temporality combination invalidIntSum = "invalidIntSum" @@ -294,6 +322,7 @@ var ( nilDataPointDoubleSum = "nilDataPointDoubleSum" nilDataPointIntHistogram = "nilDataPointIntHistogram" nilDataPointDoubleHistogram = "nilDataPointDoubleHistogram" + nilDataPointDoubleSummary = "nilDataPointDoubleSummary" // different metrics that will not pass validate metrics invalidMetrics = map[string]*otlp.Metric{ @@ -325,6 +354,10 @@ var ( Name: notMatchDoubleHistogram, Data: &otlp.Metric_DoubleHistogram{}, }, + notMatchDoubleSummary: { + Name: notMatchDoubleSummary, + Data: &otlp.Metric_DoubleSummary{}, + }, invalidIntSum: { Name: invalidIntSum, Data: &otlp.Metric_IntSum{ @@ -410,6 +443,14 @@ var ( }, }, }, + nilDataPointDoubleSummary: { + Name: nilDataPointDoubleSummary, + Data: &otlp.Metric_DoubleSummary{ + DoubleSummary: &otlp.DoubleSummary{ + DataPoints: nil, + }, + }, + }, } ) @@ -470,6 +511,17 @@ func getDoubleHistogramDataPoint(labels []commonpb.StringKeyValue, ts uint64, su } } +func getDoubleSummaryDataPoint(labels []commonpb.StringKeyValue, ts uint64, sum float64, count uint64, + quantiles []*otlp.DoubleSummaryDataPoint_ValueAtQuantile) *otlp.DoubleSummaryDataPoint { + return &otlp.DoubleSummaryDataPoint{ + Labels: labels, + TimeUnixNano: ts, + Count: count, + Sum: sum, + QuantileValues: quantiles, + } +} + // Prometheus TimeSeries func getPromLabels(lbs ...string) []prompb.Label { pbLbs := prompb.Labels{ @@ -501,3 +553,14 @@ func getTimeSeries(labels []prompb.Label, samples ...prompb.Sample) *prompb.Time Samples: samples, } } + +func getQuantiles(bounds []float64, values []float64) []*otlp.DoubleSummaryDataPoint_ValueAtQuantile { + quantiles := make([]*otlp.DoubleSummaryDataPoint_ValueAtQuantile, len(bounds)) + for i := 0; i < len(bounds); i++ { + quantiles[i] = &otlp.DoubleSummaryDataPoint_ValueAtQuantile{ + Quantile: bounds[i], + Value: values[i], + } + } + return quantiles +}