From ab7fcd18b821499d4627423007fd5e9b2c17d37f Mon Sep 17 00:00:00 2001 From: Rohithmatham12 Date: Thu, 6 Aug 2026 20:54:24 -0700 Subject: [PATCH] Fix blocksToDelete shipper initialization race Signed-off-by: Rohithmatham12 --- pkg/ingester/ingester.go | 36 ++++++++++++++++++++---------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 3a6c03449c..6813cea9e7 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -3113,26 +3113,14 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) { return nil, errors.Wrapf(err, "failed to compact TSDB: %s", udir) } - userDB.db = db - // We set the limiter here because we don't want to limit - // series during WAL replay. - userDB.limiter = i.limiter - - if db.Head().NumSeries() > 0 { - // If there are series in the head, use max time from head. If this time is too old, - // TSDB will be eligible for flushing and closing sooner, unless more data is pushed to it quickly. - userDB.setLastUpdate(util.TimeFromMillis(db.Head().MaxTime())) - } else { - // If head is empty (eg. new TSDB), don't close it right after. - userDB.setLastUpdate(time.Now()) - } - // Thanos shipper requires at least 1 external label to be set. For this reason, // we set the tenant ID as external label and we'll filter it out when reading // the series from the storage. l := labels.FromStrings(cortex_tsdb.TenantIDExternalLabel, userID, cortex_tsdb.IngesterIDExternalLabel, i.TSDBState.shipperIngesterID) - // Create a new shipper for this database + // Create a new shipper for this database before making the TSDB visible to + // callbacks. TSDB's reload goroutine can call blocksToDelete concurrently, + // and blocksToDelete reads userDB.shipper once userDB.db is set. if i.cfg.BlocksStorageConfig.TSDB.IsBlocksShippingEnabled() { udirRoot, err := os.OpenRoot(udir) if err != nil { @@ -3151,8 +3139,24 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) { shipper.WithAllowOutOfOrderUploads(true), // Allow out of order uploads. It's fine in Cortex's context. shipper.WithSkipCorruptedBlocks(true), // We allow out of order uploads. This is the same behavior. We should track error with metrics ) - userDB.shipperMetadataFilePath = filepath.Join(userDB.db.Dir(), filepath.Clean(shipper.DefaultMetaFilename)) + userDB.shipperMetadataFilePath = filepath.Join(db.Dir(), filepath.Clean(shipper.DefaultMetaFilename)) + } + + userDB.db = db + // We set the limiter here because we don't want to limit + // series during WAL replay. + userDB.limiter = i.limiter + + if db.Head().NumSeries() > 0 { + // If there are series in the head, use max time from head. If this time is too old, + // TSDB will be eligible for flushing and closing sooner, unless more data is pushed to it quickly. + userDB.setLastUpdate(util.TimeFromMillis(db.Head().MaxTime())) + } else { + // If head is empty (eg. new TSDB), don't close it right after. + userDB.setLastUpdate(time.Now()) + } + if userDB.shipper != nil { // Initialise the shipper blocks cache. if err := userDB.updateCachedShippedBlocks(); err != nil { level.Error(userLogger).Log("msg", "failed to update cached shipped blocks after shipper initialisation", "err", err)