Revert "fix(ingester): Keep read only when any number of compactions in progress" (grafana#12056)

dmwilson-grafana · web-flow · commit b3acccaec36f · 2025-07-11T14:45:45.000-04:00
Reverts grafana#11890, `cortex_ingester_tsdb_forced_compactions_in_progress` is included in mixin alerts. Will merge an updated PR with this metric re-added.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -171,7 +171,6 @@
 * [BUGFIX] Ruler: Fix rare panic when the ruler is shutting down. #11781
 * [BUGFIX] Block-builder-scheduler: Fix data loss bug in job assignment. #11785
 * [BUGFIX] Compactor: start tracking `-compactor.max-compaction-time` after the initial compaction planning phase, to avoid rare cases where planning takes longer than `-compactor.max-compaction-time` and so actual compaction never runs for a tenant. #11834
-* [BUGFIX] Ingester: Fix issue where ingesters can exit read-only mode during idle compactions, resulting in write errors. #11890
 
 ### Mixin
 
diff --git a/pkg/ingester/downscale.go b/pkg/ingester/downscale.go
@@ -52,9 +52,9 @@ func (i *Ingester) PrepareInstanceRingDownscaleHandler(w http.ResponseWriter, r
 		i.circuitBreaker.push.deactivate()
 
 	case http.MethodDelete:
-		// Don't leave read-only mode if there is any compaction pending or in progress.
-		if i.numCompactionsInProgress.Load() != 0 {
-			msg := "cannot clear read-only mode while compaction is in progress"
+		// Don't leave read-only mode if there's a forced compaction pending or in progress.
+		if len(i.forceCompactTrigger) > 0 || i.forcedCompactionInProgress.Load() {
+			msg := "cannot clear read-only mode while forced compaction is in progress"
 			level.Warn(i.logger).Log("msg", msg)
 			http.Error(w, msg, http.StatusConflict)
 			return
diff --git a/pkg/ingester/downscale_test.go b/pkg/ingester/downscale_test.go
@@ -155,7 +155,7 @@ func TestIngester_PrepareInstanceRingDownscaleHandler(t *testing.T) {
 		require.Equal(t, http.StatusMethodNotAllowed, res.Code)
 	})
 
-	t.Run("should return Conflict when any compaction is in progress", func(t *testing.T) {
+	t.Run("should return Conflict when forced compaction is in progress", func(t *testing.T) {
 		t.Parallel()
 
 		ingester, r := setup(true)
@@ -170,9 +170,7 @@ func TestIngester_PrepareInstanceRingDownscaleHandler(t *testing.T) {
 			return inst.ReadOnly
 		})
 
-		// Simulate a compation in progress.
-		cleanup := ingester.prepareCompaction()
-		defer cleanup()
+		ingester.forcedCompactionInProgress.Store(true)
 
 		// Try to switch back to read-write while compaction is in progress
 		res = httptest.NewRecorder()
diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go
@@ -340,7 +340,7 @@ type Ingester struct {
 	seriesCount atomic.Int64
 
 	// Tracks if a forced compaction is in progress
-	numCompactionsInProgress atomic.Uint32
+	forcedCompactionInProgress atomic.Bool
 
 	// For storing metadata ingested.
 	usersMetadataMtx sync.RWMutex
@@ -3252,10 +3252,6 @@ func (i *Ingester) compactionServiceRunning(ctx context.Context) error {
 	for ctx.Err() == nil {
 		select {
 		case <-tickerChan:
-			// Prepare compaction before the periodic head compaction is triggered.
-			cleanup := i.prepareCompaction()
-			defer cleanup()
-
 			// The forcedCompactionMaxTime has no meaning because force=false.
 			i.compactBlocks(ctx, false, 0, nil)
 
@@ -3273,12 +3269,6 @@ func (i *Ingester) compactionServiceRunning(ctx context.Context) error {
 			}
 
 		case req := <-i.forceCompactTrigger:
-			// Note:
-			// Prepare compaction is not done here but before the force compaction is triggered.
-			// This is because we want to track the number of compactions accurately before the
-			// downscale handler is called. This ensures that the ingester will never leave the
-			// read-only state. (See [Ingester.FlushHandler])
-
 			// Always pass math.MaxInt64 as forcedCompactionMaxTime because we want to compact the whole TSDB head.
 			i.compactBlocks(ctx, true, math.MaxInt64, req.users)
 			close(req.callback) // Notify back.
@@ -3311,28 +3301,19 @@ func (i *Ingester) compactionServiceInterval() (firstInterval, standardInterval
 	return
 }
 
-// prepareCompaction is incrementing the atomic counter of the number of compactions in progress.
-// It also exposes a metric tracking the number of compactions in progress.
-// It returns a callback that should be called when the compaction is finished, e.g. in defer statement.
-// This callback is decrementing the atomic counter of the number of compactions in progress.
-func (i *Ingester) prepareCompaction() func() {
-	// Increment the number of compactions in progress.
-	// This is used to ensure that the ingester will never leave the read-only state.
-	// (See [Ingester.PrepareInstanceRingDownscaleHandler])
-	i.numCompactionsInProgress.Inc()
-
-	// Expose a metric tracking whether there's a TSDB head compaction in progress.
+// Compacts all compactable blocks. Force flag will force compaction even if head is not compactable yet.
+func (i *Ingester) compactBlocks(ctx context.Context, force bool, forcedCompactionMaxTime int64, allowed *util.AllowList) {
+	// Expose a metric tracking whether there's a forced head compaction in progress.
 	// This metric can be used in alerts and when troubleshooting.
-	i.metrics.numCompactionsInProgress.Inc()
-
-	return func() {
-		i.numCompactionsInProgress.Dec()
-		i.metrics.numCompactionsInProgress.Dec()
+	if force {
+		i.metrics.forcedCompactionInProgress.Set(1)
+		i.forcedCompactionInProgress.Store(true)
+		defer func() {
+			i.metrics.forcedCompactionInProgress.Set(0)
+			i.forcedCompactionInProgress.Store(false)
+		}()
 	}
-}
 
-// Compacts all compactable blocks. Force flag will force compaction even if head is not compactable yet.
-func (i *Ingester) compactBlocks(ctx context.Context, force bool, forcedCompactionMaxTime int64, allowed *util.AllowList) {
 	_ = concurrency.ForEachUser(ctx, i.getTSDBUsers(), i.cfg.BlocksStorageConfig.TSDB.HeadCompactionConcurrency, func(_ context.Context, userID string) error {
 		if !allowed.IsAllowed(userID) {
 			return nil
@@ -3660,10 +3641,6 @@ func (i *Ingester) FlushHandler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		// Prepare compaction before the force compaction is triggered.
-		cleanup := i.prepareCompaction()
-		defer cleanup()
-
 		compactionCallbackCh := make(chan struct{})
 
 		level.Info(i.logger).Log("msg", "flushing TSDB blocks: triggering compaction")
diff --git a/pkg/ingester/metrics.go b/pkg/ingester/metrics.go
@@ -65,12 +65,12 @@ type ingesterMetrics struct {
 	maxLocalSeriesPerUser *prometheus.GaugeVec
 
 	// Head compactions metrics.
-	compactionsTriggered     prometheus.Counter
-	compactionsFailed        prometheus.Counter
-	numCompactionsInProgress prometheus.Gauge
-	appenderAddDuration      prometheus.Histogram
-	appenderCommitDuration   prometheus.Histogram
-	idleTsdbChecks           *prometheus.CounterVec
+	compactionsTriggered       prometheus.Counter
+	compactionsFailed          prometheus.Counter
+	forcedCompactionInProgress prometheus.Gauge
+	appenderAddDuration        prometheus.Histogram
+	appenderCommitDuration     prometheus.Histogram
+	idleTsdbChecks             *prometheus.CounterVec
 
 	// Open all existing TSDBs metrics
 	openExistingTSDB prometheus.Counter
@@ -350,9 +350,9 @@ func newIngesterMetrics(
 			Name: "cortex_ingester_tsdb_compactions_failed_total",
 			Help: "Total number of compactions that failed.",
 		}),
-		numCompactionsInProgress: promauto.With(r).NewGauge(prometheus.GaugeOpts{
-			Name: "cortex_ingester_tsdb_num_compactions_in_progress",
-			Help: "Reports the number of TSDB head compactions in progress.",
+		forcedCompactionInProgress: promauto.With(r).NewGauge(prometheus.GaugeOpts{
+			Name: "cortex_ingester_tsdb_forced_compactions_in_progress",
+			Help: "Reports 1 if there's a forced TSDB head compaction in progress, 0 otherwise.",
 		}),
 
 		appenderAddDuration: promauto.With(r).NewHistogram(prometheus.HistogramOpts{
