Support partial write on PRW2.0

CHANGELOG.md

@@ -53,6 +53,7 @@
 * [ENHANCEMENT] Compactor: Reduce object storage GET calls when updating the bucket index by skipping re-reading parquet converter markers for blocks that already have a valid-version parquet entry in the previous index. #7669
 * [ENHANCEMENT] Upgrade Thanos and promql-engine to latest. #7740
 * [ENHANCEMENT] Ruler: Adjust ruler frontend decoder to not wrap query error messages with execution prefix, this makes error responses consistent between internal and external ruler paths. #7741
+* [ENHANCEMENT] Distributor: Support partial write for Prometheus Remote Write 2.0 requests. Invalid series are now skipped and reported together in the `400` response instead of rejecting the whole batch, the valid ones are written, and the `X-Prometheus-Remote-Write-*-Written` response headers are set even when a `400` is returned. Exemplar only `TimeSeries`, which the Prometheus sender emits are also accepted now, consistently with the remote write 1.0 path. #7761
 * [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370
 * [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380
 * [BUGFIX] Distributor: Return HTTP 401 Unauthorized when tenant ID resolution fails in the Prometheus Remote Write 2.0 path. #7389

integration/remote_write_v2_test.go

@@ -879,6 +879,192 @@ func TestExemplar(t *testing.T) {
 	exemplars, err := c.QueryExemplars("test_metric", start, end)
 	require.NoError(t, err)
 	require.Equal(t, 1, len(exemplars))
+
+	// The Prometheus sender emits exemplar only TimeSeries, see
+	// https://github.com/prometheus/prometheus/issues/17857.
+	exemplarOnly := []writev2.TimeSeries{
+		{
+			LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
+			Exemplars:  []writev2.Exemplar{{LabelsRefs: []uint32{13, 14}, Value: 2, Timestamp: tsMillis + 1}},
+		},
+	}
+	writeStats, err = c.PushV2(symbols, exemplarOnly)
+	require.NoError(t, err)
+	testPushHeader(t, writeStats, 0, 0, 1)
+
+	exemplars, err = c.QueryExemplars("test_metric", start, end)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(exemplars))
+	require.Equal(t, 2, len(exemplars[0].Exemplars))
+}
+
+func TestPRW2PartialWrite(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Start dependencies.
+	consul := e2edb.NewConsulWithName("consul")
+	require.NoError(t, s.StartAndWaitReady(consul))
+
+	flags := mergeFlags(
+		AlertmanagerLocalFlags(),
+		map[string]string{
+			"-store.engine":                                     blocksStorageEngine,
+			"-blocks-storage.backend":                           "filesystem",
+			"-blocks-storage.tsdb.head-compaction-interval":     "4m",
+			"-blocks-storage.bucket-store.sync-interval":        "15m",
+			"-blocks-storage.bucket-store.index-cache.backend":  tsdb.IndexCacheBackendInMemory,
+			"-blocks-storage.bucket-store.bucket-index.enabled": "true",
+			"-blocks-storage.tsdb.ship-interval":                "1s",
+			"-blocks-storage.tsdb.enable-native-histograms":     "true",
+			// Ingester.
+			"-ring.store":             "consul",
+			"-consul.hostname":        consul.NetworkHTTPEndpoint(),
+			"-ingester.max-exemplars": "100",
+			// Distributor.
+			"-distributor.replication-factor":     "1",
+			"-distributor.remote-writev2-enabled": "true",
+			// Store-gateway.
+			"-store-gateway.sharding-enabled": "false",
+			// alert manager
+			"-alertmanager.web.external-url": "http://localhost/alertmanager",
+		},
+	)
+
+	// make alert manager config dir
+	require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{}))
+
+	path := path.Join(s.SharedDir(), "cortex-1")
+
+	flags = mergeFlags(flags, map[string]string{"-blocks-storage.filesystem.dir": path})
+	// Start Cortex replicas.
+	cortex := e2ecortex.NewSingleBinary("cortex", flags, "")
+	require.NoError(t, s.StartAndWaitReady(cortex))
+
+	// Wait until Cortex replicas have updated the ring state.
+	require.NoError(t, cortex.WaitSumMetrics(e2e.Equals(float64(512)), "cortex_ring_tokens_total"))
+
+	c, err := e2ecortex.NewClient(cortex.HTTPEndpoint(), cortex.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+
+	now := time.Now()
+	tsMillis := e2e.TimeToMilliseconds(now)
+	start := now.Add(-time.Minute)
+	end := now.Add(time.Minute)
+
+	symbols := []string{
+		"",                  // 0
+		"__name__",          // 1
+		"good_sample",       // 2
+		"good_histogram",    // 3
+		"dropped_labels",    // 4
+		"dropped_exemplar",  // 5
+		"dropped_histogram", // 6
+		"empty_series",      // 7
+		"trace_id",          // 8
+		"abc123",            // 9
+	}
+	// Any ref greater than or equal to the symbols table length is out of range.
+	const invalidRef = 10
+
+	t.Run("every series is dropped during conversion", func(t *testing.T) {
+		timeseries := []writev2.TimeSeries{
+			{LabelsRefs: []uint32{1, 7}},
+			{LabelsRefs: []uint32{1, invalidRef}, Samples: []writev2.Sample{{Value: 1, Timestamp: tsMillis}}},
+		}
+
+		writeStats, err := c.PushV2(symbols, timeseries)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "400")
+		require.Contains(t, err.Error(), "TimeSeries must contain at least one sample, histogram or exemplar")
+		require.Contains(t, err.Error(), "outside of symbols table")
+
+		// Nothing was written, so the response headers must report zero of everything.
+		testPushHeader(t, writeStats, 0, 0, 0)
+
+		result, err := c.Query("empty_series", now)
+		require.NoError(t, err)
+		require.Empty(t, result.(model.Vector))
+	})
+
+	t.Run("dropped data is excluded from the written stats headers", func(t *testing.T) {
+		h := writev2.FromIntHistogram(tsMillis, tsdbutil.GenerateTestHistogram(1))
+
+		timeseries := []writev2.TimeSeries{
+			// Written: 1 sample and 1 exemplar.
+			{
+				LabelsRefs: []uint32{1, 2},
+				Samples:    []writev2.Sample{{Value: 1, Timestamp: tsMillis}},
+				Exemplars:  []writev2.Exemplar{{LabelsRefs: []uint32{8, 9}, Value: 1, Timestamp: tsMillis}},
+			},
+			// Written: 1 histogram.
+			{
+				LabelsRefs: []uint32{1, 3},
+				Histograms: []writev2.Histogram{h},
+			},
+			// Dropped on an out of range label ref, along with its 3 samples and 2 exemplars.
+			{
+				LabelsRefs: []uint32{1, invalidRef},
+				Samples: []writev2.Sample{
+					{Value: 1, Timestamp: tsMillis},
+					{Value: 2, Timestamp: tsMillis + 1},
+					{Value: 3, Timestamp: tsMillis + 2},
+				},
+				Exemplars: []writev2.Exemplar{
+					{LabelsRefs: []uint32{8, 9}, Value: 1, Timestamp: tsMillis},
+					{LabelsRefs: []uint32{8, 9}, Value: 2, Timestamp: tsMillis + 1},
+				},
+			},
+			// Dropped on an out of range exemplar label ref, along with its 2 samples.
+			{
+				LabelsRefs: []uint32{1, 5},
+				Samples: []writev2.Sample{
+					{Value: 1, Timestamp: tsMillis},
+					{Value: 2, Timestamp: tsMillis + 1},
+				},
+				Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{8, invalidRef}, Value: 1, Timestamp: tsMillis}},
+			},
+			// Dropped on an out of range metadata unit ref, along with its histogram.
+			{
+				LabelsRefs: []uint32{1, 6},
+				Metadata:   writev2.Metadata{UnitRef: invalidRef},
+				Histograms: []writev2.Histogram{h},
+			},
+			// Dropped for holding no data at all.
+			{LabelsRefs: []uint32{1, 7}},
+		}
+
+		writeStats, err := c.PushV2(symbols, timeseries)
+
+		// The client is told the request was a bad one, even though it was partially written.
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "400")
+
+		// Only the series that survived conversion are reported as written.
+		testPushHeader(t, writeStats, 1, 1, 1)
+
+		// And they are the only ones actually stored.
+		for _, name := range []string{"good_sample", "good_histogram"} {
+			result, err := c.Query(name, now)
+			require.NoError(t, err)
+			require.Len(t, result.(model.Vector), 1, "%s must be ingested", name)
+		}
+		for _, name := range []string{"dropped_exemplar", "dropped_histogram", "empty_series"} {
+			result, err := c.Query(name, now)
+			require.NoError(t, err)
+			require.Empty(t, result.(model.Vector), "%s must not be ingested", name)
+		}
+
+		exemplars, err := c.QueryExemplars("good_sample", start, end)
+		require.NoError(t, err)
+		require.Len(t, exemplars, 1)
+		require.Len(t, exemplars[0].Exemplars, 1)
+
+		exemplars, err = c.QueryExemplars("dropped_exemplar", start, end)
+		require.NoError(t, err)
+		require.Empty(t, exemplars)
+	})
 }
 
 func Test_WriteStatWithReplication(t *testing.T) {

pkg/util/push/push.go

@@ -115,19 +115,22 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
 				req.Source = cortexpb.API
 			}
 
