Skip to content

Commit af8e168

Browse files
SungJin1212SungJin1212
committed
PRW2: support partial write
Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>
1 parent e450d52 commit af8e168

4 files changed

Lines changed: 445 additions & 39 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
* [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
5454
* [ENHANCEMENT] Upgrade Thanos and promql-engine to latest. #7740
5555
* [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
56+
* [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
5657
* [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370
5758
* [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380
5859
* [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‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -879,6 +879,192 @@ func TestExemplar(t *testing.T) {
879879
exemplars, err := c.QueryExemplars("test_metric", start, end)
880880
require.NoError(t, err)
881881
require.Equal(t, 1, len(exemplars))
882+
883+
// The Prometheus sender emits exemplar only TimeSeries, see
884+
// https://github.com/prometheus/prometheus/issues/17857.
885+
exemplarOnly := []writev2.TimeSeries{
886+
{
887+
LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
888+
Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{13, 14}, Value: 2, Timestamp: tsMillis + 1}},
889+
},
890+
}
891+
writeStats, err = c.PushV2(symbols, exemplarOnly)
892+
require.NoError(t, err)
893+
testPushHeader(t, writeStats, 0, 0, 1)
894+
895+
exemplars, err = c.QueryExemplars("test_metric", start, end)
896+
require.NoError(t, err)
897+
require.Equal(t, 1, len(exemplars))
898+
require.Equal(t, 2, len(exemplars[0].Exemplars))
899+
}
900+
901+
func TestPRW2PartialWrite(t *testing.T) {
902+
s, err := e2e.NewScenario(networkName)
903+
require.NoError(t, err)
904+
defer s.Close()
905+
906+
// Start dependencies.
907+
consul := e2edb.NewConsulWithName("consul")
908+
require.NoError(t, s.StartAndWaitReady(consul))
909+
910+
flags := mergeFlags(
911+
AlertmanagerLocalFlags(),
912+
map[string]string{
913+
"-store.engine": blocksStorageEngine,
914+
"-blocks-storage.backend": "filesystem",
915+
"-blocks-storage.tsdb.head-compaction-interval": "4m",
916+
"-blocks-storage.bucket-store.sync-interval": "15m",
917+
"-blocks-storage.bucket-store.index-cache.backend": tsdb.IndexCacheBackendInMemory,
918+
"-blocks-storage.bucket-store.bucket-index.enabled": "true",
919+
"-blocks-storage.tsdb.ship-interval": "1s",
920+
"-blocks-storage.tsdb.enable-native-histograms": "true",
921+
// Ingester.
922+
"-ring.store": "consul",
923+
"-consul.hostname": consul.NetworkHTTPEndpoint(),
924+
"-ingester.max-exemplars": "100",
925+
// Distributor.
926+
"-distributor.replication-factor": "1",
927+
"-distributor.remote-writev2-enabled": "true",
928+
// Store-gateway.
929+
"-store-gateway.sharding-enabled": "false",
930+
// alert manager
931+
"-alertmanager.web.external-url": "http://localhost/alertmanager",
932+
},
933+
)
934+
935+
// make alert manager config dir
936+
require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{}))
937+
938+
path := path.Join(s.SharedDir(), "cortex-1")
939+
940+
flags = mergeFlags(flags, map[string]string{"-blocks-storage.filesystem.dir": path})
941+
// Start Cortex replicas.
942+
cortex := e2ecortex.NewSingleBinary("cortex", flags, "")
943+
require.NoError(t, s.StartAndWaitReady(cortex))
944+
945+
// Wait until Cortex replicas have updated the ring state.
946+
require.NoError(t, cortex.WaitSumMetrics(e2e.Equals(float64(512)), "cortex_ring_tokens_total"))
947+
948+
c, err := e2ecortex.NewClient(cortex.HTTPEndpoint(), cortex.HTTPEndpoint(), "", "", "user-1")
949+
require.NoError(t, err)
950+
951+
now := time.Now()
952+
tsMillis := e2e.TimeToMilliseconds(now)
953+
start := now.Add(-time.Minute)
954+
end := now.Add(time.Minute)
955+
956+
symbols := []string{
957+
"", // 0
958+
"__name__", // 1
959+
"good_sample", // 2
960+
"good_histogram", // 3
961+
"dropped_labels", // 4
962+
"dropped_exemplar", // 5
963+
"dropped_histogram", // 6
964+
"empty_series", // 7
965+
"trace_id", // 8
966+
"abc123", // 9
967+
}
968+
// Any ref greater than or equal to the symbols table length is out of range.
969+
const invalidRef = 10
970+
971+
t.Run("every series is dropped during conversion", func(t *testing.T) {
972+
timeseries := []writev2.TimeSeries{
973+
{LabelsRefs: []uint32{1, 7}},
974+
{LabelsRefs: []uint32{1, invalidRef}, Samples: []writev2.Sample{{Value: 1, Timestamp: tsMillis}}},
975+
}
976+
977+
writeStats, err := c.PushV2(symbols, timeseries)
978+
require.Error(t, err)
979+
require.Contains(t, err.Error(), "400")
980+
require.Contains(t, err.Error(), "TimeSeries must contain at least one sample, histogram or exemplar")
981+
require.Contains(t, err.Error(), "outside of symbols table")
982+
983+
// Nothing was written, so the response headers must report zero of everything.
984+
testPushHeader(t, writeStats, 0, 0, 0)
985+
986+
result, err := c.Query("empty_series", now)
987+
require.NoError(t, err)
988+
require.Empty(t, result.(model.Vector))
989+
})
990+
991+
t.Run("dropped data is excluded from the written stats headers", func(t *testing.T) {
992+
h := writev2.FromIntHistogram(tsMillis, tsdbutil.GenerateTestHistogram(1))
993+
994+
timeseries := []writev2.TimeSeries{
995+
// Written: 1 sample and 1 exemplar.
996+
{
997+
LabelsRefs: []uint32{1, 2},
998+
Samples: []writev2.Sample{{Value: 1, Timestamp: tsMillis}},
999+
Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{8, 9}, Value: 1, Timestamp: tsMillis}},
1000+
},
1001+
// Written: 1 histogram.
1002+
{
1003+
LabelsRefs: []uint32{1, 3},
1004+
Histograms: []writev2.Histogram{h},
1005+
},
1006+
// Dropped on an out of range label ref, along with its 3 samples and 2 exemplars.
1007+
{
1008+
LabelsRefs: []uint32{1, invalidRef},
1009+
Samples: []writev2.Sample{
1010+
{Value: 1, Timestamp: tsMillis},
1011+
{Value: 2, Timestamp: tsMillis + 1},
1012+
{Value: 3, Timestamp: tsMillis + 2},
1013+
},
1014+
Exemplars: []writev2.Exemplar{
1015+
{LabelsRefs: []uint32{8, 9}, Value: 1, Timestamp: tsMillis},
1016+
{LabelsRefs: []uint32{8, 9}, Value: 2, Timestamp: tsMillis + 1},
1017+
},
1018+
},
1019+
// Dropped on an out of range exemplar label ref, along with its 2 samples.
1020+
{
1021+
LabelsRefs: []uint32{1, 5},
1022+
Samples: []writev2.Sample{
1023+
{Value: 1, Timestamp: tsMillis},
1024+
{Value: 2, Timestamp: tsMillis + 1},
1025+
},
1026+
Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{8, invalidRef}, Value: 1, Timestamp: tsMillis}},
1027+
},
1028+
// Dropped on an out of range metadata unit ref, along with its histogram.
1029+
{
1030+
LabelsRefs: []uint32{1, 6},
1031+
Metadata: writev2.Metadata{UnitRef: invalidRef},
1032+
Histograms: []writev2.Histogram{h},
1033+
},
1034+
// Dropped for holding no data at all.
1035+
{LabelsRefs: []uint32{1, 7}},
1036+
}
1037+
1038+
writeStats, err := c.PushV2(symbols, timeseries)
1039+
1040+
// The client is told the request was a bad one, even though it was partially written.
1041+
require.Error(t, err)
1042+
require.Contains(t, err.Error(), "400")
1043+
1044+
// Only the series that survived conversion are reported as written.
1045+
testPushHeader(t, writeStats, 1, 1, 1)
1046+
1047+
// And they are the only ones actually stored.
1048+
for _, name := range []string{"good_sample", "good_histogram"} {
1049+
result, err := c.Query(name, now)
1050+
require.NoError(t, err)
1051+
require.Len(t, result.(model.Vector), 1, "%s must be ingested", name)
1052+
}
1053+
for _, name := range []string{"dropped_exemplar", "dropped_histogram", "empty_series"} {
1054+
result, err := c.Query(name, now)
1055+
require.NoError(t, err)
1056+
require.Empty(t, result.(model.Vector), "%s must not be ingested", name)
1057+
}
1058+
1059+
exemplars, err := c.QueryExemplars("good_sample", start, end)
1060+
require.NoError(t, err)
1061+
require.Len(t, exemplars, 1)
1062+
require.Len(t, exemplars[0].Exemplars, 1)
1063+
1064+
exemplars, err = c.QueryExemplars("dropped_exemplar", start, end)
1065+
require.NoError(t, err)
1066+
require.Empty(t, exemplars)
1067+
})
8821068
}
8831069

