Merge pull request #23 from grafana/local-synk

Not sink the global metrics

Author: codebien

pkg/remotewrite/metrics.go

@@ -13,37 +13,62 @@ import (
 
 // metricsStorage is an in-memory gather point for metrics
 type metricsStorage struct {
-	m map[string]metrics.Sample
+	m map[string]*metrics.Metric
 }
 
 func newMetricsStorage() *metricsStorage {
 	return &metricsStorage{
-		m: make(map[string]metrics.Sample),
+		m: make(map[string]*metrics.Metric),
 	}
 }
 
 // update modifies metricsStorage and returns updated sample
 // so that the stored metric and the returned metric hold the same value
-func (ms *metricsStorage) update(sample metrics.Sample, add func(current, s metrics.Sample) metrics.Sample) metrics.Sample {
-	if current, ok := ms.m[sample.Metric.Name]; ok {
-		if add == nil {
-			current.Metric.Sink.Add(sample)
-		} else {
-			current = add(current, sample)
+func (ms *metricsStorage) update(sample metrics.Sample, add func(*metrics.Metric, metrics.Sample)) *metrics.Metric {
+	m, ok := ms.m[sample.Metric.Name]
+	if !ok {
+		var sink metrics.Sink
+		switch sample.Metric.Type {
+		case metrics.Counter:
+			sink = &metrics.CounterSink{}
+		case metrics.Gauge:
+			sink = &metrics.GaugeSink{}
+		case metrics.Trend:
+			sink = &metrics.TrendSink{}
+		case metrics.Rate:
+			sink = &metrics.RateSink{}
+		default:
+			panic("the Metric Type is not supported")
 		}
-		current.Time = sample.Time // to avoid duplicates in timestamps
-		// Sometimes remote write endpoint throws an error about duplicates even if the values
-		// sent were different. By current observations, this is a hard to repeat case and
-		// potentially a bug.
-		// Related: https://github.com/prometheus/prometheus/issues/9210
-
-		ms.m[current.Metric.Name] = current
-		return current
+
+		m = &metrics.Metric{
+			Name:     sample.Metric.Name,
+			Type:     sample.Metric.Type,
+			Contains: sample.Metric.Contains,
+			Sink:     sink,
+		}
+
+		ms.m[m.Name] = m
+	}
+
+	// TODO: https://github.com/grafana/xk6-output-prometheus-remote/issues/11
+	//
+	// Sometimes remote write endpoint throws an error about duplicates even if the values
+	// sent were different. By current observations, this is a hard to repeat case and
+	// potentially a bug.
+	// Related: https://github.com/prometheus/prometheus/issues/9210
+
+	// TODO: Trend is the unique type that benefits from this logic.
+	// so this logic can be removed just creating
+	// a new implementation in this extension
+	// for TrendSink and its Add method.
+	if add == nil {
+		m.Sink.Add(sample)
 	} else {
-		sample.Metric.Sink.Add(sample)
-		ms.m[sample.Metric.Name] = sample
-		return sample
+		add(m, sample)
 	}
+
+	return m
 }
 
 // transform k6 sample into TimeSeries for remote-write

pkg/remotewrite/prometheus.go

@@ -13,8 +13,8 @@ import (
 type PrometheusMapping struct{}
 
 func (pm *PrometheusMapping) MapCounter(ms *metricsStorage, sample metrics.Sample, labels []prompb.Label) []prompb.TimeSeries {
-	sample = ms.update(sample, nil)
-	aggr := sample.Metric.Sink.Format(0)
+	metric := ms.update(sample, nil)
+	aggr := metric.Sink.Format(0)
 
 	return []prompb.TimeSeries{
 		{
@@ -33,9 +33,6 @@ func (pm *PrometheusMapping) MapCounter(ms *metricsStorage, sample metrics.Sampl
 }
 
 func (pm *PrometheusMapping) MapGauge(ms *metricsStorage, sample metrics.Sample, labels []prompb.Label) []prompb.TimeSeries {
-	sample = ms.update(sample, nil)
-	aggr := sample.Metric.Sink.Format(0)
-
 	return []prompb.TimeSeries{
 		{
 			Labels: append(labels, prompb.Label{
@@ -44,7 +41,9 @@ func (pm *PrometheusMapping) MapGauge(ms *metricsStorage, sample metrics.Sample,
 			}),
 			Samples: []prompb.Sample{
 				{
-					Value:     aggr["value"],
+					// Gauge is just the latest value
+					// so we can skip the sink using directly the value from the sample.
+					Value:     sample.Value,
 					Timestamp: timestamp.FromTime(sample.Time),
 				},
 			},
@@ -53,8 +52,8 @@ func (pm *PrometheusMapping) MapGauge(ms *metricsStorage, sample metrics.Sample,
 }
 
 func (pm *PrometheusMapping) MapRate(ms *metricsStorage, sample metrics.Sample, labels []prompb.Label) []prompb.TimeSeries {
-	sample = ms.update(sample, nil)
-	aggr := sample.Metric.Sink.Format(0)
+	metric := ms.update(sample, nil)
+	aggr := metric.Sink.Format(0)
 
 	return []prompb.TimeSeries{
 		{
@@ -73,9 +72,13 @@ func (pm *PrometheusMapping) MapRate(ms *metricsStorage, sample metrics.Sample,
 }
 
 func (pm *PrometheusMapping) MapTrend(ms *metricsStorage, sample metrics.Sample, labels []prompb.Label) []prompb.TimeSeries {
-	sample = ms.update(sample, trendAdd)
+	metric := ms.update(sample, trendAdd)
+
+	// Prometheus metric system does not support Trend so this mapping will store gauges
+	// to keep track of key values.
+	// TODO: when Prometheus implements support for sparse histograms, re-visit this implementation
 
-	s := sample.Metric.Sink.(*metrics.TrendSink)
+	s := metric.Sink.(*metrics.TrendSink)
 	aggr := map[string]float64{
 		"min":   s.Min,
 		"max":   s.Max,
@@ -85,10 +88,6 @@ func (pm *PrometheusMapping) MapTrend(ms *metricsStorage, sample metrics.Sample,
 		"p(95)": p(s, 0.95),
 	}
 
-	// Prometheus metric system does not support Trend so this mapping will store gauges
-	// to keep track of key values.
-	// TODO: when Prometheus implements support for sparse histograms, re-visit this implementation
-
 	return []prompb.TimeSeries{
 		{
 			Labels: append(labels, prompb.Label{
@@ -169,8 +168,8 @@ func (pm *PrometheusMapping) MapTrend(ms *metricsStorage, sample metrics.Sample,
 // and are a partial copy-paste from k6/metrics.
 // TODO: re-write & refactor this once metrics refactoring progresses in k6.
 
-func trendAdd(current, s metrics.Sample) metrics.Sample {
-	t := current.Metric.Sink.(*metrics.TrendSink)
+func trendAdd(current *metrics.Metric, s metrics.Sample) {
+	t := current.Sink.(*metrics.TrendSink)
 
 	// insert into sorted array instead of sorting anew on each addition
 	index := sort.Search(len(t.Values), func(i int) bool {
@@ -197,8 +196,7 @@ func trendAdd(current, s metrics.Sample) metrics.Sample {
 		t.Med = t.Values[t.Count/2]
 	}
 
-	current.Metric.Sink = t
-	return current
+	current.Sink = t
 }
 
 func p(t *metrics.TrendSink, pct float64) float64 {

pkg/remotewrite/prometheus_test.go

@@ -1,11 +1,10 @@
 package remotewrite
 
 import (
+	"math/rand"
 	"testing"
 	"time"
 
-	"math/rand"
-
 	"github.com/stretchr/testify/assert"
 	"go.k6.io/k6/metrics"
 )
@@ -15,81 +14,97 @@ func TestTrendAdd(t *testing.T) {
 	t.Parallel()
 
 	testCases := []struct {
-		current, s metrics.Sample
+		current  *metrics.Metric
+		s        metrics.Sample
+		expected metrics.TrendSink
 	}{
 		{
-			current: metrics.Sample{Metric: &metrics.Metric{
+			current: &metrics.Metric{
 				Sink: &metrics.TrendSink{},
-			}},
+			},
 			s: metrics.Sample{Value: 2},
+			expected: metrics.TrendSink{
+				Values: []float64{2},
+				Count:  1,
+				Min:    2,
+				Max:    2,
+				Sum:    2,
+				Avg:    2,
+				Med:    2,
+			},
 		},
 		{
-			current: metrics.Sample{Metric: &metrics.Metric{
+			current: &metrics.Metric{
 				Sink: &metrics.TrendSink{
 					Values: []float64{8, 3, 1, 7, 4, 2},
 					Count:  6,
-					Min:    1, Max: 8,
-					Sum: 25, Avg: (8 + 3 + 1 + 7 + 4 + 2) / 6,
-					Med: (3 + 4) / 2,
+					Min:    1,
+					Max:    8,
+					Sum:    25,
 				},
-			}},
+			},
 			s: metrics.Sample{Value: 12.3},
+			expected: metrics.TrendSink{
+				Values: []float64{8, 3, 1, 7, 4, 2, 12.3},
+				Count:  7,
+				Min:    1,
+				Max:    12.3,
+				Sum:    37.3,
+				Avg:    37.3 / 7,
+				Med:    7,
+			},
 		},
 	}
 
 	for _, testCase := range testCases {
 		// trendAdd should result in the same values as Sink.Add
 
-		s := trendAdd(testCase.current, testCase.s)
-		sink := s.Metric.Sink.(*metrics.TrendSink)
+		trendAdd(testCase.current, testCase.s)
+		sink := testCase.current.Sink.(*metrics.TrendSink)
 
-		testCase.current.Metric.Sink.Add(testCase.s)
-		expected := testCase.current.Metric.Sink.(*metrics.TrendSink)
-
-		assert.Equal(t, expected.Count, sink.Count)
-		assert.Equal(t, expected.Min, sink.Min)
-		assert.Equal(t, expected.Max, sink.Max)
-		assert.Equal(t, expected.Sum, sink.Sum)
-		assert.Equal(t, expected.Avg, sink.Avg)
-		assert.EqualValues(t, expected.Values, sink.Values)
+		assert.Equal(t, testCase.expected.Count, sink.Count)
+		assert.Equal(t, testCase.expected.Min, sink.Min)
+		assert.Equal(t, testCase.expected.Max, sink.Max)
+		assert.Equal(t, testCase.expected.Sum, sink.Sum)
+		assert.Equal(t, testCase.expected.Avg, sink.Avg)
+		assert.Equal(t, testCase.expected.Med, sink.Med)
+		assert.Equal(t, testCase.expected.Values, sink.Values)
 	}
 }
 
 func BenchmarkTrendAdd(b *testing.B) {
-	benchF := []func(b *testing.B, start metrics.Sample){
-		func(b *testing.B, s metrics.Sample) {
+	benchF := []func(b *testing.B, start metrics.Metric){
+		func(b *testing.B, m metrics.Metric) {
 			b.ResetTimer()
 			rand.Seed(time.Now().Unix())
 
 			for i := 0; i < b.N; i++ {
-				s = trendAdd(s, metrics.Sample{Value: rand.Float64() * 1000})
-				sink := s.Metric.Sink.(*metrics.TrendSink)
+				trendAdd(&m, metrics.Sample{Value: rand.Float64() * 1000})
+				sink := m.Sink.(*metrics.TrendSink)
 				p(sink, 0.90)
 				p(sink, 0.95)
 			}
 		},
-		func(b *testing.B, start metrics.Sample) {
+		func(b *testing.B, start metrics.Metric) {
 			b.ResetTimer()
 			rand.Seed(time.Now().Unix())
 
 			for i := 0; i < b.N; i++ {
-				start.Metric.Sink.Add(metrics.Sample{Value: rand.Float64() * 1000})
-				start.Metric.Sink.Format(0)
+				start.Sink.Add(metrics.Sample{Value: rand.Float64() * 1000})
+				start.Sink.Format(0)
 			}
 		},
 	}
 
-	s := metrics.Sample{
-		Metric: &metrics.Metric{
-			Type: metrics.Trend,
-			Sink: &metrics.TrendSink{},
-		},
+	start := metrics.Metric{
+		Type: metrics.Trend,
+		Sink: &metrics.TrendSink{},
 	}
 
 	b.Run("trendAdd", func(b *testing.B) {
-		benchF[0](b, s)
+		benchF[0](b, start)
 	})
 	b.Run("TrendSink.Add", func(b *testing.B) {
-		benchF[1](b, s)
+		benchF[1](b, start)
 	})
 }