Ruler: split oversized remote distributor writes

CHANGELOG.md

@@ -9,8 +9,10 @@
 * [ENHANCEMENT] Query-frontend: Improve the stability of cardinality estimates and therefore sharding factors for queries when running splitting and caching inside MQE is enabled, or range vector splitting is enabled. #16274 #16301
   * When running splitting and caching inside MQE is enabled, the `cortex_query_frontend_cardinality_estimation_difference` metric will no longer be emitted.
 * [ENHANCEMENT] Distributor: Add the experimental `cortex_distributor_otlp_requests_with_job_or_instance_resource_attribute_total{user}` counter to track OTLP requests carrying `job` or `instance` as a resource attribute. #16285
+* [ENHANCEMENT] Ruler: Split oversized remote distributor writes to keep resulting calls within the configured gRPC maximum send size, and expose the number of generated requests in `cortex_ruler_remote_distributor_requests_per_write_request`. #16160
 * [BUGFIX] Query-frontend: Return a HTTP 500 error rather than a HTTP 400 when a querier receives a query plan that is too new. #16233
 * [BUGFIX] Compactor, Store-gateway: Fix the store-gateway always logging `num_series=0` in its `loaded new block` message. #16276
+* [BUGFIX] Ingest storage: Account for protobuf framing when splitting Remote Write 1.0 requests so generated Kafka record data stays within `-ingest-storage.kafka.producer-max-record-size-bytes` when individual series and metadata entries fit. #16160
 
 ### Mixin
 

docs/sources/mimir/references/architecture/components/ruler/index.md

@@ -57,6 +57,16 @@ To push rule-result series to remote distributors over native gRPC instead, set
 Most deployments only need to set this address.
 `ruler.distributor.remote_timeout` configures the per-request timeout, and `ruler.distributor.grpc_client_config` provides advanced standard gRPC client tuning such as TLS, message sizes, compression, retries, and cluster validation.
 
+The ruler splits remote writes along series boundaries so that they fit within `ruler.distributor.grpc_client_config.max_send_msg_size`.
+When compression is enabled, the ruler reserves a conservative amount of space for compression framing, so it can split a write slightly below the configured transport limit.
+It sends the resulting requests sequentially, with independent timeouts and retries; each request also consumes a separate gRPC client rate-limit token when rate limiting is enabled.
+Requests accepted before a later request fails aren't rolled back.
+The ruler emits one float or native-histogram sample per result series for each evaluation.
+Consequently, an individual result series exceeding the effective split limit cannot be losslessly subdivided.
+If this occurs, increase `ruler.distributor.grpc_client_config.max_send_msg_size` and configure the distributor's `server.grpc_server_max_recv_msg_size` to be at least as large.
+When compression is enabled, a series above the conservative effective limit may still succeed if its compressed payload fits within the configured transport limit; otherwise gRPC returns `ResourceExhausted`.
+The `cortex_ruler_remote_distributor_requests_per_write_request` histogram reports how many remote requests each ruler write produced.
+
 In Kubernetes deployments, point `ruler.distributor.address` at the distributor headless service on the gRPC port when you want gRPC client-side load balancing.
 A normal ClusterIP service can work for connectivity, but it doesn't provide the intended per-RPC client-side balancing across distributor endpoints.
 

pkg/mimirpb/split.go

@@ -15,6 +15,9 @@ package mimirpb
 //
 // The returned requests may still retain references to fields in the original WriteRequest, i.e. they are tied to its lifecycle.
 func SplitWriteRequestByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []*WriteRequest {
+	if maxSize <= 0 {
+		return []*WriteRequest{req}
+	}
 	if reqSize <= maxSize {
 		return []*WriteRequest{req}
 	}
@@ -131,55 +134,48 @@ func splitTimeseriesByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []
 		return nil
 	}
 
-	newPartialReq := func() (*WriteRequest, int) {
-		r := &WriteRequest{
-			Source:                          req.Source,
-			SkipLabelValidation:             req.SkipLabelValidation,
-			skipNormalizeMetadataMetricName: req.skipNormalizeMetadataMetricName,
-			skipDeduplicateMetadata:         req.skipDeduplicateMetadata,
-		}
-
-		return r, r.Size()
-	}
-
 	// The partial requests returned by this function will not contain any Metadata,
 	// so we first compute the request size without it.
 	reqSizeWithoutMetadata := reqSize - req.MetadataSize()
 	if reqSizeWithoutMetadata <= maxSize {
-		partialReq, _ := newPartialReq()
+		partialReq := newPartialWriteRequest(req)
 		partialReq.Timeseries = req.Timeseries
 		return []*WriteRequest{partialReq}
 	}
 
 	// We assume that different timeseries roughly have the same size (no huge outliers)
