diff --git a/CHANGELOG.md b/CHANGELOG.md index 75220a91e14..a9971c0dc94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ * [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `source_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828 * [FEATURE] Distributor: Add experimental OTLP metrics ingestion over gRPC (`opentelemetry.proto.collector.metrics.v1.MetricsService/Export`) on the gRPC server port. Enable it with `-distributor.otlp.grpc-enabled`. The tenant is read from the `X-Scope-OrgID` gRPC metadata, and the request size is limited by `-server.grpc-max-recv-msg-size-bytes`. #7873 * [FEATURE] Storage: Add support for Alibaba Cloud OSS as a backend for blocks, ruler, and alertmanager storage. Configured via `-.oss.*` flags with `backend: oss`. #7907 +* [FEATURE] Ingester: Add experimental push coalescing, which merges the TSDB commits of concurrent write requests for the same tenant into a single commit. Every commit takes the tenant's WAL mutex, so a tenant whose writes are sharded across many ingesters, leaving each ingester a high rate of very small requests, contends on that mutex once per request. A request only waits for a commit that was already in flight, and joins the batch that runs after it, so there is no timer and no added latency when the ingester is idle. Enable it with `-ingester.push-coalesce.enabled` and bound each batch with `-ingester.push-coalesce.max-batch-requests`, `-ingester.push-coalesce.max-batch-samples` and `-ingester.push-coalesce.max-batch-bytes`. While enabled, a sample rejected only because of another request in the same batch, such as a duplicate timestamp, is dropped without the request receiving the `400` it would get from a separate commit, so the matching `cortex_discarded_samples_total` reasons can under-report. #7928 * [ENHANCEMENT] Compactor: Add `-compactor.block-ranges-compaction-delay` to configure the compaction group graceful period per `-compactor.block-ranges` entry, matched by position. A larger block range can now be held back until its sub-ranges have settled, instead of every range sharing one period derived from `-compactor.cleanup-interval`. Leaving it empty, and any range without an entry, keeps the current behaviour of twice `-compactor.cleanup-interval`. The delay is a lower bound: groups only become eligible on a compaction run, so the effective wait can be up to `-compactor.compaction-interval` longer. Only used by the partitioning compaction strategy. #7916 * [ENHANCEMENT] Ring: Add the `cortex_ring_member_zonal_ownership_percent{name, member, zone}` metric, reporting the fraction of its own zone's token space owned by each ring member. With zone-aware replication a write is routed within a single zone, so the existing `cortex_ring_member_ownership_percent`, which is computed over the merged token list of all zones, does not reflect the hash range a member actually serves. Reported when ring detailed metrics and zone awareness are both enabled. #7910 * [ENHANCEMENT] Distributor: Support partial write for Prometheus Remote Write 2.0 requests. Invalid series are now skipped and reported together in the `400` response instead of rejecting the whole batch, the valid ones are written, and the `X-Prometheus-Remote-Write-*-Written` response headers are set even when a `400` is returned. #7761 diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 5bdae1e6923..6e9aa04089a 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -4391,6 +4391,27 @@ query_protection: # when resource thresholds are breached. # CLI flag: -ingester.query-protection.eviction.max-evictions-per-cycle [max_evictions_per_cycle: | default = 1] + +push_coalesce: + # Experimental: merge the TSDB commits of concurrent write requests for the + # same tenant into a single commit. A request only ever waits for a commit + # that was already in flight, so this adds no latency of its own. + # CLI flag: -ingester.push-coalesce.enabled + [enabled: | default = false] + + # Maximum number of write requests a single coalesced commit may carry. + # CLI flag: -ingester.push-coalesce.max-batch-requests + [max_batch_requests: | default = 32] + + # Maximum number of samples a single coalesced commit may carry. 0 to disable + # the limit. + # CLI flag: -ingester.push-coalesce.max-batch-samples + [max_batch_samples: | default = 2000] + + # Maximum encoded size, in bytes, of the write requests a single coalesced + # commit may carry. 0 to disable the limit. + # CLI flag: -ingester.push-coalesce.max-batch-bytes + [max_batch_bytes: | default = 0] ``` ### `ingester_client_config` diff --git a/docs/configuration/v1-guarantees.md b/docs/configuration/v1-guarantees.md index d1561ac9af5..ff754fec41c 100644 --- a/docs/configuration/v1-guarantees.md +++ b/docs/configuration/v1-guarantees.md @@ -158,6 +158,11 @@ Currently experimental features are: - `-ingester.head-queried-series-metrics-windows` time windows to report (default: 2h) - `-ingester.head-queried-series-metrics-window-duration` HLL sub-window size - `-ingester.head-queried-series-metrics-sample-rate` query sampling rate +- Ingester: Push coalescing + - `-ingester.push-coalesce.enabled` + - `-ingester.push-coalesce.max-batch-requests` + - `-ingester.push-coalesce.max-batch-samples` + - `-ingester.push-coalesce.max-batch-bytes` - Parquet storage - Parquet Converter: the `-parquet-converter.*` CLI flags, including `-parquet-converter.enabled`, `-parquet-converter.max-num-columns` (automatically shards parquet files when the number of columns diff --git a/ingester.test b/ingester.test new file mode 100755 index 00000000000..6b84d729217 Binary files /dev/null and b/ingester.test differ diff --git a/integration/ingester_push_coalesce_test.go b/integration/ingester_push_coalesce_test.go new file mode 100644 index 00000000000..12ce3f3040f --- /dev/null +++ b/integration/ingester_push_coalesce_test.go @@ -0,0 +1,181 @@ +//go:build requires_docker + +package integration + +import ( + "fmt" + "strconv" + "sync" + "testing" + "time" + + "github.com/prometheus/common/model" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/prompb" + "github.com/stretchr/testify/require" + "go.uber.org/atomic" + + "github.com/cortexproject/cortex/integration/e2e" + e2edb "github.com/cortexproject/cortex/integration/e2e/db" + "github.com/cortexproject/cortex/integration/e2ecortex" +) + +// TestIngesterPushCoalescing sends many concurrent small write requests for +// one tenant to a single ingester with push coalescing enabled, and checks +// that every request succeeds, that the ingester really merged commits, and +// that every sample can be queried back with the value it was written with. +func TestIngesterPushCoalescing(t *testing.T) { + for _, tc := range []struct { + name string + // pushersPerSeries is how many pushers write each series. Above 1, + // pushers interleave their timestamps on a shared series, so a batch + // regularly holds several requests for one series, some of them out + // of order. + pushersPerSeries int + }{ + {name: "disjoint series", pushersPerSeries: 1}, + {name: "shared series", pushersPerSeries: 8}, + } { + t.Run(tc.name, func(t *testing.T) { + testIngesterPushCoalescing(t, tc.pushersPerSeries) + }) + } +} + +func testIngesterPushCoalescing(t *testing.T, pushersPerSeries int) { + const ( + pushers = 64 + pushesPerPusher = 50 + seriesPerRequest = 5 + metricName = "push_coalesce_load" + + totalRequests = pushers * pushesPerPusher + totalSamples = totalRequests * seriesPerRequest + ) + seriesGroups := pushers / pushersPerSeries + totalSeries := seriesGroups * seriesPerRequest + + s, err := e2e.NewScenario(networkName) + require.NoError(t, err) + defer s.Close() + + flags := mergeFlags(BlocksStorageFlags(), map[string]string{ + // One ingester and no replication, so every request for the tenant + // lands on the same TSDB, which is where coalescing happens. + "-distributor.replication-factor": "1", + "-ingester.push-coalesce.enabled": "true", + "-ingester.out-of-order-time-window": "30m", + "-ingester.max-series-per-user": "0", + "-ingester.max-series-per-metric": "0", + "-ingester.max-global-series-per-user": "0", + "-ingester.max-global-series-per-metric": "0", + // Nothing is shipped during the test, so query only the ingester. + "-limits.query-store-after": (1 * time.Hour).String(), + }) + + consul := e2edb.NewConsul() + minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"]) + require.NoError(t, s.StartAndWaitReady(consul, minio)) + + distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "") + require.NoError(t, s.StartAndWaitReady(distributor, ingester, storeGateway, querier)) + + // Wait until the distributor and the querier see the ingester as ACTIVE. + ingesterActive := e2e.WithLabelMatchers( + labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"), + labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE")) + require.NoError(t, distributor.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, ingesterActive)) + require.NoError(t, querier.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, ingesterActive)) + + // Pushers sharing a series take turns on its timestamps, so every + // (series, timestamp) pair is written exactly once. They run + // concurrently, so samples of a shared series arrive out of order, but + // the span of one series, pushesPerPusher*pushersPerSeries seconds, stays + // well inside the out-of-order window: every sample must be accepted. + start := time.Now().Add(-10 * time.Minute).Truncate(time.Second) + + var ( + wg sync.WaitGroup + failed atomic.Int64 + firstFail atomic.String + ) + fail := func(msg string) { + failed.Inc() + firstFail.CompareAndSwap("", msg) + } + for p := range pushers { + group, slot := p/pushersPerSeries, p%pushersPerSeries + wg.Go(func() { + client, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", userID) + if err != nil { + fail(err.Error()) + return + } + for n := range pushesPerPusher { + ts := start.Add(time.Duration(n*pushersPerSeries+slot) * time.Second) + series := make([]prompb.TimeSeries, 0, seriesPerRequest) + for i := range seriesPerRequest { + series = append(series, prompb.TimeSeries{ + Labels: []prompb.Label{ + {Name: labels.MetricName, Value: metricName}, + {Name: "group", Value: strconv.Itoa(group)}, + {Name: "series", Value: strconv.Itoa(i)}, + }, + Samples: []prompb.Sample{{Value: float64(n), Timestamp: e2e.TimeToMilliseconds(ts)}}, + }) + } + res, err := client.Push(series) + switch { + case err != nil: + fail(err.Error()) + case res.StatusCode != 200: + fail(fmt.Sprintf("pusher %d push %d: status %d", p, n, res.StatusCode)) + } + } + }) + } + wg.Wait() + require.Zerof(t, failed.Load(), "%d of %d pushes failed, first: %s", failed.Load(), totalRequests, firstFail.Load()) + + // Every sample reached the TSDB. + require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(totalSamples), "cortex_ingester_ingested_samples_total")) + + // The coalescer saw every request, served some of them as followers of + // another request's commit, and so issued fewer commits than requests. + require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(totalRequests), "cortex_ingester_push_coalesce_requests_total")) + require.NoError(t, ingester.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_ingester_push_coalesce_requests_total"}, + e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "role", "follower")))) + commits, err := ingester.SumMetrics([]string{"cortex_ingester_push_coalesce_commits_total"}) + require.NoError(t, err) + require.Less(t, commits[0], float64(totalRequests), "coalescing must issue fewer commits than requests") + require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(0), "cortex_ingester_push_coalesce_commit_failures_total")) + t.Logf("%d requests committed in %.0f commits", totalRequests, commits[0]) + + // Every series and every sample can be read back with its own value. + client, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", "", userID) + require.NoError(t, err) + queryAt := start.Add(time.Duration(pushesPerPusher*pushersPerSeries) * time.Second) + + requireScalar := func(query string, want float64) { + t.Helper() + result, err := client.Query(query, queryAt) + require.NoError(t, err) + require.Equal(t, model.ValVector, result.Type(), query) + vector := result.(model.Vector) + require.Len(t, vector, 1, query) + require.Equal(t, want, float64(vector[0].Value), query) + } + requireScalar(fmt.Sprintf("count(%s)", metricName), float64(totalSeries)) + requireScalar(fmt.Sprintf("sum(count_over_time(%s[1h]))", metricName), totalSamples) + // Every pusher writes the values 0 to pushesPerPusher-1 to each of its + // seriesPerRequest series. + requireScalar(fmt.Sprintf("sum(sum_over_time(%s[1h]))", metricName), + float64(pushers*seriesPerRequest*(pushesPerPusher-1)*pushesPerPusher/2)) + + assertServiceMetricsPrefixes(t, Distributor, distributor) + assertServiceMetricsPrefixes(t, Ingester, ingester) + assertServiceMetricsPrefixes(t, Querier, querier) +} diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index d877b90a12f..d5e41c7e695 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -189,6 +189,10 @@ type Config struct { EnableRegexMatcherLimits bool `yaml:"enable_regex_matcher_limits"` QueryProtection configs.QueryProtection `yaml:"query_protection"` + + // Merges the TSDB commits of concurrent write requests for the same tenant + // into a single commit. Disabled by default. + PushCoalesce PushCoalesceConfig `yaml:"push_coalesce"` } // RegisterFlags adds the flags required to config this to the given FlagSet @@ -228,6 +232,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) { f.BoolVar(&cfg.EnableRegexMatcherLimits, "ingester.enable-regex-matcher-limits", false, "Enable regex matcher limits and metrics collection for unoptimized regex queries. When enabled, the ingester will track pattern length, label cardinality, and total value length for unoptimized regex matchers.") cfg.DefaultLimits.RegisterFlagsWithPrefix(f, "ingester.") cfg.QueryProtection.RegisterFlagsWithPrefix(f, "ingester.") + cfg.PushCoalesce.RegisterFlagsWithPrefix(f, "ingester.") } func (cfg *Config) Validate(monitoredResources flagext.StringSliceCSV) error { @@ -243,6 +248,13 @@ func (cfg *Config) Validate(monitoredResources flagext.StringSliceCSV) error { return err } + if err := cfg.PushCoalesce.Validate(); err != nil { + return err + } + if cfg.PushCoalesce.Enabled { + logutil.WarnExperimentalUse("Push coalescing Enabled") + } + // Validate active queried series metrics windows if cfg.ActiveQueriedSeriesMetricsEnabled { if len(cfg.ActiveQueriedSeriesMetricsWindows) == 0 { @@ -428,6 +440,10 @@ type userTSDB struct { // Tracks active series per configured tracker pattern. trackerCounter *trackerCounter + + // Merge point for concurrent pushes for this tenant. nil when push + // coalescing is disabled, in which case every push commits on its own. + pushCoalescer *pushCoalescer } // Explicitly wrapping the tsdb.DB functions that we use. @@ -867,7 +883,8 @@ func New(cfg Config, limits *validation.Overrides, registerer prometheus.Registe &i.maxInflightPushRequests, &i.maxInflightQueryRequests, cfg.BlocksStorageConfig.TSDB.PostingsCache.Blocks.Enabled || cfg.BlocksStorageConfig.TSDB.PostingsCache.Head.Enabled, - cfg.EnableRegexMatcherLimits) + cfg.EnableRegexMatcherLimits, + cfg.PushCoalesce.Enabled) i.validateMetrics = validation.NewValidateMetrics(registerer) // Replace specific metrics which we can't directly track but we need to read @@ -967,6 +984,7 @@ func NewForFlusher(cfg Config, limits *validation.Overrides, registerer promethe &i.maxInflightQueryRequests, cfg.BlocksStorageConfig.TSDB.PostingsCache.Blocks.Enabled || cfg.BlocksStorageConfig.TSDB.PostingsCache.Head.Enabled, cfg.EnableRegexMatcherLimits, + cfg.PushCoalesce.Enabled, ) i.TSDBState.shipperIngesterID = "flusher" @@ -1545,32 +1563,6 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte } ) - // Walk the samples, appending them to the users database - app := db.Appender(ctx).(extendedAppender) - headMaxTime = db.Head().MaxTime() - - // Ensure the appender is always released so that we don't leak TSDB head - // series references, mmap'd chunks and pending state on early returns. - // `committed` is flipped to true immediately before app.Commit() because - // Prometheus closes the appender even on Commit failure (it self-rolls - // back internally on WAL error), so the deferred Rollback must not run - // afterwards. - committed := false - defer func() { - if committed { - return - } - if rollbackErr := app.Rollback(); rollbackErr != nil { - level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "failed to rollback appender on early return", "user", userID, "err", rollbackErr) - } - }() - - // Even when OOO is enabled globally, we want to reject OOO samples in some cases. - // prometheus implementation: https://github.com/prometheus/prometheus/pull/14710 - if req.DiscardOutOfOrder { - app.SetOptions(&storage.AppendOptions{DiscardOutOfOrder: true}) - } - delayObserver := i.metrics.ingestionDelaySeconds.WithLabelValues(userID) nowMs := time.Now().UnixMilli() observeDelay := func(timestampMs int64) { @@ -1579,195 +1571,289 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte } } - for _, ts := range req.Timeseries { - // The labels must be sorted (in our case, it's guaranteed a write request - // has sorted labels once hit the ingester). - - // Look up a reference for this series. - tsLabels := cortexpb.FromLabelAdaptersToLabels(ts.Labels) - if i.isLabelSetOutOfOrder(tsLabels) { - i.metrics.oooLabelsTotal.WithLabelValues(userID).Inc() - return nil, wrapWithUser(errors.Errorf("out-of-order label set found when push: %s", tsLabels), userID) - } - tsLabelsHash := tsLabels.Hash() - ref, copiedLabels := app.GetRef(tsLabels, tsLabelsHash) - - // To find out if any sample was added to this series, we keep old value. - oldSucceededSamplesCount := succeededSamplesCount - // To find out if any histogram was added to this series, we keep old value. - oldSucceededHistogramsCount := succeededHistogramsCount - - // Copied labels will be empty if ref is 0. - if ref == 0 { - // Copy the label set because both TSDB and the active series tracker may retain it. - copiedLabels = cortexpb.FromLabelAdaptersToLabelsWithCopy(ts.Labels) - } - matchedLabelSetLimits := i.limiter.limitsPerLabelSets(userID, copiedLabels) - - for _, s := range ts.Samples { - var err error - - // Observe ingestion delay for all samples (accepted and rejected) - observeDelay(s.TimestampMs) - - if s.StartTimestampMs != 0 && s.TimestampMs != 0 { - if _, err = app.AppendSTZeroSample(ref, copiedLabels, s.TimestampMs, s.StartTimestampMs); err != nil && !errors.Is(err, storage.ErrOutOfOrderST) { - startTimestampSampleAppendFailCount++ - i.metrics.startTimestampFail.WithLabelValues(sampleMetricTypeFloat).Inc() - } + // seriesHashes caches the label hash of every series in the request. It is + // only built on the coalesced path, which validates and hashes every series + // before the request joins a batch; without coalescing the hash is computed + // in the loop below, as before. + var seriesHashes []uint64 + + // appendSamples writes this request's samples into app. + // + // It runs on the request's own goroutine when the request commits alone, + // and on a batch leader's goroutine when the commit is coalesced. An + // appender is not safe for concurrent use, so in the coalesced case only + // the leader ever runs it. For that to stay correct it must touch nothing + // but state belonging to this request: every counter it updates is one of + // this request's locals, and they are turned into metrics further down by + // this request's own goroutine, after the commit. + // + // It returns an error only for failures that require the appender to be + // discarded. Soft per-sample errors are recorded in firstPartialErr. + appendSamples := func(app extendedAppender, batchHeadMaxTime int64) error { + headMaxTime = batchHeadMaxTime + + for tsIdx, ts := range req.Timeseries { + // The labels must be sorted (in our case, it's guaranteed a write request + // has sorted labels once hit the ingester). + + // Look up a reference for this series. + tsLabels := cortexpb.FromLabelAdaptersToLabels(ts.Labels) + if i.isLabelSetOutOfOrder(tsLabels) { + i.metrics.oooLabelsTotal.WithLabelValues(userID).Inc() + return errors.Errorf("out-of-order label set found when push: %s", tsLabels) } - - // If the cached reference exists, we try to use it. - if ref != 0 { - if _, err = app.Append(ref, copiedLabels, s.TimestampMs, s.Value); err == nil { - succeededSamplesCount++ - continue - } - + var tsLabelsHash uint64 + if seriesHashes != nil { + tsLabelsHash = seriesHashes[tsIdx] } else { - // Retain the reference in case there are multiple samples for the series. - if ref, err = app.Append(0, copiedLabels, s.TimestampMs, s.Value); err == nil { - succeededSamplesCount++ - continue - } + tsLabelsHash = tsLabels.Hash() } + ref, copiedLabels := app.GetRef(tsLabels, tsLabelsHash) - failedSamplesCount++ + // To find out if any sample was added to this series, we keep old value. + oldSucceededSamplesCount := succeededSamplesCount + // To find out if any histogram was added to this series, we keep old value. + oldSucceededHistogramsCount := succeededHistogramsCount - if rollback := handleAppendFailure(err, s.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits, value.IsStaleNaN(s.Value)); !rollback { - continue + // Copied labels will be empty if ref is 0. + if ref == 0 { + // Copy the label set because both TSDB and the active series tracker may retain it. + copiedLabels = cortexpb.FromLabelAdaptersToLabelsWithCopy(ts.Labels) } - // The error looks an issue on our side, so we should rollback. - // The deferred rollback above will close the appender; nothing to do here. - return nil, wrapWithUser(err, userID) - } + matchedLabelSetLimits := i.limiter.limitsPerLabelSets(userID, copiedLabels) - if i.limits.EnableNativeHistograms(userID) { - ingestedBucketsObserver := i.metrics.ingestedHistogramBuckets.WithLabelValues(userID) - for _, hp := range ts.Histograms { - var ( - err error - h *histogram.Histogram - fh *histogram.FloatHistogram - ) - - // Observe ingestion delay for all histograms (accepted and rejected) - observeDelay(hp.TimestampMs) - - // Choose the decoder based on the histogram's proto type (the - // CountInt/CountFloat oneof), not the count value. A float - // histogram with a count of 0 (e.g. a staleness marker or an - // empty histogram) still has the CountFloat oneof set, so a - // value-based check (hp.GetCountFloat() > 0) would misroute it - // to the integer decoder, which panics. This mirrors the - // discriminator used everywhere else (e.g. util/validation). - if hp.IsFloatHistogram() { - fh = cortexpb.FloatHistogramProtoToFloatHistogram(hp.Histogram) - } else { - h = cortexpb.HistogramProtoToHistogram(hp.Histogram) - } + for _, s := range ts.Samples { + var err error - if hp.StartTimestampMs != 0 && hp.TimestampMs != 0 { - if _, err = app.AppendHistogramSTZeroSample(ref, copiedLabels, hp.TimestampMs, hp.StartTimestampMs, h, fh); err != nil && !errors.Is(err, storage.ErrOutOfOrderST) { - startTimestampHistogramAppendFailCount++ - i.metrics.startTimestampFail.WithLabelValues(sampleMetricTypeHistogram).Inc() + // Observe ingestion delay for all samples (accepted and rejected) + observeDelay(s.TimestampMs) + + if s.StartTimestampMs != 0 && s.TimestampMs != 0 { + if _, err = app.AppendSTZeroSample(ref, copiedLabels, s.TimestampMs, s.StartTimestampMs); err != nil && !errors.Is(err, storage.ErrOutOfOrderST) { + startTimestampSampleAppendFailCount++ + i.metrics.startTimestampFail.WithLabelValues(sampleMetricTypeFloat).Inc() } } + // If the cached reference exists, we try to use it. if ref != 0 { - if _, err = app.AppendHistogram(ref, copiedLabels, hp.TimestampMs, h, fh); err == nil { - succeededHistogramsCount++ - ingestedBucketsObserver.Observe(float64(hp.BucketCount())) + if _, err = app.Append(ref, copiedLabels, s.TimestampMs, s.Value); err == nil { + succeededSamplesCount++ continue } + } else { - // Copy the label set because both TSDB and the active series tracker may retain it. - copiedLabels = cortexpb.FromLabelAdaptersToLabelsWithCopy(ts.Labels) - if ref, err = app.AppendHistogram(0, copiedLabels, hp.TimestampMs, h, fh); err == nil { - succeededHistogramsCount++ - ingestedBucketsObserver.Observe(float64(hp.BucketCount())) + // Retain the reference in case there are multiple samples for the series. + if ref, err = app.Append(0, copiedLabels, s.TimestampMs, s.Value); err == nil { + succeededSamplesCount++ continue } } - failedHistogramsCount++ + failedSamplesCount++ - if rollback := handleAppendFailure(err, hp.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits, false); !rollback { + if rollback := handleAppendFailure(err, s.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits, value.IsStaleNaN(s.Value)); !rollback { continue } - // The error looks an issue on our side, so we should rollback. - // The deferred rollback above will close the appender; nothing to do here. - return nil, wrapWithUser(err, userID) + // The error looks an issue on our side, so the appender must be + // discarded. Returning says so; the caller rolls back, and in the + // coalesced case that fails every request sharing the batch. + return err } - } else { - discardedNativeHistogramCount += len(ts.Histograms) - } - isNHAppended := succeededHistogramsCount > oldSucceededHistogramsCount - shouldUpdateSeries := (succeededSamplesCount > oldSucceededSamplesCount) || isNHAppended - if i.cfg.ActiveSeriesMetricsEnabled && shouldUpdateSeries { - db.activeSeries.UpdateSeries(tsLabels, tsLabelsHash, startAppend, isNHAppended, func(l labels.Labels) labels.Labels { - // we must already have copied the labels if succeededSamplesCount or succeededHistogramsCount has been incremented. - return copiedLabels - }) - } + if i.limits.EnableNativeHistograms(userID) { + ingestedBucketsObserver := i.metrics.ingestedHistogramBuckets.WithLabelValues(userID) + for _, hp := range ts.Histograms { + var ( + err error + h *histogram.Histogram + fh *histogram.FloatHistogram + ) + + // Observe ingestion delay for all histograms (accepted and rejected) + observeDelay(hp.TimestampMs) + + // Choose the decoder based on the histogram's proto type (the + // CountInt/CountFloat oneof), not the count value. A float + // histogram with a count of 0 (e.g. a staleness marker or an + // empty histogram) still has the CountFloat oneof set, so a + // value-based check (hp.GetCountFloat() > 0) would misroute it + // to the integer decoder, which panics. This mirrors the + // discriminator used everywhere else (e.g. util/validation). + if hp.IsFloatHistogram() { + fh = cortexpb.FloatHistogramProtoToFloatHistogram(hp.Histogram) + } else { + h = cortexpb.HistogramProtoToHistogram(hp.Histogram) + } - maxExemplarsForUser := i.getMaxExemplars(userID) - if maxExemplarsForUser > 0 { - // app.AppendExemplar currently doesn't create the series, it must - // already exist. If it does not then drop. - if ref == 0 && len(ts.Exemplars) > 0 { - updateFirstPartial(func() error { - return wrappedTSDBIngestExemplarErr(errExemplarRef, - model.Time(ts.Exemplars[0].TimestampMs), ts.Labels, ts.Exemplars[0].Labels) - }) - failedExemplarsCount += len(ts.Exemplars) - } else { // Note that else is explicit, rather than a continue in the above if, in case of additional logic post exemplar processing. - for _, ex := range ts.Exemplars { - e := exemplar.Exemplar{ - Value: ex.Value, - Ts: ex.TimestampMs, - HasTs: true, - Labels: cortexpb.FromLabelAdaptersToLabelsWithCopy(ex.Labels), + if hp.StartTimestampMs != 0 && hp.TimestampMs != 0 { + if _, err = app.AppendHistogramSTZeroSample(ref, copiedLabels, hp.TimestampMs, hp.StartTimestampMs, h, fh); err != nil && !errors.Is(err, storage.ErrOutOfOrderST) { + startTimestampHistogramAppendFailCount++ + i.metrics.startTimestampFail.WithLabelValues(sampleMetricTypeHistogram).Inc() + } + } + + if ref != 0 { + if _, err = app.AppendHistogram(ref, copiedLabels, hp.TimestampMs, h, fh); err == nil { + succeededHistogramsCount++ + ingestedBucketsObserver.Observe(float64(hp.BucketCount())) + continue + } + } else { + // Copy the label set because both TSDB and the active series tracker may retain it. + copiedLabels = cortexpb.FromLabelAdaptersToLabelsWithCopy(ts.Labels) + if ref, err = app.AppendHistogram(0, copiedLabels, hp.TimestampMs, h, fh); err == nil { + succeededHistogramsCount++ + ingestedBucketsObserver.Observe(float64(hp.BucketCount())) + continue + } } - if _, err = app.AppendExemplar(ref, labels.EmptyLabels(), e); err == nil { - succeededExemplarsCount++ + failedHistogramsCount++ + + if rollback := handleAppendFailure(err, hp.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits, false); !rollback { continue } + // The error looks an issue on our side, so the appender must be + // discarded. Returning says so; the caller rolls back, and in the + // coalesced case that fails every request sharing the batch. + return err + } + } else { + discardedNativeHistogramCount += len(ts.Histograms) + } - // Error adding exemplar + isNHAppended := succeededHistogramsCount > oldSucceededHistogramsCount + shouldUpdateSeries := (succeededSamplesCount > oldSucceededSamplesCount) || isNHAppended + if i.cfg.ActiveSeriesMetricsEnabled && shouldUpdateSeries { + db.activeSeries.UpdateSeries(tsLabels, tsLabelsHash, startAppend, isNHAppended, func(l labels.Labels) labels.Labels { + // we must already have copied the labels if succeededSamplesCount or succeededHistogramsCount has been incremented. + return copiedLabels + }) + } + + maxExemplarsForUser := i.getMaxExemplars(userID) + if maxExemplarsForUser > 0 { + // app.AppendExemplar currently doesn't create the series, it must + // already exist. If it does not then drop. + if ref == 0 && len(ts.Exemplars) > 0 { updateFirstPartial(func() error { - return wrappedTSDBIngestExemplarErr(err, model.Time(ex.TimestampMs), ts.Labels, ex.Labels) + return wrappedTSDBIngestExemplarErr(errExemplarRef, + model.Time(ts.Exemplars[0].TimestampMs), ts.Labels, ts.Exemplars[0].Labels) }) - failedExemplarsCount++ + failedExemplarsCount += len(ts.Exemplars) + } else { // Note that else is explicit, rather than a continue in the above if, in case of additional logic post exemplar processing. + for _, ex := range ts.Exemplars { + e := exemplar.Exemplar{ + Value: ex.Value, + Ts: ex.TimestampMs, + HasTs: true, + Labels: cortexpb.FromLabelAdaptersToLabelsWithCopy(ex.Labels), + } + + if _, err = app.AppendExemplar(ref, labels.EmptyLabels(), e); err == nil { + succeededExemplarsCount++ + continue + } + + // Error adding exemplar + updateFirstPartial(func() error { + return wrappedTSDBIngestExemplarErr(err, model.Time(ex.TimestampMs), ts.Labels, ex.Labels) + }) + failedExemplarsCount++ + } } } } - } - // At this point all samples have been added to the appender, so we can track the time it took. - i.TSDBState.appenderAddDuration.Observe(time.Since(startAppend).Seconds()) + // At this point all samples have been added to the appender, so we can track the time it took. + i.TSDBState.appenderAddDuration.Observe(time.Since(startAppend).Seconds()) - if startTimestampSampleAppendFailCount > 0 || startTimestampHistogramAppendFailCount > 0 { - level.Debug(logutil.WithContext(ctx, i.logger)).Log( - "msg", "failed to append start timestamp in push", - "user", userID, - "sample_failures", startTimestampSampleAppendFailCount, - "histogram_failures", startTimestampHistogramAppendFailCount, - ) - } + if startTimestampSampleAppendFailCount > 0 || startTimestampHistogramAppendFailCount > 0 { + level.Debug(logutil.WithContext(ctx, i.logger)).Log( + "msg", "failed to append start timestamp in push", + "user", userID, + "sample_failures", startTimestampSampleAppendFailCount, + "histogram_failures", startTimestampHistogramAppendFailCount, + ) + } - startCommit := time.Now() - // Mark committed before calling Commit: Prometheus closes the appender on - // both success and failure of Commit (it self-rolls-back on WAL error), so - // the deferred Rollback must not fire afterwards. - committed = true - if err := app.Commit(); err != nil { - return nil, wrapWithUser(err, userID) + return nil } - i.TSDBState.appenderCommitDuration.Observe(time.Since(startCommit).Seconds()) + if coalescer := db.pushCoalescer; coalescer != nil { + task := &pushTask{ + ctx: ctx, + discardOutOfOrder: req.DiscardOutOfOrder, + appendFn: appendSamples, + } + for _, ts := range req.Timeseries { + task.samples += len(ts.Samples) + len(ts.Histograms) + } + if i.cfg.PushCoalesce.MaxBatchBytes > 0 { + task.bytes = req.Size() + } + // Validate the label order before joining a batch, rather than during + // the append. It is the one deterministic error that forces the + // appender to be discarded, and a discarded batch fails every request + // in it instead of only the offending one. The hashes are computed in + // the same pass because the append loop needs them anyway. + seriesHashes = make([]uint64, len(req.Timeseries)) + for tsIdx, ts := range req.Timeseries { + lbls := cortexpb.FromLabelAdaptersToLabels(ts.Labels) + if i.isLabelSetOutOfOrder(lbls) { + i.metrics.oooLabelsTotal.WithLabelValues(userID).Inc() + return nil, wrapWithUser(errors.Errorf("out-of-order label set found when push: %s", lbls), userID) + } + seriesHashes[tsIdx] = lbls.Hash() + } + + if err := coalescer.push(task); err != nil { + // A context error belongs to the caller, so it is reported as it + // is. Everything else is a TSDB failure for this tenant. + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil, err + } + return nil, wrapWithUser(err, userID) + } + } else { + app := db.Appender(ctx).(extendedAppender) + headMax := db.Head().MaxTime() + + // Even when OOO is enabled globally, we want to reject OOO samples in some cases. + // prometheus implementation: https://github.com/prometheus/prometheus/pull/14710 + if req.DiscardOutOfOrder { + app.SetOptions(&storage.AppendOptions{DiscardOutOfOrder: true}) + } + + // Ensure the appender is always released so that we don't leak TSDB head + // series references, mmap'd chunks and pending state on early returns. + // `committed` is flipped to true immediately before app.Commit() because + // Prometheus closes the appender even on Commit failure (it self-rolls + // back internally on WAL error), so the deferred Rollback must not run + // afterwards. + committed := false + defer func() { + if committed { + return + } + if rollbackErr := app.Rollback(); rollbackErr != nil { + level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "failed to rollback appender on early return", "user", userID, "err", rollbackErr) + } + }() + + if err := appendSamples(app, headMax); err != nil { + return nil, wrapWithUser(err, userID) + } + + startCommit := time.Now() + committed = true + if err := app.Commit(); err != nil { + return nil, wrapWithUser(err, userID) + } + + i.TSDBState.appenderCommitDuration.Observe(time.Since(startCommit).Seconds()) + } // If only invalid samples or histograms are pushed, don't change "last update", as TSDB was not modified. if succeededSamplesCount > 0 || succeededHistogramsCount > 0 { @@ -3133,6 +3219,21 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) { // series during WAL replay. userDB.limiter = i.limiter + if i.cfg.PushCoalesce.Enabled { + userDB.pushCoalescer = newPushCoalescer( + i.cfg.PushCoalesce, + i.metrics.pushCoalesce, + userLogger, + func() extendedAppender { + // The appender carries the samples of every request in a batch, + // so it belongs to no single request's context. Prometheus' + // head appender ignores the context it is handed. + return db.Appender(context.Background()).(extendedAppender) + }, + func() int64 { return db.Head().MaxTime() }, + ) + } + if db.Head().NumSeries() > 0 { // If there are series in the head, use max time from head. If this time is too old, // TSDB will be eligible for flushing and closing sooner, unless more data is pushed to it quickly. diff --git a/pkg/ingester/metrics.go b/pkg/ingester/metrics.go index e5e038f68cb..53f0d8af8ce 100644 --- a/pkg/ingester/metrics.go +++ b/pkg/ingester/metrics.go @@ -32,10 +32,15 @@ const ( ) type ingesterMetrics struct { - ingestedSamples prometheus.Counter - ingestedHistograms prometheus.Counter - ingestedExemplars prometheus.Counter - ingestedMetadata prometheus.Counter + ingestedSamples prometheus.Counter + ingestedHistograms prometheus.Counter + ingestedExemplars prometheus.Counter + ingestedMetadata prometheus.Counter + + // pushCoalesce holds the metrics of the per-tenant push coalescers. They are + // shared by every tenant so that enabling coalescing does not add per-user + // cardinality. + pushCoalesce *pushCoalescerMetrics ingestedSamplesFail prometheus.Counter ingestedHistogramsFail prometheus.Counter startTimestampFail *prometheus.CounterVec @@ -99,6 +104,7 @@ func newIngesterMetrics(r prometheus.Registerer, maxInflightQueryRequests *util_math.MaxTracker, postingsCacheEnabled bool, regexMatcherLimitsEnabled bool, + pushCoalesceEnabled bool, ) *ingesterMetrics { const ( instanceLimits = "cortex_ingester_instance_limits" @@ -414,6 +420,12 @@ func newIngesterMetrics(r prometheus.Registerer, m.startTimestampFail.WithLabelValues(sampleMetricTypeFloat) m.startTimestampFail.WithLabelValues(sampleMetricTypeHistogram) + // Registered only when the feature is on, so that a disabled feature adds + // nothing to the exposed metric set. + if pushCoalesceEnabled { + m.pushCoalesce = newPushCoalescerMetrics(r) + } + return m } diff --git a/pkg/ingester/metrics_test.go b/pkg/ingester/metrics_test.go index af09d07e2a3..48621231d47 100644 --- a/pkg/ingester/metrics_test.go +++ b/pkg/ingester/metrics_test.go @@ -22,7 +22,7 @@ func TestRegexMatcherLimitsMetricsFeatureFlag(t *testing.T) { reg := prometheus.NewRegistry() m := newIngesterMetrics(reg, false, false, false, false, func() *InstanceLimits { return &InstanceLimits{} }, - ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, false) + ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, false, false) require.Nil(t, m.unoptimizedRegexPatternLength) require.Nil(t, m.unoptimizedRegexLabelCardinality) @@ -35,7 +35,7 @@ func TestRegexMatcherLimitsMetricsFeatureFlag(t *testing.T) { reg := prometheus.NewRegistry() m := newIngesterMetrics(reg, false, false, false, false, func() *InstanceLimits { return &InstanceLimits{} }, - ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, true) + ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, true, false) require.NotNil(t, m.unoptimizedRegexPatternLength) require.NotNil(t, m.unoptimizedRegexLabelCardinality) @@ -53,7 +53,7 @@ func TestUnoptimizedRegexRejectedMetric(t *testing.T) { reg := prometheus.NewRegistry() m := newIngesterMetrics(reg, false, false, false, false, func() *InstanceLimits { return &InstanceLimits{} }, - ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, true) + ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, true, false) require.NotNil(t, m.unoptimizedRegexRejectedTotal) @@ -77,7 +77,7 @@ func TestUnoptimizedRegexRejectedMetric(t *testing.T) { reg := prometheus.NewRegistry() m := newIngesterMetrics(reg, false, false, false, false, func() *InstanceLimits { return &InstanceLimits{} }, - ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, true) + ingestionRate, &inflightPushRequests, &maxInflightQueryRequests, false, true, false) require.NotNil(t, m.unoptimizedRegexRejectedTotal) @@ -138,7 +138,8 @@ func TestIngesterMetrics(t *testing.T) { &inflightPushRequests, &maxInflightQueryRequests, false, - true) + true, + false) require.NotNil(t, m) diff --git a/pkg/ingester/push_coalesce.go b/pkg/ingester/push_coalesce.go new file mode 100644 index 00000000000..99a893597d6 --- /dev/null +++ b/pkg/ingester/push_coalesce.go @@ -0,0 +1,564 @@ +package ingester + +import ( + "context" + "flag" + "time" + + "github.com/go-kit/log" + "github.com/go-kit/log/level" + "github.com/pkg/errors" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/prometheus/storage" + "go.uber.org/atomic" +) + +// Push coalescing folds the TSDB commits of several concurrent write requests +// for the same tenant into a single commit. +// +// Every Push takes its own appender and calls Commit on it. Commit writes the +// samples to the WAL through wlog.(*WL).Log, which holds the per-tenant WAL +// mutex for the whole record loop. When the distributor shards one large +// tenant's writes by series across many ingesters, each ingester receives a +// very high rate of requests carrying only a handful of samples each, and +// takes that one mutex once per request. Once any waiter has waited longer +// than a millisecond, Go's sync.Mutex switches to starvation mode and hands +// the lock over in strict FIFO order; if a woken waiter is then slow to get +// back on a CPU, each handoff costs far more than the work done under the +// lock, throughput falls below the arrival rate and the queue of pushes grows +// without bound. +// +// Coalescing attacks the request-per-commit ratio rather than the lock itself. +// A request that arrives while no commit is in flight commits on its own and +// pays no extra latency. A request that arrives while a commit is in flight +// joins the next batch; when the in-flight commit finishes, the first queued +// request becomes the leader of that batch, appends every request in it into +// one appender, calls Commit once, and hands each request its own result. There +// is no timer and no "wait for X milliseconds" anywhere: batching only ever +// uses time a request was going to spend waiting regardless. + +const ( + // Defaults chosen so that a batch stays small enough that its WAL record + // still fits comfortably in the WAL page buffer, while cutting the commit + // rate by an order of magnitude under heavy concurrent load. + defaultPushCoalesceMaxBatchRequests = 32 + defaultPushCoalesceMaxBatchSamples = 2000 +) + +// PushCoalesceConfig configures opportunistic merging of per-request TSDB +// commits. It is disabled by default. +type PushCoalesceConfig struct { + Enabled bool `yaml:"enabled"` + + // MaxBatchRequests bounds how many write requests one commit may carry. + MaxBatchRequests int `yaml:"max_batch_requests"` + + // MaxBatchSamples bounds how many samples (floats plus histograms) one + // commit may carry. 0 disables the bound. + MaxBatchSamples int `yaml:"max_batch_samples"` + + // MaxBatchBytes bounds the encoded size of the requests one commit may + // carry. 0 disables the bound. + MaxBatchBytes int `yaml:"max_batch_bytes"` +} + +// RegisterFlagsWithPrefix registers the push coalescing flags. +func (cfg *PushCoalesceConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) { + f.BoolVar(&cfg.Enabled, prefix+"push-coalesce.enabled", false, + "Experimental: merge the TSDB commits of concurrent write requests for the same tenant into a single commit. A request only ever waits for a commit that was already in flight, so this adds no latency of its own.") + f.IntVar(&cfg.MaxBatchRequests, prefix+"push-coalesce.max-batch-requests", defaultPushCoalesceMaxBatchRequests, + "Maximum number of write requests a single coalesced commit may carry.") + f.IntVar(&cfg.MaxBatchSamples, prefix+"push-coalesce.max-batch-samples", defaultPushCoalesceMaxBatchSamples, + "Maximum number of samples a single coalesced commit may carry. 0 to disable the limit.") + f.IntVar(&cfg.MaxBatchBytes, prefix+"push-coalesce.max-batch-bytes", 0, + "Maximum encoded size, in bytes, of the write requests a single coalesced commit may carry. 0 to disable the limit.") +} + +// Validate returns an error if the configuration cannot be used. +func (cfg *PushCoalesceConfig) Validate() error { + if !cfg.Enabled { + return nil + } + if cfg.MaxBatchRequests < 1 { + return errors.New("ingester.push-coalesce.max-batch-requests must be at least 1") + } + if cfg.MaxBatchSamples < 0 { + return errors.New("ingester.push-coalesce.max-batch-samples must not be negative") + } + if cfg.MaxBatchBytes < 0 { + return errors.New("ingester.push-coalesce.max-batch-bytes must not be negative") + } + return nil +} + +// Reasons a batch stopped taking further queued requests. Reported on +// cortex_ingester_push_coalesce_batch_truncated_total. +const ( + truncateReasonMaxRequests = "max_requests" + truncateReasonMaxSamples = "max_samples" + truncateReasonMaxBytes = "max_bytes" + truncateReasonDiscardOOO = "discard_out_of_order" +) + +// How a request's samples reached the TSDB. Reported on +// cortex_ingester_push_coalesce_requests_total. +const ( + coalesceRoleLeaderImmediate = "leader_immediate" + coalesceRoleLeaderElected = "leader_elected" + coalesceRoleFollower = "follower" + coalesceRoleCanceled = "canceled" +) + +// errPushCoalesceBatchRolledBack is returned to the requests of a batch that +// was rolled back because a different request in it failed in a way that +// requires the appender to be discarded. It is deliberately retryable: the +// samples of these requests were never committed. +var errPushCoalesceBatchRolledBack = errors.New("write request was not committed: another request in the same coalesced commit failed, please retry") + +// errPushCoalesceLeaderAborted is the result a batch member is left with if the +// leader stops running the batch without setting a result, which can only +// happen if appending panics. It is retryable for the same reason. +var errPushCoalesceLeaderAborted = errors.New("write request was not committed: the coalesced commit it joined did not complete, please retry") + +type pushCoalescerMetrics struct { + requests *prometheus.CounterVec + batchRequests prometheus.Histogram + batchSamples prometheus.Histogram + waitSeconds prometheus.Histogram + commitSeconds prometheus.Histogram + commits prometheus.Counter + commitFailures prometheus.Counter + rollbacks prometheus.Counter + truncated *prometheus.CounterVec +} + +func newPushCoalescerMetrics(r prometheus.Registerer) *pushCoalescerMetrics { + return &pushCoalescerMetrics{ + requests: promauto.With(r).NewCounterVec(prometheus.CounterOpts{ + Name: "cortex_ingester_push_coalesce_requests_total", + Help: "Number of write requests that went through the push coalescer, by the role the request played in its batch.", + }, []string{"role"}), + batchRequests: promauto.With(r).NewHistogram(prometheus.HistogramOpts{ + Name: "cortex_ingester_push_coalesce_batch_requests", + Help: "Number of write requests carried by a single coalesced commit.", + Buckets: []float64{1, 2, 4, 8, 16, 32, 64, 128}, + }), + batchSamples: promauto.With(r).NewHistogram(prometheus.HistogramOpts{ + Name: "cortex_ingester_push_coalesce_batch_samples", + Help: "Number of samples carried by a single coalesced commit.", + Buckets: prometheus.ExponentialBuckets(4, 2, 12), + }), + waitSeconds: promauto.With(r).NewHistogram(prometheus.HistogramOpts{ + Name: "cortex_ingester_push_coalesce_wait_seconds", + Help: "Time a write request spent queued waiting for the in-flight commit to finish before its own batch started.", + Buckets: prometheus.ExponentialBuckets(0.0001, 2, 14), + }), + commitSeconds: promauto.With(r).NewHistogram(prometheus.HistogramOpts{ + Name: "cortex_ingester_push_coalesce_commit_seconds", + Help: "Duration of a coalesced TSDB commit.", + Buckets: prometheus.ExponentialBuckets(0.0001, 2, 14), + }), + commits: promauto.With(r).NewCounter(prometheus.CounterOpts{ + Name: "cortex_ingester_push_coalesce_commits_total", + Help: "Number of TSDB commits issued by the push coalescer.", + }), + commitFailures: promauto.With(r).NewCounter(prometheus.CounterOpts{ + Name: "cortex_ingester_push_coalesce_commit_failures_total", + Help: "Number of coalesced TSDB commits that failed. Every request in such a batch is returned the commit error.", + }), + rollbacks: promauto.With(r).NewCounter(prometheus.CounterOpts{ + Name: "cortex_ingester_push_coalesce_batch_rollbacks_total", + Help: "Number of coalesced batches rolled back because one of their requests failed in a way that requires discarding the appender.", + }), + truncated: promauto.With(r).NewCounterVec(prometheus.CounterOpts{ + Name: "cortex_ingester_push_coalesce_batch_truncated_total", + Help: "Number of times a batch stopped taking further queued requests, by reason. The remaining requests form the next batch.", + }, []string{"reason"}), + } +} + +// pushTask is one logical write request waiting to be folded into a batch. +// +// Everything a leader needs in order to act on behalf of the request lives +// here, because the appender is not safe for concurrent use and so only the +// leader may touch it. +type pushTask struct { + ctx context.Context + + samples int + bytes int + + // discardOutOfOrder mirrors the request's flag. It is set on the appender, + // not per sample, so requests that disagree on it cannot share a batch. + discardOutOfOrder bool + + // appendFn writes the request's samples into the shared appender. It + // returns a non-nil error only for failures that require the appender to be + // discarded; per-sample soft errors are recorded on the request itself and + // reported by its own goroutine after the commit. + appendFn func(app extendedAppender, headMaxTime int64) error + + // result is the batch outcome, read by the waiting goroutine once wake is + // closed. Written only by the leader of the batch this task joined. + result error + + // dropped records that the leader skipped this task because its caller had + // already given up, so later passes over the batch must leave its result + // alone. + dropped bool + + // next links queued tasks. Only the task's own goroutine writes it, before + // publishing the task, and only the leader that claimed the queue reads it + // afterwards. + next *pushTask + + // elected is set when this task has been made leader of followers, rather + // than served by somebody else's batch. + elected bool + followers []*pushTask + + // wake is closed once result is final. nil for a request that never queued. + wake chan struct{} + signaled bool + + enqueuedAt time.Time +} + +func (t *pushTask) signal() { + if t.signaled || t.wake == nil { + return + } + t.signaled = true + close(t.wake) +} + +// pushCoalescer is the per-tenant merge point. It is attached to the tenant's +// userTSDB, so a batch never spans tenants and the WAL mutex a batch reduces +// contention on is always the one belonging to that tenant. +type pushCoalescer struct { + cfg PushCoalesceConfig + metrics *pushCoalescerMetrics + logger log.Logger + + // newAppender and headMaxTime are injected so the coalescer can be tested + // without a full Ingester. + newAppender func() extendedAppender + headMaxTime func() int64 + + // head encodes both whether a batch is in flight and the queue of requests + // waiting for the next one, in one word: + // nil idle: the next request commits immediately. + // coalesceBusy a batch is in flight and nothing is queued. + // any other task a batch is in flight and this is the most recently + // queued task; next links back to coalesceBusy. + // A request therefore joins with a single compare-and-swap and never waits + // on a lock to do so. Go's sync.Mutex switches to strict FIFO handoff once + // any waiter has waited more than 1ms, and when the scheduler is saturated + // a woken waiter routinely waits longer than that just to get a P, however + // short the critical section; a queue nobody blocks on to join cannot get + // into that state. head is also non-nil across a handoff, so that requests + // arriving between the election of the next leader and that leader actually + // starting queue for the batch after it rather than racing it with a second + // appender. + head atomic.Pointer[pushTask] + // carry holds tasks the previous leader claimed from head but could not + // fit into its batch, in arrival order. It is owned by whichever goroutine + // is currently leading; ownership passes with the wake channel close. + carry []*pushTask + // queued counts tasks that joined the queue and are not in a batch yet. + queued atomic.Int64 +} + +// coalesceBusy is the sentinel stored in pushCoalescer.head while a batch is in +// flight. It is never run or signaled. +var coalesceBusy = &pushTask{} + +func newPushCoalescer(cfg PushCoalesceConfig, metrics *pushCoalescerMetrics, logger log.Logger, newAppender func() extendedAppender, headMaxTime func() int64) *pushCoalescer { + return &pushCoalescer{ + cfg: cfg, + metrics: metrics, + logger: logger, + newAppender: newAppender, + headMaxTime: headMaxTime, + } +} + +// push runs the task's append and commit, either on its own or folded into a +// batch, and returns the result for that task alone. +// +// The caller must not release the write request until push returns: a follower +// blocks until the leader has finished appending its samples precisely so that +// Push's deferred ReuseSlice cannot run while the leader still reads them. +func (c *pushCoalescer) push(t *pushTask) error { + // A caller that has already given up is told so rather than having its + // samples committed on its behalf. + if err := t.ctx.Err(); err != nil { + c.metrics.requests.WithLabelValues(coalesceRoleCanceled).Inc() + return err + } + + for { + old := c.head.Load() + if old == nil { + if !c.head.CompareAndSwap(nil, coalesceBusy) { + continue + } + // Nothing is committing, so commit immediately and alone. This + // is the path that keeps coalescing free of added latency. + c.metrics.requests.WithLabelValues(coalesceRoleLeaderImmediate).Inc() + c.lead(t, nil) + return t.result + } + + // A batch is in flight, so join the one that runs after it. Prepare, + // once, before publishing the task: after the compare-and-swap below + // the leader may read it. + if t.wake == nil { + t.wake = make(chan struct{}) + t.enqueuedAt = time.Now() + t.result = errPushCoalesceLeaderAborted + } + t.next = old + // Count the task before publishing it: once the compare-and-swap + // succeeds a leader may claim it and decrement the count, which would + // otherwise go briefly negative. + c.queued.Add(1) + if c.head.CompareAndSwap(old, t) { + return c.await(t) + } + c.queued.Add(-1) + } +} + +// await blocks a queued task until it has been served by a batch or elected to +// lead one, and runs that batch if so. +func (c *pushCoalescer) await(t *pushTask) error { + <-t.wake + + if t.elected { + c.metrics.requests.WithLabelValues(coalesceRoleLeaderElected).Inc() + c.lead(t, t.followers) + } else { + c.metrics.requests.WithLabelValues(coalesceRoleFollower).Inc() + } + return t.result +} + +// isIdle reports that no batch is in flight and nothing is queued. +// +// It reads carry, which only the current leader owns, so it is only meaningful +// when no push is in progress. Tests use it once every pusher has returned. +func (c *pushCoalescer) isIdle() bool { + return c.head.Load() == nil && len(c.carry) == 0 && c.queued.Load() == 0 +} + +// pendingLen reports how many requests are queued and not yet in a batch. +func (c *pushCoalescer) pendingLen() int { + return int(c.queued.Load()) +} + +// lead runs one batch and then hands leadership to the next queued request. +func (c *pushCoalescer) lead(leader *pushTask, followers []*pushTask) { + // Release every follower exactly once, whatever happens in runBatch, so + // that no request is left blocked. Followers start out holding a retryable + // error, so a panic cannot report an uncommitted request as succeeded. + defer func() { + for _, t := range followers { + t.signal() + } + c.handoff() + }() + + c.runBatch(leader, followers) +} + +// runBatch appends every request in the batch into one appender, commits once, +// and records each request's own result. +func (c *pushCoalescer) runBatch(leader *pushTask, followers []*pushTask) { + batch := make([]*pushTask, 0, len(followers)+1) + batch = append(batch, leader) + batch = append(batch, followers...) + + now := time.Now() + for _, t := range batch { + if !t.enqueuedAt.IsZero() { + c.metrics.waitSeconds.Observe(now.Sub(t.enqueuedAt).Seconds()) + } + } + + // The appender carries samples for every request in the batch, so it is not + // tied to any one request's context. Prometheus' head appender ignores the + // context it is given, so nothing is lost by saying so explicitly. + app := c.newAppender() + headMaxTime := c.headMaxTime() + if leader.discardOutOfOrder { + app.SetOptions(&storage.AppendOptions{DiscardOutOfOrder: true}) + } + + var ( + appended = make([]*pushTask, 0, len(batch)) + samples int + fatal error + fatalTask *pushTask + ) + + for _, t := range batch { + // Re-check: a request may have been abandoned while it sat in the + // queue. Dropping it here is safe because it is still blocked on its + // wake channel and so has not released its write request. It is also a + // deliberate difference from the uncoalesced path, which commits the + // samples of a request whose caller has already gone away; a client + // that timed out retries either way, and shedding the work is what we + // want when the ingester is behind. + if err := t.ctx.Err(); err != nil { + t.result = err + t.dropped = true + continue + } + if err := t.appendFn(app, headMaxTime); err != nil { + fatal, fatalTask = err, t + break + } + appended = append(appended, t) + samples += t.samples + } + + if fatal != nil { + // One request failed in a way that requires discarding the appender, + // and an appender cannot discard one request's samples alone. Roll the + // whole batch back and let everyone retry. This is the unexpected-error + // path; soft per-sample errors never reach it. + if err := app.Rollback(); err != nil { + level.Warn(c.logger).Log("msg", "failed to roll back coalesced push batch", "err", err) + } + c.metrics.rollbacks.Inc() + for _, t := range batch { + if t.dropped { + continue + } + if t == fatalTask { + t.result = fatal + continue + } + t.result = errPushCoalesceBatchRolledBack + } + return + } + + start := time.Now() + err := app.Commit() + c.metrics.commits.Inc() + if err != nil { + c.metrics.commitFailures.Inc() + for _, t := range appended { + t.result = err + } + return + } + + c.metrics.commitSeconds.Observe(time.Since(start).Seconds()) + c.metrics.batchRequests.Observe(float64(len(appended))) + c.metrics.batchSamples.Observe(float64(samples)) + for _, t := range appended { + t.result = nil + } +} + +// handoff elects the leader of the next batch, or marks the coalescer idle. +// +// Only the goroutine that currently leads calls it, so it alone touches carry. +func (c *pushCoalescer) handoff() { + for { + old := c.head.Load() + if old == coalesceBusy { + if len(c.carry) > 0 { + // Nothing new arrived, but the previous claim left work over. + break + } + if c.head.CompareAndSwap(coalesceBusy, nil) { + return + } + continue + } + + // Claim everything queued so far in one swap. head goes back to + // coalesceBusy, not nil, so that arrivals keep queueing while the next + // batch is elected and run. + if !c.head.CompareAndSwap(old, coalesceBusy) { + continue + } + // The claimed list is newest first. Append it to carry oldest first so + // that batches keep arrival order. + n := 0 + for p := old; p != coalesceBusy; p = p.next { + n++ + } + base := len(c.carry) + c.carry = append(c.carry, make([]*pushTask, n)...) + i := base + n - 1 + for p := old; p != coalesceBusy; { + next := p.next + p.next = nil + c.carry[i] = p + i-- + p = next + } + break + } + + next, followers, rest := c.sliceBatch(c.carry) + c.carry = rest + c.queued.Add(-int64(1 + len(followers))) + next.elected = true + next.followers = followers + + // Closing wake hands carry, and leadership, to next. + next.signal() +} + +// sliceBatch takes the longest prefix of the queue that may share one commit. +// +// It stops at the first request that does not fit rather than skipping it, so +// requests keep their arrival order and a request that can never join somebody +// else's batch still reaches the head of the queue and leads one of its own. +// Called only by the current leader. +func (c *pushCoalescer) sliceBatch(pending []*pushTask) (leader *pushTask, followers, rest []*pushTask) { + leader = pending[0] + + samples := leader.samples + bytes := leader.bytes + + n := 1 + for ; n < len(pending); n++ { + t := pending[n] + + reason := "" + switch { + case n >= c.cfg.MaxBatchRequests: + reason = truncateReasonMaxRequests + case c.cfg.MaxBatchSamples > 0 && samples+t.samples > c.cfg.MaxBatchSamples: + reason = truncateReasonMaxSamples + case c.cfg.MaxBatchBytes > 0 && bytes+t.bytes > c.cfg.MaxBatchBytes: + reason = truncateReasonMaxBytes + case t.discardOutOfOrder != leader.discardOutOfOrder: + reason = truncateReasonDiscardOOO + } + if reason != "" { + c.metrics.truncated.WithLabelValues(reason).Inc() + break + } + + followers = append(followers, t) + samples += t.samples + bytes += t.bytes + } + + // Copy the remainder so that the batch does not keep the queue's backing + // array, and the tasks it held, alive for as long as the queue lives. + if n < len(pending) { + rest = make([]*pushTask, len(pending)-n) + copy(rest, pending[n:]) + } + return leader, followers, rest +} diff --git a/pkg/ingester/push_coalesce_bench_test.go b/pkg/ingester/push_coalesce_bench_test.go new file mode 100644 index 00000000000..076462a5725 --- /dev/null +++ b/pkg/ingester/push_coalesce_bench_test.go @@ -0,0 +1,523 @@ +package ingester + +import ( + "context" + "fmt" + "runtime" + "runtime/metrics" + "slices" + "sync" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/model" + "github.com/prometheus/prometheus/model/labels" + "github.com/stretchr/testify/require" + "github.com/weaveworks/common/user" + "go.uber.org/atomic" + + "github.com/cortexproject/cortex/pkg/cortexpb" + "github.com/cortexproject/cortex/pkg/ring" + "github.com/cortexproject/cortex/pkg/util/services" + "github.com/cortexproject/cortex/pkg/util/test" +) + +// The benchmarks in this file reproduce one traffic shape: a single tenant whose write requests are +// sharded by series across a large ingester fleet, so each ingester sees a very +// high rate of very small pushes, every one of which commits on its own. +// +// Reported metrics, per configuration: +// +// commits/push TSDB commits per push. 1.0 means no coalescing happened. +// requests/commit Mean batch size. +// p50_us, p99_us Per-push latency as the pusher sees it. +// pushes/s Achieved throughput. +// us/push Wall time per push. Prefer this over ns/op, which the Go +// harness computes from b.N and which is wrong whenever +// concurrency exceeds b.N. +// +// Typical invocation: +// +// go test -tags "netgo slicelabels" ./pkg/ingester/ -run XXX \ +// -bench BenchmarkIngesterPushCoalescing -benchtime 200000x -timeout 60m +// +// To show where the contention went: +// +// go test -tags "netgo slicelabels" ./pkg/ingester/ -run XXX \ +// -bench 'BenchmarkIngesterPushCoalescing/concurrency=4096/coalescing=off' \ +// -benchtime 200000x -mutexprofile /tmp/mutex-off.out + +const ( + // Samples per request. Small on purpose: when a distributor shards a large + // tenant's remote-write requests by series across many ingesters, each + // ingester's share of a request is only a handful of samples. + benchSamplesPerRequest = 5 + + // Series each pusher owns in the disjoint workload. Kept small so that the + // total series count stays bounded at high concurrency: a benchmark that + // creates a new series on nearly every push measures series creation and + // the label interner, not the append and commit path it is meant to + // measure. Every series is created in a warm-up phase before the clock + // starts, so the measured section is steady state, as a production + // ingester is. + benchSeriesPerPusher = 8 + + // Series the shared workload draws from. Small enough that concurrent + // pushers collide on series often, which is what exercises the cost of + // keeping overlapping requests out of one batch. + benchSharedSeriesCount = 512 + + // Interval between successive samples of one series. Small enough that the + // whole run stays well inside the head's appendable window even when a + // pusher issues many thousands of requests. + benchSampleIntervalMs = 100 +) + +// benchPusher holds everything one pushing goroutine needs, so that the +// benchmark harness itself introduces no shared mutable state. A shared mutex +// here would dominate the measurement and hide the WAL contention the +// benchmark exists to measure. +type benchPusher struct { + series []labels.Labels + + // step is this pusher's own sample clock. It is deliberately not shared: + // with a global clock a goroutine that lags behind the others ends up + // pushing a timestamp the head has already moved its minValidTime past, + // which the TSDB rejects as out of bounds and which has nothing to do with + // what the benchmark measures. + step int64 + + lbls []labels.Labels + samples []cortexpb.Sample + next int +} + +func benchSeriesLabels(idx int) labels.Labels { + return labels.FromStrings( + labels.MetricName, "bench_metric", + "pod", fmt.Sprintf("pod-%d", idx%97), + "instance", fmt.Sprintf("instance-%d", idx), + ) +} + +// newBenchPushers builds one pusher per goroutine. In the disjoint workload +// every pusher owns its own series, as a distributor sharding by series +// produces. In the shared workload they all draw from one small pool, so +// concurrent requests frequently write the same series. +func newBenchPushers(concurrency int, shared bool) []*benchPusher { + pushers := make([]*benchPusher, concurrency) + + var sharedSeries []labels.Labels + if shared { + sharedSeries = make([]labels.Labels, benchSharedSeriesCount) + for i := range sharedSeries { + sharedSeries[i] = benchSeriesLabels(i) + } + } + + for p := range pushers { + pusher := &benchPusher{ + lbls: make([]labels.Labels, benchSamplesPerRequest), + samples: make([]cortexpb.Sample, benchSamplesPerRequest), + // Stagger the starting offset so that pushers do not all begin on + // the same series. + next: (p * benchSamplesPerRequest) % max(1, benchSharedSeriesCount), + } + if shared { + pusher.series = sharedSeries + } else { + pusher.series = make([]labels.Labels, benchSeriesPerPusher) + for i := range pusher.series { + pusher.series[i] = benchSeriesLabels(p*benchSeriesPerPusher + i) + } + pusher.next = 0 + } + pushers[p] = pusher + } + return pushers +} + +// request builds the next write request for this pusher: benchSamplesPerRequest +// consecutive series, the way one ingester's shard of a scrape looks. +func (p *benchPusher) request() *cortexpb.WriteRequest { + p.step++ + ts := coalesceTestBaseMs + p.step*benchSampleIntervalMs + + for i := range benchSamplesPerRequest { + idx := (p.next + i) % len(p.series) + p.lbls[i] = p.series[idx] + p.samples[i] = cortexpb.Sample{TimestampMs: ts, Value: float64(idx)} + } + p.next = (p.next + benchSamplesPerRequest) % len(p.series) + + return cortexpb.ToWriteRequest(p.lbls, p.samples, nil, nil, cortexpb.API) +} + +// warmupRequests is how many requests this pusher needs in order to have +// touched, and therefore created, every series it owns. +func (p *benchPusher) warmupRequests() int { + return (len(p.series) + benchSamplesPerRequest - 1) / benchSamplesPerRequest +} + +func startBenchIngester(b *testing.B, coalesce PushCoalesceConfig) (*Ingester, *prometheus.Registry) { + b.Helper() + + cfg := defaultIngesterTestConfig(b) + cfg.PushCoalesce = coalesce + // Active series bookkeeping is identical with and without coalescing and + // only adds noise to the comparison. + cfg.ActiveSeriesMetricsEnabled = false + + // Leave label string interning at its upstream default. With it on, the + // interner's pool mutex dominates mutex contention during series creation + // and masks the WAL contention being measured. + cfg.LabelsStringInterningEnabled = false + + limits := defaultLimitsTestConfig() + // An out-of-order window changes which code path a late sample takes, and + // is commonly enabled in production, so the benchmark runs with one too. + limits.OutOfOrderTimeWindow = model.Duration(time.Duration(coalesceTestOOOWindowMs) * time.Millisecond) + limits.MaxLocalSeriesPerUser = 0 + limits.MaxLocalSeriesPerMetric = 0 + limits.MaxGlobalSeriesPerUser = 0 + limits.MaxGlobalSeriesPerMetric = 0 + + reg := prometheus.NewPedanticRegistry() + ing, err := prepareIngesterWithBlocksStorageAndLimits(b, cfg, limits, nil, "", reg) + require.NoError(b, err) + require.NoError(b, services.StartAndAwaitRunning(context.Background(), ing)) + b.Cleanup(func() { _ = services.StopAndAwaitTerminated(context.Background(), ing) }) + + test.Poll(b, time.Second, ring.ACTIVE, func() any { + return ing.lifecycler.GetState() + }) + return ing, reg +} + +func BenchmarkIngesterPushCoalescing(b *testing.B) { + variants := []struct { + name string + cfg PushCoalesceConfig + }{ + {name: "coalescing=off", cfg: PushCoalesceConfig{}}, + {name: "coalescing=on", cfg: defaultCoalesceConfig()}, + } + + for _, concurrency := range []int{1, 64, 1024, 4096, 16384, 30000} { + for _, v := range variants { + b.Run(fmt.Sprintf("concurrency=%d/%s", concurrency, v.name), func(b *testing.B) { + benchmarkPush(b, concurrency, v.cfg, false) + }) + } + } +} + +// BenchmarkIngesterPushCoalescingSaturated compares the two coalescer queue +// implementations with the scheduler saturated the way the production trace +// showed it: GOMAXPROCS=8 and several hundred runnable goroutines that have +// nothing to do with the push path (gRPC transport goroutines in production). +// That is the condition under which a woken mutex waiter waits more than 1ms +// for a P and flips sync.Mutex into starvation mode regardless of how short +// the critical section is. Run it with -cpu 8. +func BenchmarkIngesterPushCoalescingSaturated(b *testing.B) { + variants := []struct { + name string + cfg PushCoalesceConfig + }{ + {name: "coalescing=off", cfg: PushCoalesceConfig{}}, + {name: "coalescing=on", cfg: defaultCoalesceConfig()}, + } + + for _, bg := range []int{0, 800} { + for _, concurrency := range []int{1024, 4096} { + for _, v := range variants { + b.Run(fmt.Sprintf("bg=%d/concurrency=%d/%s", bg, concurrency, v.name), func(b *testing.B) { + benchmarkPushWithLoad(b, concurrency, v.cfg, false, bg) + }) + } + } + } +} + +// benchBackgroundSpin is how long one background goroutine runs before +// yielding. With 800 of them on 8 Ps a runnable goroutine waits roughly +// 800*spin/8 for a P, which at 50us is in the 1-10ms range the production +// trace measured during a burst. +const benchBackgroundSpin = 50 * time.Microsecond + +// startBackgroundLoad keeps n goroutines runnable until stop is called. +func startBackgroundLoad(n int) (stop func()) { + if n == 0 { + return func() {} + } + var ( + done atomic.Bool + wg sync.WaitGroup + ) + for i := range n { + wg.Go(func() { + x := uint64(i) + for !done.Load() { + deadline := time.Now().Add(benchBackgroundSpin) + for time.Now().Before(deadline) { + x = x*6364136223846793005 + 1442695040888963407 + } + runtime.Gosched() + } + _ = x + }) + } + return func() { + done.Store(true) + wg.Wait() + } +} + +// mutexWaitSeconds is the process-wide time goroutines have spent blocked on +// sync.Mutex and sync.RWMutex, as the runtime accounts it. +func mutexWaitSeconds() float64 { + sample := []metrics.Sample{{Name: "/sync/mutex/wait/total:seconds"}} + metrics.Read(sample) + if sample[0].Value.Kind() != metrics.KindFloat64 { + return 0 + } + return sample[0].Value.Float64() +} + +// schedLatencies snapshots the runtime's histogram of time goroutines spent +// runnable before running. +func schedLatencies() *metrics.Float64Histogram { + sample := []metrics.Sample{{Name: "/sched/latencies:seconds"}} + metrics.Read(sample) + return sample[0].Value.Float64Histogram() +} + +// schedLatencyQuantile returns quantile q of the scheduling latencies observed +// between two snapshots. +func schedLatencyQuantile(before, after *metrics.Float64Histogram, q float64) float64 { + var total uint64 + delta := make([]uint64, len(after.Counts)) + for i := range after.Counts { + delta[i] = after.Counts[i] - before.Counts[i] + total += delta[i] + } + if total == 0 { + return 0 + } + target := uint64(q * float64(total)) + var seen uint64 + for i, c := range delta { + seen += c + if seen > target { + return after.Buckets[i+1] + } + } + return after.Buckets[len(after.Buckets)-1] +} + +// BenchmarkIngesterPushCoalescingSharedSeries measures the same thing on a +// workload where concurrent requests frequently write the same series, so that +// a batch regularly carries several requests for one series. Batch size is +// unaffected by that, but the per-series locks the appends take are, so it is +// worth measuring separately from the disjoint case. +func BenchmarkIngesterPushCoalescingSharedSeries(b *testing.B) { + variants := []struct { + name string + cfg PushCoalesceConfig + }{ + {name: "coalescing=off", cfg: PushCoalesceConfig{}}, + {name: "coalescing=on", cfg: defaultCoalesceConfig()}, + } + + for _, concurrency := range []int{1024, 4096} { + for _, v := range variants { + b.Run(fmt.Sprintf("concurrency=%d/%s", concurrency, v.name), func(b *testing.B) { + benchmarkPush(b, concurrency, v.cfg, true) + }) + } + } +} + +func benchmarkPush(b *testing.B, concurrency int, coalesce PushCoalesceConfig, sharedSeries bool) { + benchmarkPushWithLoad(b, concurrency, coalesce, sharedSeries, 0) +} + +func benchmarkPushWithLoad(b *testing.B, concurrency int, coalesce PushCoalesceConfig, sharedSeries bool, background int) { + ing, reg := startBenchIngester(b, coalesce) + ctx := user.InjectOrgID(context.Background(), coalesceTestUser) + pushers := newBenchPushers(concurrency, sharedSeries) + + // Create the tenant's TSDB outside the measured section so that TSDB + // creation and WAL replay are not attributed to the benchmark. + _, err := ing.Push(ctx, pushers[0].request()) + require.NoError(b, err) + + perPusher := max(b.N/concurrency, 1) + total := concurrency * perPusher + + // Each pusher records its own latencies, so the measurement adds no shared + // state of its own. + latencies := make([][]time.Duration, concurrency) + for i := range latencies { + latencies[i] = make([]time.Duration, 0, perPusher) + } + + var ( + failed atomic.Int64 + firstErr atomic.Value + wg sync.WaitGroup + ready sync.WaitGroup + startCh = make(chan struct{}) + ) + + for p := range concurrency { + ready.Add(1) + wg.Go(func() { + pusher := pushers[p] + + // Create every series this pusher writes, so that the measured + // section appends to existing series rather than paying series + // creation, which a production ingester does not do per push. + for range pusher.warmupRequests() { + if _, err := ing.Push(ctx, pusher.request()); err != nil { + failed.Add(1) + firstErr.CompareAndSwap(nil, err) + } + } + + ready.Done() + <-startCh + for range perPusher { + req := pusher.request() + start := time.Now() + if _, err := ing.Push(ctx, req); err != nil { + failed.Add(1) + firstErr.CompareAndSwap(nil, err) + } + latencies[p] = append(latencies[p], time.Since(start)) + } + }) + } + + // Have every goroutine spawned, warmed up and parked before the clock + // starts, so that neither goroutine creation nor series creation is + // measured and all pushers really do run at once. + ready.Wait() + require.Zerof(b, failed.Load(), "%d warm-up pushes failed, first error: %v", + failed.Load(), firstErr.Load()) + + failed.Store(0) + commitsBefore := commitCount(b, reg) + + stopBackground := startBackgroundLoad(background) + mutexWaitBefore := mutexWaitSeconds() + schedBefore := schedLatencies() + + b.ResetTimer() + wallStart := time.Now() + close(startCh) + wg.Wait() + wall := time.Since(wallStart) + b.StopTimer() + + mutexWait := mutexWaitSeconds() - mutexWaitBefore + schedAfter := schedLatencies() + stopBackground() + + require.Zerof(b, failed.Load(), "%d of %d measured pushes failed, first error: %v", + failed.Load(), total, firstErr.Load()) + + commits := commitCount(b, reg) - commitsBefore + + all := make([]time.Duration, 0, total) + for _, l := range latencies { + all = append(all, l...) + } + slices.Sort(all) + + b.ReportMetric(float64(commits)/float64(total), "commits/push") + if commits > 0 { + b.ReportMetric(float64(total)/float64(commits), "requests/commit") + } + b.ReportMetric(float64(percentile(all, 0.50).Microseconds()), "p50_us") + b.ReportMetric(float64(percentile(all, 0.99).Microseconds()), "p99_us") + b.ReportMetric(float64(total)/wall.Seconds(), "pushes/s") + b.ReportMetric(wall.Seconds()*1e6/float64(total), "us/push") + b.ReportMetric(mutexWait*1e6/float64(total), "mutexwait_us/push") + b.ReportMetric(schedLatencyQuantile(schedBefore, schedAfter, 0.50)*1e6, "sched_p50_us") + b.ReportMetric(schedLatencyQuantile(schedBefore, schedAfter, 0.99)*1e6, "sched_p99_us") +} + +func percentile(sorted []time.Duration, q float64) time.Duration { + if len(sorted) == 0 { + return 0 + } + idx := int(q * float64(len(sorted))) + if idx >= len(sorted) { + idx = len(sorted) - 1 + } + return sorted[idx] +} + +// commitCount returns the number of TSDB commits the ingester has issued. +// +// The two configurations are counted through different instruments because +// each one only observes on its own path, but both count exactly one event per +// Commit, so the ratio to the number of pushes is comparable. +func commitCount(b *testing.B, reg *prometheus.Registry) int64 { + b.Helper() + + families, err := reg.Gather() + require.NoError(b, err) + + for _, f := range families { + if f.GetName() == "cortex_ingester_push_coalesce_commits_total" { + if total := sumCounter(f); total > 0 { + return int64(total) + } + } + } + for _, f := range families { + if f.GetName() == "cortex_ingester_tsdb_appender_commit_duration_seconds" { + return int64(histogramCount(f)) + } + } + return 0 +} + +func sumCounter(f *dto.MetricFamily) float64 { + var total float64 + for _, m := range f.GetMetric() { + total += m.GetCounter().GetValue() + } + return total +} + +func histogramCount(f *dto.MetricFamily) uint64 { + var total uint64 + for _, m := range f.GetMetric() { + total += m.GetHistogram().GetSampleCount() + } + return total +} + +// BenchmarkIngesterPushCoalescingBatchSize sweeps the batch size cap at a +// concurrency where the queue is deep, which is where the cap decides how many +// commits it takes to drain the queue and therefore how long a queued request +// waits. +func BenchmarkIngesterPushCoalescingBatchSize(b *testing.B) { + for _, concurrency := range []int{1024, 4096} { + for _, maxRequests := range []int{8, 32, 128, 512} { + cfg := defaultCoalesceConfig() + cfg.MaxBatchRequests = maxRequests + // Raise the sample cap out of the way so that the request cap is + // the only thing being varied. + cfg.MaxBatchSamples = 0 + b.Run(fmt.Sprintf("concurrency=%d/max-batch-requests=%d", concurrency, maxRequests), func(b *testing.B) { + benchmarkPush(b, concurrency, cfg, false) + }) + } + } +} diff --git a/pkg/ingester/push_coalesce_test.go b/pkg/ingester/push_coalesce_test.go new file mode 100644 index 00000000000..895e8cf9df1 --- /dev/null +++ b/pkg/ingester/push_coalesce_test.go @@ -0,0 +1,1612 @@ +package ingester + +import ( + "cmp" + "context" + "fmt" + "slices" + "sync" + "testing" + "time" + + "github.com/go-kit/log" + "github.com/pkg/errors" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/model" + "github.com/prometheus/prometheus/model/exemplar" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/storage" + "github.com/prometheus/prometheus/tsdb/chunkenc" + "github.com/prometheus/prometheus/tsdb/tsdbutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/weaveworks/common/user" + + "github.com/cortexproject/cortex/pkg/cortexpb" + "github.com/cortexproject/cortex/pkg/ring" + "github.com/cortexproject/cortex/pkg/util/services" + "github.com/cortexproject/cortex/pkg/util/test" +) + +// Tests for push coalescing, the merging of concurrent Push requests for the +// same tenant into one TSDB commit (push_coalesce.go). +// +// The file is in two halves. The first drives pushCoalescer directly with fake +// appenders, so that batch composition, error delivery and cancellation can be +// controlled exactly. The second pushes real write requests through +// Ingester.Push into a real TSDB, and compares the outcome of a coalesced batch +// with the outcome of committing the same requests one at a time. + +// ---------------------------------------------------------------------------- +// pushCoalescer, driven directly with fake appenders. +// ---------------------------------------------------------------------------- + +// fakeAppender records which tasks appended to it and reports a configurable +// commit outcome. storage.Appender is embedded as a nil interface: the tests +// that use it never call the sample-appending methods, so only the few methods +// the coalescer itself uses are implemented. +type fakeAppender struct { + storage.Appender + + id int + commitErr error + + // gate blocks Commit until it is closed, which lets a test hold one commit + // open while it queues further requests behind it. + gate chan struct{} + gateHit chan struct{} + gateOnce sync.Once + + mtx sync.Mutex + taskIDs []int + committed bool + rolled bool +} + +func (a *fakeAppender) GetRef(labels.Labels, uint64) (storage.SeriesRef, labels.Labels) { + return 0, labels.EmptyLabels() +} + +func (a *fakeAppender) SetOptions(*storage.AppendOptions) {} + +func (a *fakeAppender) recordTask(id int) { + a.mtx.Lock() + defer a.mtx.Unlock() + a.taskIDs = append(a.taskIDs, id) +} + +func (a *fakeAppender) tasks() []int { + a.mtx.Lock() + defer a.mtx.Unlock() + return append([]int(nil), a.taskIDs...) +} + +func (a *fakeAppender) Commit() error { + if a.gate != nil { + a.gateOnce.Do(func() { close(a.gateHit) }) + <-a.gate + } + a.mtx.Lock() + a.committed = true + a.mtx.Unlock() + return a.commitErr +} + +func (a *fakeAppender) Rollback() error { + a.mtx.Lock() + a.rolled = true + a.mtx.Unlock() + return nil +} + +// coalesceHarness drives a pushCoalescer with fake appenders and makes batch +// composition deterministic: the first commit is held open, so every request +// started after it is guaranteed to be queued and to end up in the batch that +// runs once the gate is released. +type coalesceHarness struct { + t *testing.T + c *pushCoalescer + + commitErr error + + gate chan struct{} + gateHit chan struct{} + + mtx sync.Mutex + appenders []*fakeAppender +} + +func newCoalesceHarness(t *testing.T, cfg PushCoalesceConfig, commitErr error) *coalesceHarness { + t.Helper() + + cfg.Enabled = true + h := &coalesceHarness{ + t: t, + commitErr: commitErr, + gate: make(chan struct{}), + gateHit: make(chan struct{}), + } + + h.c = newPushCoalescer( + cfg, + newPushCoalescerMetrics(prometheus.NewPedanticRegistry()), + log.NewNopLogger(), + func() extendedAppender { + h.mtx.Lock() + defer h.mtx.Unlock() + app := &fakeAppender{id: len(h.appenders), commitErr: h.commitErr} + if len(h.appenders) == 0 { + // Only the first appender gates, so that the batch under test + // commits without further interference. + app.gate, app.gateHit = h.gate, h.gateHit + } + h.appenders = append(h.appenders, app) + return app + }, + func() int64 { return 0 }, + ) + return h +} + +// task builds a task whose append records its id on the appender it is given. +func (h *coalesceHarness) task(ctx context.Context, id int, opts ...func(*pushTask)) (*pushTask, *int32) { + var appends int32 + t := &pushTask{ + ctx: ctx, + samples: 1, + appendFn: func(app extendedAppender, _ int64) error { + app.(*fakeAppender).recordTask(id) + appends++ + return nil + }, + } + for _, o := range opts { + o(t) + } + return t, &appends +} + +func (h *coalesceHarness) pushAsync(t *pushTask) <-chan error { + res := make(chan error, 1) + go func() { res <- h.c.push(t) }() + return res +} + +// waitForFirstCommit blocks until the gated first commit has been entered, at +// which point the coalescer is busy and anything pushed next will queue. +func (h *coalesceHarness) waitForFirstCommit() { + h.t.Helper() + select { + case <-h.gateHit: + case <-time.After(10 * time.Second): + h.t.Fatal("first commit was never entered") + } +} + +// waitForPending blocks until exactly n tasks are queued. +func (h *coalesceHarness) waitForPending(n int) { + h.t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + got := h.c.pendingLen() + if got == n { + return + } + if time.Now().After(deadline) { + h.t.Fatalf("expected %d queued requests, got %d", n, got) + } + time.Sleep(time.Millisecond) + } +} + +func (h *coalesceHarness) release() { close(h.gate) } + +func (h *coalesceHarness) appenderList() []*fakeAppender { + h.mtx.Lock() + defer h.mtx.Unlock() + return append([]*fakeAppender(nil), h.appenders...) +} + +// batches returns, per appender after the gated one, the task ids it carried. +func (h *coalesceHarness) batches() [][]int { + var out [][]int + for _, app := range h.appenderList()[1:] { + out = append(out, app.tasks()) + } + return out +} + +func mustRecv(t *testing.T, ch <-chan error) error { + t.Helper() + select { + case err := <-ch: + return err + case <-time.After(10 * time.Second): + t.Fatal("push did not return") + return nil + } +} + +// errTestFatalAppend stands in for an append failure that requires the +// appender to be discarded. +var errTestFatalAppend = errors.New("test: append failed fatally") + +func defaultCoalesceConfig() PushCoalesceConfig { + return PushCoalesceConfig{ + Enabled: true, + MaxBatchRequests: defaultPushCoalesceMaxBatchRequests, + MaxBatchSamples: defaultPushCoalesceMaxBatchSamples, + MaxBatchBytes: 0, + } +} + +// A request that arrives while nothing is committing must commit on its own +// and not wait for anybody: that is what keeps coalescing free of added +// latency. +func TestPushCoalescer_LoneRequestCommitsImmediately(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + h.release() // nothing to hold open + + task, appends := h.task(context.Background(), 1) + require.NoError(t, h.c.push(task)) + + require.Equal(t, int32(1), *appends) + apps := h.appenderList() + require.Len(t, apps, 1, "a lone request must not create more than one appender") + require.Equal(t, []int{1}, apps[0].tasks()) + require.True(t, apps[0].committed) +} + +// Requests that arrive while a commit is in flight must be folded into one +// commit behind it. +func TestPushCoalescer_QueuedRequestsShareOneCommit(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + + seed, _ := h.task(context.Background(), 0) + seedDone := h.pushAsync(seed) + h.waitForFirstCommit() + + const n = 5 + dones := make([]<-chan error, 0, n) + for id := 1; id <= n; id++ { + task, _ := h.task(context.Background(), id) + dones = append(dones, h.pushAsync(task)) + // Enqueue one at a time so that the recorded order is the arrival + // order; which requests share the commit is what matters here, but an + // exact assertion reads better than a set comparison. + h.waitForPending(id) + } + + h.release() + require.NoError(t, mustRecv(t, seedDone)) + for _, d := range dones { + require.NoError(t, mustRecv(t, d)) + } + + require.Equal(t, [][]int{{1, 2, 3, 4, 5}}, h.batches(), + "all five queued requests should have shared a single commit") +} + +// A commit failure is the one case where every request in the batch has to be +// told, because none of their samples reached the WAL. +func TestPushCoalescer_CommitFailureFailsWholeBatch(t *testing.T) { + t.Parallel() + + commitErr := errors.New("wal is full") + h := newCoalesceHarness(t, defaultCoalesceConfig(), commitErr) + + seed, _ := h.task(context.Background(), 0) + seedDone := h.pushAsync(seed) + h.waitForFirstCommit() + + const n = 4 + dones := make([]<-chan error, 0, n) + for id := 1; id <= n; id++ { + task, _ := h.task(context.Background(), id) + dones = append(dones, h.pushAsync(task)) + } + h.waitForPending(n) + h.release() + + require.ErrorIs(t, mustRecv(t, seedDone), commitErr) + for i, d := range dones { + require.ErrorIsf(t, mustRecv(t, d), commitErr, "request %d", i+1) + } +} + +// A request that fails in a way that requires discarding the appender takes the +// whole batch down with it, because an appender cannot discard one request's +// samples alone. The offender gets its own error; everybody else gets a +// retryable one, since their samples were never committed. +func TestPushCoalescer_FatalAppendRollsBackWholeBatch(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + + seed, _ := h.task(context.Background(), 0) + seedDone := h.pushAsync(seed) + h.waitForFirstCommit() + + fatalErr := errors.New("label set is out of order") + + okTask, _ := h.task(context.Background(), 1) + okDone := h.pushAsync(okTask) + h.waitForPending(1) + + badTask, _ := h.task(context.Background(), 2) + badTask.appendFn = func(extendedAppender, int64) error { return fatalErr } + badDone := h.pushAsync(badTask) + h.waitForPending(2) + + lateTask, _ := h.task(context.Background(), 3) + lateDone := h.pushAsync(lateTask) + h.waitForPending(3) + + h.release() + + require.NoError(t, mustRecv(t, seedDone)) + require.ErrorIs(t, mustRecv(t, badDone), fatalErr) + require.ErrorIs(t, mustRecv(t, okDone), errPushCoalesceBatchRolledBack) + require.ErrorIs(t, mustRecv(t, lateDone), errPushCoalesceBatchRolledBack) + + apps := h.appenderList() + require.Len(t, apps, 2) + require.True(t, apps[1].rolled, "the batch appender should have been rolled back") + require.False(t, apps[1].committed) +} + +// A caller that has already given up is told so, and its samples are not +// committed on its behalf. +func TestPushCoalescer_ContextCanceledBeforeJoining(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + h.release() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + task, appends := h.task(ctx, 1) + require.ErrorIs(t, h.c.push(task), context.Canceled) + require.Equal(t, int32(0), *appends, "a canceled request must not be appended") + require.Empty(t, h.appenderList(), "a canceled request must not create an appender") +} + +// A request abandoned while it sat in the queue is dropped when the batch runs, +// without affecting the rest of the batch. +func TestPushCoalescer_ContextCanceledWhileQueued(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + + seed, _ := h.task(context.Background(), 0) + seedDone := h.pushAsync(seed) + h.waitForFirstCommit() + + keepCtx := context.Background() + dropCtx, cancel := context.WithCancel(context.Background()) + + keep, keepAppends := h.task(keepCtx, 1) + keepDone := h.pushAsync(keep) + h.waitForPending(1) + + drop, dropAppends := h.task(dropCtx, 2) + dropDone := h.pushAsync(drop) + h.waitForPending(2) + + // Give up only after the request is safely queued. + cancel() + h.release() + + require.NoError(t, mustRecv(t, seedDone)) + require.NoError(t, mustRecv(t, keepDone)) + require.ErrorIs(t, mustRecv(t, dropDone), context.Canceled) + + require.Equal(t, int32(1), *keepAppends) + require.Equal(t, int32(0), *dropAppends, "an abandoned request must not be appended") + require.Equal(t, [][]int{{1}}, h.batches()) +} + +// A request abandoned while queued keeps its own context error even when the +// batch it was queued in is rolled back because a different request failed. +// Without the explicit dropped flag, re-reading the context during the +// rollback could reclassify it. +func TestPushCoalescer_CanceledRequestKeepsItsErrorThroughRollback(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + + seed, _ := h.task(context.Background(), 0) + seedDone := h.pushAsync(seed) + h.waitForFirstCommit() + + dropCtx, cancel := context.WithCancel(context.Background()) + drop, dropAppends := h.task(dropCtx, 1) + dropDone := h.pushAsync(drop) + h.waitForPending(1) + + fatalErr := errTestFatalAppend + bad, _ := h.task(context.Background(), 2) + bad.appendFn = func(extendedAppender, int64) error { return fatalErr } + badDone := h.pushAsync(bad) + h.waitForPending(2) + + other, _ := h.task(context.Background(), 3) + otherDone := h.pushAsync(other) + h.waitForPending(3) + + // Abandon the first queued request only after it is safely in the queue, + // so the leader meets it already canceled. + cancel() + h.release() + + require.NoError(t, mustRecv(t, seedDone)) + require.ErrorIs(t, mustRecv(t, dropDone), context.Canceled, + "a dropped request must keep its context error, not the rollback error") + require.ErrorIs(t, mustRecv(t, badDone), fatalErr) + require.ErrorIs(t, mustRecv(t, otherDone), errPushCoalesceBatchRolledBack) + require.Equal(t, int32(0), *dropAppends) +} + +func TestPushCoalescer_BatchLimits(t *testing.T) { + t.Parallel() + + withSamples := func(n int) func(*pushTask) { + return func(t *pushTask) { t.samples = n } + } + withBytes := func(n int) func(*pushTask) { + return func(t *pushTask) { t.bytes = n } + } + withDiscardOOO := func(v bool) func(*pushTask) { + return func(t *pushTask) { t.discardOutOfOrder = v } + } + + for _, tc := range []struct { + name string + cfg func(*PushCoalesceConfig) + opts [][]func(*pushTask) + batches [][]int + }{ + { + name: "max requests truncates and the remainder forms the next batch", + cfg: func(c *PushCoalesceConfig) { c.MaxBatchRequests = 2 }, + opts: [][]func(*pushTask){nil, nil, nil, nil, nil}, + // Requests 1 and 2 share a commit; 3 and 4 the next; 5 the last. + batches: [][]int{{1, 2}, {3, 4}, {5}}, + }, + { + name: "max samples truncates", + cfg: func(c *PushCoalesceConfig) { c.MaxBatchSamples = 10 }, + opts: [][]func(*pushTask){ + {withSamples(6)}, + {withSamples(4)}, + {withSamples(5)}, + }, + batches: [][]int{{1, 2}, {3}}, + }, + { + name: "a request larger than the sample limit still leads its own batch", + cfg: func(c *PushCoalesceConfig) { c.MaxBatchSamples = 10 }, + opts: [][]func(*pushTask){ + {withSamples(4)}, + {withSamples(100)}, + {withSamples(4)}, + }, + batches: [][]int{{1}, {2}, {3}}, + }, + { + name: "max bytes truncates", + cfg: func(c *PushCoalesceConfig) { c.MaxBatchBytes = 1000 }, + opts: [][]func(*pushTask){ + {withBytes(600)}, + {withBytes(300)}, + {withBytes(400)}, + }, + batches: [][]int{{1, 2}, {3}}, + }, + { + name: "requests disagreeing on discarding out of order samples do not share a commit", + cfg: func(*PushCoalesceConfig) {}, + opts: [][]func(*pushTask){ + {withDiscardOOO(false)}, + {withDiscardOOO(false)}, + {withDiscardOOO(true)}, + {withDiscardOOO(true)}, + }, + batches: [][]int{{1, 2}, {3, 4}}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + cfg := defaultCoalesceConfig() + tc.cfg(&cfg) + + h := newCoalesceHarness(t, cfg, nil) + seed, _ := h.task(context.Background(), 0) + seedDone := h.pushAsync(seed) + h.waitForFirstCommit() + + dones := make([]<-chan error, 0, len(tc.opts)) + for idx, opts := range tc.opts { + task, _ := h.task(context.Background(), idx+1, opts...) + dones = append(dones, h.pushAsync(task)) + // Enqueue one at a time so that arrival order is the order the + // batch slicer sees. + h.waitForPending(idx + 1) + } + + h.release() + require.NoError(t, mustRecv(t, seedDone)) + for _, d := range dones { + require.NoError(t, mustRecv(t, d)) + } + + assert.Equal(t, tc.batches, h.batches()) + }) + } +} + +// Under concurrency the coalescer must commit every request exactly once and +// leave nobody blocked. Run with -race. +func TestPushCoalescer_Concurrent(t *testing.T) { + t.Parallel() + + h := newCoalesceHarness(t, defaultCoalesceConfig(), nil) + h.release() + + const ( + goroutines = 64 + perRoutine = 50 + ) + + var wg sync.WaitGroup + for g := range goroutines { + wg.Go(func() { + for n := range perRoutine { + task, appends := h.task(context.Background(), g*perRoutine+n) + require.NoError(t, h.c.push(task)) + require.Equal(t, int32(1), *appends) + } + }) + } + wg.Wait() + + // Every request appended exactly once, across however many commits the + // coalescer decided to use. + seen := map[int]int{} + total := 0 + for _, app := range h.appenderList() { + for _, id := range app.tasks() { + seen[id]++ + total++ + } + } + require.Equal(t, goroutines*perRoutine, total) + require.Len(t, seen, goroutines*perRoutine) + for id, count := range seen { + require.Equalf(t, 1, count, "task %d appended %d times", id, count) + } + + // The coalescer must be idle again, with nothing queued. + require.True(t, h.c.isIdle()) + + // It must also have done real batching, otherwise the test proves nothing + // about the mechanism it is meant to exercise. + require.Less(t, len(h.appenderList()), goroutines*perRoutine, + "expected fewer commits than requests") + t.Logf("%d requests committed in %d commits", total, len(h.appenderList())) +} + +func TestPushCoalesceConfig_Validate(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + cfg PushCoalesceConfig + err string + }{ + {name: "disabled config is never rejected", cfg: PushCoalesceConfig{}}, + { + name: "max batch requests must be positive", + cfg: PushCoalesceConfig{Enabled: true, MaxBatchRequests: 0}, + err: "ingester.push-coalesce.max-batch-requests must be at least 1", + }, + { + name: "max batch samples must not be negative", + cfg: PushCoalesceConfig{Enabled: true, MaxBatchRequests: 1, MaxBatchSamples: -1}, + err: "ingester.push-coalesce.max-batch-samples must not be negative", + }, + { + name: "max batch bytes must not be negative", + cfg: PushCoalesceConfig{Enabled: true, MaxBatchRequests: 1, MaxBatchBytes: -1}, + err: "ingester.push-coalesce.max-batch-bytes must not be negative", + }, + {name: "valid config", cfg: defaultCoalesceConfig()}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + err := tc.cfg.Validate() + if tc.err == "" { + require.NoError(t, err) + return + } + require.EqualError(t, err, tc.err) + }) + } +} + +func TestPushCoalescer_SliceBatchReportsTruncationReason(t *testing.T) { + t.Parallel() + + cfg := defaultCoalesceConfig() + cfg.MaxBatchRequests = 2 + reg := prometheus.NewPedanticRegistry() + c := newPushCoalescer(cfg, newPushCoalescerMetrics(reg), log.NewNopLogger(), + func() extendedAppender { return &fakeAppender{} }, + func() int64 { return 0 }) + + pending := []*pushTask{{samples: 1}, {samples: 1}, {samples: 1}} + leader, followers, rest := c.sliceBatch(pending) + + require.Same(t, pending[0], leader) + require.Len(t, followers, 1) + require.Len(t, rest, 1) + require.Equal(t, 1.0, promCounterValue(t, reg, + "cortex_ingester_push_coalesce_batch_truncated_total", truncateReasonMaxRequests)) +} + +func promCounterValue(t *testing.T, reg *prometheus.Registry, name, label string) float64 { + t.Helper() + families, err := reg.Gather() + require.NoError(t, err) + for _, f := range families { + if f.GetName() != name { + continue + } + for _, m := range f.GetMetric() { + for _, l := range m.GetLabel() { + if l.GetValue() == label { + return m.GetCounter().GetValue() + } + } + } + } + t.Fatalf("metric %s with label %s not found", name, label) + return 0 +} + +// ---------------------------------------------------------------------------- +// Ingester.Push into a real TSDB: helpers. +// ---------------------------------------------------------------------------- + +const ( + coalesceTestUser = "coalesce-user" + + // The out-of-order window the tests run with, in milliseconds. + coalesceTestOOOWindowMs = int64(30 * 60 * 1000) + + // Base timestamp the scenarios build on. Far enough from zero that samples + // can be placed before it without becoming out of bounds. + coalesceTestBaseMs = int64(10 * 60 * 60 * 1000) +) + +// writeRequestSpec describes one write request. Requests are rebuilt from the +// spec for every run, because Push returns the request's time series to a pool. +type writeRequestSpec struct { + series []labels.Labels + // samples[i] belongs to series[i]. + samples []cortexpb.Sample + // histograms[i] belongs to series[i]. + histograms []cortexpb.WrappedHistogram + // exemplars are all attached to the first series. + exemplars []cortexpb.Exemplar +} + +func (s writeRequestSpec) build() *cortexpb.WriteRequest { + req := cortexpb.ToWriteRequest(s.series, s.samples, nil, s.histograms, cortexpb.API) + for _, ex := range s.exemplars { + // The exemplar's label adapters must be copied per build. Push returns + // the time series to a pool, and cortexpb.ReuseTimeseries clears + // exemplar label names and values in place, which would write through + // to the spec's own slice and leave a later build with empty labels. + copied := cortexpb.Exemplar{TimestampMs: ex.TimestampMs, Value: ex.Value} + copied.Labels = append(copied.Labels, ex.Labels...) + req.Timeseries[0].Exemplars = append(req.Timeseries[0].Exemplars, copied) + } + return req +} + +func coalesceTestSeries(name string) labels.Labels { + return labels.FromStrings(labels.MetricName, name) +} + +// oneSample is a request carrying a single sample for a single series. +func oneSample(seriesName string, tMs int64, v float64) writeRequestSpec { + return writeRequestSpec{ + series: []labels.Labels{coalesceTestSeries(seriesName)}, + samples: []cortexpb.Sample{{TimestampMs: tMs, Value: v}}, + } +} + +// histogramRequest builds a request carrying one native histogram. +func histogramRequest(seriesName string, tMs int64, seed int) writeRequestSpec { + return writeRequestSpec{ + series: []labels.Labels{coalesceTestSeries(seriesName)}, + histograms: []cortexpb.WrappedHistogram{ + cortexpb.WrapHistogram(cortexpb.HistogramToHistogramProto(tMs, tsdbutil.GenerateTestHistogram(int64(seed)))), + }, + } +} + +// exemplarRequest builds a request carrying one float sample and one exemplar +// attached to it. +func exemplarRequest(seriesName string, tMs int64, v float64, traceID string) writeRequestSpec { + return writeRequestSpec{ + series: []labels.Labels{coalesceTestSeries(seriesName)}, + samples: []cortexpb.Sample{{TimestampMs: tMs, Value: v}}, + exemplars: []cortexpb.Exemplar{{ + Labels: []cortexpb.LabelAdapter{{Name: "traceID", Value: traceID}}, + TimestampMs: tMs, + Value: v, + }}, + } +} + +// startCoalesceTestIngester starts an ingester with the given push coalescing +// configuration, an out-of-order window, native histograms and exemplars +// enabled, and waits until it is ACTIVE in the ring. +func startCoalesceTestIngester(t *testing.T, coalesce PushCoalesceConfig) (*Ingester, *prometheus.Registry) { + t.Helper() + + cfg := defaultIngesterTestConfig(t) + cfg.PushCoalesce = coalesce + + limits := defaultLimitsTestConfig() + limits.OutOfOrderTimeWindow = model.Duration(time.Duration(coalesceTestOOOWindowMs) * time.Millisecond) + limits.EnableNativeHistograms = true + limits.MaxExemplars = 10 + + reg := prometheus.NewPedanticRegistry() + ing, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, nil, "", reg) + require.NoError(t, err) + require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing)) + t.Cleanup(func() { _ = services.StopAndAwaitTerminated(context.Background(), ing) }) + + test.Poll(t, time.Second, ring.ACTIVE, func() any { + return ing.lifecycler.GetState() + }) + return ing, reg +} + +// pushSequentially pushes each request on its own and returns the error each +// one produced. This is the behaviour coalescing has to reproduce. +func pushSequentially(t *testing.T, ing *Ingester, specs []writeRequestSpec) []error { + t.Helper() + + ctx := user.InjectOrgID(context.Background(), coalesceTestUser) + errs := make([]error, 0, len(specs)) + for _, spec := range specs { + _, err := ing.Push(ctx, spec.build()) + errs = append(errs, err) + } + return errs +} + +// mirrorBatchHarnessPushes replays, on an ingester driven sequentially, the two +// requests pushAsOneBatch issues of its own accord: one to create the tenant's +// TSDB and one to hold a commit open. Without this the two sides would have +// pushed a different number of requests and their counters could not be +// compared. +func mirrorBatchHarnessPushes(t *testing.T, ing *Ingester) { + t.Helper() + for _, err := range pushSequentially(t, ing, []writeRequestSpec{ + oneSample("coalesce_gate_seed", coalesceTestBaseMs, 1), + oneSample("coalesce_gate_holder", coalesceTestBaseMs, 1), + }) { + require.NoError(t, err) + } +} + +// pushAsOneBatch pushes every request into a single coalesced commit, and +// returns the error each one produced. +// +// Batch composition is made deterministic rather than left to chance: a gate +// holds one commit open, every request under test is queued behind it, and only +// then is the gate released, so the queued requests necessarily form one batch. +func pushAsOneBatch(t *testing.T, ing *Ingester, specs []writeRequestSpec) []error { + t.Helper() + + ctx := user.InjectOrgID(context.Background(), coalesceTestUser) + + // The tenant's TSDB, and with it the coalescer, is created by the first + // push. Seed it with a series none of the scenarios touch. + _, err := ing.Push(ctx, oneSample("coalesce_gate_seed", coalesceTestBaseMs, 1).build()) + require.NoError(t, err) + + db, err := ing.getTSDB(coalesceTestUser) + require.NoError(t, err) + require.NotNil(t, db.pushCoalescer, "push coalescing must be enabled for this test") + + // Replace the coalescer with one whose first appender blocks in Commit. + // Everything else about it, including the real TSDB appender, is unchanged. + gate := make(chan struct{}) + gateHit := make(chan struct{}) + var ( + mtx sync.Mutex + appenders int + ) + db.pushCoalescer = newPushCoalescer( + ing.cfg.PushCoalesce, + ing.metrics.pushCoalesce, + log.NewNopLogger(), + func() extendedAppender { + real := db.Appender(context.Background()).(extendedAppender) + mtx.Lock() + first := appenders == 0 + appenders++ + mtx.Unlock() + if !first { + return real + } + return &gatedAppender{extendedAppender: real, gate: gate, gateHit: gateHit} + }, + func() int64 { return db.Head().MaxTime() }, + ) + + // Hold a commit open on a series no scenario uses. + gateDone := make(chan error, 1) + go func() { + _, err := ing.Push(ctx, oneSample("coalesce_gate_holder", coalesceTestBaseMs, 1).build()) + gateDone <- err + }() + select { + case <-gateHit: + case <-time.After(30 * time.Second): + t.Fatal("gate commit was never entered") + } + + // Queue every request under test behind the held commit, one at a time so + // that they reach the batch in the order the scenario declares them. + results := make([]chan error, len(specs)) + for idx, spec := range specs { + results[idx] = make(chan error, 1) + go func(spec writeRequestSpec, out chan error) { + _, err := ing.Push(ctx, spec.build()) + out <- err + }(spec, results[idx]) + + deadline := time.Now().Add(30 * time.Second) + for { + queued := db.pushCoalescer.pendingLen() + if queued == idx+1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("expected %d queued requests, got %d", idx+1, queued) + } + time.Sleep(time.Millisecond) + } + } + + close(gate) + require.NoError(t, <-gateDone) + + errs := make([]error, len(specs)) + for idx, res := range results { + select { + case errs[idx] = <-res: + case <-time.After(30 * time.Second): + t.Fatalf("request %d did not return", idx) + } + } + return errs +} + +// gatedAppender blocks in Commit until its gate is closed. +type gatedAppender struct { + extendedAppender + + gate chan struct{} + gateHit chan struct{} + once sync.Once +} + +func (a *gatedAppender) Commit() error { + a.once.Do(func() { close(a.gateHit) }) + <-a.gate + return a.extendedAppender.Commit() +} + +var _ storage.Appender = (*gatedAppender)(nil) + +// headSample is one sample as stored in the head, in-order or out-of-order. +type headSample struct { + T int64 + V float64 +} + +// dumpHead returns every sample the tenant's head holds, keyed by series. The +// querier covers the out-of-order head as well, so a sample that landed there +// shows up here. Histogram samples are reported by their count plus sum, so +// that a comparison does not depend on bucket layout details. +func dumpHead(t *testing.T, ing *Ingester) map[string][]headSample { + t.Helper() + + db, err := ing.getTSDB(coalesceTestUser) + require.NoError(t, err) + + q, err := db.Querier(0, coalesceTestBaseMs*10) + require.NoError(t, err) + t.Cleanup(func() { _ = q.Close() }) + + matcher, err := labels.NewMatcher(labels.MatchRegexp, labels.MetricName, ".+") + require.NoError(t, err) + + out := map[string][]headSample{} + set := q.Select(context.Background(), true, nil, matcher) + for set.Next() { + s := set.At() + key := s.Labels().String() + it := s.Iterator(nil) + for vt := it.Next(); vt != chunkenc.ValNone; vt = it.Next() { + switch vt { + case chunkenc.ValFloat: + ts, v := it.At() + out[key] = append(out[key], headSample{T: ts, V: v}) + case chunkenc.ValHistogram: + ts, h := it.AtHistogram(nil) + out[key] = append(out[key], headSample{T: ts, V: float64(h.Count) + h.Sum}) + case chunkenc.ValFloatHistogram: + ts, fh := it.AtFloatHistogram(nil) + out[key] = append(out[key], headSample{T: ts, V: fh.Count + fh.Sum}) + } + } + require.NoError(t, it.Err()) + } + require.NoError(t, set.Err()) + require.Empty(t, set.Warnings().AsErrors()) + + for key := range out { + slices.SortFunc(out[key], func(a, b headSample) int { return cmp.Compare(a.T, b.T) }) + } + return out +} + +// dumpExemplars returns every exemplar stored for the named series, sorted so +// the comparison does not depend on query order. +func dumpExemplars(t *testing.T, ing *Ingester, seriesNames ...string) []string { + t.Helper() + + db, err := ing.getTSDB(coalesceTestUser) + require.NoError(t, err) + + q, err := db.ExemplarQuerier(context.Background()) + require.NoError(t, err) + + var matchers [][]*labels.Matcher + for _, name := range seriesNames { + m, err := labels.NewMatcher(labels.MatchEqual, labels.MetricName, name) + require.NoError(t, err) + matchers = append(matchers, []*labels.Matcher{m}) + } + + res, err := q.Select(0, coalesceTestBaseMs*10, matchers...) + require.NoError(t, err) + + var out []string + for _, r := range res { + for _, e := range r.Exemplars { + out = append(out, r.SeriesLabels.String()+" "+exemplarString(e)) + } + } + slices.Sort(out) + return out +} + +func exemplarString(e exemplar.Exemplar) string { + return e.Labels.String() + "@" + model.Time(e.Ts).String() +} + +func errStrings(errs []error) []string { + out := make([]string, len(errs)) + for i, err := range errs { + if err == nil { + out[i] = "" + continue + } + out[i] = err.Error() + } + return out +} + +// ---------------------------------------------------------------------------- +// Ingester.Push: what a coalesced batch changes compared with separate commits. +// ---------------------------------------------------------------------------- + +// coalesceScenario is a set of write requests whose interaction with each other +// is what makes merging their commits interesting, together with what merging +// them is expected to change. +// +// A batch shares one appender, and Append only sees committed state while +// Commit re-checks against the state earlier samples in the same batch left +// behind, counting rather than returning what it then rejects. So a batch can +// lose an error a separate commit would have returned, and in one case can +// accept a sample a separate commit would have rejected. Every scenario below +// declares which, so that any change in that set fails here rather than being +// discovered in production. +type coalesceScenario struct { + name string + specs []writeRequestSpec + + // errorsDiffer records that at least one request gets a different error + // than it would from a separate commit. + errorsDiffer bool + // dataDiffers records that the head ends up holding different samples. + dataDiffers bool +} + +func coalesceScenarios() []coalesceScenario { + const ( + base = coalesceTestBaseMs + win = coalesceTestOOOWindowMs + ) + + return []coalesceScenario{ + { + // The control: nothing shares a series, so merging changes nothing. + name: "disjoint series", + specs: []writeRequestSpec{ + oneSample("disjoint_a", base+1000, 1), + oneSample("disjoint_b", base+1000, 2), + oneSample("disjoint_c", base+1000, 3), + }, + }, + { + // Separate commits reject the second sample at Append time with a + // duplicate error. A shared appender only finds out at Commit + // time, where the error is counted and dropped rather than + // returned, so the client is told its write succeeded. The sample + // is dropped either way. + // + // Note that Cortex already behaves this way when the duplicate + // arrives inside one request: see + // TestIngesterPush_DuplicateIsAlreadySilentWithinOneRequest. + name: "same series, same timestamp, different value", + specs: []writeRequestSpec{ + oneSample("dup", base+1000, 1), + oneSample("dup", base+1000, 2), + }, + errorsDiffer: true, + }, + { + // An exact duplicate is accepted by Prometheus on purpose, so both + // paths must report success. + name: "same series, same timestamp, same value", + specs: []writeRequestSpec{ + oneSample("exact_dup", base+1000, 7), + oneSample("exact_dup", base+1000, 7), + }, + }, + { + // In-order then out-of-order, inside the window: accepted into the + // out-of-order head either way. + name: "same series, out of order within the window", + specs: []writeRequestSpec{ + oneSample("ooo_in_window", base+1000, 1), + oneSample("ooo_in_window", base+1000-win/2, 2), + }, + }, + { + // Out-of-order beyond the window. Separate commits return a + // too-old error on the second request; the batch drops the sample + // without reporting it. Same data either way. + name: "same series, out of order beyond the window", + specs: []writeRequestSpec{ + oneSample("ooo_too_old", base+1000, 1), + oneSample("ooo_too_old", base+1000-2*win, 2), + }, + errorsDiffer: true, + }, + { + // The only scenario where the stored data differs: a sample that + // Append accepts against committed state, and that a larger sample + // earlier in the same batch would have pushed outside the window + // had it been committed first. The batch accepts it into the + // out-of-order head; separate commits reject it as too old. + // + // The timestamps here exaggerate the effect to make it reachable at + // all. TestIngesterPushCoalescing_OutOfOrderEscapeIsBoundedByInBatchHeadAdvance + // bounds how far it can actually reach. + name: "same series, in-batch sample pushes a later one out of the window", + specs: []writeRequestSpec{ + oneSample("ooo_pushed_out", base+1000, 1), + oneSample("ooo_pushed_out", base+1000+4*win, 2), + oneSample("ooo_pushed_out", base+1000+win, 3), + }, + errorsDiffer: true, + dataDiffers: true, + }, + { + // Several samples for one series inside a single request behave + // identically in both modes, because a single request has always + // shared one appender. + name: "several samples for one series inside one request", + specs: []writeRequestSpec{ + { + series: []labels.Labels{ + coalesceTestSeries("intra"), + coalesceTestSeries("intra"), + }, + samples: []cortexpb.Sample{ + {TimestampMs: base + 1000, Value: 1}, + {TimestampMs: base + 1000, Value: 2}, + }, + }, + oneSample("intra_other", base+1000, 3), + }, + }, + { + // A new series created by one request and written again by another + // in the same batch: the second request finds the series through + // GetRef even though it has not been committed yet. + name: "new series written by two requests", + specs: []writeRequestSpec{ + oneSample("fresh", base+1000, 1), + oneSample("fresh", base+2000, 2), + oneSample("fresh", base+1000, 3), + }, + }, + } +} + +// A coalesced commit must change exactly what each scenario says it changes, +// and nothing else: the same error for every other request, and the same data +// in the head afterwards, including the out-of-order head. +func TestIngesterPushCoalescing_DivergenceFromSeparateCommits(t *testing.T) { + for _, sc := range coalesceScenarios() { + t.Run(sc.name, func(t *testing.T) { + sequential, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + coalesced, _ := startCoalesceTestIngester(t, defaultCoalesceConfig()) + + // The sequential run gets the same seed and gate-holder requests + // the coalesced run needs, so the two heads start out identical. + seedErrs := pushSequentially(t, sequential, []writeRequestSpec{ + oneSample("coalesce_gate_seed", coalesceTestBaseMs, 1), + oneSample("coalesce_gate_holder", coalesceTestBaseMs, 1), + }) + for _, err := range seedErrs { + require.NoError(t, err) + } + + wantErrs := errStrings(pushSequentially(t, sequential, sc.specs)) + gotErrs := errStrings(pushAsOneBatch(t, coalesced, sc.specs)) + wantHead := dumpHead(t, sequential) + gotHead := dumpHead(t, coalesced) + + if sc.errorsDiffer { + require.NotEqual(t, wantErrs, gotErrs, + "this scenario is expected to lose an error; if it no longer does, "+ + "the upstream TSDB has changed and this expectation should be dropped") + t.Logf("separate: %v", wantErrs) + t.Logf("coalesced: %v", gotErrs) + } else { + require.Equal(t, wantErrs, gotErrs, + "per-request errors must match separate commits") + } + + if sc.dataDiffers { + require.NotEqual(t, wantHead, gotHead, + "this scenario is expected to store different data; if it no longer does, "+ + "the upstream TSDB has changed and this expectation should be dropped") + t.Logf("separate head: %v", wantHead) + t.Logf("coalesced head: %v", gotHead) + } else { + require.Equal(t, wantHead, gotHead, + "head contents must match separate commits") + } + }) + } +} + +// How far outside the out-of-order window can a sample get in, when two +// requests for the same series share one batch? +// +// The divergence window is derivable: a batch's appender keeps the head max +// time H it snapshotted before the batch, while separate commits would let the +// first request advance it to H' = max(H, t_X). A sample at t_Y is therefore +// accepted by the batch but rejected by separate commits exactly when +// +// H - W <= t_Y < t_X - W +// +// so the band of timestamps that behave differently is (t_X - H) wide: the +// amount the batch's own newest sample advances the head max. In steady state +// that is one scrape interval, not a window's worth. +// +// This test pins that bound by placing t_Y just inside and just outside the +// band and showing where the stored data starts to differ. +func TestIngesterPushCoalescing_OutOfOrderEscapeIsBoundedByInBatchHeadAdvance(t *testing.T) { + const ( + base = coalesceTestBaseMs + win = coalesceTestOOOWindowMs + + // How far beyond the pre-batch head max the batch's newest sample sits. + // A scrape interval, which is what it is in steady state. + advance = int64(15000) + ) + + for _, tc := range []struct { + name string + // Offset of t_Y relative to the band's lower edge, base-win. + offsetIntoBand int64 + inBand bool + }{ + { + name: "t_Y inside the band: the batch stores a sample separate commits reject", + offsetIntoBand: advance / 2, + inBand: true, + }, + { + name: "t_Y below the band: both reject it, only the error differs", + offsetIntoBand: -5000, + inBand: false, + }, + { + name: "t_Y above the band: both accept it", + offsetIntoBand: advance + 5000, + inBand: false, + }, + } { + t.Run(tc.name, func(t *testing.T) { + series := fmt.Sprintf("escape_%d", tc.offsetIntoBand) + tX := base + advance + tY := base - win + tc.offsetIntoBand + + specs := []writeRequestSpec{ + oneSample(series, tX, 1), + oneSample(series, tY, 2), + } + + sequential, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + for _, err := range pushSequentially(t, sequential, []writeRequestSpec{ + oneSample("coalesce_gate_seed", base, 1), + oneSample("coalesce_gate_holder", base, 1), + }) { + require.NoError(t, err) + } + wantErrs := pushSequentially(t, sequential, specs) + + coalesced, _ := startCoalesceTestIngester(t, defaultCoalesceConfig()) + gotErrs := pushAsOneBatch(t, coalesced, specs) + + wantSamples := dumpHead(t, sequential)[coalesceTestSeries(series).String()] + gotSamples := dumpHead(t, coalesced)[coalesceTestSeries(series).String()] + + t.Logf("t_X - H = %d ms, t_Y - (H - W) = %d ms", advance, tc.offsetIntoBand) + t.Logf("separate: errs=%v samples=%v", errStrings(wantErrs), wantSamples) + t.Logf("coalesced: errs=%v samples=%v", errStrings(gotErrs), gotSamples) + + if tc.inBand { + require.Len(t, wantSamples, 1, "separate commits reject the out-of-window sample") + require.Len(t, gotSamples, 2, "the batch accepts it") + return + } + require.Equal(t, wantSamples, gotSamples, + "outside the band the stored data must be identical") + }) + } +} + +// Cortex already reports a duplicate differently depending on whether it +// arrives in one request or two, with coalescing nowhere in the picture: one +// request sharing one appender hides it, two requests each committing on their +// own surface it as a 400. Coalescing widens the first case from "one request" +// to "one batch", so this is the baseline the lost duplicate error has to be +// judged against. +func TestIngesterPush_DuplicateIsAlreadySilentWithinOneRequest(t *testing.T) { + ing, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + + withinOneRequest := pushSequentially(t, ing, []writeRequestSpec{{ + series: []labels.Labels{ + coalesceTestSeries("intra_dup"), + coalesceTestSeries("intra_dup"), + }, + samples: []cortexpb.Sample{ + {TimestampMs: coalesceTestBaseMs + 1000, Value: 1}, + {TimestampMs: coalesceTestBaseMs + 1000, Value: 2}, + }, + }}) + require.NoError(t, withinOneRequest[0], + "a duplicate inside one request is silently dropped and reported as success") + require.Equal(t, + []headSample{{T: coalesceTestBaseMs + 1000, V: 1}}, + dumpHead(t, ing)[coalesceTestSeries("intra_dup").String()], + "only the first value is stored") + + other, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + acrossTwoRequests := pushSequentially(t, other, []writeRequestSpec{ + oneSample("split_dup", coalesceTestBaseMs+1000, 1), + oneSample("split_dup", coalesceTestBaseMs+1000, 2), + }) + require.NoError(t, acrossTwoRequests[0]) + require.ErrorContains(t, acrossTwoRequests[1], "duplicate sample for timestamp", + "the same duplicate split across two requests is a 400") +} + +// ---------------------------------------------------------------------------- +// Ingester.Push: per-request attribution of errors and metrics. +// ---------------------------------------------------------------------------- + +// A leader runs the append phase on behalf of every request in its batch, so +// everything that phase records has to stay attributed to the request it +// belongs to. The implementation relies on the append phase being a closure +// over each request's own locals, with the metrics emitted afterwards by the +// request's own goroutine. These tests check that it actually works out that +// way, rather than taking the closure scoping on trust. +// +// The requests below write distinct series on purpose: a request whose verdict +// depends on another request in the same batch would be testing the divergence +// covered by TestIngesterPushCoalescing_DivergenceFromSeparateCommits instead +// of testing attribution. + +// attributionSpecs builds a batch where several requests fail for different +// reasons, interleaved with requests that succeed. Every failure is decided +// against committed state, so a batch must produce the same per-request result +// as separate commits. +func attributionSpecs() []writeRequestSpec { + const ( + base = coalesceTestBaseMs + win = coalesceTestOOOWindowMs + ) + return []writeRequestSpec{ + oneSample("attr_ok_0", base+1000, 1), + // Far enough back that the out-of-order window cannot take it. + oneSample("attr_too_old", base-4*win, 2), + oneSample("attr_ok_1", base+1000, 3), + // attr_dup is seeded with value 1 before the batch, so this is a + // duplicate against committed state and is rejected at Append time. + oneSample("attr_dup", base+1000, 99), + oneSample("attr_ok_2", base+1000, 4), + oneSample("attr_too_old_2", base-5*win, 5), + oneSample("attr_ok_3", base+1000, 6), + } +} + +// seedAttributionState gives attr_dup a committed value, so that the batch's +// duplicate is decided against committed state rather than against another +// request in the same batch. +func seedAttributionState(t *testing.T, ing *Ingester) { + t.Helper() + for _, err := range pushSequentially(t, ing, []writeRequestSpec{ + oneSample("attr_dup", coalesceTestBaseMs+1000, 1), + }) { + require.NoError(t, err) + } +} + +// Each request in a batch must receive its own error, and only its own. +func TestIngesterPushCoalescing_ErrorsStayAttributedToTheirRequest(t *testing.T) { + specs := attributionSpecs() + + coalesced, _ := startCoalesceTestIngester(t, defaultCoalesceConfig()) + seedAttributionState(t, coalesced) + got := pushAsOneBatch(t, coalesced, specs) + + require.Len(t, got, 7) + + // The successful requests must not have picked up a neighbour's error. + for _, idx := range []int{0, 2, 4, 6} { + require.NoErrorf(t, got[idx], "request %d writes only a good sample", idx) + } + + // Each failing request must carry the error for its own series and reason. + require.ErrorContains(t, got[1], "too old sample") + require.ErrorContains(t, got[1], `series={__name__="attr_too_old"}`) + + require.ErrorContains(t, got[3], "duplicate sample for timestamp") + require.ErrorContains(t, got[3], `series={__name__="attr_dup"}`) + + require.ErrorContains(t, got[5], "too old sample") + require.ErrorContains(t, got[5], `series={__name__="attr_too_old_2"}`) + + // And separate commits must agree, so the attribution is not merely + // self-consistent but identical to the uncoalesced path. + sequential, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + seedAttributionState(t, sequential) + mirrorBatchHarnessPushes(t, sequential) + want := pushSequentially(t, sequential, specs) + require.Equal(t, errStrings(want), errStrings(got)) +} + +// The per-user counters the append phase feeds must total the same whether one +// leader ran seven append phases or seven goroutines ran their own. A shared +// counter in the wrong scope would show up here as double counting or as +// counts that went missing. +func TestIngesterPushCoalescing_PerUserMetricTotalsMatchSeparateCommits(t *testing.T) { + specs := attributionSpecs() + + // Metrics that must agree. Deliberately not a whole-registry comparison: + // the commit duration histogram counts commits, not requests, and the + // coalescer's own metrics exist on one side only. + names := []string{ + "cortex_ingester_ingested_samples_total", + "cortex_ingester_ingested_samples_failures_total", + "cortex_discarded_samples_total", + } + + sequential, seqReg := startCoalesceTestIngester(t, PushCoalesceConfig{}) + seedAttributionState(t, sequential) + mirrorBatchHarnessPushes(t, sequential) + for _, err := range pushSequentially(t, sequential, specs) { + _ = err // errors are asserted by the test above + } + + coalesced, coalReg := startCoalesceTestIngester(t, defaultCoalesceConfig()) + seedAttributionState(t, coalesced) + pushAsOneBatch(t, coalesced, specs) + + want := gatherValues(t, seqReg, names) + got := gatherValues(t, coalReg, names) + + require.NotEmpty(t, want, "the scenario must move the metrics it claims to compare") + require.Equal(t, want, got) + + // Spell out what the scenario is expected to have produced, so that a + // change which happens to keep the two sides equal but wrong still fails. + require.Equal(t, 7.0, want["cortex_ingester_ingested_samples_total{}"], + "4 good samples in the batch, plus attr_dup and the two harness pushes") + require.Equal(t, 3.0, want["cortex_ingester_ingested_samples_failures_total{}"]) + require.Equal(t, 2.0, want[`cortex_discarded_samples_total{reason="sample-too-old",user="`+coalesceTestUser+`"}`]) + require.Equal(t, 1.0, want[`cortex_discarded_samples_total{reason="new-value-for-timestamp",user="`+coalesceTestUser+`"}`]) +} + +// gatherValues flattens the named metric families into label-keyed values. +func gatherValues(t *testing.T, reg *prometheus.Registry, names []string) map[string]float64 { + t.Helper() + + wanted := make(map[string]struct{}, len(names)) + for _, n := range names { + wanted[n] = struct{}{} + } + + families, err := reg.Gather() + require.NoError(t, err) + + out := map[string]float64{} + for _, f := range families { + if _, ok := wanted[f.GetName()]; !ok { + continue + } + for _, m := range f.GetMetric() { + key := f.GetName() + "{" + for i, l := range m.GetLabel() { + if i > 0 { + key += "," + } + key += l.GetName() + `="` + l.GetValue() + `"` + } + key += "}" + out[key] = metricValue(m) + } + } + return out +} + +func metricValue(m *dto.Metric) float64 { + switch { + case m.GetCounter() != nil: + return m.GetCounter().GetValue() + case m.GetGauge() != nil: + return m.GetGauge().GetValue() + default: + return 0 + } +} + +// ---------------------------------------------------------------------------- +// Ingester.Push: native histograms, exemplars and concurrency. +// ---------------------------------------------------------------------------- + +// Push appends native histograms and exemplars as well as float samples, and a +// batch leader runs all of it on behalf of other requests. The scenarios above +// carry only float samples, so these two branches are covered here. + +// Native histograms appended by a leader on another request's behalf must end +// up where separate commits would have put them. +func TestIngesterPushCoalescing_NativeHistogramsMatchSeparateCommits(t *testing.T) { + const base = coalesceTestBaseMs + + specs := []writeRequestSpec{ + histogramRequest("hist_a", base+1000, 1), + histogramRequest("hist_b", base+1000, 2), + // Same series as the first request, later timestamp: in order, so both + // paths must store both. + histogramRequest("hist_a", base+2000, 3), + // A float sample in the same batch, to exercise both branches together. + oneSample("hist_float", base+1000, 7), + } + + sequential, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + mirrorBatchHarnessPushes(t, sequential) + wantErrs := pushSequentially(t, sequential, specs) + + coalesced, _ := startCoalesceTestIngester(t, defaultCoalesceConfig()) + gotErrs := pushAsOneBatch(t, coalesced, specs) + + require.Equal(t, errStrings(wantErrs), errStrings(gotErrs)) + require.Equal(t, dumpHead(t, sequential), dumpHead(t, coalesced)) + + // The scenario has to have actually stored histograms, or it proves nothing. + stored := dumpHead(t, coalesced) + require.Len(t, stored[coalesceTestSeries("hist_a").String()], 2) + require.Len(t, stored[coalesceTestSeries("hist_b").String()], 1) +} + +// Exemplars appended by a leader on another request's behalf must end up where +// separate commits would have put them. +func TestIngesterPushCoalescing_ExemplarsMatchSeparateCommits(t *testing.T) { + const base = coalesceTestBaseMs + + specs := []writeRequestSpec{ + exemplarRequest("ex_a", base+1000, 1, "trace-a"), + exemplarRequest("ex_b", base+1000, 2, "trace-b"), + exemplarRequest("ex_a", base+2000, 3, "trace-c"), + } + + sequential, _ := startCoalesceTestIngester(t, PushCoalesceConfig{}) + mirrorBatchHarnessPushes(t, sequential) + wantErrs := pushSequentially(t, sequential, specs) + + coalesced, _ := startCoalesceTestIngester(t, defaultCoalesceConfig()) + gotErrs := pushAsOneBatch(t, coalesced, specs) + + require.Equal(t, errStrings(wantErrs), errStrings(gotErrs)) + + want := dumpExemplars(t, sequential, "ex_a", "ex_b") + got := dumpExemplars(t, coalesced, "ex_a", "ex_b") + require.Equal(t, want, got) + require.Len(t, got, 3, "the scenario must have stored exemplars") +} + +// Concurrent pushes for the same tenant must not lose, duplicate or corrupt +// samples. Run with -race. +func TestIngesterPushCoalescing_ConcurrentPushesAreConsistent(t *testing.T) { + ing, _ := startCoalesceTestIngester(t, defaultCoalesceConfig()) + ctx := user.InjectOrgID(context.Background(), coalesceTestUser) + + const ( + goroutines = 32 + samplesPerSerie = 20 + ) + + var wg sync.WaitGroup + for g := range goroutines { + wg.Go(func() { + name := fmt.Sprintf("concurrent_%d", g) + for n := range samplesPerSerie { + // Ascending timestamps per series, so nothing is out of order + // and every sample must be stored. + tMs := coalesceTestBaseMs + int64(n)*1000 + _, err := ing.Push(ctx, oneSample(name, tMs, float64(n)).build()) + require.NoError(t, err) + } + }) + } + wg.Wait() + + head := dumpHead(t, ing) + for g := range goroutines { + key := coalesceTestSeries(fmt.Sprintf("concurrent_%d", g)).String() + got := head[key] + require.Lenf(t, got, samplesPerSerie, "series %s", key) + for n, s := range got { + require.Equal(t, coalesceTestBaseMs+int64(n)*1000, s.T) + require.Equal(t, float64(n), s.V) + } + } +} diff --git a/pkg/ingester/user_metrics_metadata_test.go b/pkg/ingester/user_metrics_metadata_test.go index 0946e4869c7..37ae0c204c8 100644 --- a/pkg/ingester/user_metrics_metadata_test.go +++ b/pkg/ingester/user_metrics_metadata_test.go @@ -34,6 +34,7 @@ func Test_UserMetricsMetadata(t *testing.T) { &inflightPushRequests, &maxInflightQueryRequests, false, + false, false) limits := validation.Limits{} diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index 776aa5ee08d..9d5384cfcae 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -5478,6 +5478,35 @@ "x-cli-flag": "ingester.metadata-retain-period", "x-format": "duration" }, + "push_coalesce": { + "properties": { + "enabled": { + "default": false, + "description": "Experimental: merge the TSDB commits of concurrent write requests for the same tenant into a single commit. A request only ever waits for a commit that was already in flight, so this adds no latency of its own.", + "type": "boolean", + "x-cli-flag": "ingester.push-coalesce.enabled" + }, + "max_batch_bytes": { + "default": 0, + "description": "Maximum encoded size, in bytes, of the write requests a single coalesced commit may carry. 0 to disable the limit.", + "type": "number", + "x-cli-flag": "ingester.push-coalesce.max-batch-bytes" + }, + "max_batch_requests": { + "default": 32, + "description": "Maximum number of write requests a single coalesced commit may carry.", + "type": "number", + "x-cli-flag": "ingester.push-coalesce.max-batch-requests" + }, + "max_batch_samples": { + "default": 2000, + "description": "Maximum number of samples a single coalesced commit may carry. 0 to disable the limit.", + "type": "number", + "x-cli-flag": "ingester.push-coalesce.max-batch-samples" + } + }, + "type": "object" + }, "query_protection": { "properties": { "eviction": {