Merge pull request #34 from pablochacin/send-trend-sem-metric

Send trend sum as metric

Author: codebien

pkg/remotewrite/prometheus.go

@@ -76,12 +76,14 @@ func (pm *PrometheusMapping) MapTrend(ms *metricsStorage, sample metrics.Sample,
 
 	// Prometheus metric system does not support Trend so this mapping will
 	// store a counter for the number of reported values and gauges to keep
-	// track of aggregated values.
+	// track of aggregated values. Also store a sum of the values to allow
+	// the calculation of moving averages.
 	// TODO: when Prometheus implements support for sparse histograms, re-visit this implementation
 
 	s := metric.Sink.(*metrics.TrendSink)
 	aggr := map[string]float64{
 		"count": float64(s.Count),
+		"sum":   s.Sum,
 		"min":   s.Min,
 		"max":   s.Max,
 		"avg":   s.Avg,
@@ -103,6 +105,18 @@ func (pm *PrometheusMapping) MapTrend(ms *metricsStorage, sample metrics.Sample,
 				},
 			},
 		},
+		{
+			Labels: append(labels, prompb.Label{
+				Name:  "__name__",
+				Value: fmt.Sprintf("%s%s_sum", defaultMetricPrefix, sample.Metric.Name),
+			}),
+			Samples: []prompb.Sample{
+				{
+					Value:     aggr["sum"],
+					Timestamp: timestamp.FromTime(sample.Time),
+				},
+			},
+		},
 		{
 			Labels: append(labels, prompb.Label{
 				Name:  "__name__",

pkg/remotewrite/prometheus_test.go

@@ -2,9 +2,11 @@ package remotewrite
 
 import (
 	"math/rand"
+	"sort"
 	"testing"
 	"time"
 
+	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"go.k6.io/k6/metrics"
 )
@@ -108,3 +110,115 @@ func BenchmarkTrendAdd(b *testing.B) {
 		benchF[1](b, start)
 	})
 }
+
+// buildTimeSeries creates a TimSeries with the given name, value and timestamp
+func buildTimeSeries(name string, value float64, timestamp time.Time) prompb.TimeSeries {
+	return prompb.TimeSeries{
+		Labels: []prompb.Label{
+			{
+				Name:  "__name__",
+				Value: name,
+			},
+		},
+		Samples: []prompb.Sample{
+			{
+				Value:     value,
+				Timestamp: timestamp.Unix(),
+			},
+		},
+	}
+}
+
+// getTimeSeriesName returs the name of the timeseries defined in the '__name__' label
+func getTimeSeriesName(ts prompb.TimeSeries) string {
+	for _, l := range ts.Labels {
+		if l.Name == "__name__" {
+			return l.Value
+		}
+	}
+	return ""
+}
+
+// assertTimeSeriesEqual compares if two TimeSeries has the same name and value.
+// Assumes only one sample per TimeSeries
+func assertTimeSeriesEqual(t *testing.T, expected prompb.TimeSeries, actual prompb.TimeSeries) {
+	expectedName := getTimeSeriesName(expected)
+	actualName := getTimeSeriesName(actual)
+	if expectedName != actualName {
+		t.Errorf("names do not match expected: %s actual: %s", expectedName, actualName)
+	}
+
+	expectedValue := expected.Samples[0].Value
+	actualValue := actual.Samples[0].Value
+	if expectedValue != actualValue {
+		t.Errorf("values do not match expected: %f actual: %f", expectedValue, actualValue)
+	}
+}
+
+// sortTimeSeries sorts an array of TimeSeries by name
+func sortTimeSeries(ts []prompb.TimeSeries) []prompb.TimeSeries {
+	sorted := make([]prompb.TimeSeries, len(ts))
+	copy(sorted, ts)
+	sort.Slice(sorted, func(i int, j int) bool {
+		return getTimeSeriesName(sorted[i]) < getTimeSeriesName(sorted[j])
+	})
+
+	return sorted
+}
+
+// assertTimeSeriesMatch asserts if the elements of two arrays of TimeSeries match not considering order
+func assertTimeSeriesMatch(t *testing.T, expected []prompb.TimeSeries, actual []prompb.TimeSeries) {
+	if len(expected) != len(actual) {
+		t.Errorf("timeseries length does not match. expected %d actual: %d", len(expected), len(actual))
+	}
+
+	//sort arrays
+	se := sortTimeSeries(expected)
+	sa := sortTimeSeries(actual)
+
+	//return false if any element does not match
+	for i := 0; i < len(se); i++ {
+		assertTimeSeriesEqual(t, se[i], sa[i])
+	}
+
+}
+
+func TestMapTrend(t *testing.T) {
+	t.Parallel()
+
+	now := time.Now()
+	testCases := []struct {
+		storage  *metricsStorage
+		sample   metrics.Sample
+		labels   []prompb.Label
+		expected []prompb.TimeSeries
+	}{
+		{
+			storage: newMetricsStorage(),
+			sample: metrics.Sample{
+				Metric: &metrics.Metric{
+					Name: "test",
+					Type: metrics.Trend,
+				},
+				Value: 1.0,
+				Time:  now,
+			},
+			expected: []prompb.TimeSeries{
+				buildTimeSeries("k6_test_count", 1.0, now),
+				buildTimeSeries("k6_test_sum", 1.0, now),
+				buildTimeSeries("k6_test_min", 1.0, now),
+				buildTimeSeries("k6_test_max", 1.0, now),
+				buildTimeSeries("k6_test_avg", 1.0, now),
+				buildTimeSeries("k6_test_med", 1.0, now),
+				buildTimeSeries("k6_test_p90", 1.0, now),
+				buildTimeSeries("k6_test_p95", 1.0, now),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		m := &PrometheusMapping{}
+		ts := m.MapTrend(tc.storage, tc.sample, tc.labels)
+		assertTimeSeriesMatch(t, tc.expected, ts)
+	}
+}