From e3e3ac8d34cb21d36764083a1029f7321bfe8549 Mon Sep 17 00:00:00 2001 From: Arve Knudsen Date: Mon, 20 Jul 2026 16:22:07 +0200 Subject: [PATCH 1/4] Ruler: split oversized remote distributor writes Remote ruler writes were previously sent as a single gRPC message, so large rule evaluation results could exceed the configured transport limit. Split Remote Write 1.0 requests by series at the gRPC maximum send size and send each part sequentially with independent retries. Preserve request ownership and ordering, expose the number of generated requests, document partial-write semantics, and fix the RW1 splitter's protobuf size accounting and capacity estimates. Signed-off-by: Arve Knudsen --- CHANGELOG.md | 2 + .../architecture/components/ruler/index.md | 10 + pkg/mimirpb/split.go | 76 +++-- pkg/mimirpb/split_test.go | 83 +++++ pkg/ruler/distributor_client.go | 114 ++++++- pkg/ruler/distributor_client_test.go | 317 +++++++++++++++++- 6 files changed, 555 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9680080833b..f97c60eb9c6 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/sources/mimir/references/architecture/components/ruler/index.md b/docs/sources/mimir/references/architecture/components/ruler/index.md index 0815c282f66..91903c2258e 100644 --- a/docs/sources/mimir/references/architecture/components/ruler/index.md +++ b/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. diff --git a/pkg/mimirpb/split.go b/pkg/mimirpb/split.go index e8101be7d97..8200cca08ef 100644 --- a/pkg/mimirpb/split.go +++ b/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) { diff --git a/pkg/mimirpb/split_test.go b/pkg/mimirpb/split_test.go index 4afa1c9ab72..60e6250c4d6 100644 --- a/pkg/mimirpb/split_test.go +++ b/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...) } diff --git a/pkg/ruler/distributor_client.go b/pkg/ruler/distributor_client.go index 8da43fe9af9..04de8db82fd 100644 --- a/pkg/ruler/distributor_client.go +++ b/pkg/ruler/distributor_client.go @@ -7,6 +7,8 @@ import ( "errors" "flag" "fmt" + "math" + "sort" "strings" "sync" "time" @@ -15,13 +17,16 @@ import ( "github.com/go-kit/log/level" "github.com/grafana/dskit/backoff" "github.com/grafana/dskit/grpcclient" + "github.com/grafana/dskit/grpcencoding/snappy" "github.com/grafana/dskit/grpcutil" "github.com/grafana/dskit/middleware" "github.com/grafana/dskit/services" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "google.golang.org/grpc" "google.golang.org/grpc/codes" + grpcgzip "google.golang.org/grpc/encoding/gzip" "github.com/grafana/mimir/pkg/distributor/distributorpb" "github.com/grafana/mimir/pkg/mimirpb" @@ -63,6 +68,9 @@ func (c *DistributorConfig) Validate() error { if err := c.GRPCClientConfig.Validate(); err != nil { return fmt.Errorf("ruler's distributor client gRPC settings: %w", err) } + if _, err := maxUncompressedPayloadSize(c.GRPCClientConfig.MaxSendMsgSize, c.GRPCClientConfig.GRPCCompression); err != nil { + return fmt.Errorf("ruler's distributor client gRPC settings: %w", err) + } if c.Address == "" { return nil @@ -95,10 +103,14 @@ type DistributorGRPCClient struct { logger log.Logger cfg DistributorConfig invalidClusterValidation *prometheus.CounterVec + requestsPerWriteRequest prometheus.Histogram mu sync.RWMutex conn *grpc.ClientConn client distributorpb.DistributorClient + // maxWriteRequestSize is the largest uncompressed protobuf payload guaranteed + // to fit within the configured gRPC send limit after compression. + maxWriteRequestSize int } func NewDistributorGRPCClient(cfg DistributorConfig, reg prometheus.Registerer, logger log.Logger) (*DistributorGRPCClient, error) { @@ -106,6 +118,11 @@ func NewDistributorGRPCClient(cfg DistributorConfig, reg prometheus.Registerer, logger: logger, cfg: cfg, invalidClusterValidation: util.NewRequestInvalidClusterValidationLabelsTotalCounter(reg, "ruler-distributor", util.GRPCProtocol), + requestsPerWriteRequest: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{ + Name: "cortex_ruler_remote_distributor_requests_per_write_request", + Help: "The number of remote distributor requests a single ruler write request has been split into.", + Buckets: prometheus.ExponentialBuckets(1, 2, 8), + }), } c.Service = services.NewIdleService(c.start, c.stop).WithName("ruler distributor client") return c, nil @@ -115,6 +132,10 @@ func (c *DistributorGRPCClient) start(context.Context) error { if err := c.cfg.Validate(); err != nil { return err } + maxWriteRequestSize, err := maxUncompressedPayloadSize(c.cfg.GRPCClientConfig.MaxSendMsgSize, c.cfg.GRPCClientConfig.GRPCCompression) + if err != nil { + return err + } opts, err := c.cfg.GRPCClientConfig.DialOption( []grpc.UnaryClientInterceptor{ @@ -138,6 +159,7 @@ func (c *DistributorGRPCClient) start(context.Context) error { c.mu.Lock() c.conn = conn c.client = distributorpb.NewDistributorClient(conn) + c.maxWriteRequestSize = maxWriteRequestSize c.mu.Unlock() return nil @@ -162,11 +184,47 @@ func (c *DistributorGRPCClient) Push(ctx context.Context, req *mimirpb.WriteRequ c.mu.RLock() client := c.client + maxWriteRequestSize := c.maxWriteRequestSize c.mu.RUnlock() if client == nil { return nil, errDistributorClientNotRunning } + requests := splitWriteRequest(req, maxWriteRequestSize) + c.requestsPerWriteRequest.Observe(float64(len(requests))) + + var resp *mimirpb.WriteResponse + for idx, request := range requests { + var err error + resp, err = c.pushWithRetries(ctx, client, request, idx+1, len(requests)) + if err != nil { + return nil, err + } + } + + return resp, nil +} + +func splitWriteRequest(req *mimirpb.WriteRequest, maxSize int) []*mimirpb.WriteRequest { + // The ruler currently creates Remote Write 1.0 requests. Preserve an RW2 + // request instead of partially splitting an unsupported shape. + if maxSize <= 0 || len(req.TimeseriesRW2) > 0 { + return []*mimirpb.WriteRequest{req} + } + + reqSize := req.Size() + if reqSize <= maxSize { + return []*mimirpb.WriteRequest{req} + } + + requests := mimirpb.SplitWriteRequestByMaxMarshalSize(req, reqSize, maxSize) + if len(requests) == 0 { + return []*mimirpb.WriteRequest{req} + } + return requests +} + +func (c *DistributorGRPCClient) pushWithRetries(ctx context.Context, client distributorpb.DistributorClient, req *mimirpb.WriteRequest, requestNumber, totalRequests int) (*mimirpb.WriteResponse, error) { pushAttempt := func() (*mimirpb.WriteResponse, error) { attemptCtx, cancel := context.WithTimeout(ctx, c.cfg.RemoteTimeout) defer cancel() @@ -184,7 +242,7 @@ func (c *DistributorGRPCClient) Push(ctx context.Context, req *mimirpb.WriteRequ } retryable := isRetryableDistributorPushError(err) - level.Warn(c.logger).Log("msg", "failed to write to remote distributor", "err", err, "retryable", retryable, "attempt", retry.NumRetries()+1, "max_attempts", maxAttempts) + level.Warn(c.logger).Log("msg", "failed to write to remote distributor", "err", err, "retryable", retryable, "attempt", retry.NumRetries()+1, "max_attempts", maxAttempts, "request", requestNumber, "requests", totalRequests) if !retryable { return nil, err } @@ -226,6 +284,7 @@ func (c *DistributorGRPCClient) Close() error { conn := c.conn c.conn = nil c.client = nil + c.maxWriteRequestSize = 0 c.mu.Unlock() if conn == nil { @@ -233,3 +292,56 @@ func (c *DistributorGRPCClient) Close() error { } return conn.Close() } + +const ( + // The common bound covers the gzip header, trailer, and final empty DEFLATE + // block, plus the per-block overhead of gzip, framed Snappy, and framed S2. + // Go's default DEFLATE writer can fill a block after 1<<14 literal tokens, + // which is the smallest block size among these compressors. + compressedPayloadFixedOverhead = 23 + compressedPayloadBlockSize = 1 << 14 + compressedPayloadMaxBlockOverhead = 8 +) + +// maxUncompressedPayloadSize returns the largest uncompressed protobuf payload +// guaranteed to fit within maxSendMsgSize after applying compression. It returns +// an error when no worst-case size bound is known for the compressor. +func maxUncompressedPayloadSize(maxSendMsgSize int, compression string) (int, error) { + switch compression { + case "": + case grpcgzip.Name, snappy.Name, s2.Name: + default: + return 0, fmt.Errorf("compression type %q has no payload-size bound", compression) + } + if maxSendMsgSize <= 0 { + return 0, nil + } + if compression == "" { + return maxSendMsgSize, nil + } + + // Index i represents payload size i+1, so the first non-fitting index is the + // largest fitting size. This avoids computing maxSendMsgSize+1. + return sort.Search(maxSendMsgSize, func(i int) bool { + upperBound, ok := compressedPayloadSizeUpperBound(i + 1) + return !ok || upperBound > maxSendMsgSize + }), nil +} + +func compressedPayloadSizeUpperBound(uncompressedSize int) (int, bool) { + if uncompressedSize < 0 || uncompressedSize > math.MaxInt-compressedPayloadFixedOverhead { + return 0, false + } + + blocks := uncompressedSize / compressedPayloadBlockSize + if uncompressedSize%compressedPayloadBlockSize != 0 { + blocks++ + } + + remaining := math.MaxInt - uncompressedSize - compressedPayloadFixedOverhead + if blocks > remaining/compressedPayloadMaxBlockOverhead { + return 0, false + } + + return uncompressedSize + compressedPayloadFixedOverhead + blocks*compressedPayloadMaxBlockOverhead, true +} diff --git a/pkg/ruler/distributor_client_test.go b/pkg/ruler/distributor_client_test.go index 7883f16deae..f3b208b2810 100644 --- a/pkg/ruler/distributor_client_test.go +++ b/pkg/ruler/distributor_client_test.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "errors" + "math/rand" "net" "strconv" "strings" @@ -16,13 +17,17 @@ import ( "github.com/go-kit/log" "github.com/gogo/status" "github.com/grafana/dskit/flagext" + grpcsnappy "github.com/grafana/dskit/grpcencoding/snappy" "github.com/grafana/dskit/services" "github.com/grafana/dskit/user" + dto "github.com/prometheus/client_model/go" "github.com/prometheus/common/model" "github.com/prometheus/prometheus/model/labels" "github.com/stretchr/testify/require" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/encoding" + grpcgzip "google.golang.org/grpc/encoding/gzip" "google.golang.org/grpc/mem" "github.com/grafana/mimir/pkg/distributor/distributorpb" @@ -33,10 +38,11 @@ import ( type mockDistributorServer struct { distributorpb.UnimplementedDistributorServer - mu sync.Mutex - requests []*mimirpb.WriteRequest - userIDs []string - errs []error + mu sync.Mutex + requests []*mimirpb.WriteRequest + userIDs []string + errs []error + errsByCall map[int]error blockUntilContextDone bool onPush func(calls int) @@ -62,6 +68,9 @@ func (m *mockDistributorServer) Push(ctx context.Context, req *mimirpb.WriteRequ if m.onPush != nil { m.onPush(calls) } + if err := m.errsByCall[calls]; err != nil { + return nil, err + } if len(m.errs) > 0 { err := m.errs[0] @@ -89,10 +98,22 @@ func (m *mockDistributorServer) lastUserID() string { return m.userIDs[len(m.userIDs)-1] } -func setupDistributorGRPCClient(t *testing.T, srv *mockDistributorServer, logger log.Logger, configure func(*DistributorConfig)) *DistributorGRPCClient { +func (m *mockDistributorServer) allRequests() []*mimirpb.WriteRequest { + m.mu.Lock() + defer m.mu.Unlock() + return append([]*mimirpb.WriteRequest(nil), m.requests...) +} + +func (m *mockDistributorServer) allUserIDs() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.userIDs...) +} + +func setupDistributorGRPCClient(t *testing.T, srv *mockDistributorServer, logger log.Logger, configure func(*DistributorConfig), serverOptions ...grpc.ServerOption) *DistributorGRPCClient { t.Helper() - grpcServer := grpc.NewServer() + grpcServer := grpc.NewServer(serverOptions...) distributorpb.RegisterDistributorServer(grpcServer, srv) listener, err := net.Listen("tcp", "127.0.0.1:0") @@ -130,17 +151,145 @@ func setupDistributorGRPCClient(t *testing.T, srv *mockDistributorServer, logger } func newTestWriteRequest() *mimirpb.WriteRequest { + return newTestWriteRequestWithSeries(1) +} + +func newTestWriteRequestWithSeries(numSeries int) *mimirpb.WriteRequest { + seriesLabels := make([][]mimirpb.LabelAdapter, 0, numSeries) + samples := make([]mimirpb.Sample, 0, numSeries) + for i := range numSeries { + seriesLabels = append(seriesLabels, mimirpb.FromLabelsToLabelAdapters(labels.FromStrings( + model.MetricNameLabel, "test_metric_"+strconv.Itoa(i), + "padding", strings.Repeat("x", 32), + ))) + samples = append(samples, mimirpb.Sample{TimestampMs: 123, Value: float64(i)}) + } + return mimirpb.ToWriteRequest( - [][]mimirpb.LabelAdapter{ - mimirpb.FromLabelsToLabelAdapters(labels.FromStrings(model.MetricNameLabel, "test_metric")), - }, - []mimirpb.Sample{{TimestampMs: 123, Value: 456}}, + seriesLabels, + samples, nil, nil, mimirpb.RULE, ) } +func maxWriteRequestSizeForOneSeries(req *mimirpb.WriteRequest) int { + maxSize := 0 + for idx := range req.Timeseries { + oneSeriesReq := &mimirpb.WriteRequest{ + Timeseries: req.Timeseries[idx : idx+1], + Source: req.Source, + SkipLabelValidation: req.SkipLabelValidation, + } + maxSize = max(maxSize, oneSeriesReq.Size()) + } + return maxSize +} + +func metricNamesFromWriteRequests(requests []*mimirpb.WriteRequest) []string { + metricNames := make([]string, 0) + for _, req := range requests { + for _, ts := range req.Timeseries { + for _, lbl := range ts.Labels { + if lbl.Name == model.MetricNameLabel { + metricNames = append(metricNames, lbl.Value) + break + } + } + } + } + return metricNames +} + +func requireRequestsPerWriteMetric(t *testing.T, client *DistributorGRPCClient, expectedCount uint64, expectedSum float64) { + t.Helper() + + metric := &dto.Metric{} + require.NoError(t, client.requestsPerWriteRequest.Write(metric)) + require.Equal(t, expectedCount, metric.GetHistogram().GetSampleCount()) + require.Equal(t, expectedSum, metric.GetHistogram().GetSampleSum()) +} + +func compressPayload(t *testing.T, compressorName string, payload []byte) int { + t.Helper() + + compressor := encoding.GetCompressor(compressorName) + require.NotNil(t, compressor) + + var compressed bytes.Buffer + writer, err := compressor.Compress(&compressed) + require.NoError(t, err) + _, err = writer.Write(payload) + require.NoError(t, err) + require.NoError(t, writer.Close()) + return compressed.Len() +} + +func TestMaxUncompressedPayloadSize(t *testing.T) { + t.Run("uncompressed payload uses the configured limit", func(t *testing.T) { + for _, limit := range []int{-1, 0, 1, 100, int(^uint(0) >> 1)} { + actual, err := maxUncompressedPayloadSize(limit, "") + require.NoError(t, err) + require.Equal(t, max(limit, 0), actual) + } + }) + + t.Run("compressed payload reserves framing overhead", func(t *testing.T) { + for _, tc := range []struct { + limit int + expected int + }{ + {limit: 0, expected: 0}, + {limit: 31, expected: 0}, + {limit: 32, expected: 1}, + {limit: 16415, expected: 16384}, + {limit: 16423, expected: 16384}, + {limit: 16424, expected: 16385}, + } { + for _, compressorName := range []string{grpcgzip.Name, grpcsnappy.Name, s2.Name} { + actual, err := maxUncompressedPayloadSize(tc.limit, compressorName) + require.NoError(t, err) + require.Equal(t, tc.expected, actual, "compressor %s", compressorName) + } + } + }) + + t.Run("compressed payload calculation does not overflow", func(t *testing.T) { + maxInt := int(^uint(0) >> 1) + actual, err := maxUncompressedPayloadSize(maxInt, grpcgzip.Name) + require.NoError(t, err) + require.Positive(t, actual) + + upperBound, ok := compressedPayloadSizeUpperBound(actual) + require.True(t, ok) + require.LessOrEqual(t, upperBound, maxInt) + _, ok = compressedPayloadSizeUpperBound(actual + 1) + require.False(t, ok) + }) + + t.Run("unsupported compressor fails closed", func(t *testing.T) { + _, err := maxUncompressedPayloadSize(100, "future-compressor") + require.EqualError(t, err, `compression type "future-compressor" has no payload-size bound`) + }) +} + +func TestCompressedPayloadSizeUpperBound(t *testing.T) { + random := rand.New(rand.NewSource(12345)) + for _, size := range []int{0, 1, 16383, 16384, 16385, 65534, 65535, 65536, 2 * 65535, 2*65535 + 1, 1<<20 - 1, 1 << 20, 1<<20 + 1} { + payload := make([]byte, size) + _, err := random.Read(payload) + require.NoError(t, err) + + upperBound, ok := compressedPayloadSizeUpperBound(size) + require.True(t, ok) + for _, compressorName := range []string{grpcgzip.Name, grpcsnappy.Name, s2.Name} { + compressedSize := compressPayload(t, compressorName, payload) + require.LessOrEqual(t, compressedSize, upperBound, "compressor %s, payload size %d", compressorName, size) + } + } +} + func TestDistributorGRPCClient(t *testing.T) { t.Run("push before service start", func(t *testing.T) { var cfg DistributorConfig @@ -167,6 +316,106 @@ func TestDistributorGRPCClient(t *testing.T) { require.Equal(t, 1, srv.calls()) require.Equal(t, "test-user", srv.lastUserID()) require.Equal(t, mimirpb.RULE, srv.lastRequest().Source) + requireRequestsPerWriteMetric(t, client, 1, 1) + }) + + t.Run("push splits oversized request", func(t *testing.T) { + srv := &mockDistributorServer{} + req := newTestWriteRequestWithSeries(4) + expectedMetricNames := metricNamesFromWriteRequests([]*mimirpb.WriteRequest{req}) + maxSize := maxWriteRequestSizeForOneSeries(req) + req.SetBuffer(mem.SliceBuffer([]byte("request buffer"))) + + client := setupDistributorGRPCClient(t, srv, log.NewNopLogger(), func(cfg *DistributorConfig) { + cfg.GRPCClientConfig.MaxSendMsgSize = maxSize + }) + + _, err := client.Push(user.InjectOrgID(t.Context(), "test-user"), req) + require.NoError(t, err) + require.Nil(t, req.Buffer()) + + requests := srv.allRequests() + require.Len(t, requests, 4) + for _, request := range requests { + require.LessOrEqual(t, request.Size(), maxSize) + require.Equal(t, mimirpb.RULE, request.Source) + } + require.Equal(t, expectedMetricNames, metricNamesFromWriteRequests(requests)) + require.Equal(t, []string{"test-user", "test-user", "test-user", "test-user"}, srv.allUserIDs()) + requireRequestsPerWriteMetric(t, client, 1, 4) + }) + + for _, compressorName := range []string{grpcgzip.Name, grpcsnappy.Name, s2.Name} { + t.Run("push splits oversized request with "+compressorName+" compression", func(t *testing.T) { + srv := &mockDistributorServer{} + req := newTestWriteRequestWithSeries(4) + expectedMetricNames := metricNamesFromWriteRequests([]*mimirpb.WriteRequest{req}) + maxUncompressedSize := maxWriteRequestSizeForOneSeries(req) + maxSendMsgSize, ok := compressedPayloadSizeUpperBound(maxUncompressedSize) + require.True(t, ok) + + client := setupDistributorGRPCClient(t, srv, log.NewNopLogger(), func(cfg *DistributorConfig) { + cfg.GRPCClientConfig.MaxSendMsgSize = maxSendMsgSize + cfg.GRPCClientConfig.GRPCCompression = compressorName + }, grpc.MaxRecvMsgSize(maxSendMsgSize)) + + _, err := client.Push(user.InjectOrgID(t.Context(), "test-user"), req) + require.NoError(t, err) + + requests := srv.allRequests() + require.Len(t, requests, 4) + for _, request := range requests { + require.LessOrEqual(t, request.Size(), maxUncompressedSize) + } + require.Equal(t, expectedMetricNames, metricNamesFromWriteRequests(requests)) + requireRequestsPerWriteMetric(t, client, 1, 4) + }) + } + + t.Run("push retries split requests independently", func(t *testing.T) { + srv := &mockDistributorServer{ + errsByCall: map[int]error{2: status.Error(codes.Unavailable, "try again")}, + } + req := newTestWriteRequestWithSeries(4) + maxSize := maxWriteRequestSizeForOneSeries(req) + client := setupDistributorGRPCClient(t, srv, log.NewNopLogger(), func(cfg *DistributorConfig) { + cfg.GRPCClientConfig.MaxSendMsgSize = maxSize + }) + + _, err := client.Push(user.InjectOrgID(t.Context(), "test-user"), req) + require.NoError(t, err) + require.Equal(t, []string{"test_metric_0", "test_metric_1", "test_metric_1", "test_metric_2", "test_metric_3"}, metricNamesFromWriteRequests(srv.allRequests())) + requireRequestsPerWriteMetric(t, client, 1, 4) + }) + + t.Run("push stops after split request failure", func(t *testing.T) { + srv := &mockDistributorServer{ + errsByCall: map[int]error{2: status.Error(codes.ResourceExhausted, "limited")}, + } + var logs bytes.Buffer + req := newTestWriteRequestWithSeries(4) + maxSize := maxWriteRequestSizeForOneSeries(req) + client := setupDistributorGRPCClient(t, srv, log.NewLogfmtLogger(&logs), func(cfg *DistributorConfig) { + cfg.GRPCClientConfig.MaxSendMsgSize = maxSize + }) + + _, err := client.Push(user.InjectOrgID(t.Context(), "test-user"), req) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + require.Equal(t, []string{"test_metric_0", "test_metric_1"}, metricNamesFromWriteRequests(srv.allRequests())) + require.Contains(t, logs.String(), "request=2 requests=4") + requireRequestsPerWriteMetric(t, client, 1, 4) + }) + + t.Run("push reports an unsplittable request", func(t *testing.T) { + srv := &mockDistributorServer{} + client := setupDistributorGRPCClient(t, srv, log.NewNopLogger(), func(cfg *DistributorConfig) { + cfg.GRPCClientConfig.MaxSendMsgSize = 1 + }) + + _, err := client.Push(user.InjectOrgID(t.Context(), "test-user"), newTestWriteRequest()) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + require.Equal(t, 0, srv.calls()) + requireRequestsPerWriteMetric(t, client, 1, 1) }) t.Run("push retries", func(t *testing.T) { @@ -284,6 +533,8 @@ func TestDistributorGRPCClient(t *testing.T) { require.Contains(t, logOutput, "retryable=false") require.Contains(t, logOutput, "attempt=1") require.Contains(t, logOutput, "max_attempts=2") + require.Contains(t, logOutput, "request=1") + require.Contains(t, logOutput, "requests=1") }) t.Run("push returns context error when context is canceled before first attempt", func(t *testing.T) { @@ -373,6 +624,43 @@ func TestDistributorGRPCClient(t *testing.T) { }) } +func TestSplitWriteRequest(t *testing.T) { + t.Run("does not split at the size limit", func(t *testing.T) { + req := newTestWriteRequest() + t.Cleanup(func() { + req.FreeBuffer() + mimirpb.ReuseSlice(req.Timeseries) + }) + + requests := splitWriteRequest(req, req.Size()) + require.Len(t, requests, 1) + require.Same(t, req, requests[0]) + }) + + t.Run("does not split with a non-positive limit", func(t *testing.T) { + req := &mimirpb.WriteRequest{} + + for _, maxSize := range []int{0, -1} { + requests := splitWriteRequest(req, maxSize) + require.Len(t, requests, 1) + require.Same(t, req, requests[0]) + } + }) + + t.Run("does not drop an unsupported oversized request", func(t *testing.T) { + req := newTestWriteRequest() + req.TimeseriesRW2 = []mimirpb.TimeSeriesRW2{{LabelsRefs: []uint32{1, 2}}} + t.Cleanup(func() { + req.FreeBuffer() + mimirpb.ReuseSlice(req.Timeseries) + }) + + requests := splitWriteRequest(req, 1) + require.Len(t, requests, 1) + require.Same(t, req, requests[0]) + }) +} + func TestDistributorConfig_Validate(t *testing.T) { t.Run("address validation", func(t *testing.T) { for _, tc := range []struct { @@ -446,6 +734,15 @@ func TestDistributorConfig_Validate(t *testing.T) { require.EqualError(t, cfg.Validate(), `ruler's distributor client gRPC settings: unsupported compression type: "unsupported"`) }) + t.Run("rejects a custom compressor without a payload-size bound", func(t *testing.T) { + var cfg DistributorConfig + flagext.DefaultValues(&cfg) + cfg.GRPCClientConfig.CustomCompressors = append(cfg.GRPCClientConfig.CustomCompressors, "future-compressor") + cfg.GRPCClientConfig.GRPCCompression = "future-compressor" + + require.EqualError(t, cfg.Validate(), `ruler's distributor client gRPC settings: compression type "future-compressor" has no payload-size bound`) + }) + t.Run("validate does not mutate custom compressors", func(t *testing.T) { var cfg DistributorConfig cfg.GRPCClientConfig.GRPCCompression = s2.Name From 4200bc862205ce5f97544593f9f9663cc05926c3 Mon Sep 17 00:00:00 2001 From: Arve Knudsen Date: Wed, 5 Aug 2026 14:40:41 +0200 Subject: [PATCH 2/4] Use math.MaxInt Signed-off-by: Arve Knudsen --- pkg/ruler/distributor_client_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/ruler/distributor_client_test.go b/pkg/ruler/distributor_client_test.go index f3b208b2810..f039aad848a 100644 --- a/pkg/ruler/distributor_client_test.go +++ b/pkg/ruler/distributor_client_test.go @@ -6,6 +6,7 @@ import ( "bytes" "context" "errors" + "math" "math/rand" "net" "strconv" @@ -256,14 +257,13 @@ func TestMaxUncompressedPayloadSize(t *testing.T) { }) t.Run("compressed payload calculation does not overflow", func(t *testing.T) { - maxInt := int(^uint(0) >> 1) - actual, err := maxUncompressedPayloadSize(maxInt, grpcgzip.Name) + actual, err := maxUncompressedPayloadSize(math.MaxInt, grpcgzip.Name) require.NoError(t, err) require.Positive(t, actual) upperBound, ok := compressedPayloadSizeUpperBound(actual) require.True(t, ok) - require.LessOrEqual(t, upperBound, maxInt) + require.LessOrEqual(t, upperBound, math.MaxInt) _, ok = compressedPayloadSizeUpperBound(actual + 1) require.False(t, ok) }) From 6a7cd95345b2c1eb5e78d2f83b83930763e047c5 Mon Sep 17 00:00:00 2001 From: Arve Knudsen Date: Wed, 5 Aug 2026 15:08:18 +0200 Subject: [PATCH 3/4] Update pkg/ruler/distributor_client.go Co-authored-by: Julius Hinze --- pkg/ruler/distributor_client.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/ruler/distributor_client.go b/pkg/ruler/distributor_client.go index 04de8db82fd..5fb6f4f8a0c 100644 --- a/pkg/ruler/distributor_client.go +++ b/pkg/ruler/distributor_client.go @@ -121,7 +121,7 @@ func NewDistributorGRPCClient(cfg DistributorConfig, reg prometheus.Registerer, requestsPerWriteRequest: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{ Name: "cortex_ruler_remote_distributor_requests_per_write_request", Help: "The number of remote distributor requests a single ruler write request has been split into.", - Buckets: prometheus.ExponentialBuckets(1, 2, 8), + Buckets: prometheus.LinearBuckets(1, 1, 8), }), } c.Service = services.NewIdleService(c.start, c.stop).WithName("ruler distributor client") From b83b72c378e4e257e052e5ea9fe52ba633c9846e Mon Sep 17 00:00:00 2001 From: Arve Knudsen Date: Wed, 5 Aug 2026 15:09:18 +0200 Subject: [PATCH 4/4] Ruler: refine request split histogram buckets Signed-off-by: Arve Knudsen --- pkg/ruler/distributor_client.go | 2 +- pkg/ruler/distributor_client_test.go | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/pkg/ruler/distributor_client.go b/pkg/ruler/distributor_client.go index 5fb6f4f8a0c..2070af3215b 100644 --- a/pkg/ruler/distributor_client.go +++ b/pkg/ruler/distributor_client.go @@ -121,7 +121,7 @@ func NewDistributorGRPCClient(cfg DistributorConfig, reg prometheus.Registerer, requestsPerWriteRequest: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{ Name: "cortex_ruler_remote_distributor_requests_per_write_request", Help: "The number of remote distributor requests a single ruler write request has been split into.", - Buckets: prometheus.LinearBuckets(1, 1, 8), + Buckets: append(prometheus.LinearBuckets(1, 1, 8), 16, 32, 64, 128), }), } c.Service = services.NewIdleService(c.start, c.stop).WithName("ruler distributor client") diff --git a/pkg/ruler/distributor_client_test.go b/pkg/ruler/distributor_client_test.go index f039aad848a..e20741a9749 100644 --- a/pkg/ruler/distributor_client_test.go +++ b/pkg/ruler/distributor_client_test.go @@ -210,6 +210,13 @@ func requireRequestsPerWriteMetric(t *testing.T, client *DistributorGRPCClient, require.NoError(t, client.requestsPerWriteRequest.Write(metric)) require.Equal(t, expectedCount, metric.GetHistogram().GetSampleCount()) require.Equal(t, expectedSum, metric.GetHistogram().GetSampleSum()) + + buckets := metric.GetHistogram().GetBucket() + actualUpperBounds := make([]float64, 0, len(buckets)) + for _, bucket := range buckets { + actualUpperBounds = append(actualUpperBounds, bucket.GetUpperBound()) + } + require.Equal(t, []float64{1, 2, 3, 4, 5, 6, 7, 8, 16, 32, 64, 128}, actualUpperBounds) } func compressPayload(t *testing.T, compressorName string, payload []byte) int {