Remote write 2.0 (part 1/2): move start timestamp, keep writing legacy field for rolling upgrade

CHANGELOG.md

@@ -8,6 +8,7 @@
 * [CHANGE] Removed the following deprecated config: `-querier.filter-queryables-enabled`, `-query-frontend.cache-samples-processed-stats`, `-ingest-storage.kafka.write-clients`, `-blocks-storage.tsdb.head-postings-for-matchers-cache-size`, `-blocks-storage.tsdb.block-postings-for-matchers-cache-size`. #16352
 * [CHANGE] The `bucket` label of the `thanos_objstore_bucket_*` metrics, previously always empty, is now set to the name of the bucket the metrics refer to. This lets a component that accesses more than one bucket report each of them separately. The `thanos_store_bucket_cache_*` and `cortex_bucket_index_load*` metrics gained a `bucket` label carrying the same bucket name, for the same reason. #16265
 * [CHANGE] Compactor: Stabilize `-compactor.first-level-compaction-skip-future-max-time` to `true` and `-compactor.first-level-compaction-ooo-wait-period` to 5 minutes, both of which have been shown to improve batching during the split phase and reduce the total volume of L2 blocks in deployments with lots of out-of-order writes. #16464
+* [CHANGE] Distributor: Remote Write 2.0's `created_timestamp` field on `TimeSeries` has moved to a `start_timestamp` field on each `Sample` and `Histogram`, matching the upstream Remote Write 2.0 specification. The old per-series field is now reserved, but is still marshalled (derived from the first sample's or histogram's start timestamp) for compatibility with not-yet-upgraded internal components during a rolling upgrade. #16475
 * [ENHANCEMENT] Compactor: Add the experimental `-compactor.block-health-validation-concurrency` option to limit how many blocks are validated concurrently within a compaction job. #16269
 * [ENHANCEMENT] Query-frontend: Improve the stability of cardinality estimates and therefore sharding factors for queries when running splitting and caching inside MQE is enabled, or range vector splitting is enabled. #16274 #16301 #16305 #16311
   * When running splitting and caching inside MQE is enabled, the `cortex_query_frontend_cardinality_estimation_difference` metric will no longer be emitted.

integration/distributor_test.go

@@ -1140,6 +1140,118 @@ func TestDistributor_RW2_RC3_CreatedTimestamp(t *testing.T) {
 	require.Equal(t, want.String(), got.String())
 }
 
