Fail ingestion when a flush is going on

CHANGELOG.md

@@ -8,6 +8,7 @@
   - `cortex_ingester_tsdb_head_truncations_total`
   - `cortex_ingester_tsdb_head_gc_duration_seconds`
 * [ENHANCEMENT] Added `cortex_alertmanager_config_hash` metric to expose hash of Alertmanager Config loaded per user. #3388
+* [BUGFIX] Blocks storage: Fix the race between ingestion and `/flush` call resulting in overlapping blocks. #3422
 
 ## 1.5.0 in progress
 

pkg/ingester/ingester_v2.go

@@ -54,6 +54,10 @@ type userTSDB struct {
 	seriesInMetric *metricCounter
 	limiter        *Limiter
 
+	forcedCompactionInProgress    bool
+	forcedCompactionInProgressMtx sync.RWMutex
+	pushesInFlight                sync.WaitGroup
+
 	// Used to detect idle TSDBs.
 	lastUpdate *atomic.Int64
 
@@ -93,6 +97,16 @@ func (u *userTSDB) Compact() error {
 
 // compactHead compacts the Head block at specified block durations avoiding a single huge block.
 func (u *userTSDB) compactHead(blockDuration int64) error {
+	u.forcedCompactionInProgressMtx.Lock()
+	u.forcedCompactionInProgress = true
+	u.forcedCompactionInProgressMtx.Unlock()
+	defer func() {
+		u.forcedCompactionInProgress = false
+	}()
+	// Ingestion of samples in parallel with forced compaction can lead to overlapping blocks.
+	// So we wait for existing in-flight requests to finish. Future push requests would fail until compaction is over.
+	u.pushesInFlight.Wait()
+
 	h := u.Head()
 
 	minTime, maxTime := h.MinTime(), h.MaxTime()
@@ -501,6 +515,11 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 	}
 	i.userStatesMtx.RUnlock()
 
+	if err := db.acquireAppendLock(); err != nil {
+		return &client.WriteResponse{}, httpgrpc.Errorf(http.StatusServiceUnavailable, wrapWithUser(err, userID).Error())
+	}
+	defer db.releaseAppendLock()
+
 	// Given metadata is a best-effort approach, and we don't halt on errors
 	// process it before samples. Otherwise, we risk returning an error before ingestion.
 	i.pushMetadata(ctx, userID, req.GetMetadata())
@@ -651,6 +670,22 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 	return &client.WriteResponse{}, nil
 }
 
+func (u *userTSDB) acquireAppendLock() error {
+	u.forcedCompactionInProgressMtx.RLock()
+	defer u.forcedCompactionInProgressMtx.RUnlock()
+
+	if u.forcedCompactionInProgress {
+		return errors.New("forced compaction in progress")
+	}
+
+	u.pushesInFlight.Add(1)
+	return nil
+}
+
+func (u *userTSDB) releaseAppendLock() {
+	u.pushesInFlight.Done()
+}
+
 func (i *Ingester) v2Query(ctx context.Context, req *client.QueryRequest) (*client.QueryResponse, error) {
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {

pkg/ingester/ingester_v2_test.go

@@ -2422,3 +2422,39 @@ func TestIngesterNotDeleteUnshippedBlocks(t *testing.T) {
 		require.NotEqual(t, b.Meta().ULID, newBlocks2[1].Meta().ULID)
 	}
 }
+
+func TestIngesterPushErrorDuringForcedCompaction(t *testing.T) {
+	i, cleanup, err := newIngesterMockWithTSDBStorage(defaultIngesterTestConfig(), nil)
+	require.NoError(t, err)
+	t.Cleanup(cleanup)
+
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	t.Cleanup(func() {
+		_ = services.StopAndAwaitTerminated(context.Background(), i)
+	})
+
+	// Wait until it's ACTIVE
+	test.Poll(t, 10*time.Millisecond, ring.ACTIVE, func() interface{} {
+		return i.lifecycler.GetState()
+	})
+
+	// Push a sample, it should succeed.
+	pushSingleSample(t, i)
+
+	// We mock a flushing by setting the boolean.
+	db := i.getTSDB(userID)
+	require.NotNil(t, db)
+	require.False(t, db.forcedCompactionInProgress)
+	db.forcedCompactionInProgress = true
+
+	// Ingestion should fail with a 503.
+	req, _, _ := mockWriteRequest(labels.Labels{{Name: labels.MetricName, Value: "test"}}, 0, util.TimeToMillis(time.Now()))
+	ctx := user.InjectOrgID(context.Background(), userID)
+	_, err = i.v2Push(ctx, req)
+	require.Equal(t, httpgrpc.Errorf(http.StatusServiceUnavailable, wrapWithUser(errors.New("forced compaction in progress"), userID).Error()), err)
+
+	// Ingestion is successful after a flush.
+	require.True(t, db.forcedCompactionInProgress)
+	db.forcedCompactionInProgress = false
+	pushSingleSample(t, i)
+}