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)