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..2070af3215b 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: append(prometheus.LinearBuckets(1, 1, 8), 16, 32, 64, 128), + }), } 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..e20741a9749 100644 --- a/pkg/ruler/distributor_client_test.go +++ b/pkg/ruler/distributor_client_test.go @@ -6,6 +6,8 @@ import ( "bytes" "context" "errors" + "math" + "math/rand" "net" "strconv" "strings" @@ -16,13 +18,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 +39,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 +69,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 +99,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 +152,151 @@ 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()) + + 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 { + 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) { + 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, math.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 +323,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 +540,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 +631,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 +741,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