1ms to the stale marker to guarantee the accuracy

Author: codebien

pkg/remote/client_test.go

@@ -10,8 +10,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/snappy"
 	"github.com/grafana/xk6-output-prometheus-remote/pkg/stale"
+
+	"github.com/golang/snappy"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	prompb "go.buf.build/grpc/go/prometheus/prometheus"

pkg/remotewrite/remotewrite.go

@@ -25,6 +25,7 @@ type Output struct {
 
 	config             Config
 	logger             logrus.FieldLogger
+	now                func() time.Time
 	periodicFlusher    *output.PeriodicFlusher
 	tsdb               map[metrics.TimeSeries]*seriesWithMeasure
 	trendStatsResolver map[string]func(*metrics.TrendSink) float64
@@ -55,6 +56,10 @@ func New(params output.Params) (*Output, error) {
 	o := &Output{
 		client: wc,
 		config: config,
+		// TODO: consider to do this function millisecond-based
+		// so we don't need to truncate all the time we invoke it.
+		// Before we should analyze if in some cases is it useful to have it in ns.
+		now:    time.Now,
 		logger: logger,
 		tsdb:   make(map[metrics.TimeSeries]*seriesWithMeasure),
 	}
@@ -93,8 +98,7 @@ func (o *Output) Stop() error {
 	if !o.config.StaleMarkers.Bool {
 		return nil
 	}
-
-	staleMarkers := o.staleMarkers(time.Now())
+	staleMarkers := o.staleMarkers()
 	if len(staleMarkers) < 1 {
 		o.logger.Debug("No time series to mark as stale")
 		return nil
@@ -109,10 +113,18 @@ func (o *Output) Stop() error {
 }
 
 // staleMarkers maps all the seen time series with a stale marker.
-func (o *Output) staleMarkers(t time.Time) []*prompb.TimeSeries {
-	timestamp := t.UnixMilli()
-	staleMarkers := make([]*prompb.TimeSeries, 0, len(o.tsdb))
+func (o *Output) staleMarkers() []*prompb.TimeSeries {
+	// Add 1ms so in the extreme case that the time frame
+	// between the last and the next flush operation is under-millisecond,
+	// we can avoid the sample being seen as a duplicate,
+	// if we force it in the future.
+	// It is essential because if it overlaps, the remote write discards the last sample,
+	// so the stale marker and the metric will remain active for the next 5 min
+	// as the default logic without stale markers.
+	timestamp := o.now().
+		Truncate(time.Millisecond).Add(1 * time.Millisecond).UnixMilli()
 
+	staleMarkers := make([]*prompb.TimeSeries, 0, len(o.tsdb))
 	for _, swm := range o.tsdb {
 		series := swm.MapPrompb()
 		// series' length is expected to be equal to 1 for most of the cases

pkg/remotewrite/remotewrite_test.go

@@ -312,7 +312,11 @@ func TestOutputStaleMarkers(t *testing.T) {
 		Tags:   registry.RootTagSet(),
 	}
 
-	o := Output{}
+	o := Output{
+		now: func() time.Time {
+			return time.Unix(1, 0)
+		},
+	}
 	err := o.setTrendStatsResolver([]string{"p(99)"})
 	require.NoError(t, err)
 	trendSink, err := newExtendedTrendSink(o.trendStatsResolver)
@@ -321,27 +325,25 @@ func TestOutputStaleMarkers(t *testing.T) {
 	o.tsdb = map[metrics.TimeSeries]*seriesWithMeasure{
 		trendSinkSeries: {
 			TimeSeries: trendSinkSeries,
-			Latest:     time.Now(),
-			// TODO: if Measure would be a lighter interface
-			// then it could be just a mapper mock.
+			// TODO: if Measure is a lighter interface
+			// then it can be just a mapper mock.
 			Measure: trendSink,
 		},
 		counterSinkSeries: {
 			TimeSeries: counterSinkSeries,
-			Latest:     time.Now(),
 			Measure:    &metrics.CounterSink{},
 		},
 	}
 
-	now := time.Now()
-	markers := o.staleMarkers(now)
+	markers := o.staleMarkers()
 	require.Len(t, markers, 2)
 
 	sortByNameLabel(markers)
 	expNameLabels := []string{"k6_metric1_p99", "k6_metric2_total"}
+	expTimestamp := time.Unix(1, int64(1*time.Millisecond)).UnixMilli()
 	for i, expName := range expNameLabels {
 		assert.Equal(t, expName, markers[i].Labels[0].Value)
-		assert.Equal(t, now.UnixMilli(), markers[i].Samples[0].Timestamp)
+		assert.Equal(t, expTimestamp, markers[i].Samples[0].Timestamp)
 		assert.True(t, math.IsNaN(markers[i].Samples[0].Value), "it isn't a StaleNaN value")
 	}
 }
@@ -360,9 +362,12 @@ func TestOutputStopWithStaleMarkers(t *testing.T) {
 			logger: logger,
 			config: Config{
 				// setting a large interval so it does not trigger
+				// and the trigger can be inoked only when Stop is
+				// invoked.
 				PushInterval: types.NullDurationFrom(1 * time.Hour),
 				StaleMarkers: null.BoolFrom(tc),
 			},
+			now: time.Now,
 		}
 
 		err := o.Start()

pkg/stale/stale.go

@@ -16,5 +16,9 @@ import "math"
 // The value is the same used by the Prometheus package.
 // https://pkg.go.dev/github.com/prometheus/prometheus/pkg/value#pkg-constants
 //
+// It isn't imported directly to avoid the direct dependency
+// from the big Prometheus project that would bring more
+// dependencies.
+//
 //nolint:gochecknoglobals
 var Marker = math.Float64frombits(0x7ff0000000000002)