8841070
func Test_WriteStatWithReplication(t *testing.T) {

‎pkg/util/push/push.go‎

Lines changed: 77 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -115,19 +115,22 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
115115
req.Source = cortexpb.API
116116
}
117117

118-
v1Req, err := convertV2RequestToV1(req, overrides.EnableTypeAndUnitLabels(userID), overrides.EnableStartTimestamp(userID))
119-
if err != nil {
120-
level.Error(logger).Log("err", err.Error())
121-
http.Error(w, err.Error(), http.StatusBadRequest)
122-
return
118+
// convertErr is a non-retriable bad request error. The series that converted
119+
// successfully are still pushed, so the request is partially written.
120+
v1Req, convertErr := convertV2RequestToV1(req, overrides.EnableTypeAndUnitLabels(userID), overrides.EnableStartTimestamp(userID))
121+
if convertErr != nil {
122+
level.Warn(logger).Log("msg", "remote write v2 request partially converted", "err", convertErr)
123123
}
124124

125125
v1Req.SkipLabelNameValidation = false
126126
if v1Req.Source == 0 {
127127
v1Req.Source = cortexpb.API
128128
}
129129

130-
if writeResp, err := push(ctx, &v1Req.WriteRequest); err != nil {
130+
// The Distributor owns the pooled TimeSeries, so push must be called even when
131+
// every series was skipped.
132+
writeResp, err := push(ctx, &v1Req.WriteRequest)
133+
if err != nil {
131134
if errors.Is(err, context.Canceled) {
132135
err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error())
133136
}
@@ -148,11 +151,18 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool,
148151
} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest {
149152
level.Warn(logger).Log("msg", "push refused", "err", err)
150153
}
154+
// The push error takes precedence over convertErr: a 5xx must be retried as a
155+
// whole, and a 4xx is at least as informative.
151156
http.Error(w, string(resp.Body), int(resp.Code))
152-
} else {
153-
setPRW2RespHeader(w, writeResp.Samples, writeResp.Histograms, writeResp.Exemplars)
154-
w.WriteHeader(http.StatusNoContent)
157+
return
155158
}
159+
160+
setPRW2RespHeader(w, writeResp.Samples, writeResp.Histograms, writeResp.Exemplars)
161+
if convertErr != nil {
162+
http.Error(w, convertErr.Error(), http.StatusBadRequest)
163+
return
164+
}
165+
w.WriteHeader(http.StatusNoContent)
156166
}
157167