+// TestDistributor_RW2_StartTimestamp checks that Remote Write 2.0's per-sample start_timestamp
+// (Sample.StartTimestamp), unlike the older per-series created_timestamp, lets each sample in a
+// batch carry its own start timestamp, and that each distinct one independently triggers its own
+// zero sample.
+func TestDistributor_RW2_StartTimestamp(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	previousRuntimeConfig := ""
+	require.NoError(t, writeFileToSharedDir(s, "runtime.yaml", []byte(previousRuntimeConfig)))
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, blocksBucketName)
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	baseFlags := map[string]string{
+		"-distributor.ingestion-tenant-shard-size":           "0",
+		"-ingester.ring.heartbeat-period":                    "1s",
+		"-distributor.ha-tracker.enable":                     "true",
+		"-distributor.ha-tracker.enable-for-all-users":       "true",
+		"-distributor.ha-tracker.store":                      "consul",
+		"-distributor.ha-tracker.consul.hostname":            consul.NetworkHTTPEndpoint(),
+		"-distributor.ha-tracker.prefix":                     "prom_ha/",
+		"-timeseries-unmarshal-caching-optimization-enabled": strconv.FormatBool(true),
+	}
+
+	flags := mergeFlags(
+		BlocksStorageFlags(),
+		BlocksStorageS3Flags(),
+		baseFlags,
+	)
+
+	// We want only distributor to be reloading runtime config.
+	distributorFlags := mergeFlags(flags, map[string]string{
+		"-runtime-config.file":          filepath.Join(e2e.ContainerSharedDir, "runtime.yaml"),
+		"-runtime-config.reload-period": "100ms",
+		// Set non-zero default for number of exemplars. That way our values used in the test (0 and 100) will show up in runtime config diff.
+		"-ingester.max-global-exemplars-per-user": "3",
+	})
+
+	// Ingester will not reload runtime config.
+	ingesterFlags := mergeFlags(flags, map[string]string{
+		// Ingester will always see exemplars enabled. We do this to avoid waiting for ingester to apply new setting to TSDB.
+		"-ingester.max-global-exemplars-per-user": "100",
+	})
+
+	// Start Mimir components.
+	distributor := e2emimir.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), distributorFlags)
+	ingester := e2emimir.NewIngester("ingester", consul.NetworkHTTPEndpoint(), ingesterFlags)
+	querier := e2emimir.NewQuerier("querier", consul.NetworkHTTPEndpoint(), flags)
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester, querier))
+
+	// Wait until distributor has updated the ring.
+	require.NoError(t, distributor.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	// Wait until querier has updated the ring.
+	require.NoError(t, querier.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	client, err := e2emimir.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", userID)
+	require.NoError(t, err)
+
+	queryEnd := time.Now().Round(time.Second)
+	queryStart := queryEnd.Add(-1 * time.Hour)
+	queryStep := 5 * time.Minute
+
+	// Two samples in the same request, each carrying its own StartTimestamp, 15 minutes apart:
+	// this is only expressible via Remote Write 2.0's per-sample start_timestamp, not the older
+	// per-series created_timestamp, which could only ever describe a single counter generation
+	// per request. Each distinct start timestamp should independently trigger its own zero
+	// sample, not just the first one in the batch.
+	rw2req := &promRW2.Request{
+		Timeseries: []promRW2.TimeSeries{
+			{
+				LabelsRefs: []uint32{1, 2},
+				Samples: []promRW2.Sample{
+					{Timestamp: queryStart.Add(1 * time.Second).UnixMilli(), Value: 100, StartTimestamp: queryStart.UnixMilli()},
+					{Timestamp: queryStart.Add(15*time.Minute + 1*time.Second).UnixMilli(), Value: 200, StartTimestamp: queryStart.Add(15 * time.Minute).UnixMilli()},
+				},
+				Metadata: promRW2.Metadata{
+					Type:    promRW2.Metadata_METRIC_TYPE_COUNTER,
+					HelpRef: 3,
+					UnitRef: 4,
+				},
+			},
+		},
+		Symbols: []string{"", "__name__", "foobarST_total", "some helpST", "someunitST"},
+	}
+
+	writeRes, err := client.PushRW2(rw2req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, writeRes.StatusCode)
+
+	want := model.Matrix{{
+		Metric: model.Metric{"__name__": "foobarST_total"},
+		Values: []model.SamplePair{
+			{Timestamp: model.Time(queryStart.UnixMilli()), Value: model.SampleValue(0)},
+			{Timestamp: model.Time(queryStart.Add(5 * time.Minute).UnixMilli()), Value: model.SampleValue(100)},
+			{Timestamp: model.Time(queryStart.Add(15 * time.Minute).UnixMilli()), Value: model.SampleValue(0)},
+			{Timestamp: model.Time(queryStart.Add(20 * time.Minute).UnixMilli()), Value: model.SampleValue(200)},
+		},
+	}}
+	got, _, _, err := client.QueryRange("foobarST_total", queryStart, queryEnd, queryStep)
+	require.NoError(t, err)
+	require.Equal(t, want.String(), got.String())
+}
+
 func testDistributorCases(t *testing.T, cachingUnmarshalDataEnabled bool, rwVersion string, queryStart, queryEnd time.Time, queryStep time.Duration, testCases map[string]distributorTestCase) {
 	s, err := e2e.NewScenario(networkName)
 	require.NoError(t, err)

pkg/blockbuilder/tsdb.go

@@ -141,18 +141,18 @@ func (b *TSDBBuilder) PushToStorageAndReleaseRequest(ctx context.Context, req *m
 		// and NOT the stable hashing because that's what TSDB expects. We don't need stable hashing in block builder.
 		ref, copiedLabels := app.GetRef(nonCopiedLabels, hash)
 
-		ingestCreatedTimestamp := ts.CreatedTimestamp > 0
+		var prevSampleStartTimestamp int64
 
 		for _, s := range ts.Samples {
-			if ingestCreatedTimestamp && ts.CreatedTimestamp < s.TimestampMs &&
+			if s.StartTimestamp > 0 && s.StartTimestamp != prevSampleStartTimestamp && s.StartTimestamp < s.TimestampMs &&
 				(!nativeHistogramsIngestionEnabled || len(ts.Histograms) == 0 || ts.Histograms[0].Timestamp >= s.TimestampMs) {
 				if ref != 0 {
 					// If the cached reference exists, we try to use it.
-					_, err = app.AppendSTZeroSample(ref, copiedLabels, s.TimestampMs, ts.CreatedTimestamp)
+					_, err = app.AppendSTZeroSample(ref, copiedLabels, s.TimestampMs, s.StartTimestamp)
 				} else {
 					// Copy the label set because TSDB may retain it.
 					copiedLabels = mimirpb.CopyLabels(nonCopiedLabels)
-					ref, err = app.AppendSTZeroSample(0, copiedLabels, s.TimestampMs, ts.CreatedTimestamp)
+					ref, err = app.AppendSTZeroSample(0, copiedLabels, s.TimestampMs, s.StartTimestamp)
 				}
 				if err != nil && !errors.Is(err, storage.ErrDuplicateSampleForTimestamp) && !errors.Is(err, storage.ErrOutOfOrderST) && !errors.Is(err, storage.ErrOutOfOrderSample) {
 					// According to OTEL spec: https://opentelemetry.io/docs/specs/otel/metrics/data-model/#cumulative-streams-handling-unknown-start-time
@@ -164,7 +164,7 @@ func (b *TSDBBuilder) PushToStorageAndReleaseRequest(ctx context.Context, req *m
 					level.Warn(b.logger).Log("msg", "failed to store zero float sample for created timestamp", "tenant", tenantID, "err", err)
 					discardedSamples++
 				}
-				ingestCreatedTimestamp = false // Only try to append created timestamp once per series.
+				prevSampleStartTimestamp = s.StartTimestamp // Only try to append a given start timestamp once per series.
 			}
 
 			if ref != 0 {
@@ -194,8 +194,10 @@ func (b *TSDBBuilder) PushToStorageAndReleaseRequest(ctx context.Context, req *m
 			continue
 		}
 
+		var prevHistogramStartTimestamp int64
+
 		for _, h := range ts.Histograms {
-			if ingestCreatedTimestamp && ts.CreatedTimestamp < h.Timestamp {
+			if h.StartTimestamp > 0 && h.StartTimestamp != prevHistogramStartTimestamp && h.StartTimestamp < h.Timestamp {
 				var (
 					ih *histogram.Histogram
 					fh *histogram.FloatHistogram
@@ -208,11 +210,11 @@ func (b *TSDBBuilder) PushToStorageAndReleaseRequest(ctx context.Context, req *m
 					ih = zeroHistogram
 				}
 				if ref != 0 {
-					_, err = app.AppendHistogramSTZeroSample(ref, copiedLabels, h.Timestamp, ts.CreatedTimestamp, ih, fh)
+					_, err = app.AppendHistogramSTZeroSample(ref, copiedLabels, h.Timestamp, h.StartTimestamp, ih, fh)
 				} else {
 					// Copy the label set because both TSDB and the active series tracker may retain it.
 					copiedLabels = mimirpb.CopyLabels(nonCopiedLabels)
-					ref, err = app.AppendHistogramSTZeroSample(0, copiedLabels, h.Timestamp, ts.CreatedTimestamp, ih, fh)
+					ref, err = app.AppendHistogramSTZeroSample(0, copiedLabels, h.Timestamp, h.StartTimestamp, ih, fh)
 				}
 				if err != nil && !errors.Is(err, storage.ErrDuplicateSampleForTimestamp) && !errors.Is(err, storage.ErrOutOfOrderST) && !errors.Is(err, storage.ErrOutOfOrderSample) {
 					// According to OTEL spec: https://opentelemetry.io/docs/specs/otel/metrics/data-model/#cumulative-streams-handling-unknown-start-time
@@ -224,7 +226,7 @@ func (b *TSDBBuilder) PushToStorageAndReleaseRequest(ctx context.Context, req *m
 					level.Warn(b.logger).Log("msg", "failed to store zero histogram sample for created timestamp", "tenant", tenantID, "err", err)
 					discardedSamples++
 				}
-				ingestCreatedTimestamp = false // Only try to append created timestamp once per series.
+				prevHistogramStartTimestamp = h.StartTimestamp // Only try to append a given start timestamp once per series.
 			}
 			var (
 				ih *histogram.Histogram