From 74d6d0968aa98d8b6e1fe639d732587ef3ae0b9e Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Mon, 10 Mar 2025 11:02:12 +0900 Subject: [PATCH 1/4] Support remote write v2 by converting request Signed-off-by: SungJin1212 --- .github/workflows/test-build-deploy.yml | 1 + .golangci.yml | 1 + CHANGELOG.md | 1 + docs/configuration/config-file-reference.md | 5 + docs/configuration/v1-guarantees.md | 1 + integration/e2e/util.go | 76 +++++ integration/e2ecortex/client.go | 34 ++ integration/remote_write_v2_test.go | 327 ++++++++++++++++++++ pkg/api/api.go | 8 +- pkg/cortexpb/cortex.proto | 8 + pkg/cortexpb/histograms.go | 34 ++ pkg/distributor/distributor.go | 30 +- pkg/distributor/write_stats.go | 62 ++++ pkg/distributor/write_stats_test.go | 41 +++ pkg/ingester/ingester.go | 8 +- pkg/util/push/push.go | 266 +++++++++++++++- pkg/util/push/push_test.go | 314 +++++++++++++++++-- 17 files changed, 1170 insertions(+), 47 deletions(-) create mode 100644 integration/remote_write_v2_test.go create mode 100644 pkg/distributor/write_stats.go create mode 100644 pkg/distributor/write_stats_test.go diff --git a/.github/workflows/test-build-deploy.yml b/.github/workflows/test-build-deploy.yml index c082e73d5d9..34b9527f664 100644 --- a/.github/workflows/test-build-deploy.yml +++ b/.github/workflows/test-build-deploy.yml @@ -162,6 +162,7 @@ jobs: - integration_querier - integration_ruler - integration_query_fuzz + - integration_remote_write_v2 steps: - name: Upgrade golang uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0 diff --git a/.golangci.yml b/.golangci.yml index dd0398764be..e9bad40696a 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -51,3 +51,4 @@ run: - integration_querier - integration_ruler - integration_query_fuzz + - integration_remote_write_v2 diff --git a/CHANGELOG.md b/CHANGELOG.md index 524121bd486..fef61bf0b9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ * [FEATURE] Querier/Ruler: Add `query_partial_data` and `rules_partial_data` limits to allow queries/rules to be evaluated with data from a single zone, if other zones are not available. #6526 * [FEATURE] Update prometheus alertmanager version to v0.28.0 and add new integration msteamsv2, jira, and rocketchat. #6590 * [FEATURE] Ingester/StoreGateway: Add `ResourceMonitor` module in Cortex, and add `ResourceBasedLimiter` in Ingesters and StoreGateways. #6674 +* [FEATURE] Support Prometheus remote write 2.0. #6330 * [FEATURE] Ingester: Support out-of-order native histogram ingestion. It automatically enabled when `-ingester.out-of-order-time-window > 0` and `-blocks-storage.tsdb.enable-native-histograms=true`. #6626 #6663 * [FEATURE] Ruler: Add support for percentage based sharding for rulers. #6680 * [FEATURE] Ruler: Add support for group labels. #6665 diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 292c67970c9..3ebc69d9435 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -2690,6 +2690,11 @@ ha_tracker: # CLI flag: -distributor.sign-write-requests [sign_write_requests: | default = false] +# EXPERIMENTAL: If true, accept prometheus remote write v2 protocol push +# request. +# CLI flag: -distributor.remote-write2-enabled +[remote_write2_enabled: | default = false] + # EXPERIMENTAL: If enabled, distributor would use stream connection to send # requests to ingesters. # CLI flag: -distributor.use-stream-push diff --git a/docs/configuration/v1-guarantees.md b/docs/configuration/v1-guarantees.md index 77a66e4d293..a8cfb7bc891 100644 --- a/docs/configuration/v1-guarantees.md +++ b/docs/configuration/v1-guarantees.md @@ -59,6 +59,7 @@ Currently experimental features are: - Distributor: - Do not extend writes on unhealthy ingesters (`-distributor.extend-writes=false`) - Accept multiple HA pairs in the same request (enabled via `-experimental.distributor.ha-tracker.mixed-ha-samples=true`) + - Accept Prometheus remote write 2.0 request (`-distributor.remote-write2-enabled=true`) - Tenant Deletion in Purger, for blocks storage. - Query-frontend: query stats tracking (`-frontend.query-stats-enabled`) - Blocks storage bucket index diff --git a/integration/e2e/util.go b/integration/e2e/util.go index 1f0cb7a5707..793e4150e99 100644 --- a/integration/e2e/util.go +++ b/integration/e2e/util.go @@ -19,6 +19,7 @@ import ( "github.com/prometheus/prometheus/model/histogram" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/prompb" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/tsdb" "github.com/prometheus/prometheus/tsdb/tsdbutil" @@ -423,3 +424,78 @@ func CreateBlock( return id, nil } + +func GenerateHistogramSeriesV2(name string, ts time.Time, i uint32, floatHistogram bool, additionalLabels ...prompb.Label) (symbols []string, series []writev2.TimeSeries) { + tsMillis := TimeToMilliseconds(ts) + + st := writev2.NewSymbolTable() + + lbs := labels.Labels{labels.Label{Name: "__name__", Value: name}} + for _, lbl := range additionalLabels { + lbs = append(lbs, labels.Label{Name: lbl.Name, Value: lbl.Value}) + } + + var ( + h *histogram.Histogram + fh *histogram.FloatHistogram + ph writev2.Histogram + ) + if floatHistogram { + fh = tsdbutil.GenerateTestFloatHistogram(int64(i)) + ph = writev2.FromFloatHistogram(tsMillis, fh) + } else { + h = tsdbutil.GenerateTestHistogram(int64(i)) + ph = writev2.FromIntHistogram(tsMillis, h) + } + + // Generate the series + series = append(series, writev2.TimeSeries{ + LabelsRefs: st.SymbolizeLabels(lbs, nil), + Histograms: []writev2.Histogram{ph}, + }) + + symbols = st.Symbols() + + return +} + +func GenerateSeriesV2(name string, ts time.Time, additionalLabels ...prompb.Label) (symbols []string, series []writev2.TimeSeries, vector model.Vector) { + tsMillis := TimeToMilliseconds(ts) + value := rand.Float64() + + st := writev2.NewSymbolTable() + lbs := labels.Labels{{Name: labels.MetricName, Value: name}} + + for _, label := range additionalLabels { + lbs = append(lbs, labels.Label{ + Name: label.Name, + Value: label.Value, + }) + } + series = append(series, writev2.TimeSeries{ + // Generate the series + LabelsRefs: st.SymbolizeLabels(lbs, nil), + Samples: []writev2.Sample{ + {Value: value, Timestamp: tsMillis}, + }, + Metadata: writev2.Metadata{ + Type: writev2.Metadata_METRIC_TYPE_GAUGE, + }, + }) + symbols = st.Symbols() + + // Generate the expected vector when querying it + metric := model.Metric{} + metric[labels.MetricName] = model.LabelValue(name) + for _, lbl := range additionalLabels { + metric[model.LabelName(lbl.Name)] = model.LabelValue(lbl.Value) + } + + vector = append(vector, &model.Sample{ + Metric: metric, + Value: model.SampleValue(value), + Timestamp: model.Time(tsMillis), + }) + + return +} diff --git a/integration/e2ecortex/client.go b/integration/e2ecortex/client.go index 9067b60c078..a6d7184dab3 100644 --- a/integration/e2ecortex/client.go +++ b/integration/e2ecortex/client.go @@ -24,6 +24,7 @@ import ( "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/model/rulefmt" "github.com/prometheus/prometheus/prompb" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/storage/remote" yaml "gopkg.in/yaml.v3" @@ -147,6 +148,39 @@ func (c *Client) Push(timeseries []prompb.TimeSeries, metadata ...prompb.MetricM return res, nil } +// PushV2 the input timeseries to the remote endpoint +func (c *Client) PushV2(symbols []string, timeseries []writev2.TimeSeries) (*http.Response, error) { + // Create write request + data, err := proto.Marshal(&writev2.Request{Symbols: symbols, Timeseries: timeseries}) + if err != nil { + return nil, err + } + + // Create HTTP request + compressed := snappy.Encode(nil, data) + req, err := http.NewRequest("POST", fmt.Sprintf("http://%s/api/prom/push", c.distributorAddress), bytes.NewReader(compressed)) + if err != nil { + return nil, err + } + + req.Header.Add("Content-Encoding", "snappy") + req.Header.Set("Content-Type", "application/x-protobuf;proto=io.prometheus.write.v2.Request") + req.Header.Set("X-Prometheus-Remote-Write-Version", "2.0.0") + req.Header.Set("X-Scope-OrgID", c.orgID) + + ctx, cancel := context.WithTimeout(context.Background(), c.timeout) + defer cancel() + + // Execute HTTP request + res, err := c.httpClient.Do(req.WithContext(ctx)) + if err != nil { + return nil, err + } + + defer res.Body.Close() + return res, nil +} + func getNameAndAttributes(ts prompb.TimeSeries) (string, map[string]any) { var metricName string attributes := make(map[string]any) diff --git a/integration/remote_write_v2_test.go b/integration/remote_write_v2_test.go new file mode 100644 index 00000000000..88ca384fb04 --- /dev/null +++ b/integration/remote_write_v2_test.go @@ -0,0 +1,327 @@ +//go:build integration_remote_write_v2 +// +build integration_remote_write_v2 + +package integration + +import ( + "math/rand" + "net/http" + "path" + "testing" + "time" + + "github.com/prometheus/common/model" + "github.com/prometheus/prometheus/prompb" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" + "github.com/prometheus/prometheus/tsdb/tsdbutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/cortexproject/cortex/integration/e2e" + e2edb "github.com/cortexproject/cortex/integration/e2e/db" + "github.com/cortexproject/cortex/integration/e2ecortex" + "github.com/cortexproject/cortex/pkg/storage/tsdb" +) + +func TestIngesterRollingUpdate(t *testing.T) { + // Test ingester rolling update situation: when -distributor.remote-write2-enabled is true, and ingester uses the v1.19.0 image. + // Expected: remote write 2.0 push success, but response header values are set to "0". + const blockRangePeriod = 5 * time.Second + ingesterImage := "quay.io/cortexproject/cortex:v1.19.0" + + 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", + "-querier.query-store-for-labels-enabled": "true", + "-blocks-storage.tsdb.block-ranges-period": blockRangePeriod.String(), + "-blocks-storage.tsdb.ship-interval": "1s", + "-blocks-storage.tsdb.retention-period": ((blockRangePeriod * 2) - 1).String(), + "-blocks-storage.tsdb.enable-native-histograms": "true", + // Ingester. + "-ring.store": "consul", + "-consul.hostname": consul.NetworkHTTPEndpoint(), + // Distributor. + "-distributor.replication-factor": "1", + // Store-gateway. + "-store-gateway.sharding-enabled": "false", + // alert manager + "-alertmanager.web.external-url": "http://localhost/alertmanager", + }, + ) + + distributorFlag := mergeFlags(flags, map[string]string{ + "-distributor.remote-write2-enabled": "true", + }) + + // 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. + // Start all other services. + ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, ingesterImage) + distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), distributorFlag, "") + storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{ + "-querier.store-gateway-addresses": storeGateway.NetworkGRPCEndpoint()}), "") + + require.NoError(t, s.StartAndWaitReady(querier, ingester, distributor, storeGateway)) + + // Wait until Cortex replicas have updated the ring state. + require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + + c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", "user-1") + require.NoError(t, err) + + now := time.Now() + + // series push + symbols1, series, expectedVector := e2e.GenerateSeriesV2("test_series", now, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "foo", Value: "bar"}) + res, err := c.PushV2(symbols1, series) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "0", "0", "0") + + // sample + result, err := c.Query("test_series", now) + require.NoError(t, err) + assert.Equal(t, expectedVector, result.(model.Vector)) + + // metadata + metadata, err := c.Metadata("test_series", "") + require.NoError(t, err) + require.Equal(t, 1, len(metadata["test_series"])) + + // histogram + histogramIdx := rand.Uint32() + symbols2, histogramSeries := e2e.GenerateHistogramSeriesV2("test_histogram", now, histogramIdx, false, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "float", Value: "false"}) + res, err = c.PushV2(symbols2, histogramSeries) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "0", "0", "0") + + symbols3, histogramFloatSeries := e2e.GenerateHistogramSeriesV2("test_histogram", now, histogramIdx, false, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "float", Value: "true"}) + res, err = c.PushV2(symbols3, histogramFloatSeries) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "0", "0", "0") + + testHistogramTimestamp := now.Add(blockRangePeriod * 2) + expectedHistogram := tsdbutil.GenerateTestHistogram(int64(histogramIdx)) + result, err = c.Query(`test_histogram`, testHistogramTimestamp) + require.NoError(t, err) + require.Equal(t, model.ValVector, result.Type()) + v := result.(model.Vector) + require.Equal(t, 2, v.Len()) + for _, s := range v { + require.NotNil(t, s.Histogram) + require.Equal(t, float64(expectedHistogram.Count), float64(s.Histogram.Count)) + require.Equal(t, float64(expectedHistogram.Sum), float64(s.Histogram.Sum)) + } +} + +func TestIngest(t *testing.T) { + const blockRangePeriod = 5 * time.Second + + 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", + "-querier.query-store-for-labels-enabled": "true", + "-blocks-storage.tsdb.block-ranges-period": blockRangePeriod.String(), + "-blocks-storage.tsdb.ship-interval": "1s", + "-blocks-storage.tsdb.retention-period": ((blockRangePeriod * 2) - 1).String(), + "-blocks-storage.tsdb.enable-native-histograms": "true", + // Ingester. + "-ring.store": "consul", + "-consul.hostname": consul.NetworkHTTPEndpoint(), + // Distributor. + "-distributor.replication-factor": "1", + "-distributor.remote-write2-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() + + // series push + symbols1, series, expectedVector := e2e.GenerateSeriesV2("test_series", now, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "foo", Value: "bar"}) + res, err := c.PushV2(symbols1, series) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "1", "0", "0") + + // sample + result, err := c.Query("test_series", now) + require.NoError(t, err) + assert.Equal(t, expectedVector, result.(model.Vector)) + + // metadata + metadata, err := c.Metadata("test_series", "") + require.NoError(t, err) + require.Equal(t, 1, len(metadata["test_series"])) + + // histogram + histogramIdx := rand.Uint32() + symbols2, histogramSeries := e2e.GenerateHistogramSeriesV2("test_histogram", now, histogramIdx, false, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "float", Value: "false"}) + res, err = c.PushV2(symbols2, histogramSeries) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "0", "1", "0") + + symbols3, histogramFloatSeries := e2e.GenerateHistogramSeriesV2("test_histogram", now, histogramIdx, false, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "float", Value: "true"}) + res, err = c.PushV2(symbols3, histogramFloatSeries) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "0", "1", "0") + + testHistogramTimestamp := now.Add(blockRangePeriod * 2) + expectedHistogram := tsdbutil.GenerateTestHistogram(int64(histogramIdx)) + result, err = c.Query(`test_histogram`, testHistogramTimestamp) + require.NoError(t, err) + require.Equal(t, model.ValVector, result.Type()) + v := result.(model.Vector) + require.Equal(t, 2, v.Len()) + for _, s := range v { + require.NotNil(t, s.Histogram) + require.Equal(t, float64(expectedHistogram.Count), float64(s.Histogram.Count)) + require.Equal(t, float64(expectedHistogram.Sum), float64(s.Histogram.Sum)) + } +} + +func TestExemplar(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", + "-querier.query-store-for-labels-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-write2-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) + + symbols := []string{"", "__name__", "test_metric", "b", "c", "baz", "qux", "d", "e", "foo", "bar", "f", "g", "h", "i", "Test gauge for test purposes", "Maybe op/sec who knows (:", "Test counter for test purposes"} + timeseries := []writev2.TimeSeries{ + { + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, // Symbolized writeRequestFixture.Timeseries[0].Labels + Metadata: writev2.Metadata{ + Type: writev2.Metadata_METRIC_TYPE_COUNTER, // writeV2RequestSeries1Metadata.Type. + + HelpRef: 15, // Symbolized writeV2RequestSeries1Metadata.Help. + UnitRef: 16, // Symbolized writeV2RequestSeries1Metadata.Unit. + }, + Samples: []writev2.Sample{{Value: 1, Timestamp: tsMillis}}, + Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: tsMillis}}, + }, + } + + res, err := c.PushV2(symbols, timeseries) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + testPushHeader(t, res.Header, "1", "0", "1") + + start := time.Now().Add(-time.Minute) + end := now.Add(time.Minute) + + exemplars, err := c.QueryExemplars("test_metric", start, end) + require.NoError(t, err) + require.Equal(t, 1, len(exemplars)) +} + +func testPushHeader(t *testing.T, header http.Header, expectedSamples, expectedHistogram, expectedExemplars string) { + require.Equal(t, expectedSamples, header.Get("X-Prometheus-Remote-Write-Samples-Written")) + require.Equal(t, expectedHistogram, header.Get("X-Prometheus-Remote-Write-Histograms-Written")) + require.Equal(t, expectedExemplars, header.Get("X-Prometheus-Remote-Write-Exemplars-Written")) +} diff --git a/pkg/api/api.go b/pkg/api/api.go index 13843c3e64a..c8c9390f481 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -277,7 +277,7 @@ func (a *API) RegisterRuntimeConfig(runtimeConfigHandler http.HandlerFunc) { func (a *API) RegisterDistributor(d *distributor.Distributor, pushConfig distributor.Config, overrides *validation.Overrides) { distributorpb.RegisterDistributorServer(a.server.GRPC, d) - a.RegisterRoute("/api/v1/push", push.Handler(pushConfig.MaxRecvMsgSize, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST") + a.RegisterRoute("/api/v1/push", push.Handler(pushConfig.RemoteWrite2Enabled, pushConfig.MaxRecvMsgSize, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST") a.RegisterRoute("/api/v1/otlp/v1/metrics", push.OTLPHandler(pushConfig.OTLPMaxRecvMsgSize, overrides, pushConfig.OTLPConfig, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST") a.indexPage.AddLink(SectionAdminEndpoints, "/distributor/ring", "Distributor Ring Status") @@ -289,7 +289,7 @@ func (a *API) RegisterDistributor(d *distributor.Distributor, pushConfig distrib a.RegisterRoute("/distributor/ha_tracker", d.HATracker, false, "GET") // Legacy Routes - a.RegisterRoute(path.Join(a.cfg.LegacyHTTPPrefix, "/push"), push.Handler(pushConfig.MaxRecvMsgSize, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST") + a.RegisterRoute(path.Join(a.cfg.LegacyHTTPPrefix, "/push"), push.Handler(pushConfig.RemoteWrite2Enabled, pushConfig.MaxRecvMsgSize, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST") a.RegisterRoute("/all_user_stats", http.HandlerFunc(d.AllUserStatsHandler), false, "GET") a.RegisterRoute("/ha-tracker", d.HATracker, false, "GET") } @@ -322,12 +322,12 @@ func (a *API) RegisterIngester(i Ingester, pushConfig distributor.Config) { a.RegisterRoute("/ingester/renewTokens", http.HandlerFunc(i.RenewTokenHandler), false, "GET", "POST") a.RegisterRoute("/ingester/all_user_stats", http.HandlerFunc(i.AllUserStatsHandler), false, "GET") a.RegisterRoute("/ingester/mode", http.HandlerFunc(i.ModeHandler), false, "GET", "POST") - a.RegisterRoute("/ingester/push", push.Handler(pushConfig.MaxRecvMsgSize, a.sourceIPs, i.Push), true, "POST") // For testing and debugging. + a.RegisterRoute("/ingester/push", push.Handler(pushConfig.RemoteWrite2Enabled, pushConfig.MaxRecvMsgSize, a.sourceIPs, i.Push), true, "POST") // For testing and debugging. // Legacy Routes a.RegisterRoute("/flush", http.HandlerFunc(i.FlushHandler), false, "GET", "POST") a.RegisterRoute("/shutdown", http.HandlerFunc(i.ShutdownHandler), false, "GET", "POST") - a.RegisterRoute("/push", push.Handler(pushConfig.MaxRecvMsgSize, a.sourceIPs, i.Push), true, "POST") // For testing and debugging. + a.RegisterRoute("/push", push.Handler(pushConfig.RemoteWrite2Enabled, pushConfig.MaxRecvMsgSize, a.sourceIPs, i.Push), true, "POST") // For testing and debugging. } func (a *API) RegisterTenantDeletion(api *purger.TenantDeletionAPI) { diff --git a/pkg/cortexpb/cortex.proto b/pkg/cortexpb/cortex.proto index 91773fbc0f1..189e8955f7c 100644 --- a/pkg/cortexpb/cortex.proto +++ b/pkg/cortexpb/cortex.proto @@ -30,6 +30,14 @@ message WriteResponse { int32 code = 1; string message = 2; } +message WriteResponse { + // Samples represents X-Prometheus-Remote-Write-Written-Samples + int64 Samples = 1; + // Histograms represents X-Prometheus-Remote-Write-Written-Histograms + int64 Histograms = 2; + // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars + int64 Exemplars = 3; +} message TimeSeries { repeated LabelPair labels = 1 [(gogoproto.nullable) = false, (gogoproto.customtype) = "LabelAdapter"]; diff --git a/pkg/cortexpb/histograms.go b/pkg/cortexpb/histograms.go index 60e7207a19a..d05dbaa7727 100644 --- a/pkg/cortexpb/histograms.go +++ b/pkg/cortexpb/histograms.go @@ -16,6 +16,7 @@ package cortexpb import ( "github.com/prometheus/prometheus/model/histogram" "github.com/prometheus/prometheus/prompb" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" ) func (h Histogram) IsFloatHistogram() bool { @@ -23,6 +24,30 @@ func (h Histogram) IsFloatHistogram() bool { return ok } +func HistogramWriteV2ProtoToHistogramProto(h writev2.Histogram) Histogram { + ph := Histogram{ + Sum: h.Sum, + Schema: h.Schema, + ZeroThreshold: h.ZeroThreshold, + NegativeSpans: spansWriteV2ProtoToSpansProto(h.NegativeSpans), + NegativeDeltas: h.NegativeDeltas, + NegativeCounts: h.NegativeCounts, + PositiveSpans: spansWriteV2ProtoToSpansProto(h.PositiveSpans), + PositiveDeltas: h.PositiveDeltas, + PositiveCounts: h.PositiveCounts, + ResetHint: Histogram_ResetHint(h.ResetHint), + TimestampMs: h.Timestamp, + } + if h.IsFloatHistogram() { + ph.Count = &Histogram_CountFloat{CountFloat: h.GetCountFloat()} + ph.ZeroCount = &Histogram_ZeroCountFloat{ZeroCountFloat: h.GetZeroCountFloat()} + } else { + ph.Count = &Histogram_CountInt{CountInt: h.GetCountInt()} + ph.ZeroCount = &Histogram_ZeroCountInt{ZeroCountInt: h.GetZeroCountInt()} + } + return ph +} + // HistogramPromProtoToHistogramProto converts a prometheus protobuf Histogram to cortex protobuf Histogram. func HistogramPromProtoToHistogramProto(h prompb.Histogram) Histogram { ph := Histogram{ @@ -155,3 +180,12 @@ func spansPromProtoToSpansProto(s []prompb.BucketSpan) []BucketSpan { return spans } + +func spansWriteV2ProtoToSpansProto(s []writev2.BucketSpan) []BucketSpan { + spans := make([]BucketSpan, len(s)) + for i := 0; i < len(s); i++ { + spans[i] = BucketSpan{Offset: s[i].Offset, Length: s[i].Length} + } + + return spans +} diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go index 734a5fa0343..68dfc1fe519 100644 --- a/pkg/distributor/distributor.go +++ b/pkg/distributor/distributor.go @@ -152,6 +152,7 @@ type Config struct { ExtendWrites bool `yaml:"extend_writes"` SignWriteRequestsEnabled bool `yaml:"sign_write_requests"` UseStreamPush bool `yaml:"use_stream_push"` + RemoteWrite2Enabled bool `yaml:"remote_write2_enabled"` // Distributors ring DistributorRing RingConfig `yaml:"ring"` @@ -211,6 +212,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) { f.BoolVar(&cfg.ExtendWrites, "distributor.extend-writes", true, "Try writing to an additional ingester in the presence of an ingester not in the ACTIVE state. It is useful to disable this along with -ingester.unregister-on-shutdown=false in order to not spread samples to extra ingesters during rolling restarts with consistent naming.") f.BoolVar(&cfg.ZoneResultsQuorumMetadata, "distributor.zone-results-quorum-metadata", false, "Experimental, this flag may change in the future. If zone awareness and this both enabled, when querying metadata APIs (labels names and values for now), only results from quorum number of zones will be included.") f.IntVar(&cfg.NumPushWorkers, "distributor.num-push-workers", 0, "EXPERIMENTAL: Number of go routines to handle push calls from distributors to ingesters. When no workers are available, a new goroutine will be spawned automatically. If set to 0 (default), workers are disabled, and a new goroutine will be created for each push request.") + f.BoolVar(&cfg.RemoteWrite2Enabled, "distributor.remote-write2-enabled", false, "EXPERIMENTAL: If true, accept prometheus remote write v2 protocol push request.") f.Float64Var(&cfg.InstanceLimits.MaxIngestionRate, "distributor.instance-limits.max-ingestion-rate", 0, "Max ingestion rate (samples/sec) that this distributor will accept. This limit is per-distributor, not per-tenant. Additional push requests will be rejected. Current ingestion rate is computed as exponentially weighted moving average, updated every second. 0 = unlimited.") f.IntVar(&cfg.InstanceLimits.MaxInflightPushRequests, "distributor.instance-limits.max-inflight-push-requests", 0, "Max inflight push requests that this distributor can handle. This limit is per-distributor, not per-tenant. Additional requests will be rejected. 0 = unlimited.") @@ -801,12 +803,21 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co keys := append(seriesKeys, metadataKeys...) initialMetadataIndex := len(seriesKeys) - err = d.doBatch(ctx, req, subRing, keys, initialMetadataIndex, validatedMetadata, validatedTimeseries, userID) + ws := WriteStats{} + + err = d.doBatch(ctx, req, subRing, keys, initialMetadataIndex, validatedMetadata, validatedTimeseries, userID, &ws) if err != nil { return nil, err } - return &cortexpb.WriteResponse{}, firstPartialErr + resp := &cortexpb.WriteResponse{} + if d.cfg.RemoteWrite2Enabled { + resp.Samples = ws.LoadSamples() + resp.Histograms = ws.LoadHistogram() + resp.Exemplars = ws.LoadExemplars() + } + + return resp, firstPartialErr } func (d *Distributor) updateLabelSetMetrics() { @@ -868,7 +879,7 @@ func (d *Distributor) cleanStaleIngesterMetrics() { } } -func (d *Distributor) doBatch(ctx context.Context, req *cortexpb.WriteRequest, subRing ring.ReadRing, keys []uint32, initialMetadataIndex int, validatedMetadata []*cortexpb.MetricMetadata, validatedTimeseries []cortexpb.PreallocTimeseries, userID string) error { +func (d *Distributor) doBatch(ctx context.Context, req *cortexpb.WriteRequest, subRing ring.ReadRing, keys []uint32, initialMetadataIndex int, validatedMetadata []*cortexpb.MetricMetadata, validatedTimeseries []cortexpb.PreallocTimeseries, userID string, ws *WriteStats) error { span, _ := opentracing.StartSpanFromContext(ctx, "doBatch") defer span.Finish() @@ -903,7 +914,7 @@ func (d *Distributor) doBatch(ctx context.Context, req *cortexpb.WriteRequest, s } } - return d.send(localCtx, ingester, timeseries, metadata, req.Source) + return d.send(localCtx, ingester, timeseries, metadata, req.Source, ws) }, func() { cortexpb.ReuseSlice(req.Timeseries) cancel() @@ -1129,7 +1140,7 @@ func sortLabelsIfNeeded(labels []cortexpb.LabelAdapter) { }) } -func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, timeseries []cortexpb.PreallocTimeseries, metadata []*cortexpb.MetricMetadata, source cortexpb.WriteRequest_SourceEnum) error { +func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, timeseries []cortexpb.PreallocTimeseries, metadata []*cortexpb.MetricMetadata, source cortexpb.WriteRequest_SourceEnum, ws *WriteStats) error { h, err := d.ingesterPool.GetClientFor(ingester.Addr) if err != nil { return err @@ -1158,7 +1169,7 @@ func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, time req.Metadata = metadata req.Source = source - _, err = c.PushPreAlloc(ctx, req) + resp, err = c.PushPreAlloc(ctx, req) // We should not reuse the req in case of errors: // See: https://github.com/grpc/grpc-go/issues/6355 @@ -1180,6 +1191,13 @@ func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, time } } + if resp != nil { + // track write stats + ws.SetSamples(resp.Samples) + ws.SetHistograms(resp.Histograms) + ws.SetExemplars(resp.Exemplars) + } + return err } diff --git a/pkg/distributor/write_stats.go b/pkg/distributor/write_stats.go new file mode 100644 index 00000000000..0f7fbc332d0 --- /dev/null +++ b/pkg/distributor/write_stats.go @@ -0,0 +1,62 @@ +package distributor + +import ( + "go.uber.org/atomic" +) + +type WriteStats struct { + // Samples represents X-Prometheus-Remote-Write-Written-Samples + Samples atomic.Int64 + // Histograms represents X-Prometheus-Remote-Write-Written-Histograms + Histograms atomic.Int64 + // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars + Exemplars atomic.Int64 +} + +func (w *WriteStats) SetSamples(samples int64) { + if w == nil { + return + } + + w.Samples.Store(samples) +} + +func (w *WriteStats) SetHistograms(histograms int64) { + if w == nil { + return + } + + w.Histograms.Store(histograms) +} + +func (w *WriteStats) SetExemplars(exemplars int64) { + if w == nil { + return + } + + w.Exemplars.Store(exemplars) +} + +func (w *WriteStats) LoadSamples() int64 { + if w == nil { + return 0 + } + + return w.Samples.Load() +} + +func (w *WriteStats) LoadHistogram() int64 { + if w == nil { + return 0 + } + + return w.Histograms.Load() +} + +func (w *WriteStats) LoadExemplars() int64 { + if w == nil { + return 0 + } + + return w.Exemplars.Load() +} diff --git a/pkg/distributor/write_stats_test.go b/pkg/distributor/write_stats_test.go new file mode 100644 index 00000000000..523f16788fe --- /dev/null +++ b/pkg/distributor/write_stats_test.go @@ -0,0 +1,41 @@ +package distributor + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func Test_SetAndLoad(t *testing.T) { + ws := &WriteStats{} + + t.Run("Samples", func(t *testing.T) { + ws.SetSamples(3) + assert.Equal(t, int64(3), ws.LoadSamples()) + }) + t.Run("Histograms", func(t *testing.T) { + ws.SetHistograms(10) + assert.Equal(t, int64(10), ws.LoadHistogram()) + }) + t.Run("Exemplars", func(t *testing.T) { + ws.SetExemplars(2) + assert.Equal(t, int64(2), ws.LoadExemplars()) + }) +} + +func Test_NilReceiver(t *testing.T) { + var ws *WriteStats + + t.Run("Samples", func(t *testing.T) { + ws.SetSamples(3) + assert.Equal(t, int64(0), ws.LoadSamples()) + }) + t.Run("Histograms", func(t *testing.T) { + ws.SetHistograms(10) + assert.Equal(t, int64(0), ws.LoadHistogram()) + }) + t.Run("Exemplars", func(t *testing.T) { + ws.SetExemplars(2) + assert.Equal(t, int64(0), ws.LoadExemplars()) + }) +} diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 885daea5e67..cc5d73784bd 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -1533,7 +1533,13 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte return &cortexpb.WriteResponse{}, httpgrpc.Errorf(code, "%s", wrapWithUser(firstPartialErr, userID).Error()) } - return &cortexpb.WriteResponse{}, nil + writeResponse := &cortexpb.WriteResponse{ + Samples: int64(succeededSamplesCount), + Histograms: int64(succeededHistogramsCount), + Exemplars: int64(succeededExemplarsCount), + } + + return writeResponse, nil } func (i *Ingester) PushStream(srv client.Ingester_PushStreamServer) error { diff --git a/pkg/util/push/push.go b/pkg/util/push/push.go index 9cabb395228..730b3d46193 100644 --- a/pkg/util/push/push.go +++ b/pkg/util/push/push.go @@ -2,22 +2,45 @@ package push import ( "context" + "fmt" "net/http" + "strconv" + "strings" "github.com/go-kit/log/level" + "github.com/prometheus/prometheus/config" + "github.com/prometheus/prometheus/model/labels" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" + "github.com/prometheus/prometheus/storage/remote" "github.com/weaveworks/common/httpgrpc" "github.com/weaveworks/common/middleware" "github.com/cortexproject/cortex/pkg/cortexpb" "github.com/cortexproject/cortex/pkg/util" + "github.com/cortexproject/cortex/pkg/util/extract" "github.com/cortexproject/cortex/pkg/util/log" ) +const ( + remoteWriteVersionHeader = "X-Prometheus-Remote-Write-Version" + remoteWriteVersion1HeaderValue = "0.1.0" + remoteWriteVersion20HeaderValue = "2.0.0" + appProtoContentType = "application/x-protobuf" + appProtoV1ContentType = "application/x-protobuf;proto=prometheus.WriteRequest" + appProtoV2ContentType = "application/x-protobuf;proto=io.prometheus.write.v2.Request" + + rw20WrittenSamplesHeader = "X-Prometheus-Remote-Write-Samples-Written" + rw20WrittenHistogramsHeader = "X-Prometheus-Remote-Write-Histograms-Written" + rw20WrittenExemplarsHeader = "X-Prometheus-Remote-Write-Exemplars-Written" + + errMsgNotEnabledPRW2 = "Not enabled prometheus remote write v2 push request" +) + // Func defines the type of the push. It is similar to http.HandlerFunc. type Func func(context.Context, *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) // Handler is a http.Handler which accepts WriteRequests. -func Handler(maxRecvMsgSize int, sourceIPs *middleware.SourceIPExtractor, push Func) http.Handler { +func Handler(remoteWrite2Enabled bool, maxRecvMsgSize int, sourceIPs *middleware.SourceIPExtractor, push Func) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ctx := r.Context() logger := log.WithContext(ctx, log.Logger) @@ -28,31 +51,240 @@ func Handler(maxRecvMsgSize int, sourceIPs *middleware.SourceIPExtractor, push F logger = log.WithSourceIPs(source, logger) } } - var req cortexpb.PreallocWriteRequest - err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy) + + // follow Prometheus https://github.com/prometheus/prometheus/blob/main/storage/remote/write_handler.go + contentType := r.Header.Get("Content-Type") + if contentType == "" { + contentType = appProtoContentType + } + + msgType, err := parseProtoMsg(contentType) if err != nil { - level.Error(logger).Log("err", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) + level.Error(logger).Log("Error decoding remote write request", "err", err) + http.Error(w, err.Error(), http.StatusUnsupportedMediaType) return } - req.SkipLabelNameValidation = false - if req.Source == 0 { - req.Source = cortexpb.API + if msgType != config.RemoteWriteProtoMsgV1 && msgType != config.RemoteWriteProtoMsgV2 { + level.Error(logger).Log("Not accepted msg type", "msgType", msgType, "err", err) + http.Error(w, err.Error(), http.StatusUnsupportedMediaType) + return + } + + enc := r.Header.Get("Content-Encoding") + if enc == "" { + } else if enc != string(remote.SnappyBlockCompression) { + err := fmt.Errorf("%v encoding (compression) is not accepted by this server; only %v is acceptable", enc, remote.SnappyBlockCompression) + level.Error(logger).Log("Error decoding remote write request", "err", err) + http.Error(w, err.Error(), http.StatusUnsupportedMediaType) + return } - if _, err := push(ctx, &req.WriteRequest); err != nil { - resp, ok := httpgrpc.HTTPResponseFromError(err) - if !ok { - http.Error(w, err.Error(), http.StatusInternalServerError) + switch msgType { + case config.RemoteWriteProtoMsgV1: + var req cortexpb.PreallocWriteRequest + err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy) + if err != nil { + level.Error(logger).Log("err", err.Error()) + http.Error(w, err.Error(), http.StatusBadRequest) return } - if resp.GetCode()/100 == 5 { - level.Error(logger).Log("msg", "push error", "err", err) - } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { - level.Warn(logger).Log("msg", "push refused", "err", err) + + req.SkipLabelNameValidation = false + if req.Source == 0 { + req.Source = cortexpb.API + } + + if _, err := push(ctx, &req.WriteRequest); err != nil { + resp, ok := httpgrpc.HTTPResponseFromError(err) + if !ok { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if resp.GetCode()/100 == 5 { + level.Error(logger).Log("msg", "push error", "err", err) + } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { + level.Warn(logger).Log("msg", "push refused", "err", err) + } + http.Error(w, string(resp.Body), int(resp.Code)) + } + case config.RemoteWriteProtoMsgV2: + if remoteWrite2Enabled { + var req writev2.Request + err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy) + if err != nil { + level.Error(logger).Log("err", err.Error()) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + v1Req, err := convertV2RequestToV1(&req) + if err != nil { + level.Error(logger).Log("err", err.Error()) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + v1Req.SkipLabelNameValidation = false + // Current source is only API + if v1Req.Source == 0 { + v1Req.Source = cortexpb.API + } + + if resp, err := push(ctx, &v1Req.WriteRequest); err != nil { + resp, ok := httpgrpc.HTTPResponseFromError(err) + setHeader(w, 0, 0, 0) + if !ok { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + if resp.GetCode()/100 == 5 { + level.Error(logger).Log("msg", "push error", "err", err) + } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { + level.Warn(logger).Log("msg", "push refused", "err", err) + } + http.Error(w, string(resp.Body), int(resp.Code)) + } else { + setHeader(w, resp.Samples, resp.Histograms, resp.Exemplars) + } + } else { + level.Error(logger).Log(errMsgNotEnabledPRW2) + http.Error(w, errMsgNotEnabledPRW2, http.StatusUnsupportedMediaType) + return } - http.Error(w, string(resp.Body), int(resp.Code)) } }) } + +func setHeader(w http.ResponseWriter, samples, histograms, exemplars int64) { + w.Header().Set(rw20WrittenSamplesHeader, strconv.FormatInt(samples, 10)) + w.Header().Set(rw20WrittenHistogramsHeader, strconv.FormatInt(histograms, 10)) + w.Header().Set(rw20WrittenExemplarsHeader, strconv.FormatInt(exemplars, 10)) +} + +// Refer to parseProtoMsg in https://github.com/prometheus/prometheus/blob/main/storage/remote/write_handler.go +func parseProtoMsg(contentType string) (config.RemoteWriteProtoMsg, error) { + contentType = strings.TrimSpace(contentType) + + parts := strings.Split(contentType, ";") + if parts[0] != appProtoContentType { + return "", fmt.Errorf("expected %v as the first (media) part, got %v content-type", appProtoContentType, contentType) + } + // Parse potential https://www.rfc-editor.org/rfc/rfc9110#parameter + for _, p := range parts[1:] { + pair := strings.Split(p, "=") + if len(pair) != 2 { + return "", fmt.Errorf("as per https://www.rfc-editor.org/rfc/rfc9110#parameter expected parameters to be key-values, got %v in %v content-type", p, contentType) + } + if pair[0] == "proto" { + ret := config.RemoteWriteProtoMsg(pair[1]) + if err := ret.Validate(); err != nil { + return "", fmt.Errorf("got %v content type; %w", contentType, err) + } + return ret, nil + } + } + // No "proto=" parameter, assuming v1. + return config.RemoteWriteProtoMsgV1, nil +} + +func convertV2RequestToV1(req *writev2.Request) (cortexpb.PreallocWriteRequest, error) { + var v1Req cortexpb.PreallocWriteRequest + v1Timeseries := make([]cortexpb.PreallocTimeseries, 0, len(req.Timeseries)) + var v1Metadata []*cortexpb.MetricMetadata + + b := labels.NewScratchBuilder(0) + symbols := req.Symbols + for _, v2Ts := range req.Timeseries { + lbs := v2Ts.ToLabels(&b, symbols) + v1Timeseries = append(v1Timeseries, cortexpb.PreallocTimeseries{ + TimeSeries: &cortexpb.TimeSeries{ + Labels: cortexpb.FromLabelsToLabelAdapters(lbs), + Samples: convertV2ToV1Samples(v2Ts.Samples), + Exemplars: convertV2ToV1Exemplars(b, symbols, v2Ts.Exemplars), + Histograms: convertV2ToV1Histograms(v2Ts.Histograms), + }, + }) + + if shouldConvertV2Metadata(v2Ts.Metadata) { + metricName, err := extract.MetricNameFromLabels(lbs) + if err != nil { + return v1Req, err + } + v1Metadata = append(v1Metadata, convertV2ToV1Metadata(metricName, symbols, v2Ts.Metadata)) + } + } + + v1Req.Timeseries = v1Timeseries + v1Req.Metadata = v1Metadata + + return v1Req, nil +} + +func shouldConvertV2Metadata(metadata writev2.Metadata) bool { + return !(metadata.HelpRef == 0 && metadata.UnitRef == 0 && metadata.Type == writev2.Metadata_METRIC_TYPE_UNSPECIFIED) +} + +func convertV2ToV1Histograms(histograms []writev2.Histogram) []cortexpb.Histogram { + v1Histograms := make([]cortexpb.Histogram, 0, len(histograms)) + + for _, h := range histograms { + v1Histograms = append(v1Histograms, cortexpb.HistogramWriteV2ProtoToHistogramProto(h)) + } + + return v1Histograms +} + +func convertV2ToV1Samples(samples []writev2.Sample) []cortexpb.Sample { + v1Samples := make([]cortexpb.Sample, 0, len(samples)) + + for _, s := range samples { + v1Samples = append(v1Samples, cortexpb.Sample{ + Value: s.Value, + TimestampMs: s.Timestamp, + }) + } + + return v1Samples +} + +func convertV2ToV1Metadata(name string, symbols []string, metadata writev2.Metadata) *cortexpb.MetricMetadata { + t := cortexpb.UNKNOWN + + switch metadata.Type { + case writev2.Metadata_METRIC_TYPE_COUNTER: + t = cortexpb.COUNTER + case writev2.Metadata_METRIC_TYPE_GAUGE: + t = cortexpb.GAUGE + case writev2.Metadata_METRIC_TYPE_HISTOGRAM: + t = cortexpb.HISTOGRAM + case writev2.Metadata_METRIC_TYPE_GAUGEHISTOGRAM: + t = cortexpb.GAUGEHISTOGRAM + case writev2.Metadata_METRIC_TYPE_SUMMARY: + t = cortexpb.SUMMARY + case writev2.Metadata_METRIC_TYPE_INFO: + t = cortexpb.INFO + case writev2.Metadata_METRIC_TYPE_STATESET: + t = cortexpb.STATESET + } + + return &cortexpb.MetricMetadata{ + Type: t, + MetricFamilyName: name, + Unit: symbols[metadata.UnitRef], + Help: symbols[metadata.HelpRef], + } +} + +func convertV2ToV1Exemplars(b labels.ScratchBuilder, symbols []string, v2Exemplars []writev2.Exemplar) []cortexpb.Exemplar { + v1Exemplars := make([]cortexpb.Exemplar, 0, len(v2Exemplars)) + for _, e := range v2Exemplars { + promExemplar := e.ToExemplar(&b, symbols) + v1Exemplars = append(v1Exemplars, cortexpb.Exemplar{ + Labels: cortexpb.FromLabelsToLabelAdapters(promExemplar.Labels), + Value: e.Value, + TimestampMs: e.Timestamp, + }) + } + return v1Exemplars +} diff --git a/pkg/util/push/push_test.go b/pkg/util/push/push_test.go index b806011a611..85fcaf29f68 100644 --- a/pkg/util/push/push_test.go +++ b/pkg/util/push/push_test.go @@ -10,6 +10,8 @@ import ( "github.com/golang/snappy" "github.com/prometheus/prometheus/prompb" + writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" + "github.com/prometheus/prometheus/tsdb/tsdbutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/weaveworks/common/middleware" @@ -17,30 +19,239 @@ import ( "github.com/cortexproject/cortex/pkg/cortexpb" ) +func Test_convertV2RequestToV1(t *testing.T) { + var v2Req writev2.Request + + fh := tsdbutil.GenerateTestFloatHistogram(1) + ph := writev2.FromFloatHistogram(4, fh) + + symbols := []string{"", "__name__", "test_metric", "b", "c", "baz", "qux", "d", "e", "foo", "bar", "f", "g", "h", "i", "Test gauge for test purposes", "Maybe op/sec who knows (:", "Test counter for test purposes"} + timeseries := []writev2.TimeSeries{ + { + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + Metadata: writev2.Metadata{ + Type: writev2.Metadata_METRIC_TYPE_COUNTER, + + HelpRef: 15, + UnitRef: 16, + }, + Samples: []writev2.Sample{{Value: 1, Timestamp: 1}}, + Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: 1}}, + }, + { + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + Samples: []writev2.Sample{{Value: 2, Timestamp: 2}}, + }, + { + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + Samples: []writev2.Sample{{Value: 3, Timestamp: 3}}, + }, + { + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + Histograms: []writev2.Histogram{ph, ph}, + Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: 1}}, + }, + } + + v2Req.Symbols = symbols + v2Req.Timeseries = timeseries + v1Req, err := convertV2RequestToV1(&v2Req) + assert.NoError(t, err) + expectedSamples := 3 + expectedExemplars := 2 + expectedHistograms := 2 + countSamples := 0 + countExemplars := 0 + countHistograms := 0 + + for _, ts := range v1Req.Timeseries { + countSamples += len(ts.Samples) + countExemplars += len(ts.Exemplars) + countHistograms += len(ts.Histograms) + } + + 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, 1, len(v1Req.Metadata)) +} + func TestHandler_remoteWrite(t *testing.T) { - req := createRequest(t, createPrometheusRemoteWriteProtobuf(t)) - resp := httptest.NewRecorder() - handler := Handler(100000, nil, verifyWriteRequestHandler(t, cortexpb.API)) - handler.ServeHTTP(resp, req) - assert.Equal(t, 200, resp.Code) + t.Run("remote write v1", func(t *testing.T) { + handler := Handler(true, 100000, nil, verifyWriteRequestHandler(t, cortexpb.API)) + req := createRequest(t, createPrometheusRemoteWriteProtobuf(t), false) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, http.StatusOK, resp.Code) + }) + t.Run("remote write v2", func(t *testing.T) { + handler := Handler(true, 100000, nil, verifyWriteRequestHandler(t, cortexpb.API)) + req := createRequest(t, createPrometheusRemoteWriteV2Protobuf(t), true) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, http.StatusOK, resp.Code) + + // test header value + respHeader := resp.Header() + assert.Equal(t, "1", respHeader[rw20WrittenSamplesHeader][0]) + assert.Equal(t, "1", respHeader[rw20WrittenHistogramsHeader][0]) + assert.Equal(t, "1", respHeader[rw20WrittenExemplarsHeader][0]) + }) + t.Run("remote write v2 with not support remote write 2.0", func(t *testing.T) { + handler := Handler(false, 100000, nil, verifyWriteRequestHandler(t, cortexpb.API)) + req := createRequest(t, createPrometheusRemoteWriteV2Protobuf(t), true) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, http.StatusUnsupportedMediaType, resp.Code) + }) +} + +func TestHandler_ContentTypeAndEncoding(t *testing.T) { + sourceIPs, _ := middleware.NewSourceIPs("SomeField", "(.*)") + handler := Handler(true, 100000, sourceIPs, verifyWriteRequestHandler(t, cortexpb.API)) + + tests := []struct { + description string + reqHeaders map[string]string + expectedCode int + isV2 bool + }{ + { + description: "[RW 2.0] correct content-type", + reqHeaders: map[string]string{ + "Content-Type": appProtoV2ContentType, + "Content-Encoding": "snappy", + remoteWriteVersionHeader: "2.0.0", + }, + expectedCode: http.StatusOK, + isV2: true, + }, + { + description: "[RW 1.0] correct content-type", + reqHeaders: map[string]string{ + "Content-Type": appProtoV1ContentType, + "Content-Encoding": "snappy", + remoteWriteVersionHeader: "0.1.0", + }, + expectedCode: http.StatusOK, + isV2: false, + }, + { + description: "[RW 2.0] wrong content-type", + reqHeaders: map[string]string{ + "Content-Type": "yolo", + "Content-Encoding": "snappy", + remoteWriteVersionHeader: "2.0.0", + }, + expectedCode: http.StatusUnsupportedMediaType, + isV2: true, + }, + { + description: "[RW 2.0] wrong content-type", + reqHeaders: map[string]string{ + "Content-Type": "application/x-protobuf;proto=yolo", + "Content-Encoding": "snappy", + remoteWriteVersionHeader: "2.0.0", + }, + expectedCode: http.StatusUnsupportedMediaType, + isV2: true, + }, + { + description: "[RW 2.0] wrong content-encoding", + reqHeaders: map[string]string{ + "Content-Type": "application/x-protobuf;proto=io.prometheus.write.v2.Request", + "Content-Encoding": "zstd", + remoteWriteVersionHeader: "2.0.0", + }, + expectedCode: http.StatusUnsupportedMediaType, + isV2: true, + }, + { + description: "no header, should treated as RW 1.0", + expectedCode: http.StatusOK, + isV2: false, + }, + { + description: "missing content-type, should treated as RW 1.0", + reqHeaders: map[string]string{ + "Content-Encoding": "snappy", + remoteWriteVersionHeader: "2.0.0", + }, + expectedCode: http.StatusOK, + isV2: false, + }, + { + description: "missing content-encoding", + reqHeaders: map[string]string{ + "Content-Type": appProtoV2ContentType, + remoteWriteVersionHeader: "2.0.0", + }, + expectedCode: http.StatusOK, + isV2: true, + }, + { + description: "missing remote write version, should treated based on Content-type", + reqHeaders: map[string]string{ + "Content-Type": appProtoV2ContentType, + "Content-Encoding": "snappy", + }, + expectedCode: http.StatusOK, + isV2: true, + }, + { + description: "missing remote write version, should treated based on Content-type", + reqHeaders: map[string]string{ + "Content-Type": appProtoV1ContentType, + "Content-Encoding": "snappy", + }, + expectedCode: http.StatusOK, + isV2: false, + }, + } + + for _, test := range tests { + t.Run(test.description, func(t *testing.T) { + if test.isV2 { + req := createRequestWithHeaders(t, test.reqHeaders, createCortexRemoteWriteV2Protobuf(t, false, cortexpb.API)) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, test.expectedCode, resp.Code) + } else { + req := createRequestWithHeaders(t, test.reqHeaders, createCortexWriteRequestProtobuf(t, false, cortexpb.API)) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, test.expectedCode, resp.Code) + } + }) + } } func TestHandler_cortexWriteRequest(t *testing.T) { - req := createRequest(t, createCortexWriteRequestProtobuf(t, false)) - resp := httptest.NewRecorder() sourceIPs, _ := middleware.NewSourceIPs("SomeField", "(.*)") - handler := Handler(100000, sourceIPs, verifyWriteRequestHandler(t, cortexpb.RULE)) - handler.ServeHTTP(resp, req) - assert.Equal(t, 200, resp.Code) + handler := Handler(true, 100000, sourceIPs, verifyWriteRequestHandler(t, cortexpb.API)) + + t.Run("remote write v1", func(t *testing.T) { + req := createRequest(t, createCortexWriteRequestProtobuf(t, false, cortexpb.API), false) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, 200, resp.Code) + }) + t.Run("remote write v2", func(t *testing.T) { + req := createRequest(t, createCortexRemoteWriteV2Protobuf(t, false, cortexpb.API), true) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(t, 200, resp.Code) + }) } func TestHandler_ignoresSkipLabelNameValidationIfSet(t *testing.T) { for _, req := range []*http.Request{ - createRequest(t, createCortexWriteRequestProtobuf(t, true)), - createRequest(t, createCortexWriteRequestProtobuf(t, false)), + createRequest(t, createCortexWriteRequestProtobuf(t, true, cortexpb.RULE), false), + createRequest(t, createCortexWriteRequestProtobuf(t, true, cortexpb.RULE), false), } { resp := httptest.NewRecorder() - handler := Handler(100000, nil, verifyWriteRequestHandler(t, cortexpb.RULE)) + handler := Handler(true, 100000, nil, verifyWriteRequestHandler(t, cortexpb.RULE)) handler.ServeHTTP(resp, req) assert.Equal(t, 200, resp.Code) } @@ -54,21 +265,86 @@ func verifyWriteRequestHandler(t *testing.T, expectSource cortexpb.WriteRequest_ assert.Equal(t, "foo", request.Timeseries[0].Labels[0].Value) assert.Equal(t, expectSource, request.Source) assert.False(t, request.SkipLabelNameValidation) - return &cortexpb.WriteResponse{}, nil + + resp := &cortexpb.WriteResponse{ + Samples: 1, + Histograms: 1, + Exemplars: 1, + } + + return resp, nil + } +} + +func createRequestWithHeaders(t *testing.T, headers map[string]string, protobuf []byte) *http.Request { + t.Helper() + inoutBytes := snappy.Encode(nil, protobuf) + req, err := http.NewRequest("POST", "http://localhost/", bytes.NewReader(inoutBytes)) + require.NoError(t, err) + + for k, v := range headers { + req.Header.Set(k, v) } + return req } -func createRequest(t *testing.T, protobuf []byte) *http.Request { +func createRequest(t *testing.T, protobuf []byte, isV2 bool) *http.Request { t.Helper() inoutBytes := snappy.Encode(nil, protobuf) req, err := http.NewRequest("POST", "http://localhost/", bytes.NewReader(inoutBytes)) require.NoError(t, err) + req.Header.Add("Content-Encoding", "snappy") - req.Header.Set("Content-Type", "application/x-protobuf") - req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0") + + if isV2 { + req.Header.Set("Content-Type", appProtoV2ContentType) + req.Header.Set("X-Prometheus-Remote-Write-Version", remoteWriteVersion20HeaderValue) + return req + } + + req.Header.Set("Content-Type", appProtoContentType) + req.Header.Set("X-Prometheus-Remote-Write-Version", remoteWriteVersion1HeaderValue) return req } +func createCortexRemoteWriteV2Protobuf(t *testing.T, skipLabelNameValidation bool, source cortexpb.WriteRequest_SourceEnum) []byte { + t.Helper() + input := writev2.Request{ + Symbols: []string{"", "__name__", "foo"}, + Timeseries: []writev2.TimeSeries{ + { + LabelsRefs: []uint32{1, 2}, + Samples: []writev2.Sample{ + {Value: 1, Timestamp: time.Date(2020, 4, 1, 0, 0, 0, 0, time.UTC).UnixNano()}, + }, + }, + }, + } + + inoutBytes, err := input.Marshal() + require.NoError(t, err) + return inoutBytes +} + +func createPrometheusRemoteWriteV2Protobuf(t *testing.T) []byte { + t.Helper() + input := writev2.Request{ + Symbols: []string{"", "__name__", "foo"}, + Timeseries: []writev2.TimeSeries{ + { + LabelsRefs: []uint32{1, 2}, + Samples: []writev2.Sample{ + {Value: 1, Timestamp: time.Date(2020, 4, 1, 0, 0, 0, 0, time.UTC).UnixNano()}, + }, + }, + }, + } + + inoutBytes, err := input.Marshal() + require.NoError(t, err) + return inoutBytes +} + func createPrometheusRemoteWriteProtobuf(t *testing.T) []byte { t.Helper() input := prompb.WriteRequest{ @@ -87,7 +363,7 @@ func createPrometheusRemoteWriteProtobuf(t *testing.T) []byte { require.NoError(t, err) return inoutBytes } -func createCortexWriteRequestProtobuf(t *testing.T, skipLabelNameValidation bool) []byte { +func createCortexWriteRequestProtobuf(t *testing.T, skipLabelNameValidation bool, source cortexpb.WriteRequest_SourceEnum) []byte { t.Helper() ts := cortexpb.PreallocTimeseries{ TimeSeries: &cortexpb.TimeSeries{ @@ -101,7 +377,7 @@ func createCortexWriteRequestProtobuf(t *testing.T, skipLabelNameValidation bool } input := cortexpb.WriteRequest{ Timeseries: []cortexpb.PreallocTimeseries{ts}, - Source: cortexpb.RULE, + Source: source, SkipLabelNameValidation: skipLabelNameValidation, } inoutBytes, err := input.Marshal() From dad7b59a83576a03dfd8b61e9d249a56ef341cc7 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Mon, 14 Apr 2025 10:53:24 +0900 Subject: [PATCH 2/4] Change to not break exist behavior Signed-off-by: SungJin1212 --- integration/remote_write_v2_test.go | 63 ++++++++++++ pkg/util/push/push.go | 143 ++++++++++++++-------------- pkg/util/push/push_test.go | 7 -- 3 files changed, 136 insertions(+), 77 deletions(-) diff --git a/integration/remote_write_v2_test.go b/integration/remote_write_v2_test.go index 88ca384fb04..4ebcc142077 100644 --- a/integration/remote_write_v2_test.go +++ b/integration/remote_write_v2_test.go @@ -137,6 +137,69 @@ func TestIngesterRollingUpdate(t *testing.T) { } } +func TestIngest_SenderSendPRW2_DistributorNotAllowPRW2(t *testing.T) { + const blockRangePeriod = 5 * time.Second + + 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", + "-querier.query-store-for-labels-enabled": "true", + "-blocks-storage.tsdb.block-ranges-period": blockRangePeriod.String(), + "-blocks-storage.tsdb.ship-interval": "1s", + "-blocks-storage.tsdb.retention-period": ((blockRangePeriod * 2) - 1).String(), + "-blocks-storage.tsdb.enable-native-histograms": "true", + // Ingester. + "-ring.store": "consul", + "-consul.hostname": consul.NetworkHTTPEndpoint(), + // Distributor. + "-distributor.replication-factor": "1", + "-distributor.remote-write2-enabled": "false", + // 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() + + // series push + symbols1, series, _ := e2e.GenerateSeriesV2("test_series", now, prompb.Label{Name: "job", Value: "test"}, prompb.Label{Name: "foo", Value: "bar"}) + res, err := c.PushV2(symbols1, series) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) +} + func TestIngest(t *testing.T) { const blockRangePeriod = 5 * time.Second diff --git a/pkg/util/push/push.go b/pkg/util/push/push.go index 730b3d46193..bbe3e8d489c 100644 --- a/pkg/util/push/push.go +++ b/pkg/util/push/push.go @@ -32,8 +32,6 @@ const ( rw20WrittenSamplesHeader = "X-Prometheus-Remote-Write-Samples-Written" rw20WrittenHistogramsHeader = "X-Prometheus-Remote-Write-Histograms-Written" rw20WrittenExemplarsHeader = "X-Prometheus-Remote-Write-Exemplars-Written" - - errMsgNotEnabledPRW2 = "Not enabled prometheus remote write v2 push request" ) // Func defines the type of the push. It is similar to http.HandlerFunc. @@ -52,36 +50,7 @@ func Handler(remoteWrite2Enabled bool, maxRecvMsgSize int, sourceIPs *middleware } } - // follow Prometheus https://github.com/prometheus/prometheus/blob/main/storage/remote/write_handler.go - contentType := r.Header.Get("Content-Type") - if contentType == "" { - contentType = appProtoContentType - } - - msgType, err := parseProtoMsg(contentType) - if err != nil { - level.Error(logger).Log("Error decoding remote write request", "err", err) - http.Error(w, err.Error(), http.StatusUnsupportedMediaType) - return - } - - if msgType != config.RemoteWriteProtoMsgV1 && msgType != config.RemoteWriteProtoMsgV2 { - level.Error(logger).Log("Not accepted msg type", "msgType", msgType, "err", err) - http.Error(w, err.Error(), http.StatusUnsupportedMediaType) - return - } - - enc := r.Header.Get("Content-Encoding") - if enc == "" { - } else if enc != string(remote.SnappyBlockCompression) { - err := fmt.Errorf("%v encoding (compression) is not accepted by this server; only %v is acceptable", enc, remote.SnappyBlockCompression) - level.Error(logger).Log("Error decoding remote write request", "err", err) - http.Error(w, err.Error(), http.StatusUnsupportedMediaType) - return - } - - switch msgType { - case config.RemoteWriteProtoMsgV1: + handlePRW1 := func() { var req cortexpb.PreallocWriteRequest err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy) if err != nil { @@ -108,55 +77,89 @@ func Handler(remoteWrite2Enabled bool, maxRecvMsgSize int, sourceIPs *middleware } http.Error(w, string(resp.Body), int(resp.Code)) } - case config.RemoteWriteProtoMsgV2: - if remoteWrite2Enabled { - var req writev2.Request - err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy) - if err != nil { - level.Error(logger).Log("err", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return - } + } - v1Req, err := convertV2RequestToV1(&req) - if err != nil { - level.Error(logger).Log("err", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return - } + handlePRW2 := func() { + var req writev2.Request + err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy) + if err != nil { + level.Error(logger).Log("err", err.Error()) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } - v1Req.SkipLabelNameValidation = false - // Current source is only API - if v1Req.Source == 0 { - v1Req.Source = cortexpb.API - } + v1Req, err := convertV2RequestToV1(&req) + if err != nil { + level.Error(logger).Log("err", err.Error()) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } - if resp, err := push(ctx, &v1Req.WriteRequest); err != nil { - resp, ok := httpgrpc.HTTPResponseFromError(err) - setHeader(w, 0, 0, 0) - if !ok { - http.Error(w, err.Error(), http.StatusInternalServerError) - return - } - if resp.GetCode()/100 == 5 { - level.Error(logger).Log("msg", "push error", "err", err) - } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { - level.Warn(logger).Log("msg", "push refused", "err", err) - } - http.Error(w, string(resp.Body), int(resp.Code)) - } else { - setHeader(w, resp.Samples, resp.Histograms, resp.Exemplars) + v1Req.SkipLabelNameValidation = false + if v1Req.Source == 0 { + v1Req.Source = cortexpb.API + } + + if resp, err := push(ctx, &v1Req.WriteRequest); err != nil { + resp, ok := httpgrpc.HTTPResponseFromError(err) + setPRW2RespHeader(w, 0, 0, 0) + if !ok { + http.Error(w, err.Error(), http.StatusInternalServerError) + return } + if resp.GetCode()/100 == 5 { + level.Error(logger).Log("msg", "push error", "err", err) + } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { + level.Warn(logger).Log("msg", "push refused", "err", err) + } + http.Error(w, string(resp.Body), int(resp.Code)) } else { - level.Error(logger).Log(errMsgNotEnabledPRW2) - http.Error(w, errMsgNotEnabledPRW2, http.StatusUnsupportedMediaType) + setPRW2RespHeader(w, resp.Samples, resp.Histograms, resp.Exemplars) + } + } + + if remoteWrite2Enabled { + // follow Prometheus https://github.com/prometheus/prometheus/blob/main/storage/remote/write_handler.go + contentType := r.Header.Get("Content-Type") + if contentType == "" { + contentType = appProtoContentType + } + + msgType, err := parseProtoMsg(contentType) + if err != nil { + level.Error(logger).Log("Error decoding remote write request", "err", err) + http.Error(w, err.Error(), http.StatusUnsupportedMediaType) return } + + if msgType != config.RemoteWriteProtoMsgV1 && msgType != config.RemoteWriteProtoMsgV2 { + level.Error(logger).Log("Not accepted msg type", "msgType", msgType, "err", err) + http.Error(w, err.Error(), http.StatusUnsupportedMediaType) + return + } + + enc := r.Header.Get("Content-Encoding") + if enc == "" { + } else if enc != string(remote.SnappyBlockCompression) { + err := fmt.Errorf("%v encoding (compression) is not accepted by this server; only %v is acceptable", enc, remote.SnappyBlockCompression) + level.Error(logger).Log("Error decoding remote write request", "err", err) + http.Error(w, err.Error(), http.StatusUnsupportedMediaType) + return + } + + switch msgType { + case config.RemoteWriteProtoMsgV1: + handlePRW1() + case config.RemoteWriteProtoMsgV2: + handlePRW2() + } + } else { + handlePRW1() } }) } -func setHeader(w http.ResponseWriter, samples, histograms, exemplars int64) { +func setPRW2RespHeader(w http.ResponseWriter, samples, histograms, exemplars int64) { w.Header().Set(rw20WrittenSamplesHeader, strconv.FormatInt(samples, 10)) w.Header().Set(rw20WrittenHistogramsHeader, strconv.FormatInt(histograms, 10)) w.Header().Set(rw20WrittenExemplarsHeader, strconv.FormatInt(exemplars, 10)) diff --git a/pkg/util/push/push_test.go b/pkg/util/push/push_test.go index 85fcaf29f68..fd23ac1dff6 100644 --- a/pkg/util/push/push_test.go +++ b/pkg/util/push/push_test.go @@ -98,13 +98,6 @@ func TestHandler_remoteWrite(t *testing.T) { assert.Equal(t, "1", respHeader[rw20WrittenHistogramsHeader][0]) assert.Equal(t, "1", respHeader[rw20WrittenExemplarsHeader][0]) }) - t.Run("remote write v2 with not support remote write 2.0", func(t *testing.T) { - handler := Handler(false, 100000, nil, verifyWriteRequestHandler(t, cortexpb.API)) - req := createRequest(t, createPrometheusRemoteWriteV2Protobuf(t), true) - resp := httptest.NewRecorder() - handler.ServeHTTP(resp, req) - assert.Equal(t, http.StatusUnsupportedMediaType, resp.Code) - }) } func TestHandler_ContentTypeAndEncoding(t *testing.T) { From d6e2d96f81726db521a47bdca2af0c6a03dc1f4c Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 15 Apr 2025 13:36:07 +0900 Subject: [PATCH 3/4] Add benchmarks Signed-off-by: SungJin1212 --- pkg/util/push/push_test.go | 144 +++++++++++++++++++++++++++++++++++++ 1 file changed, 144 insertions(+) diff --git a/pkg/util/push/push_test.go b/pkg/util/push/push_test.go index fd23ac1dff6..46cb0770f75 100644 --- a/pkg/util/push/push_test.go +++ b/pkg/util/push/push_test.go @@ -3,12 +3,14 @@ package push import ( "bytes" "context" + "fmt" "net/http" "net/http/httptest" "testing" "time" "github.com/golang/snappy" + "github.com/prometheus/prometheus/model/histogram" "github.com/prometheus/prometheus/prompb" writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2" "github.com/prometheus/prometheus/tsdb/tsdbutil" @@ -19,6 +21,148 @@ import ( "github.com/cortexproject/cortex/pkg/cortexpb" ) +var ( + testHistogram = histogram.Histogram{ + Schema: 2, + ZeroThreshold: 1e-128, + ZeroCount: 0, + Count: 3, + Sum: 20, + PositiveSpans: []histogram.Span{{Offset: 0, Length: 1}}, + PositiveBuckets: []int64{1}, + NegativeSpans: []histogram.Span{{Offset: 0, Length: 1}}, + NegativeBuckets: []int64{2}, + } +) + +func makeV2ReqWithSeries(num int) *writev2.Request { + ts := make([]writev2.TimeSeries, 0, num) + symbols := []string{"", "__name__", "test_metric1", "b", "c", "baz", "qux", "d", "e", "foo", "bar", "f", "g", "h", "i", "Test gauge for test purposes", "Maybe op/sec who knows (:", "Test counter for test purposes"} + for i := 0; i < num; i++ { + ts = append(ts, writev2.TimeSeries{ + LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + Metadata: writev2.Metadata{ + Type: writev2.Metadata_METRIC_TYPE_GAUGE, + + HelpRef: 15, + UnitRef: 16, + }, + Samples: []writev2.Sample{{Value: 1, Timestamp: 10}}, + Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: 10}}, + Histograms: []writev2.Histogram{ + writev2.FromIntHistogram(10, &testHistogram), + writev2.FromFloatHistogram(20, testHistogram.ToFloat(nil)), + }, + }) + } + + return &writev2.Request{ + Symbols: symbols, + Timeseries: ts, + } +} + +func createPRW1HTTPRequest(seriesNum int) (*http.Request, error) { + series := makeV2ReqWithSeries(seriesNum) + v1Req, err := convertV2RequestToV1(series) + if err != nil { + return nil, err + } + protobuf, err := v1Req.Marshal() + if err != nil { + return nil, err + } + + body := snappy.Encode(nil, protobuf) + req, err := http.NewRequest("POST", "http://localhost/", newResetReader(body)) + if err != nil { + return nil, err + } + + req.Header.Add("Content-Encoding", "snappy") + req.Header.Set("Content-Type", appProtoContentType) + req.Header.Set("X-Prometheus-Remote-Write-Version", remoteWriteVersion1HeaderValue) + req.ContentLength = int64(len(body)) + return req, nil +} + +func createPRW2HTTPRequest(seriesNum int) (*http.Request, error) { + series := makeV2ReqWithSeries(seriesNum) + protobuf, err := series.Marshal() + if err != nil { + return nil, err + } + + body := snappy.Encode(nil, protobuf) + req, err := http.NewRequest("POST", "http://localhost/", newResetReader(body)) + if err != nil { + return nil, err + } + + req.Header.Add("Content-Encoding", "snappy") + req.Header.Set("Content-Type", appProtoV2ContentType) + req.Header.Set("X-Prometheus-Remote-Write-Version", remoteWriteVersion20HeaderValue) + req.ContentLength = int64(len(body)) + return req, nil +} + +func Benchmark_Handler(b *testing.B) { + mockHandler := func(context.Context, *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) { + // Nothing to do. + return &cortexpb.WriteResponse{}, nil + } + testSeriesNums := []int{10, 100, 500, 1000} + for _, seriesNum := range testSeriesNums { + b.Run(fmt.Sprintf("PRW1 with %d series", seriesNum), func(b *testing.B) { + handler := Handler(true, 1000000, nil, mockHandler) + req, err := createPRW1HTTPRequest(seriesNum) + require.NoError(b, err) + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(b, http.StatusOK, resp.Code) + req.Body.(*resetReader).Reset() + } + }) + b.Run(fmt.Sprintf("PRW2 with %d series", seriesNum), func(b *testing.B) { + handler := Handler(true, 1000000, nil, mockHandler) + req, err := createPRW2HTTPRequest(seriesNum) + require.NoError(b, err) + + b.ResetTimer() + b.ReportAllocs() + + for i := 0; i < b.N; i++ { + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + assert.Equal(b, http.StatusOK, resp.Code) + req.Body.(*resetReader).Reset() + } + }) + } +} + +func Benchmark_convertV2RequestToV1(b *testing.B) { + testSeriesNums := []int{100, 500, 1000} + + for _, seriesNum := range testSeriesNums { + b.Run(fmt.Sprintf("%d series", seriesNum), func(b *testing.B) { + series := makeV2ReqWithSeries(seriesNum) + + b.ResetTimer() + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _, err := convertV2RequestToV1(series) + require.NoError(b, err) + } + }) + } +} + func Test_convertV2RequestToV1(t *testing.T) { var v2Req writev2.Request From 71291b0f16648b25cb84be321ac0b1579cf943d4 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Fri, 16 May 2025 13:36:53 +0900 Subject: [PATCH 4/4] rebase from stream connection Signed-off-by: SungJin1212 --- docs/configuration/config-file-reference.md | 10 +- pkg/cortexpb/cortex.pb.go | 267 ++++++++++++++------ pkg/cortexpb/cortex.proto | 8 +- pkg/distributor/distributor.go | 3 +- 4 files changed, 206 insertions(+), 82 deletions(-) diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 3ebc69d9435..17049620f68 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -2690,16 +2690,16 @@ ha_tracker: # CLI flag: -distributor.sign-write-requests [sign_write_requests: | default = false] -# EXPERIMENTAL: If true, accept prometheus remote write v2 protocol push -# request. -# CLI flag: -distributor.remote-write2-enabled -[remote_write2_enabled: | default = false] - # EXPERIMENTAL: If enabled, distributor would use stream connection to send # requests to ingesters. # CLI flag: -distributor.use-stream-push [use_stream_push: | default = false] +# EXPERIMENTAL: If true, accept prometheus remote write v2 protocol push +# request. +# CLI flag: -distributor.remote-write2-enabled +[remote_write2_enabled: | default = false] + ring: kvstore: # Backend storage to use for the ring. Supported values are: consul, etcd, diff --git a/pkg/cortexpb/cortex.pb.go b/pkg/cortexpb/cortex.pb.go index d1caafe11e7..8a820fcc744 100644 --- a/pkg/cortexpb/cortex.pb.go +++ b/pkg/cortexpb/cortex.pb.go @@ -229,6 +229,12 @@ func (m *StreamWriteRequest) GetRequest() *WriteRequest { type WriteResponse struct { Code int32 `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"` Message string `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"` + // Samples represents X-Prometheus-Remote-Write-Written-Samples + Samples int64 `protobuf:"varint,3,opt,name=Samples,proto3" json:"Samples,omitempty"` + // Histograms represents X-Prometheus-Remote-Write-Written-Histograms + Histograms int64 `protobuf:"varint,4,opt,name=Histograms,proto3" json:"Histograms,omitempty"` + // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars + Exemplars int64 `protobuf:"varint,5,opt,name=Exemplars,proto3" json:"Exemplars,omitempty"` } func (m *WriteResponse) Reset() { *m = WriteResponse{} } @@ -277,6 +283,27 @@ func (m *WriteResponse) GetMessage() string { return "" } +func (m *WriteResponse) GetSamples() int64 { + if m != nil { + return m.Samples + } + return 0 +} + +func (m *WriteResponse) GetHistograms() int64 { + if m != nil { + return m.Histograms + } + return 0 +} + +func (m *WriteResponse) GetExemplars() int64 { + if m != nil { + return m.Exemplars + } + return 0 +} + type TimeSeries struct { Labels []LabelAdapter `protobuf:"bytes,1,rep,name=labels,proto3,customtype=LabelAdapter" json:"labels"` // Sorted by time, oldest sample first. @@ -910,76 +937,78 @@ func init() { func init() { proto.RegisterFile("cortex.proto", fileDescriptor_893a47d0a749d749) } var fileDescriptor_893a47d0a749d749 = []byte{ - // 1090 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x56, 0x3b, 0x6f, 0x1b, 0x47, - 0x17, 0xdd, 0xe1, 0x9b, 0x97, 0x0f, 0xaf, 0xe7, 0x13, 0xfc, 0x2d, 0x04, 0x78, 0x45, 0x6f, 0x90, - 0x84, 0x08, 0x02, 0x25, 0x50, 0x90, 0x04, 0x36, 0x94, 0x00, 0xa4, 0x4d, 0x3d, 0x60, 0x93, 0x12, - 0x86, 0x54, 0x0c, 0xa7, 0x21, 0x46, 0xe4, 0x88, 0x5c, 0x78, 0x5f, 0xd9, 0x19, 0x0a, 0x56, 0xaa, - 0x54, 0x41, 0xca, 0xd4, 0x69, 0xd3, 0xe4, 0x17, 0xe4, 0x37, 0xa8, 0x54, 0x69, 0xa4, 0x10, 0x22, - 0xaa, 0x71, 0xe9, 0x22, 0x3f, 0x20, 0x98, 0xd9, 0x97, 0x64, 0xd9, 0x48, 0xe3, 0x6e, 0xee, 0xb9, - 0xe7, 0xde, 0x39, 0x7b, 0xef, 0xd9, 0x25, 0xa1, 0x3e, 0xf1, 0x43, 0xc1, 0x5e, 0xac, 0x07, 0xa1, - 0x2f, 0x7c, 0x5c, 0x89, 0xa2, 0xe0, 0x70, 0x75, 0x65, 0xe6, 0xcf, 0x7c, 0x05, 0x7e, 0x26, 0x4f, - 0x51, 0xde, 0xfa, 0x33, 0x07, 0xf5, 0xa7, 0xa1, 0x2d, 0x18, 0x61, 0x3f, 0x2c, 0x18, 0x17, 0x78, - 0x1f, 0x40, 0xd8, 0x2e, 0xe3, 0x2c, 0xb4, 0x19, 0x37, 0x50, 0x2b, 0xdf, 0xae, 0x6d, 0xac, 0xac, - 0x27, 0x5d, 0xd6, 0x47, 0xb6, 0xcb, 0x86, 0x2a, 0xd7, 0x5d, 0x3d, 0x3d, 0x5f, 0xd3, 0xfe, 0x3a, - 0x5f, 0xc3, 0xfb, 0x21, 0xa3, 0x8e, 0xe3, 0x4f, 0x46, 0x69, 0x1d, 0xb9, 0xd2, 0x03, 0xdf, 0x87, - 0xd2, 0xd0, 0x5f, 0x84, 0x13, 0x66, 0xe4, 0x5a, 0xa8, 0xdd, 0xdc, 0xb8, 0x97, 0x75, 0xbb, 0x7a, - 0xf3, 0x7a, 0x44, 0xea, 0x79, 0x0b, 0x97, 0xc4, 0x05, 0xf8, 0x01, 0x54, 0x5c, 0x26, 0xe8, 0x94, - 0x0a, 0x6a, 0xe4, 0x95, 0x14, 0x23, 0x2b, 0xee, 0x33, 0x11, 0xda, 0x93, 0x7e, 0x9c, 0xef, 0x16, - 0x4e, 0xcf, 0xd7, 0x10, 0x49, 0xf9, 0x78, 0x13, 0x56, 0xf9, 0x73, 0x3b, 0x18, 0x3b, 0xf4, 0x90, - 0x39, 0x63, 0x8f, 0xba, 0x6c, 0x7c, 0x4c, 0x1d, 0x7b, 0x4a, 0x85, 0xed, 0x7b, 0xc6, 0xab, 0x72, - 0x0b, 0xb5, 0x2b, 0xe4, 0xff, 0x92, 0xf2, 0x44, 0x32, 0x06, 0xd4, 0x65, 0xdf, 0xa5, 0x79, 0x6b, - 0x0d, 0x20, 0xd3, 0x83, 0xcb, 0x90, 0xef, 0xec, 0xef, 0xea, 0x1a, 0xae, 0x40, 0x81, 0x1c, 0x3c, - 0xe9, 0xe9, 0xc8, 0x3a, 0x04, 0x3c, 0x14, 0x21, 0xa3, 0xee, 0xb5, 0xe9, 0xad, 0x42, 0x65, 0xc4, - 0x3c, 0xea, 0x89, 0xdd, 0x47, 0x06, 0x6a, 0xa1, 0x76, 0x95, 0xa4, 0x31, 0xfe, 0x1c, 0xca, 0x31, - 0x4d, 0x0d, 0xa2, 0xb6, 0x71, 0xe7, 0xed, 0x83, 0x20, 0x09, 0xcd, 0xfa, 0x06, 0x1a, 0x71, 0x82, - 0x07, 0xbe, 0xc7, 0x19, 0xc6, 0x50, 0x98, 0xf8, 0x53, 0xa6, 0x5a, 0x17, 0x89, 0x3a, 0x63, 0x03, - 0xca, 0x2e, 0xe3, 0x9c, 0xce, 0xa2, 0xf9, 0x56, 0x49, 0x12, 0x5a, 0xff, 0x20, 0x80, 0x6c, 0x5f, - 0xb8, 0x03, 0x25, 0x35, 0x8b, 0x64, 0xab, 0xff, 0xcb, 0xae, 0x57, 0x13, 0xd8, 0xa7, 0x76, 0xd8, - 0x5d, 0x89, 0x97, 0x5a, 0x57, 0x50, 0x67, 0x4a, 0x03, 0xc1, 0x42, 0x12, 0x17, 0xca, 0x47, 0xe0, - 0xd4, 0x0d, 0x1c, 0xc6, 0x8d, 0x9c, 0xea, 0xa1, 0x67, 0x3d, 0x86, 0x2a, 0xa1, 0xd6, 0xa0, 0x91, - 0x84, 0x86, 0xbf, 0x82, 0x2a, 0x7b, 0xc1, 0xdc, 0xc0, 0xa1, 0x21, 0x8f, 0x57, 0x88, 0xb3, 0x9a, - 0x5e, 0x9c, 0x8a, 0xab, 0x32, 0x2a, 0xbe, 0x0f, 0x30, 0xb7, 0xb9, 0xf0, 0x67, 0x21, 0x75, 0xb9, - 0x51, 0x78, 0x53, 0xf0, 0x4e, 0x92, 0x8b, 0x2b, 0xaf, 0x90, 0xad, 0x2f, 0xa1, 0x9a, 0x3e, 0x8f, - 0x9c, 0x98, 0x5c, 0xbd, 0x9a, 0x58, 0x9d, 0xa8, 0x33, 0x5e, 0x81, 0xe2, 0x31, 0x75, 0x16, 0xd1, - 0xbc, 0xea, 0x24, 0x0a, 0xac, 0x0e, 0x94, 0xa2, 0x47, 0xc8, 0xf2, 0xb2, 0x08, 0xc5, 0x79, 0x7c, - 0x0f, 0xea, 0xca, 0xd4, 0x82, 0xba, 0xc1, 0xd8, 0xe5, 0xaa, 0x38, 0x4f, 0x6a, 0x29, 0xd6, 0xe7, - 0xd6, 0x6f, 0x39, 0x68, 0x5e, 0x77, 0x25, 0xfe, 0x1a, 0x0a, 0xe2, 0x24, 0x88, 0x5a, 0x35, 0x37, - 0x3e, 0x78, 0x97, 0x7b, 0xe3, 0x70, 0x74, 0x12, 0x30, 0xa2, 0x0a, 0xf0, 0xa7, 0x80, 0x5d, 0x85, - 0x8d, 0x8f, 0xa8, 0x6b, 0x3b, 0x27, 0xca, 0xc1, 0xf1, 0x86, 0xf5, 0x28, 0xb3, 0xa5, 0x12, 0xd2, - 0xb8, 0xf2, 0x31, 0xe7, 0xcc, 0x09, 0x8c, 0x82, 0xca, 0xab, 0xb3, 0xc4, 0x16, 0x9e, 0x2d, 0x8c, - 0x62, 0x84, 0xc9, 0xb3, 0x75, 0x02, 0x90, 0xdd, 0x84, 0x6b, 0x50, 0x3e, 0x18, 0x3c, 0x1e, 0xec, - 0x3d, 0x1d, 0xe8, 0x9a, 0x0c, 0x1e, 0xee, 0x1d, 0x0c, 0x46, 0x3d, 0xa2, 0x23, 0x5c, 0x85, 0xe2, - 0x76, 0xe7, 0x60, 0xbb, 0xa7, 0xe7, 0x70, 0x03, 0xaa, 0x3b, 0xbb, 0xc3, 0xd1, 0xde, 0x36, 0xe9, - 0xf4, 0xf5, 0x3c, 0xc6, 0xd0, 0x54, 0x99, 0x0c, 0x2b, 0xc8, 0xd2, 0xe1, 0x41, 0xbf, 0xdf, 0x21, - 0xcf, 0xf4, 0xa2, 0x7c, 0x45, 0x76, 0x07, 0x5b, 0x7b, 0x7a, 0x09, 0xd7, 0xa1, 0x32, 0x1c, 0x75, - 0x46, 0xbd, 0x61, 0x6f, 0xa4, 0x97, 0xad, 0xc7, 0x50, 0x8a, 0xae, 0x7e, 0x0f, 0x46, 0xb4, 0x7e, - 0x46, 0x50, 0x49, 0xcc, 0xf3, 0x3e, 0x8c, 0x7d, 0xcd, 0x12, 0xef, 0x5c, 0x79, 0xfe, 0xe6, 0xca, - 0xcf, 0x8a, 0x50, 0x4d, 0xcd, 0x88, 0xef, 0x42, 0x75, 0xe2, 0x2f, 0x3c, 0x31, 0xb6, 0x3d, 0xa1, - 0x56, 0x5e, 0xd8, 0xd1, 0x48, 0x45, 0x41, 0xbb, 0x9e, 0xc0, 0xf7, 0xa0, 0x16, 0xa5, 0x8f, 0x1c, - 0x9f, 0x46, 0x5f, 0x01, 0xb4, 0xa3, 0x11, 0x50, 0xe0, 0x96, 0xc4, 0xb0, 0x0e, 0x79, 0xbe, 0x70, - 0xd5, 0x4d, 0x88, 0xc8, 0x23, 0xbe, 0x03, 0x25, 0x3e, 0x99, 0x33, 0x97, 0xaa, 0xe5, 0xde, 0x26, - 0x71, 0x84, 0x3f, 0x84, 0xe6, 0x8f, 0x2c, 0xf4, 0xc7, 0x62, 0x1e, 0x32, 0x3e, 0xf7, 0x9d, 0xa9, - 0x5a, 0x34, 0x22, 0x0d, 0x89, 0x8e, 0x12, 0x10, 0x7f, 0x14, 0xd3, 0x32, 0x5d, 0x25, 0xa5, 0x0b, - 0x91, 0xba, 0xc4, 0x1f, 0x26, 0xda, 0x3e, 0x01, 0xfd, 0x0a, 0x2f, 0x12, 0x58, 0x56, 0x02, 0x11, - 0x69, 0xa6, 0xcc, 0x48, 0x64, 0x07, 0x9a, 0x1e, 0x9b, 0x51, 0x61, 0x1f, 0xb3, 0x31, 0x0f, 0xa8, - 0xc7, 0x8d, 0xca, 0x9b, 0xbf, 0x13, 0xdd, 0xc5, 0xe4, 0x39, 0x13, 0xc3, 0x80, 0x7a, 0xf1, 0x1b, - 0xda, 0x48, 0x2a, 0x24, 0xc6, 0xf1, 0xc7, 0x70, 0x2b, 0x6d, 0x31, 0x65, 0x8e, 0xa0, 0xdc, 0xa8, - 0xb6, 0xf2, 0x6d, 0x4c, 0xd2, 0xce, 0x8f, 0x14, 0x7a, 0x8d, 0xa8, 0xb4, 0x71, 0x03, 0x5a, 0xf9, - 0x36, 0xca, 0x88, 0x4a, 0x98, 0xfc, 0xbc, 0x35, 0x03, 0x9f, 0xdb, 0x57, 0x44, 0xd5, 0xfe, 0x5b, - 0x54, 0x52, 0x91, 0x8a, 0x4a, 0x5b, 0xc4, 0xa2, 0xea, 0x91, 0xa8, 0x04, 0xce, 0x44, 0xa5, 0xc4, - 0x58, 0x54, 0x23, 0x12, 0x95, 0xc0, 0xb1, 0xa8, 0x4d, 0x80, 0x90, 0x71, 0x26, 0xc6, 0x73, 0x39, - 0xf9, 0xa6, 0xfa, 0x08, 0xdc, 0x7d, 0xcb, 0x67, 0x6c, 0x9d, 0x48, 0xd6, 0x8e, 0xed, 0x09, 0x52, - 0x0d, 0x93, 0xe3, 0x0d, 0xff, 0xdd, 0xba, 0xe9, 0xbf, 0x07, 0x50, 0x4d, 0x4b, 0xaf, 0xbf, 0xcf, - 0x65, 0xc8, 0x3f, 0xeb, 0x0d, 0x75, 0x84, 0x4b, 0x90, 0x1b, 0xec, 0xe9, 0xb9, 0xec, 0x9d, 0xce, - 0xaf, 0x16, 0x7e, 0xf9, 0xdd, 0x44, 0xdd, 0x32, 0x14, 0x95, 0xf8, 0x6e, 0x1d, 0x20, 0xdb, 0xbd, - 0xb5, 0x09, 0x90, 0x0d, 0x4a, 0xda, 0xcf, 0x3f, 0x3a, 0xe2, 0x2c, 0xf2, 0xf3, 0x6d, 0x12, 0x47, - 0x12, 0x77, 0x98, 0x37, 0x13, 0x73, 0x65, 0xe3, 0x06, 0x89, 0xa3, 0xee, 0xb7, 0x67, 0x17, 0xa6, - 0xf6, 0xf2, 0xc2, 0xd4, 0x5e, 0x5f, 0x98, 0xe8, 0xa7, 0xa5, 0x89, 0xfe, 0x58, 0x9a, 0xe8, 0x74, - 0x69, 0xa2, 0xb3, 0xa5, 0x89, 0xfe, 0x5e, 0x9a, 0xe8, 0xd5, 0xd2, 0xd4, 0x5e, 0x2f, 0x4d, 0xf4, - 0xeb, 0xa5, 0xa9, 0x9d, 0x5d, 0x9a, 0xda, 0xcb, 0x4b, 0x53, 0xfb, 0x3e, 0xfd, 0x9b, 0x72, 0x58, - 0x52, 0xff, 0x4b, 0xbe, 0xf8, 0x37, 0x00, 0x00, 0xff, 0xff, 0x62, 0x2d, 0x55, 0x17, 0xc7, 0x08, - 0x00, 0x00, + // 1121 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x56, 0x4b, 0x8f, 0x1b, 0x45, + 0x10, 0x9e, 0xf6, 0xf8, 0x35, 0xb5, 0xb6, 0x33, 0x69, 0x56, 0x61, 0xb4, 0x22, 0xb3, 0xce, 0x20, + 0xc0, 0x42, 0x68, 0x41, 0x8b, 0x00, 0x25, 0x8a, 0x90, 0xec, 0xc4, 0xc9, 0xae, 0x12, 0x7b, 0x57, + 0x6d, 0x2f, 0x51, 0xb8, 0x58, 0xbd, 0x76, 0xaf, 0x3d, 0xca, 0xbc, 0x98, 0x6e, 0x47, 0x59, 0x4e, + 0x9c, 0x10, 0x47, 0x2e, 0x5c, 0xb8, 0x72, 0xe1, 0x17, 0xf0, 0x1b, 0x72, 0xdc, 0x63, 0xc4, 0x21, + 0x22, 0xce, 0x25, 0xc7, 0x1c, 0xf8, 0x01, 0xa8, 0x7b, 0x5e, 0xde, 0x3c, 0xc4, 0x25, 0xb7, 0xae, + 0xaf, 0x1e, 0xf3, 0x75, 0xd5, 0x57, 0x6d, 0x43, 0x63, 0x1a, 0xc6, 0x82, 0x3d, 0xda, 0x89, 0xe2, + 0x50, 0x84, 0xb8, 0x9e, 0x58, 0xd1, 0xf1, 0xd6, 0xe6, 0x3c, 0x9c, 0x87, 0x0a, 0xfc, 0x5c, 0x9e, + 0x12, 0xbf, 0xf3, 0x57, 0x09, 0x1a, 0xf7, 0x62, 0x57, 0x30, 0xc2, 0x7e, 0x58, 0x32, 0x2e, 0xf0, + 0x21, 0x80, 0x70, 0x7d, 0xc6, 0x59, 0xec, 0x32, 0x6e, 0xa1, 0xb6, 0xde, 0xd9, 0xd8, 0xdd, 0xdc, + 0xc9, 0xaa, 0xec, 0x8c, 0x5d, 0x9f, 0x8d, 0x94, 0xaf, 0xb7, 0xf5, 0xf8, 0xe9, 0xb6, 0xf6, 0xf7, + 0xd3, 0x6d, 0x7c, 0x18, 0x33, 0xea, 0x79, 0xe1, 0x74, 0x9c, 0xe7, 0x91, 0xb5, 0x1a, 0xf8, 0x2a, + 0x54, 0x47, 0xe1, 0x32, 0x9e, 0x32, 0xab, 0xd4, 0x46, 0x9d, 0xd6, 0xee, 0x95, 0xa2, 0xda, 0xfa, + 0x97, 0x77, 0x92, 0xa0, 0x7e, 0xb0, 0xf4, 0x49, 0x9a, 0x80, 0xaf, 0x41, 0xdd, 0x67, 0x82, 0xce, + 0xa8, 0xa0, 0x96, 0xae, 0xa8, 0x58, 0x45, 0xf2, 0x80, 0x89, 0xd8, 0x9d, 0x0e, 0x52, 0x7f, 0xaf, + 0xfc, 0xf8, 0xe9, 0x36, 0x22, 0x79, 0x3c, 0xbe, 0x0e, 0x5b, 0xfc, 0x81, 0x1b, 0x4d, 0x3c, 0x7a, + 0xcc, 0xbc, 0x49, 0x40, 0x7d, 0x36, 0x79, 0x48, 0x3d, 0x77, 0x46, 0x85, 0x1b, 0x06, 0xd6, 0x8b, + 0x5a, 0x1b, 0x75, 0xea, 0xe4, 0x7d, 0x19, 0x72, 0x57, 0x46, 0x0c, 0xa9, 0xcf, 0xbe, 0xcb, 0xfd, + 0xce, 0x36, 0x40, 0xc1, 0x07, 0xd7, 0x40, 0xef, 0x1e, 0xee, 0x9b, 0x1a, 0xae, 0x43, 0x99, 0x1c, + 0xdd, 0xed, 0x9b, 0xc8, 0x39, 0x06, 0x3c, 0x12, 0x31, 0xa3, 0xfe, 0xb9, 0xee, 0x6d, 0x41, 0x7d, + 0xcc, 0x02, 0x1a, 0x88, 0xfd, 0x9b, 0x16, 0x6a, 0xa3, 0x8e, 0x41, 0x72, 0x1b, 0x7f, 0x01, 0xb5, + 0x34, 0x4c, 0x35, 0x62, 0x63, 0xf7, 0xd2, 0x9b, 0x1b, 0x41, 0xb2, 0x30, 0xe7, 0x37, 0x04, 0xcd, + 0xd4, 0xc3, 0xa3, 0x30, 0xe0, 0x0c, 0x63, 0x28, 0x4f, 0xc3, 0x19, 0x53, 0xb5, 0x2b, 0x44, 0x9d, + 0xb1, 0x05, 0x35, 0x9f, 0x71, 0x4e, 0xe7, 0x49, 0x83, 0x0d, 0x92, 0x99, 0xd2, 0x33, 0xa2, 0x7e, + 0xe4, 0x31, 0x6e, 0xe9, 0x6d, 0xd4, 0xd1, 0x49, 0x66, 0x62, 0x1b, 0x60, 0xcf, 0xe5, 0x22, 0x9c, + 0xc7, 0xd4, 0xe7, 0x56, 0x59, 0x39, 0xd7, 0x10, 0xfc, 0x01, 0x18, 0xfd, 0x47, 0xcc, 0x8f, 0x3c, + 0x1a, 0x73, 0xab, 0xa2, 0xdc, 0x05, 0xe0, 0xfc, 0x8b, 0x00, 0x0a, 0x21, 0xe0, 0x2e, 0x54, 0x55, + 0x93, 0x33, 0xb9, 0xbc, 0x57, 0xdc, 0x4b, 0xb5, 0xf6, 0x90, 0xba, 0x71, 0x6f, 0x33, 0x55, 0x4b, + 0x43, 0x41, 0xdd, 0x19, 0x8d, 0x04, 0x8b, 0x49, 0x9a, 0x28, 0x7b, 0xc3, 0x53, 0xa6, 0x25, 0x55, + 0xc3, 0x2c, 0x6a, 0x24, 0x9c, 0xd5, 0x7c, 0x35, 0x92, 0x85, 0xe1, 0xaf, 0xc1, 0x60, 0x39, 0xc3, + 0x44, 0x1b, 0xb8, 0xc8, 0xc9, 0xb8, 0xa6, 0x59, 0x45, 0x28, 0xbe, 0x0a, 0xb0, 0x58, 0xbf, 0xf9, + 0x2b, 0x84, 0xf3, 0x1e, 0xa4, 0x99, 0x6b, 0xc1, 0xce, 0x57, 0x60, 0xe4, 0xf7, 0x91, 0x93, 0x90, + 0x9a, 0x52, 0x93, 0x68, 0x10, 0x75, 0xc6, 0x9b, 0x50, 0x79, 0x48, 0xbd, 0x65, 0x32, 0x87, 0x06, + 0x49, 0x0c, 0xa7, 0x0b, 0xd5, 0xe4, 0x0a, 0x85, 0x5f, 0x26, 0xa1, 0xd4, 0x8f, 0xaf, 0x40, 0x43, + 0x6d, 0x8b, 0xa0, 0x7e, 0x34, 0xf1, 0xb9, 0x4a, 0xd6, 0xc9, 0x46, 0x8e, 0x0d, 0xb8, 0xf3, 0x7b, + 0x09, 0x5a, 0xe7, 0xe5, 0x8e, 0xbf, 0x81, 0xb2, 0x38, 0x8d, 0x92, 0x52, 0xad, 0xdd, 0x0f, 0xdf, + 0xb6, 0x16, 0xa9, 0x39, 0x3e, 0x8d, 0x18, 0x51, 0x09, 0xf8, 0x33, 0xc0, 0xbe, 0xc2, 0x26, 0x27, + 0xd4, 0x77, 0xbd, 0x53, 0xb5, 0x1a, 0xa9, 0x72, 0xcc, 0xc4, 0x73, 0x4b, 0x39, 0xe4, 0x46, 0xc8, + 0x6b, 0x2e, 0x98, 0x17, 0x29, 0x89, 0x18, 0x44, 0x9d, 0x25, 0xb6, 0x0c, 0x5c, 0xa1, 0x74, 0x61, + 0x10, 0x75, 0x76, 0x4e, 0x01, 0x8a, 0x2f, 0xe1, 0x0d, 0xa8, 0x1d, 0x0d, 0xef, 0x0c, 0x0f, 0xee, + 0x0d, 0x4d, 0x4d, 0x1a, 0x37, 0x0e, 0x8e, 0x86, 0xe3, 0x3e, 0x31, 0x11, 0x36, 0xa0, 0x72, 0xbb, + 0x7b, 0x74, 0xbb, 0x6f, 0x96, 0x70, 0x13, 0x8c, 0xbd, 0xfd, 0xd1, 0xf8, 0xe0, 0x36, 0xe9, 0x0e, + 0x4c, 0x1d, 0x63, 0x68, 0x29, 0x4f, 0x81, 0x95, 0x65, 0xea, 0xe8, 0x68, 0x30, 0xe8, 0x92, 0xfb, + 0x66, 0x45, 0xee, 0xde, 0xfe, 0xf0, 0xd6, 0x81, 0x59, 0xc5, 0x0d, 0xa8, 0x8f, 0xc6, 0xdd, 0x71, + 0x7f, 0xd4, 0x1f, 0x9b, 0x35, 0xe7, 0x0e, 0x54, 0x93, 0x4f, 0xbf, 0x03, 0x21, 0x3a, 0x3f, 0x23, + 0xa8, 0x67, 0xe2, 0x79, 0x17, 0xc2, 0x3e, 0x27, 0x89, 0xb7, 0x8e, 0x5c, 0x7f, 0x7d, 0xe4, 0x67, + 0x15, 0x30, 0x72, 0x31, 0xe2, 0xcb, 0x60, 0x4c, 0xc3, 0x65, 0x20, 0x26, 0x6e, 0x20, 0xd4, 0xc8, + 0xcb, 0x7b, 0x1a, 0xa9, 0x2b, 0x68, 0x3f, 0x10, 0xf8, 0x0a, 0x6c, 0x24, 0xee, 0x13, 0x2f, 0xa4, + 0xc9, 0xf3, 0x82, 0xf6, 0x34, 0x02, 0x0a, 0xbc, 0x25, 0x31, 0x6c, 0x82, 0xce, 0x97, 0xbe, 0xfa, + 0x12, 0x22, 0xf2, 0x88, 0x2f, 0x41, 0x95, 0x4f, 0x17, 0xcc, 0xa7, 0x6a, 0xb8, 0x17, 0x49, 0x6a, + 0xe1, 0x8f, 0xa0, 0xf5, 0x23, 0x8b, 0xc3, 0x89, 0x58, 0xc4, 0x8c, 0x2f, 0x42, 0x6f, 0xa6, 0x06, + 0x8d, 0x48, 0x53, 0xa2, 0xe3, 0x0c, 0xc4, 0x1f, 0xa7, 0x61, 0x05, 0xaf, 0xaa, 0xe2, 0x85, 0x48, + 0x43, 0xe2, 0x37, 0x32, 0x6e, 0x9f, 0x82, 0xb9, 0x16, 0x97, 0x10, 0xac, 0x29, 0x82, 0x88, 0xb4, + 0xf2, 0xc8, 0x84, 0x64, 0x17, 0x5a, 0x01, 0x9b, 0x53, 0xe1, 0x3e, 0x64, 0x13, 0x1e, 0xd1, 0x80, + 0x5b, 0xf5, 0x57, 0x7f, 0x80, 0x7a, 0xcb, 0xe9, 0x03, 0x26, 0x46, 0x11, 0x0d, 0xd2, 0x0d, 0x6d, + 0x66, 0x19, 0x12, 0xe3, 0xf8, 0x13, 0xb8, 0x90, 0x97, 0x98, 0x31, 0x4f, 0x50, 0x6e, 0x19, 0x6d, + 0xbd, 0x83, 0x49, 0x5e, 0xf9, 0xa6, 0x42, 0xcf, 0x05, 0x2a, 0x6e, 0xdc, 0x82, 0xb6, 0xde, 0x41, + 0x45, 0xa0, 0x22, 0x26, 0x9f, 0xb7, 0x56, 0x14, 0x72, 0x77, 0x8d, 0xd4, 0xc6, 0xff, 0x93, 0xca, + 0x32, 0x72, 0x52, 0x79, 0x89, 0x94, 0x54, 0x23, 0x21, 0x95, 0xc1, 0x05, 0xa9, 0x3c, 0x30, 0x25, + 0xd5, 0x4c, 0x48, 0x65, 0x70, 0x4a, 0xea, 0x3a, 0x40, 0xcc, 0x38, 0x13, 0x93, 0x85, 0xec, 0x7c, + 0x4b, 0x3d, 0x02, 0x97, 0xdf, 0xf0, 0x8c, 0xed, 0x10, 0x19, 0xb5, 0xe7, 0x06, 0x82, 0x18, 0x71, + 0x76, 0x7c, 0x4d, 0x7f, 0x17, 0x5e, 0xd7, 0xdf, 0x35, 0x30, 0xf2, 0xd4, 0xf3, 0xfb, 0x5c, 0x03, + 0xfd, 0x7e, 0x7f, 0x64, 0x22, 0x5c, 0x85, 0xd2, 0xf0, 0xc0, 0x2c, 0x15, 0x3b, 0xad, 0x6f, 0x95, + 0x7f, 0xf9, 0xc3, 0x46, 0xbd, 0x1a, 0x54, 0x14, 0xf9, 0x5e, 0x03, 0xa0, 0x98, 0xbd, 0x73, 0x1d, + 0xa0, 0x68, 0x94, 0x94, 0x5f, 0x78, 0x72, 0xc2, 0x59, 0xa2, 0xe7, 0x8b, 0x24, 0xb5, 0x24, 0xee, + 0xb1, 0x60, 0x2e, 0x16, 0x4a, 0xc6, 0x4d, 0x92, 0x5a, 0xbd, 0x6f, 0xcf, 0x9e, 0xd9, 0xda, 0x93, + 0x67, 0xb6, 0xf6, 0xf2, 0x99, 0x8d, 0x7e, 0x5a, 0xd9, 0xe8, 0xcf, 0x95, 0x8d, 0x1e, 0xaf, 0x6c, + 0x74, 0xb6, 0xb2, 0xd1, 0x3f, 0x2b, 0x1b, 0xbd, 0x58, 0xd9, 0xda, 0xcb, 0x95, 0x8d, 0x7e, 0x7d, + 0x6e, 0x6b, 0x67, 0xcf, 0x6d, 0xed, 0xc9, 0x73, 0x5b, 0xfb, 0x3e, 0xff, 0xff, 0x73, 0x5c, 0x55, + 0x7f, 0x78, 0xbe, 0xfc, 0x2f, 0x00, 0x00, 0xff, 0xff, 0x31, 0x1d, 0xb9, 0x5e, 0x20, 0x09, 0x00, + 0x00, } func (x WriteRequest_SourceEnum) String() string { @@ -1098,6 +1127,15 @@ func (this *WriteResponse) Equal(that interface{}) bool { if this.Message != that1.Message { return false } + if this.Samples != that1.Samples { + return false + } + if this.Histograms != that1.Histograms { + return false + } + if this.Exemplars != that1.Exemplars { + return false + } return true } func (this *TimeSeries) Equal(that interface{}) bool { @@ -1561,10 +1599,13 @@ func (this *WriteResponse) GoString() string { if this == nil { return "nil" } - s := make([]string, 0, 6) + s := make([]string, 0, 9) s = append(s, "&cortexpb.WriteResponse{") s = append(s, "Code: "+fmt.Sprintf("%#v", this.Code)+",\n") s = append(s, "Message: "+fmt.Sprintf("%#v", this.Message)+",\n") + s = append(s, "Samples: "+fmt.Sprintf("%#v", this.Samples)+",\n") + s = append(s, "Histograms: "+fmt.Sprintf("%#v", this.Histograms)+",\n") + s = append(s, "Exemplars: "+fmt.Sprintf("%#v", this.Exemplars)+",\n") s = append(s, "}") return strings.Join(s, "") } @@ -1875,6 +1916,21 @@ func (m *WriteResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l + if m.Exemplars != 0 { + i = encodeVarintCortex(dAtA, i, uint64(m.Exemplars)) + i-- + dAtA[i] = 0x28 + } + if m.Histograms != 0 { + i = encodeVarintCortex(dAtA, i, uint64(m.Histograms)) + i-- + dAtA[i] = 0x20 + } + if m.Samples != 0 { + i = encodeVarintCortex(dAtA, i, uint64(m.Samples)) + i-- + dAtA[i] = 0x18 + } if len(m.Message) > 0 { i -= len(m.Message) copy(dAtA[i:], m.Message) @@ -2475,6 +2531,15 @@ func (m *WriteResponse) Size() (n int) { if l > 0 { n += 1 + l + sovCortex(uint64(l)) } + if m.Samples != 0 { + n += 1 + sovCortex(uint64(m.Samples)) + } + if m.Histograms != 0 { + n += 1 + sovCortex(uint64(m.Histograms)) + } + if m.Exemplars != 0 { + n += 1 + sovCortex(uint64(m.Exemplars)) + } return n } @@ -2758,6 +2823,9 @@ func (this *WriteResponse) String() string { s := strings.Join([]string{`&WriteResponse{`, `Code:` + fmt.Sprintf("%v", this.Code) + `,`, `Message:` + fmt.Sprintf("%v", this.Message) + `,`, + `Samples:` + fmt.Sprintf("%v", this.Samples) + `,`, + `Histograms:` + fmt.Sprintf("%v", this.Histograms) + `,`, + `Exemplars:` + fmt.Sprintf("%v", this.Exemplars) + `,`, `}`, }, "") return s @@ -3299,6 +3367,63 @@ func (m *WriteResponse) Unmarshal(dAtA []byte) error { } m.Message = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Samples", wireType) + } + m.Samples = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowCortex + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Samples |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Histograms", wireType) + } + m.Histograms = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowCortex + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Histograms |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } + case 5: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Exemplars", wireType) + } + m.Exemplars = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowCortex + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Exemplars |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := skipCortex(dAtA[iNdEx:]) diff --git a/pkg/cortexpb/cortex.proto b/pkg/cortexpb/cortex.proto index 189e8955f7c..c83501a1d51 100644 --- a/pkg/cortexpb/cortex.proto +++ b/pkg/cortexpb/cortex.proto @@ -29,14 +29,12 @@ message StreamWriteRequest { message WriteResponse { int32 code = 1; string message = 2; -} -message WriteResponse { // Samples represents X-Prometheus-Remote-Write-Written-Samples - int64 Samples = 1; + int64 Samples = 3; // Histograms represents X-Prometheus-Remote-Write-Written-Histograms - int64 Histograms = 2; + int64 Histograms = 4; // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars - int64 Exemplars = 3; + int64 Exemplars = 5; } message TimeSeries { diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go index 68dfc1fe519..8317eed6e31 100644 --- a/pkg/distributor/distributor.go +++ b/pkg/distributor/distributor.go @@ -1156,13 +1156,14 @@ func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, time d.inflightClientRequests.Inc() defer d.inflightClientRequests.Dec() + var resp *cortexpb.WriteResponse if d.cfg.UseStreamPush { req := &cortexpb.WriteRequest{ Timeseries: timeseries, Metadata: metadata, Source: source, } - _, err = c.PushStreamConnection(ctx, req) + resp, err = c.PushStreamConnection(ctx, req) } else { req := cortexpb.PreallocWriteRequestFromPool() req.Timeseries = timeseries