-			v1Req, err := convertV2RequestToV1(req, overrides.EnableTypeAndUnitLabels(userID), overrides.EnableStartTimestamp(userID))
-			if err != nil {
-				level.Error(logger).Log("err", err.Error())
-				http.Error(w, err.Error(), http.StatusBadRequest)
-				return
+			// convertErr is a non-retriable bad request error. The series that converted
+			// successfully are still pushed, so the request is partially written.
+			v1Req, convertErr := convertV2RequestToV1(req, overrides.EnableTypeAndUnitLabels(userID), overrides.EnableStartTimestamp(userID))
+			if convertErr != nil {
+				level.Warn(logger).Log("msg", "remote write v2 request partially converted", "err", convertErr)
 			}
 
 			v1Req.SkipLabelNameValidation = false
 			if v1Req.Source == 0 {
 				v1Req.Source = cortexpb.API
 			}
 
-			if writeResp, err := push(ctx, &v1Req.WriteRequest); err != nil {
+			// The Distributor owns the pooled TimeSeries, so push must be called even when
+			// every series was skipped.
+			writeResp, err := push(ctx, &v1Req.WriteRequest)
+			if err != nil {
 				if errors.Is(err, context.Canceled) {
 					err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error())
 				}
@@ -148,11 +151,17 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
 				} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest {
 					level.Warn(logger).Log("msg", "push refused", "err", err)
 				}
+				// The push error takes precedence over convertErr: a 5xx must be retried.
 				http.Error(w, string(resp.Body), int(resp.Code))
-			} else {
-				setPRW2RespHeader(w, writeResp.Samples, writeResp.Histograms, writeResp.Exemplars)
-				w.WriteHeader(http.StatusNoContent)
+				return
 			}
