diff --git a/CHANGELOG.md b/CHANGELOG.md index abd8bcc792..071d1e658a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/integration/remote_write_v2_test.go b/integration/remote_write_v2_test.go index f1455848b8..ea9927cee7 100644 --- a/integration/remote_write_v2_test.go +++ b/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) { diff --git a/pkg/util/push/push.go b/pkg/util/push/push.go index 7f2ae9dd16..93c2bb1ab5 100644 --- a/pkg/util/push/push.go +++ b/pkg/util/push/push.go @@ -115,11 +115,11 @@ 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 @@ -127,7 +127,10 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, 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 { diff --git a/pkg/util/push/push_test.go b/pkg/util/push/push_test.go index 6ad88c9589..c7faac28c5 100644 --- a/pkg/util/push/push_test.go +++ b/pkg/util/push/push_test.go @@ -6,6 +6,8 @@ import ( "fmt" "net/http" "net/http/httptest" + "strconv" + "strings" "testing" "time" @@ -555,6 +557,14 @@ func Test_convertV2RequestToV1(t *testing.T) { Exemplars: []cortexpb.ExemplarV2{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: 1}}, }, }, + { + // The Prometheus sender emits exemplar only TimeSeries, see + // https://github.com/prometheus/prometheus/issues/17857 + TimeSeriesV2: &cortexpb.TimeSeriesV2{ + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + Exemplars: []cortexpb.ExemplarV2{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: 1}}, + }, + }, } v2Req.Symbols = symbols @@ -562,7 +572,7 @@ func Test_convertV2RequestToV1(t *testing.T) { v1Req, err := convertV2RequestToV1(&v2Req, false, false) assert.NoError(t, err) expectedSamples := 3 - expectedExemplars := 2 + expectedExemplars := 3 expectedHistograms := 2 countSamples := 0 countExemplars := 0 @@ -577,7 +587,7 @@ func Test_convertV2RequestToV1(t *testing.T) { assert.Equal(t, expectedSamples, countSamples) assert.Equal(t, expectedExemplars, countExemplars) assert.Equal(t, expectedHistograms, countHistograms) - assert.Equal(t, 4, len(v1Req.Timeseries)) + assert.Equal(t, 5, len(v1Req.Timeseries)) assert.Equal(t, 1, len(v1Req.Metadata)) } @@ -753,9 +763,8 @@ func TestHandler_remoteWrite(t *testing.T) { }, }, { - name: "remote write v2 with empty samples and histograms should return 400", + name: "remote write v2 with an exemplar only series is accepted", createBody: func() ([]byte, bool) { - // Create a request with a TimeSeries that has no samples and no histograms reqProto := writev2.Request{ Symbols: []string{"", "__name__", "foo"}, Timeseries: []writev2.TimeSeries{ @@ -775,8 +784,13 @@ func TestHandler_remoteWrite(t *testing.T) { require.NoError(t, err) return reqBytes, true }, - expectedStatus: http.StatusBadRequest, - expectedBody: "TimeSeries must contain at least one sample or histogram for series {__name__=\"foo\"}", + expectedStatus: http.StatusNoContent, + verifyResponse: func(resp *httptest.ResponseRecorder) { + respHeader := resp.Header() + assert.Equal(t, "1", respHeader[rw20WrittenSamplesHeader][0]) + assert.Equal(t, "1", respHeader[rw20WrittenHistogramsHeader][0]) + assert.Equal(t, "1", respHeader[rw20WrittenExemplarsHeader][0]) + }, }, { name: "remote write v1 with oversized histogram returns 400", @@ -1550,3 +1564,164 @@ func TestHandler_remoteWriteV2_UnauthorizedWithoutTenantID(t *testing.T) { assert.Contains(t, resp.Body.String(), user.ErrNoOrgID.Error()) assert.False(t, pushCalled, "push function must not be called when tenant ID is missing") } + +func Test_convertV2RequestToV1_ExemplarOnlySeries(t *testing.T) { + var v2Req cortexpb.PreallocWriteRequestV2 + v2Req.Symbols = []string{"", "__name__", "test_metric", "trace_id", "abc123"} + v2Req.Timeseries = []cortexpb.PreallocTimeseriesV2{ + { + TimeSeriesV2: &cortexpb.TimeSeriesV2{ + LabelsRefs: []uint32{1, 2}, + Exemplars: []cortexpb.ExemplarV2{{LabelsRefs: []uint32{3, 4}, Value: 1.5, Timestamp: 20}}, + }, + }, + } + + v1Req, err := convertV2RequestToV1(&v2Req, false, false) + require.NoError(t, err) + require.Len(t, v1Req.Timeseries, 1) + + ts := v1Req.Timeseries[0] + assert.Empty(t, ts.Samples) + assert.Empty(t, ts.Histograms) + assert.Equal(t, []cortexpb.LabelAdapter{{Name: "__name__", Value: "test_metric"}}, ts.Labels) + require.Len(t, ts.Exemplars, 1) + assert.Equal(t, []cortexpb.LabelAdapter{{Name: "trace_id", Value: "abc123"}}, ts.Exemplars[0].Labels) + assert.Equal(t, 1.5, ts.Exemplars[0].Value) + assert.Equal(t, int64(20), ts.Exemplars[0].TimestampMs) +} + +func TestHandler_remoteWriteV2_PartialWrite(t *testing.T) { + var limits validation.Limits + flagext.DefaultValues(&limits) + overrides := validation.NewOverrides(limits, nil) + + // Symbol 5 is out of range for a symbols table of length 5. + const invalidRef = 5 + + tests := []struct { + name string + // pushedSeries lists the metric names expected to reach the Distributor, in order. + pushedSeries []string + timeseries []writev2.TimeSeries + expectedErrs []string + expectedStats []string + }{ + { + name: "a series holding no data at all is skipped, the rest of the batch is pushed", + timeseries: []writev2.TimeSeries{ + {LabelsRefs: []uint32{1, 2}}, + {LabelsRefs: []uint32{1, 3}, Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}}, + }, + pushedSeries: []string{"bar"}, + expectedErrs: []string{`TimeSeries must contain at least one sample, histogram or exemplar for series {__name__="foo"}`}, + }, + { + name: "multiple invalid series produce a joined error", + timeseries: []writev2.TimeSeries{ + {LabelsRefs: []uint32{1, 2}}, + {LabelsRefs: []uint32{1, 3}, Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}}, + {LabelsRefs: []uint32{1, 4}}, + }, + pushedSeries: []string{"bar"}, + expectedErrs: []string{ + `TimeSeries must contain at least one sample, histogram or exemplar for series {__name__="foo"}`, + `TimeSeries must contain at least one sample, histogram or exemplar for series {__name__="baz"}`, + }, + }, + { + name: "every series is invalid, push is still called with an empty request", + timeseries: []writev2.TimeSeries{ + {LabelsRefs: []uint32{1, 2}}, + }, + pushedSeries: nil, + expectedErrs: []string{`TimeSeries must contain at least one sample, histogram or exemplar for series {__name__="foo"}`}, + }, + { + name: "a series with an out of range label ref is skipped", + timeseries: []writev2.TimeSeries{ + {LabelsRefs: []uint32{1, invalidRef}, Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}}, + {LabelsRefs: []uint32{1, 3}, Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}}, + }, + pushedSeries: []string{"bar"}, + expectedErrs: []string{"outside of symbols table"}, + }, + { + name: "a series with an out of range exemplar label ref is skipped", + timeseries: []writev2.TimeSeries{ + { + LabelsRefs: []uint32{1, 2}, + Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}, + Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{1, invalidRef}, Value: 1, Timestamp: 10}}, + }, + {LabelsRefs: []uint32{1, 3}, Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}}, + }, + pushedSeries: []string{"bar"}, + expectedErrs: []string{"outside of symbols table"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var pushed []string + pushCalled := false + pushFunc := func(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) { + pushCalled = true + for _, ts := range req.Timeseries { + pushed = append(pushed, ts.Labels[0].Value) + } + return &cortexpb.WriteResponse{Samples: int64(len(req.Timeseries))}, nil + } + + reqProto := writev2.Request{ + Symbols: []string{"", "__name__", "foo", "bar", "baz"}, + Timeseries: test.timeseries, + } + reqBytes, err := reqProto.Marshal() + require.NoError(t, err) + + handler := Handler(true, false, 100000, overrides, nil, pushFunc, nil) + httpReq := createRequest(t, reqBytes, true).WithContext(user.InjectOrgID(context.Background(), "user-1")) + + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, httpReq) + + assert.Equal(t, http.StatusBadRequest, resp.Code) + for _, expectedErr := range test.expectedErrs { + assert.Contains(t, resp.Body.String(), expectedErr) + } + + // The valid part of the request must still be written. + assert.True(t, pushCalled, "push function must always be called") + assert.Equal(t, test.pushedSeries, pushed) + + // The written stats headers must be set even when responding with a 400. + respHeader := resp.Header() + assert.Equal(t, strconv.Itoa(len(test.pushedSeries)), respHeader.Get(rw20WrittenSamplesHeader)) + assert.Equal(t, "0", respHeader.Get(rw20WrittenHistogramsHeader)) + assert.Equal(t, "0", respHeader.Get(rw20WrittenExemplarsHeader)) + }) + } +} + +func Test_convertV2RequestToV1_BoundsReportedErrors(t *testing.T) { + const badSeries = maxConversionErrs + 5 + + var v2Req cortexpb.PreallocWriteRequestV2 + v2Req.Symbols = []string{"", "__name__", "test_metric"} + for range badSeries { + // A series holding no data at all. + v2Req.Timeseries = append(v2Req.Timeseries, cortexpb.PreallocTimeseriesV2{ + TimeSeriesV2: &cortexpb.TimeSeriesV2{LabelsRefs: []uint32{1, 2}}, + }) + } + + v1Req, err := convertV2RequestToV1(&v2Req, false, false) + require.Error(t, err) + assert.Empty(t, v1Req.Timeseries) + + // Only maxConversionErrs errors are reported, the rest are summarized. + lines := strings.Split(err.Error(), "\n") + assert.Len(t, lines, maxConversionErrs+1) + assert.Equal(t, fmt.Sprintf("%d more errors omitted", badSeries-maxConversionErrs), lines[len(lines)-1]) +}