Ingester: Coalesce concurrent pushes for the same tenant into one TSDB commit

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 `-<prefix>.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

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: <int> | 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: <boolean> | default = false]
+
+  # Maximum number of write requests a single coalesced commit may carry.
+  # CLI flag: -ingester.push-coalesce.max-batch-requests
+  [max_batch_requests: <int> | 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: <int> | 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: <int> | default = 0]
 ```
 
 ### `ingester_client_config`

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

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)
+}