Not sink the global metric

The global metric is controlled by the k6's ingester output, so the sink operation on the global metric in the sample is already executed. Use a local metric in the metricsStorage for executing a local and dedicated sink in the extension. It fixes the data race where both the extension and the k6 ingester calls Add and sink for the same metrics.

Author: codebien

pkg/remotewrite/metrics.go

@@ -13,37 +13,66 @@ 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: this is just avoiding duplicates with the previous
+// Implement a better and complete solution
+// maybe discard any timestamp < latest?
+//
+//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
+
+// 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 {