-	// so we preallocate the returned slice just adding 1 extra item (+2 because a +1 is to round up).
-	estimatedPartialReqs := (reqSizeWithoutMetadata / maxSize) + 2
+	// so we preallocate the returned slice just adding 1 extra item (+2 because a +1 is to round up),
+	// capped at the number of timeseries.
+	estimatedPartialReqs := min((reqSizeWithoutMetadata/maxSize)+2, len(req.Timeseries))
 	partialReqs := make([]*WriteRequest, 0, estimatedPartialReqs)
 
 	// Split timeseries into partial write requests.
-	nextReq, nextReqSize := newPartialReq()
+	nextReq := newPartialWriteRequest(req)
+	nextReqSize := nextReq.Size()
 	nextReqTimeseriesStart := 0
 	nextReqTimeseriesLength := 0
 
 	for i := 0; i < len(req.Timeseries); i++ {
 		seriesSize := req.Timeseries[i].Size()
+		seriesFieldSize := embeddedMessageFieldSize(seriesSize)
 
 		// Check if the next partial request is full (or close to be full), and so it's time to finalize it and create a new one.
 		// If the next partial request doesn't have any timeseries yet, we add the series anyway, in order to avoid an infinite loop
 		// if a single timeseries is bigger than the limit.
-		if nextReqSize+seriesSize > maxSize && nextReqTimeseriesLength > 0 {
+		if nextReqSize+seriesFieldSize > maxSize && nextReqTimeseriesLength > 0 {
 			// Finalize the next partial request.
 			nextReq.Timeseries = req.Timeseries[nextReqTimeseriesStart : nextReqTimeseriesStart+nextReqTimeseriesLength]
 			partialReqs = append(partialReqs, nextReq)
 
 			// Initialize a new partial request.
-			nextReq, nextReqSize = newPartialReq()
+			nextReq = newPartialWriteRequest(req)
+			nextReqSize = nextReq.Size()
 			nextReqTimeseriesStart = i
 			nextReqTimeseriesLength = 0
 		}
 
 		// Add the current series to next partial request.
-		nextReqSize += seriesSize + 1 + sovMimir(uint64(seriesSize)) // Math copied from Size().
+		nextReqSize += seriesFieldSize
 		nextReqTimeseriesLength++
 	}
 
@@ -197,55 +193,48 @@ func splitMetadataByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []*W
 		return nil
 	}
 
-	newPartialReq := func() (*WriteRequest, int) {
-		r := &WriteRequest{
-			Source:                          req.Source,
-			SkipLabelValidation:             req.SkipLabelValidation,
-			skipUnmarshalingExemplars:       req.skipUnmarshalingExemplars,
-			skipNormalizeMetadataMetricName: req.skipNormalizeMetadataMetricName,
-			skipDeduplicateMetadata:         req.skipDeduplicateMetadata,
-		}
-		return r, r.Size()
-	}
-
 	// The partial requests returned by this function will not contain any Timeseries,
 	// so we first compute the request size without it.
 	reqSizeWithoutTimeseries := reqSize - req.TimeseriesSize()
 	if reqSizeWithoutTimeseries <= maxSize {
-		partialReq, _ := newPartialReq()
+		partialReq := newPartialWriteRequest(req)
 		partialReq.Metadata = req.Metadata
 		return []*WriteRequest{partialReq}
 	}
 
 	// We assume that different metadata roughly have the same size (no huge outliers)
-	// so we preallocate the returned slice just adding 1 extra item (+2 because a +1 is to round up).
-	estimatedPartialReqs := (reqSizeWithoutTimeseries / maxSize) + 2
+	// so we preallocate the returned slice just adding 1 extra item (+2 because a +1 is to round up),
+	// capped at the number of metadata entries.
+	estimatedPartialReqs := min((reqSizeWithoutTimeseries/maxSize)+2, len(req.Metadata))
 	partialReqs := make([]*WriteRequest, 0, estimatedPartialReqs)
 
 	// Split metadata into partial write requests.
