Merge pull request #790 from grafana/revert-766-start-time-quiet

Revert "TSDB: Add 'quiet zero' for Otel start-time handling"

Author: jesusvazquez

model/value/value.go

@@ -26,9 +26,6 @@ const (
 	// complicated values in the future. It is 2 rather than 1 to make
 	// it easier to distinguish from the NormalNaN by a human when debugging.
 	StaleNaN uint64 = 0x7ff0000000000002
-
-	// QuietZeroNaN signals TSDB to add a zero, but do nothing if there is already a value at that timestamp.
-	QuietZeroNaN uint64 = 0x7ff0000000000003
 )
 
 // IsStaleNaN returns true when the provided NaN value is a stale marker.

storage/remote/otlptranslator/prometheusremotewrite/helper.go

@@ -595,10 +595,9 @@ const defaultIntervalForStartTimestamps = int64(300_000)
 // handleStartTime adds a zero sample at startTs only if startTs is within validIntervalForStartTimestamps of the sample timestamp.
 // The reason for doing this is that PRW v1 doesn't support Created Timestamps. After switching to PRW v2's direct CT support,
 // make use of its direct support fort Created Timestamps instead.
-// See https://github.com/prometheus/prometheus/issues/14600 for context.
 // See https://opentelemetry.io/docs/specs/otel/metrics/data-model/#resets-and-gaps to know more about how OTel handles
 // resets for cumulative metrics.
-func (c *PrometheusConverter) handleStartTime(startTs, ts int64, labels []prompb.Label, settings Settings, typ string, val float64, logger *slog.Logger) {
+func (c *PrometheusConverter) handleStartTime(startTs, ts int64, labels []prompb.Label, settings Settings, typ string, value float64, logger *slog.Logger) {
 	if !settings.EnableCreatedTimestampZeroIngestion {
 		return
 	}
@@ -620,9 +619,10 @@ func (c *PrometheusConverter) handleStartTime(startTs, ts int64, labels []prompb
 		return
 	}
 
-	logger.Debug("adding zero value at start_ts", "type", typ, "labels", labelsStringer(labels), "start_ts", startTs, "sample_ts", ts, "sample_value", val)
+	logger.Debug("adding zero value at start_ts", "type", typ, "labels", labelsStringer(labels), "start_ts", startTs, "sample_ts", ts, "sample_value", value)
 
-	c.addSample(&prompb.Sample{Timestamp: startTs, Value: math.Float64frombits(value.QuietZeroNaN)}, labels)
+	// See https://github.com/prometheus/prometheus/issues/14600 for context.
+	c.addSample(&prompb.Sample{Timestamp: startTs}, labels)
 }
 
 // handleHistogramStartTime similar to the method above but for native histograms..

storage/remote/otlptranslator/prometheusremotewrite/helper_test.go

@@ -18,11 +18,9 @@ package prometheusremotewrite
 
 import (
 	"context"
-	"math"
 	"testing"
 	"time"
 
-	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -31,9 +29,7 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/promslog"
 
-	"github.com/prometheus/prometheus/model/value"
 	"github.com/prometheus/prometheus/prompb"
-	"github.com/prometheus/prometheus/util/testutil"
 )
 
 func TestCreateAttributes(t *testing.T) {
@@ -317,14 +313,14 @@ func TestPrometheusConverter_AddSummaryDataPoints(t *testing.T) {
 					timeSeriesSignature(labels): {
 						Labels: labels,
 						Samples: []prompb.Sample{
-							{Value: math.Float64frombits(value.QuietZeroNaN), Timestamp: convertTimeStamp(nowMinus2m30s)},
+							{Value: 0, Timestamp: convertTimeStamp(nowMinus2m30s)},
 							{Value: 0, Timestamp: convertTimeStamp(nowUnixNano)},
 						},
 					},
 					timeSeriesSignature(sumLabels): {
 						Labels: sumLabels,
 						Samples: []prompb.Sample{
-							{Value: math.Float64frombits(value.QuietZeroNaN), Timestamp: convertTimeStamp(nowMinus2m30s)},
+							{Value: 0, Timestamp: convertTimeStamp(nowMinus2m30s)},
 							{Value: 0, Timestamp: convertTimeStamp(nowUnixNano)},
 						},
 					},
@@ -365,14 +361,14 @@ func TestPrometheusConverter_AddSummaryDataPoints(t *testing.T) {
 					timeSeriesSignature(labels): {
 						Labels: labels,
 						Samples: []prompb.Sample{
-							{Value: math.Float64frombits(value.QuietZeroNaN), Timestamp: convertTimeStamp(nowMinus6m)},
+							{Value: 0, Timestamp: convertTimeStamp(nowMinus6m)},
 							{Value: 0, Timestamp: convertTimeStamp(nowUnixNano)},
 						},
 					},
 					timeSeriesSignature(sumLabels): {
 						Labels: sumLabels,
 						Samples: []prompb.Sample{
-							{Value: math.Float64frombits(value.QuietZeroNaN), Timestamp: convertTimeStamp(nowMinus6m)},
+							{Value: 0, Timestamp: convertTimeStamp(nowMinus6m)},
 							{Value: 0, Timestamp: convertTimeStamp(nowUnixNano)},
 						},
 					},
@@ -485,17 +481,12 @@ func TestPrometheusConverter_AddSummaryDataPoints(t *testing.T) {
 			)
 			require.NoError(t, err)
 
-			testutil.RequireEqualWithOptions(t, tt.want(), converter.unique, []cmp.Option{cmp.Comparer(equalSamples)})
+			assert.Equal(t, tt.want(), converter.unique)
 			assert.Empty(t, converter.conflicts)
 		})
 	}
 }
 