+
+			setPRW2RespHeader(w, writeResp.Samples, writeResp.Histograms, writeResp.Exemplars)
+			if convertErr != nil {
+				http.Error(w, convertErr.Error(), http.StatusBadRequest)
+				return
+			}
+			w.WriteHeader(http.StatusNoContent)
 		}
 
 		// follow Prometheus https://github.com/prometheus/prometheus/blob/v3.3.1/storage/remote/write_handler.go#L121
@@ -214,35 +223,67 @@ func setPRW2RespHeader(w http.ResponseWriter, samples, histograms, exemplars int
 	w.Header().Set(rw20WrittenExemplarsHeader, strconv.FormatInt(exemplars, 10))
 }
 
-func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUnitLabels bool, enableStartTimestamp bool) (v1Req cortexpb.PreallocWriteRequest, err error) {
+// maxConversionErrs is the maximum number of per-series conversion errors reported back
+// to the client.
+const maxConversionErrs = 10
+
+// conversionErrs collects per-series conversion errors, keeping at most maxConversionErrs
+// of them while still reporting the total count.
+type conversionErrs struct {
+	errs  []error
+	count int
+}
+
+func (c *conversionErrs) add(err error) {
+	c.count++
+	if len(c.errs) < maxConversionErrs {
+		c.errs = append(c.errs, err)
+	}
+}
+
+// err joins the collected errors, summarizing the ones that were left out.
+func (c *conversionErrs) err() error {
+	if c.count == 0 {
+		return nil
+	}
+	if omitted := c.count - len(c.errs); omitted > 0 {
+		return errors.Join(append(c.errs, fmt.Errorf("%d more errors omitted", omitted))...)
+	}
+	return errors.Join(c.errs...)
+}
+
+// convertV2RequestToV1 converts a remote write v2 request into a v1 one.
+//
+// Malformed series are skipped and their errors are joined into the returned error,
+// which is always a non-retriable bad request error. Series that converted successfully
+// are still returned, so the request can be partially written, following the Prometheus
+// receiver behavior.
+func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUnitLabels bool, enableStartTimestamp bool) (cortexpb.PreallocWriteRequest, error) {
+	var v1Req cortexpb.PreallocWriteRequest
 	v1Timeseries := make([]cortexpb.PreallocTimeseries, 0, len(req.Timeseries))
 	var v1Metadata []*cortexpb.MetricMetadata
-
-	// Release any pulled TimeSeries back to the pool to prevent memory leaks in case of an error.
-	defer func() {
-		if err != nil {
-			for _, pts := range v1Timeseries {
-				if pts.TimeSeries != nil {
-					cortexpb.ReuseTimeseries(pts.TimeSeries)
-				}
-			}
-		}
-	}()
+	var badRequestErrs conversionErrs
 
 	b := labels.NewScratchBuilder(0)
 	symbols := req.Symbols
 	for _, v2Ts := range req.Timeseries {
 		lbs, err := v2Ts.ToLabels(&b, symbols)
 		if err != nil {
-			return v1Req, err
+			badRequestErrs.add(err)
+			continue
 		}
 
-		if len(v2Ts.Samples) == 0 && len(v2Ts.Histograms) == 0 {
-			return v1Req, fmt.Errorf("TimeSeries must contain at least one sample or histogram for series %v", lbs.String())
+		// The remote write 2.0 spec requires a TimeSeries to hold at least one sample or
+		// histogram, but the Prometheus sender emits exemplar only TimeSeries, see
+		// https://github.com/prometheus/prometheus/issues/17857.
+		if len(v2Ts.Samples) == 0 && len(v2Ts.Histograms) == 0 && len(v2Ts.Exemplars) == 0 {
+			badRequestErrs.add(fmt.Errorf("TimeSeries must contain at least one sample, histogram or exemplar for series %v", lbs.String()))
+			continue
 		}
 
 		if int(v2Ts.Metadata.UnitRef) >= len(symbols) {
-			return v1Req, fmt.Errorf("invalid UnitRef %d: exceeds symbols length %d", v2Ts.Metadata.UnitRef, len(symbols))
+			badRequestErrs.add(fmt.Errorf("invalid UnitRef %d: exceeds symbols length %d", v2Ts.Metadata.UnitRef, len(symbols)))
+			continue
 		}
 
 		unit := symbols[v2Ts.Metadata.UnitRef]
@@ -281,7 +322,8 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
 		if err != nil {
 			// Current ts is not appended to the v1Timeseries, so we should call reuse here.
 			cortexpb.ReuseTimeseries(ts)
-			return v1Req, err
+			badRequestErrs.add(err)
+			continue
 		}
 
 		ts.Histograms = ts.Histograms[:0]
@@ -302,16 +344,17 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
 		})
 
 		if shouldConvertV2Metadata(v2Ts.Metadata) {
-			var metricName string
-			metricName, err = extract.MetricNameFromLabels(lbs)
+			// The series has already been appended above, so only its metadata is dropped here.
+			metricName, err := extract.MetricNameFromLabels(lbs)
 			if err != nil {
-				return v1Req, err
+				badRequestErrs.add(err)
+				continue
 			}
 
-			var metadata *cortexpb.MetricMetadata
-			metadata, err = convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
+			metadata, err := convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
 			if err != nil {
-				return v1Req, err
+				badRequestErrs.add(err)
+				continue
 			}
 			v1Metadata = append(v1Metadata, metadata)
 		}
@@ -320,7 +363,7 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
 	v1Req.Timeseries = v1Timeseries
 	v1Req.Metadata = v1Metadata
 
-	return v1Req, nil
+	return v1Req, badRequestErrs.err()
 }
 
 func shouldConvertV2Metadata(metadata cortexpb.MetadataV2) bool {