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..17049620f68 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -2695,6 +2695,11 @@ ha_tracker: # 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/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..4ebcc142077 --- /dev/null +++ b/integration/remote_write_v2_test.go @@ -0,0 +1,390 @@ +//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_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 + + 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.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 91773fbc0f1..c83501a1d51 100644 --- a/pkg/cortexpb/cortex.proto +++ b/pkg/cortexpb/cortex.proto @@ -29,6 +29,12 @@ message StreamWriteRequest { message WriteResponse { int32 code = 1; string message = 2; + // Samples represents X-Prometheus-Remote-Write-Written-Samples + int64 Samples = 3; + // Histograms represents X-Prometheus-Remote-Write-Written-Histograms + int64 Histograms = 4; + // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars + int64 Exemplars = 5; } message TimeSeries { 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..8317eed6e31 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 @@ -1145,20 +1156,21 @@ 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 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 +1192,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..bbe3e8d489c 100644 --- a/pkg/util/push/push.go +++ b/pkg/util/push/push.go @@ -2,22 +2,43 @@ 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" +) + // 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 +49,245 @@ 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) - if err != nil { - level.Error(logger).Log("err", err.Error()) - http.Error(w, err.Error(), http.StatusBadRequest) - return + + handlePRW1 := func() { + 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 + } + + 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)) + } } - req.SkipLabelNameValidation = false - if req.Source == 0 { - req.Source = cortexpb.API + 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, err := convertV2RequestToV1(&req) + if err != nil { + level.Error(logger).Log("err", err.Error()) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + 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 { + setPRW2RespHeader(w, resp.Samples, resp.Histograms, resp.Exemplars) + } } - if _, err := push(ctx, &req.WriteRequest); err != nil { - resp, ok := httpgrpc.HTTPResponseFromError(err) - if !ok { - http.Error(w, err.Error(), http.StatusInternalServerError) + 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 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) + + 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 } - http.Error(w, string(resp.Body), int(resp.Code)) + + 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 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)) +} + +// 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..46cb0770f75 100644 --- a/pkg/util/push/push_test.go +++ b/pkg/util/push/push_test.go @@ -3,13 +3,17 @@ 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" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/weaveworks/common/middleware" @@ -17,30 +21,374 @@ 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 + + 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]) + }) +} + +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 +402,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 createRequest(t *testing.T, protobuf []byte) *http.Request { +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, 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 +500,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 +514,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()