From a866acc07b530e405df5184ab68b1b987e32cc76 Mon Sep 17 00:00:00 2001 From: Kyle Stang Date: Wed, 16 Sep 2026 22:09:03 +0000 Subject: [PATCH] Gracefully skip compaction during shipping This commit makes ingesters prevented from performing head compaction due to block shipping skip compaction rather than treating it as a failure. The shipping/compaction race is intentional functionality and not an error. With the old functionality, we would get alarms when these two events overlapped, even though there were no issues and compaction succeeded on the next attempt. Signed-off-by: Kyle Stang --- CHANGELOG.md | 1 + pkg/ingester/ingester.go | 27 +++++++---- pkg/ingester/ingester_test.go | 88 +++++++++++++++++++++++++++++++++-- 3 files changed, 103 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b6bcaa78a8..7dbb0603951 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -120,6 +120,7 @@ * [BUGFIX] Config: Fix CSV-list flags/YAML fields (e.g. `-compactor.enabled-tenants`) treating an explicitly empty string as a one-element list containing an empty tenant name instead of an empty list. #7714 * [BUGFIX] Tenant Federation: Fix regex tenant federation dropping tenants when `-blocks-storage.users-scanner.cache-ttl` is set. The regex resolver sorted the user list returned by the users scanner in place, corrupting the scanner cache and progressively losing tenants on every sync until the cache expired. #7812 * [BUGFIX] Tenant Federation: Fix regex tenant federation resolving to an empty user list right after startup. #7811 +* [BUGFIX] Ingester: Don't count a forced head compaction skipped because blocks shipping is in progress as a failure. Previously such skips incremented `cortex_ingester_tsdb_compactions_failed_total`, producing spurious alerts. #7842 ## 1.21.1 2026-06-04 diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 933a2e31835..033822045bf 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -111,6 +111,7 @@ var ( errIngesterStopping = errors.New("ingester stopping") errNoUserDb = errors.New("no user db") errLabelsOutOfOrder = errors.New("labels out of order") + errTsdbShipping = errors.New("tsdb is in state activeShipping") tsChunksPool zeropool.Pool[[]client.TimeSeriesChunk] @@ -471,21 +472,24 @@ func (u *userTSDB) StartTime() (int64, error) { return u.db.StartTime() } -func (u *userTSDB) casState(from, to tsdbState) bool { +func (u *userTSDB) casState(from, to tsdbState) (bool, tsdbState) { u.stateMtx.Lock() defer u.stateMtx.Unlock() if u.state != from { - return false + return false, u.state } u.state = to - return true + return true, u.state } // compactHead compacts the Head block at specified block durations avoiding a single huge block. func (u *userTSDB) compactHead(ctx context.Context, blockDuration int64) error { - if !u.casState(active, forceCompacting) { - return errors.New("TSDB head cannot be compacted because it is not in active state (possibly being closed or blocks shipping in progress)") + if success, state := u.casState(active, forceCompacting); !success { + if state == activeShipping { + return errTsdbShipping + } + return fmt.Errorf("TSDB head cannot be compacted because it is not in active state (state: %d)", state) } defer u.casState(forceCompacting, active) @@ -3409,7 +3413,7 @@ func (i *Ingester) shipBlocks(ctx context.Context, allowed *users.AllowedTenants // Run the shipper's Sync() to upload unshipped blocks. Make sure the TSDB state is active, in order to // avoid any race condition with closing idle TSDBs. - if !userDB.casState(active, activeShipping) { + if success, _ := userDB.casState(active, activeShipping); !success { level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "shipper skipped because the TSDB is not active", "user", userID) return nil } @@ -3523,8 +3527,13 @@ func (i *Ingester) compactBlocks(ctx context.Context, force bool, allowed *users } if err != nil { - i.TSDBState.compactionsFailed.Inc() - level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction for user has failed", "user", userID, "err", err, "compactReason", reason) + // Don't treat blocks shipping as a failure + if errors.Is(err, errTsdbShipping) { + level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction for user was skipped", "user", userID, "err", err, "compactReason", reason) + } else { + i.TSDBState.compactionsFailed.Inc() + level.Warn(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction for user has failed", "user", userID, "err", err, "compactReason", reason) + } } else { level.Debug(logutil.WithContext(ctx, i.logger)).Log("msg", "TSDB blocks compaction completed successfully", "user", userID, "compactReason", reason) } @@ -3574,7 +3583,7 @@ func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string) tsdbCloseCheckRes } // This disables pushes and force-compactions. Not allowed to close while shipping is in progress. - if !userDB.casState(active, closing) { + if success, _ := userDB.casState(active, closing); !success { return tsdbNotActive } diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index d91088e05b0..68ca3d23577 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -5388,8 +5388,8 @@ func TestIngester_ReadNotFailWhenTSDBIsBeingDeleted(t *testing.T) { err = db.Close() require.NoError(t, err) - b := db.casState(active, c.state) - require.True(t, b) + casSuccess, _ := db.casState(active, c.state) + require.True(t, casSuccess) // Mock request ctx = user.InjectOrgID(context.Background(), userID) @@ -6395,7 +6395,8 @@ func TestIngesterPushErrorDuringForcedCompaction(t *testing.T) { db, err := i.getTSDB(userID) require.NoError(t, err) require.NotNil(t, db) - require.True(t, db.casState(active, forceCompacting)) + casSuccess, _ := db.casState(active, forceCompacting) + require.True(t, casSuccess) // Ingestion should fail with a 503. req, _ := mockWriteRequest(t, labels.FromStrings(labels.MetricName, "test"), 0, util.TimeToMillis(time.Now())) @@ -6404,10 +6405,89 @@ func TestIngesterPushErrorDuringForcedCompaction(t *testing.T) { require.Equal(t, httpgrpc.Errorf(http.StatusServiceUnavailable, "%s", wrapWithUser(errors.New("forced compaction in progress"), userID).Error()), err) // Ingestion is successful after a flush. - require.True(t, db.casState(forceCompacting, active)) + casSuccess, _ = db.casState(forceCompacting, active) + require.True(t, casSuccess) pushSingleSampleWithMetadata(t, i) } +func TestUserTSDB_compactHead_returnsShippingError(t *testing.T) { + i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), prometheus.NewRegistry()) + require.NoError(t, err) + + require.NoError(t, services.StartAndAwaitRunning(context.Background(), i)) + t.Cleanup(func() { + _ = services.StopAndAwaitTerminated(context.Background(), i) + }) + + // Wait until it's ACTIVE + test.Poll(t, 1*time.Second, ring.ACTIVE, func() any { + return i.lifecycler.GetState() + }) + + pushSingleSampleWithMetadata(t, i) + + db, err := i.getTSDB(userID) + require.NoError(t, err) + require.NotNil(t, db) + + blockDuration := i.cfg.BlocksStorageConfig.TSDB.BlockRanges[0].Milliseconds() + + // When the TSDB is shipping blocks, compactHead should return the sentinel errTsdbShipping + // so callers can distinguish it from a genuine compaction failure. + casSuccess, _ := db.casState(active, activeShipping) + require.True(t, casSuccess) + err = db.compactHead(context.Background(), blockDuration) + require.ErrorIs(t, err, errTsdbShipping) + casSuccess, _ = db.casState(activeShipping, active) + require.True(t, casSuccess) + + // For any other non-active state, compactHead returns a generic error (not errTsdbShipping) + // reporting the current state. + casSuccess, _ = db.casState(active, closing) + require.True(t, casSuccess) + err = db.compactHead(context.Background(), blockDuration) + require.Error(t, err) + require.NotErrorIs(t, err, errTsdbShipping) + require.Contains(t, err.Error(), "not in active state") + casSuccess, _ = db.casState(closing, active) + require.True(t, casSuccess) +} + +func TestIngester_compactBlocks_shippingIsNotCountedAsFailure(t *testing.T) { + i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), prometheus.NewRegistry()) + require.NoError(t, err) + + require.NoError(t, services.StartAndAwaitRunning(context.Background(), i)) + t.Cleanup(func() { + _ = services.StopAndAwaitTerminated(context.Background(), i) + }) + + // Wait until it's ACTIVE + test.Poll(t, 1*time.Second, ring.ACTIVE, func() any { + return i.lifecycler.GetState() + }) + + pushSingleSampleWithMetadata(t, i) + + db, err := i.getTSDB(userID) + require.NoError(t, err) + require.NotNil(t, db) + + // Simulate blocks shipping being in progress. Forced compaction will fail the CAS to + // forceCompacting, but this must not be counted as a compaction failure. + casSuccess, _ := db.casState(active, activeShipping) + require.True(t, casSuccess) + t.Cleanup(func() { db.casState(activeShipping, active) }) + + require.Equal(t, float64(0), testutil.ToFloat64(i.TSDBState.compactionsFailed)) + + i.compactBlocks(context.Background(), true, nil) + + // Compaction was triggered but the shipping skip must not increment the failure counter. + require.Equal(t, float64(1), testutil.ToFloat64(i.TSDBState.compactionsTriggered)) + require.Equal(t, float64(0), testutil.ToFloat64(i.TSDBState.compactionsFailed)) +} + func TestIngesterNoFlushWithInFlightRequest(t *testing.T) { registry := prometheus.NewRegistry() i, err := prepareIngesterWithBlocksStorage(t, defaultIngesterTestConfig(t), registry)