-	nextReq, nextReqSize := newPartialReq()
+	nextReq := newPartialWriteRequest(req)
+	nextReqSize := nextReq.Size()
 	nextReqMetadataStart := 0
 	nextReqMetadataLength := 0
 
 	for i := 0; i < len(req.Metadata); i++ {
 		metadataSize := req.Metadata[i].Size()
+		metadataFieldSize := embeddedMessageFieldSize(metadataSize)
 
 		// Check if the next partial request is full (or close to be full), and so it's time to finalize it and create a new one.
 		// If the next partial request doesn't have any metadata yet, we add the metadata anyway, in order to avoid an infinite loop
 		// if a single metadata is bigger than the limit.
-		if nextReqSize+metadataSize > maxSize && nextReqMetadataLength > 0 {
+		if nextReqSize+metadataFieldSize > maxSize && nextReqMetadataLength > 0 {
 			// Finalize the next partial request.
 			nextReq.Metadata = req.Metadata[nextReqMetadataStart : nextReqMetadataStart+nextReqMetadataLength]
 			partialReqs = append(partialReqs, nextReq)
 
 			// Initialize a new partial request.
-			nextReq, nextReqSize = newPartialReq()
+			nextReq = newPartialWriteRequest(req)
+			nextReqSize = nextReq.Size()
 			nextReqMetadataStart = i
 			nextReqMetadataLength = 0
 		}
 
 		// Add the current metadata to next partial request.
-		nextReqSize += metadataSize + 1 + sovMimir(uint64(metadataSize)) // Math copied from Size().
+		nextReqSize += metadataFieldSize
 		nextReqMetadataLength++
 	}
 
@@ -258,6 +247,21 @@ func splitMetadataByMaxMarshalSize(req *WriteRequest, reqSize, maxSize int) []*W
 	return partialReqs
 }
 
+func newPartialWriteRequest(req *WriteRequest) *WriteRequest {
+	return &WriteRequest{
+		Source:                          req.Source,
+		SkipLabelValidation:             req.SkipLabelValidation,
+		SkipLabelCountValidation:        req.SkipLabelCountValidation,
+		skipUnmarshalingExemplars:       req.skipUnmarshalingExemplars,
+		skipNormalizeMetadataMetricName: req.skipNormalizeMetadataMetricName,
+		skipDeduplicateMetadata:         req.skipDeduplicateMetadata,
+	}
+}
+
+func embeddedMessageFieldSize(messageSize int) int {
+	return 1 + messageSize + sovMimir(uint64(messageSize))
+}
+
 // maxSeriesSizeAfterResymbolization calculates an upper bound for the size of the given TimeSeries, and its referenced symbols.
 // It is only an upper bound. The actual series might end up being smaller if it re-uses symbols or has low magnitude references.
 func maxRW2SeriesSizeAfterResymbolization(ts *TimeSeriesRW2, symbols []string, symbolOffset uint32) (seriesSize int, symbolsSize int) {

pkg/mimirpb/split_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/mem"
 )
 
 func TestSplitWriteRequestByMaxMarshalSize(t *testing.T) {
@@ -57,6 +58,14 @@ func TestSplitWriteRequestByMaxMarshalSize(t *testing.T) {
 		assert.Equal(t, reqv2, partials[0])
 	})
 
+	t.Run("should return the input WriteRequest for a non-positive size limit", func(t *testing.T) {
+		for _, limit := range []int{0, -1} {
+			partials := SplitWriteRequestByMaxMarshalSize(reqv1, reqv1.Size(), limit)
+			require.Len(t, partials, 1)
+			require.Same(t, reqv1, partials[0])
+		}
+	})
+
 	t.Run("should split the input WriteRequest into multiple requests, honoring the size limit", func(t *testing.T) {
 		const limit = 100
 
@@ -286,6 +295,77 @@ func TestSplitWriteRequestByMaxMarshalSize(t *testing.T) {
 		}
 	})
 