-func equalSamples(a, b prompb.Sample) bool {
-	// Compare Float64bits so NaN values which are exactly the same will compare equal.
-	return a.Timestamp == b.Timestamp && math.Float64bits(a.Value) == math.Float64bits(b.Value)
-}
-
 func TestPrometheusConverter_AddHistogramDataPoints(t *testing.T) {
 	ts := pcommon.Timestamp(time.Now().UnixNano())
 	tests := []struct {

tsdb/db_test.go

@@ -105,15 +105,6 @@ func openTestDB(t testing.TB, opts *Options, rngs []int64) (db *DB) {
 	return db
 }
 
-// queryHead is a helper to query the head for a given time range and labelset.
-func queryHead(t testing.TB, head *Head, mint, maxt int64, label labels.Label) (map[string][]chunks.Sample, error) {
-	q, err := NewBlockQuerier(head, mint, maxt)
-	if err != nil {
-		return nil, err
-	}
-	return query(t, q, labels.MustNewMatcher(labels.MatchEqual, label.Name, label.Value)), nil
-}
-
 // query runs a matcher query against the querier and fully expands its data.
 func query(t testing.TB, q storage.Querier, matchers ...*labels.Matcher) map[string][]chunks.Sample {
 	ss := q.Select(context.Background(), false, nil, matchers...)

tsdb/head_append.go

@@ -497,7 +497,7 @@ func (s *memSeries) appendable(t int64, v float64, headMaxt, minValidTime, oooTi
 			if s.lastHistogramValue != nil || s.lastFloatHistogramValue != nil {
 				return false, 0, storage.NewDuplicateHistogramToFloatErr(t, v)
 			}
-			if math.Float64bits(s.lastValue) != math.Float64bits(v) && math.Float64bits(v) != value.QuietZeroNaN {
+			if math.Float64bits(s.lastValue) != math.Float64bits(v) {
 				return false, 0, storage.NewDuplicateFloatErr(t, s.lastValue, v)
 			}
 			// Sample is identical (ts + value) with most current (highest ts) sample in sampleBuf.
@@ -1141,10 +1141,6 @@ func (a *headAppender) commitSamples(acc *appenderCommitContext) {
 			handleAppendableError(err, &acc.floatsAppended, &acc.floatOOORejected, &acc.floatOOBRejected, &acc.floatTooOldRejected)
 		}
 
-		if math.Float64bits(s.V) == value.QuietZeroNaN {
-			s.V = 0
-		}
-
 		switch {
 		case err != nil:
 			// Do nothing here.

tsdb/head_test.go

@@ -547,6 +547,15 @@ func TestHead_HighConcurrencyReadAndWrite(t *testing.T) {
 		})
 	}
 
+	// queryHead is a helper to query the head for a given time range and labelset.
+	queryHead := func(mint, maxt uint64, label labels.Label) (map[string][]chunks.Sample, error) {
+		q, err := NewBlockQuerier(head, int64(mint), int64(maxt))
+		if err != nil {
+			return nil, err
+		}
+		return query(t, q, labels.MustNewMatcher(labels.MatchEqual, label.Name, label.Value)), nil
+	}
+
 	// readerTsCh will be used by the coordinator go routine to coordinate which timestamps the reader should read.
 	readerTsCh := make(chan uint64)
 
@@ -574,7 +583,7 @@ func TestHead_HighConcurrencyReadAndWrite(t *testing.T) {
 				lbls.Range(func(l labels.Label) {
 					lbl = l
 				})
-				samples, err := queryHead(t, head, int64(ts-qryRange), int64(ts), lbl)
+				samples, err := queryHead(ts-qryRange, ts, lbl)
 				if err != nil {
 					return false, err
 				}
@@ -6089,42 +6098,6 @@ func TestCuttingNewHeadChunks(t *testing.T) {
 	}
 }
 
-func TestAppendDuplicates(t *testing.T) {
-	ts := int64(1695209650)
-	lbls := labels.FromStrings("foo", "bar")
-	h, _ := newTestHead(t, DefaultBlockDuration, wlog.CompressionNone, false)
-	defer func() {
-		require.NoError(t, h.Close())
-	}()
-
-	a := h.Appender(context.Background())
-	_, err := a.Append(0, lbls, ts, 42.0)
-	require.NoError(t, err)
-	_, err = a.Append(0, lbls, ts, 42.0) // Exactly the same value.
-	require.NoError(t, err)
-	_, err = a.Append(0, lbls, ts, math.Float64frombits(value.QuietZeroNaN)) // Should be a no-op.
-	require.NoError(t, err)
-	require.NoError(t, a.Commit())
-
-	result, err := queryHead(t, h, math.MinInt64, math.MaxInt64, labels.Label{Name: "foo", Value: "bar"})
-	require.NoError(t, err)
-	expectedSamples := []chunks.Sample{sample{t: ts, f: 42.0}}
-	require.Equal(t, expectedSamples, result[`{foo="bar"}`])
-
-	a = h.Appender(context.Background())
-	_, err = a.Append(0, lbls, ts+10, math.Float64frombits(value.QuietZeroNaN)) // This is at a different timestamp so should append a real zero.
-	require.NoError(t, err)
-	require.NoError(t, a.Commit())
-
-	result, err = queryHead(t, h, math.MinInt64, math.MaxInt64, labels.Label{Name: "foo", Value: "bar"})
-	require.NoError(t, err)
-	expectedSamples = []chunks.Sample{
-		sample{t: ts, f: 42.0},
-		sample{t: ts + 10, f: 0},
-	}
-	require.Equal(t, expectedSamples, result[`{foo="bar"}`])
-}
-
 // TestHeadDetectsDuplicateSampleAtSizeLimit tests a regression where a duplicate sample
 // is appended to the head, right when the head chunk is at the size limit.
 // The test adds all samples as duplicate, thus expecting that the result has