158168
// follow Prometheus https://github.com/prometheus/prometheus/blob/v3.3.1/storage/remote/write_handler.go#L121
@@ -214,35 +224,67 @@ func setPRW2RespHeader(w http.ResponseWriter, samples, histograms, exemplars int
214224
w.Header().Set(rw20WrittenExemplarsHeader, strconv.FormatInt(exemplars, 10))
215225
}
216226

217-
func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUnitLabels bool, enableStartTimestamp bool) (v1Req cortexpb.PreallocWriteRequest, err error) {
227+
// maxConversionErrs is the maximum number of per-series conversion errors reported back
228+
// to the client.
229+
const maxConversionErrs = 10
230+
231+
// conversionErrs collects per-series conversion errors, keeping at most maxConversionErrs
232+
// of them while still reporting the total count.
233+
type conversionErrs struct {
234+
errs []error
235+
count int
236+
}
237+
238+
func (c *conversionErrs) add(err error) {
239+
c.count++
240+
if len(c.errs) < maxConversionErrs {
241+
c.errs = append(c.errs, err)
242+
}
243+
}
244+
245+
// err joins the collected errors, summarizing the ones that were left out.
246+
func (c *conversionErrs) err() error {
247+
if c.count == 0 {
248+
return nil
249+
}
250+
if omitted := c.count - len(c.errs); omitted > 0 {
251+
return errors.Join(append(c.errs, fmt.Errorf("%d more errors omitted", omitted))...)
252+
}
253+
return errors.Join(c.errs...)
254+
}
255+
256+
// convertV2RequestToV1 converts a remote write v2 request into a v1 one.
257+
//
258+
// Malformed series are skipped and their errors are joined into the returned error,
259+
// which is always a non-retriable bad request error. Series that converted successfully
260+
// are still returned, so the request can be partially written, following the Prometheus
261+
// receiver behavior.
262+
func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUnitLabels bool, enableStartTimestamp bool) (cortexpb.PreallocWriteRequest, error) {
263+
var v1Req cortexpb.PreallocWriteRequest
218264
v1Timeseries := make([]cortexpb.PreallocTimeseries, 0, len(req.Timeseries))
219265
var v1Metadata []*cortexpb.MetricMetadata
220-
221-
// Release any pulled TimeSeries back to the pool to prevent memory leaks in case of an error.
222-
defer func() {
223-
if err != nil {
224-
for _, pts := range v1Timeseries {
225-
if pts.TimeSeries != nil {
226-
cortexpb.ReuseTimeseries(pts.TimeSeries)
227-
}
228-
}
229-
}
230-
}()
266+
var badRequestErrs conversionErrs
231267

232268
b := labels.NewScratchBuilder(0)
233269
symbols := req.Symbols
234270
for _, v2Ts := range req.Timeseries {
235271
lbs, err := v2Ts.ToLabels(&b, symbols)
236272
if err != nil {
237-
return v1Req, err
273+
badRequestErrs.add(err)
274+
continue
238275
}
239276

240-
if len(v2Ts.Samples) == 0 && len(v2Ts.Histograms) == 0 {
241-
return v1Req, fmt.Errorf("TimeSeries must contain at least one sample or histogram for series %v", lbs.String())
277+
// The remote write 2.0 spec requires a TimeSeries to hold at least one sample or
278+
// histogram, but the Prometheus sender emits exemplar only TimeSeries, see
279+
// https://github.com/prometheus/prometheus/issues/17857.
280+
if len(v2Ts.Samples) == 0 && len(v2Ts.Histograms) == 0 && len(v2Ts.Exemplars) == 0 {
281+
badRequestErrs.add(fmt.Errorf("TimeSeries must contain at least one sample, histogram or exemplar for series %v", lbs.String()))
282+
continue
242283
}
243284

244285
if int(v2Ts.Metadata.UnitRef) >= len(symbols) {
245-
return v1Req, fmt.Errorf("invalid UnitRef %d: exceeds symbols length %d", v2Ts.Metadata.UnitRef, len(symbols))
286+
badRequestErrs.add(fmt.Errorf("invalid UnitRef %d: exceeds symbols length %d", v2Ts.Metadata.UnitRef, len(symbols)))
287+
continue
246288
}
247289

248290
unit := symbols[v2Ts.Metadata.UnitRef]
@@ -281,7 +323,8 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
281323
if err != nil {
282324
// Current ts is not appended to the v1Timeseries, so we should call reuse here.
283325
cortexpb.ReuseTimeseries(ts)
284-
return v1Req, err
326+
badRequestErrs.add(err)
327+
continue
285328
}
286329

287330
ts.Histograms = ts.Histograms[:0]
@@ -302,16 +345,17 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
302345
})
303346

304347
if shouldConvertV2Metadata(v2Ts.Metadata) {
305-
var metricName string
306-
metricName, err = extract.MetricNameFromLabels(lbs)
348+
// The series has already been appended above, so only its metadata is dropped here.
349+
metricName, err := extract.MetricNameFromLabels(lbs)
307350
if err != nil {
308-
return v1Req, err
351+
badRequestErrs.add(err)
352+
continue
309353
}
310354

311-
var metadata *cortexpb.MetricMetadata
312-
metadata, err = convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
355+
metadata, err := convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)
313356
if err != nil {
314-
return v1Req, err
357+
badRequestErrs.add(err)
358+
continue
315359
}
316360
v1Metadata = append(v1Metadata, metadata)
317361
}
@@ -320,7 +364,7 @@ func convertV2RequestToV1(req *cortexpb.PreallocWriteRequestV2, enableTypeAndUni
320364
v1Req.Timeseries = v1Timeseries
321365
v1Req.Metadata = v1Metadata
322366

323-
return v1Req, nil
367+
return v1Req, badRequestErrs.err()
324368
}
325369

326370
func shouldConvertV2Metadata(metadata cortexpb.MetadataV2) bool {

0 commit comments

Comments
 (0)