+	t.Run("should not preallocate more partial request slots than entities", func(t *testing.T) {
+		const limit = 1
+
+		timeseriesReq := &WriteRequest{Timeseries: reqv1.Timeseries}
+		timeseriesPartials := SplitWriteRequestByMaxMarshalSize(timeseriesReq, timeseriesReq.Size(), limit)
+		assert.LessOrEqual(t, cap(timeseriesPartials), len(timeseriesReq.Timeseries))
+
+		metadataReq := &WriteRequest{Metadata: reqv1.Metadata}
+		metadataPartials := SplitWriteRequestByMaxMarshalSize(metadataReq, metadataReq.Size(), limit)
+		assert.LessOrEqual(t, cap(metadataPartials), len(metadataReq.Metadata))
+
+	})
+
+	t.Run("should account for embedded message framing when selecting a partial request", func(t *testing.T) {
+		timeseriesReq := &WriteRequest{
+			Source:     RULE,
+			Timeseries: reqv1.Timeseries,
+		}
+		baseSize := newPartialWriteRequest(timeseriesReq).Size()
+		limit := baseSize + embeddedMessageFieldSize(timeseriesReq.Timeseries[0].Size()) + timeseriesReq.Timeseries[1].Size()
+		timeseriesPartials := SplitWriteRequestByMaxMarshalSize(timeseriesReq, timeseriesReq.Size(), limit)
+		require.Len(t, timeseriesPartials, 2)
+		for _, partial := range timeseriesPartials {
+			require.LessOrEqual(t, partial.Size(), limit)
+		}
+
+		metadataReq := &WriteRequest{
+			Source:   RULE,
+			Metadata: reqv1.Metadata[:2],
+		}
+		baseSize = newPartialWriteRequest(metadataReq).Size()
+		limit = baseSize + embeddedMessageFieldSize(metadataReq.Metadata[0].Size()) + metadataReq.Metadata[1].Size()
+		metadataPartials := SplitWriteRequestByMaxMarshalSize(metadataReq, metadataReq.Size(), limit)
+		require.Len(t, metadataPartials, 2)
+		for _, partial := range metadataPartials {
+			require.LessOrEqual(t, partial.Size(), limit)
+		}
+	})
+
+	t.Run("should preserve request settings without transferring buffer ownership", func(t *testing.T) {
+		req := generateWriteRequest(2, 2, 1, 2)
+		t.Cleanup(req.FreeBuffer)
+		req.Source = RULE
+		req.SkipLabelValidation = true
+		req.SkipLabelCountValidation = true
+		req.skipUnmarshalingExemplars = true
+		req.skipNormalizeMetadataMetricName = true
+		req.skipDeduplicateMetadata = true
+		req.SetBuffer(mem.SliceBuffer([]byte("request buffer")))
+
+		source := &WriteRequest{}
+		source.SetBuffer(mem.SliceBuffer([]byte("source buffer")))
+		req.AddSourceBufferHolder(&source.BufferHolder)
+		source.FreeBuffer()
+
+		partials := SplitWriteRequestByMaxMarshalSize(req, req.Size(), 1)
+		require.Greater(t, len(partials), 1)
+		for _, partial := range partials {
+			require.Equal(t, RULE, partial.Source)
+			require.True(t, partial.SkipLabelValidation)
+			require.True(t, partial.SkipLabelCountValidation)
+			require.True(t, partial.skipUnmarshalingExemplars)
+			require.True(t, partial.skipNormalizeMetadataMetricName)
+			require.True(t, partial.skipDeduplicateMetadata)
+			require.Nil(t, partial.Buffer())
+			require.Nil(t, partial.sourceBufferHolders)
+			require.False(t, partial.unmarshalFromRW2)
+			require.Empty(t, partial.rw2symbols.pages)
+		}
+	})
+
 	t.Run("should split the input WriteRequest into multiple requests with size bigger than limit, if limit > size(symbols) but each request < limit", func(t *testing.T) {
 		const limit = 70
 		reqv2 := testReqV2Static(t)
@@ -385,6 +465,9 @@ func TestSplitWriteRequestByMaxMarshalSize_Fuzzy(t *testing.T) {
 			}
 
 			for _, partial := range partials {
+				if partial.Size() > maxSize {
+					require.Equal(t, 1, len(partial.Timeseries)+len(partial.Metadata), "only an individually oversized entity may exceed the limit")
+				}
 				merged.Timeseries = append(merged.Timeseries, partial.Timeseries...)
 				merged.Metadata = append(merged.Metadata, partial.Metadata...)
 			}