Skip corrupt blocks on shipper (#6786)

* Skip corrupt blocks on shipper

Signed-off-by: Daniel Deluiggi <[email protected]>

* changelog

Signed-off-by: Daniel Deluiggi <[email protected]>

---------

Signed-off-by: Daniel Deluiggi <[email protected]>

Author: danielblando

CHANGELOG.md

@@ -50,6 +50,7 @@
 * [BUGFIX] Add `__markers__` tenant ID validation. #6761
 * [BUGFIX] Ring: Fix nil pointer exception when token is shared. #6768
 * [BUGFIX] Fix race condition in active user. #6773
+* [BUGFIX] Ingester: Allow shipper to skip corrupted blocks. #6786
 
 ## 1.19.0 2025-02-27
 

go.mod

@@ -52,7 +52,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	github.com/thanos-io/objstore v0.0.0-20250317105316-a0136a6f898d
 	github.com/thanos-io/promql-engine v0.0.0-20250522103302-dd83bd8fdb50
-	github.com/thanos-io/thanos v0.37.3-0.20250529092349-12649d8be797
+	github.com/thanos-io/thanos v0.37.3-0.20250603135757-4ad45948cd10
 	github.com/uber/jaeger-client-go v2.30.0+incompatible
 	github.com/weaveworks/common v0.0.0-20230728070032-dd9e68f319d5
 	go.etcd.io/etcd/api/v3 v3.5.17

go.sum

@@ -1697,8 +1697,8 @@ github.com/thanos-io/objstore v0.0.0-20241111205755-d1dd89d41f97 h1:VjG0mwhN1Dkn
 github.com/thanos-io/objstore v0.0.0-20241111205755-d1dd89d41f97/go.mod h1:vyzFrBXgP+fGNG2FopEGWOO/zrIuoy7zt3LpLeezRsw=
 github.com/thanos-io/promql-engine v0.0.0-20250522103302-dd83bd8fdb50 h1:RGdaDAyFOjrFJSjaPT2z8robLvQ3KxNiNEN3DojpLOs=
 github.com/thanos-io/promql-engine v0.0.0-20250522103302-dd83bd8fdb50/go.mod h1:agUazAk1yHLYSL87MdEcRbjN12DJ9OZfSUcfFLqy+F8=
-github.com/thanos-io/thanos v0.37.3-0.20250529092349-12649d8be797 h1:Co+TgEgln2gBoQJ7cjzD9f8Uj3+8MmqPnp33FvNOtYw=
-github.com/thanos-io/thanos v0.37.3-0.20250529092349-12649d8be797/go.mod h1:hqWLLR6Yd7remOR33hRgVkg28Gx40Nh3mhWryB8RVJs=
+github.com/thanos-io/thanos v0.37.3-0.20250603135757-4ad45948cd10 h1:mtmcivEm0EoXeHTJAgjXTthyQSTLNFWrPTzpiovau3Y=
+github.com/thanos-io/thanos v0.37.3-0.20250603135757-4ad45948cd10/go.mod h1:2NvA8ZJtoGcOTriumDnJQzDmbxJz1ISGPovVAGGYDbg=
 github.com/tjhop/slog-gokit v0.1.4 h1:uj/vbDt3HaF0Py8bHPV4ti/s0utnO0miRbO277FLBKM=
 github.com/tjhop/slog-gokit v0.1.4/go.mod h1:Bbu5v2748qpAWH7k6gse/kw3076IJf6owJmh7yArmJs=
 github.com/trivago/tgo v1.0.7 h1:uaWH/XIy9aWYWpjm2CU3RpcqZXmX2ysQ9/Go+d9gyrM=

pkg/ingester/ingester.go

@@ -2566,18 +2566,16 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	// Create a new shipper for this database
 	if i.cfg.BlocksStorageConfig.TSDB.IsBlocksShippingEnabled() {
 		userDB.shipper = shipper.New(
-			userLogger,
-			tsdbPromReg,
-			udir,
 			bucket.NewUserBucketClient(userID, i.TSDBState.bucket, i.limits),
-			func() labels.Labels { return l },
-			metadata.ReceiveSource,
-			func() bool {
-				return i.cfg.UploadCompactedBlocksEnabled
-			},
-			true, // Allow out of order uploads. It's fine in Cortex's context.
-			metadata.NoneFunc,
-			"",
+			udir,
+			shipper.WithLogger(userLogger),
+			shipper.WithRegisterer(tsdbPromReg),
+			shipper.WithLabels(func() labels.Labels { return l }),
+			shipper.WithSource(metadata.ReceiveSource),
+			shipper.WithHashFunc(metadata.NoneFunc),
+			shipper.WithUploadCompacted(i.cfg.UploadCompactedBlocksEnabled),
+			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))
 

pkg/ingester/metrics.go

@@ -312,6 +312,7 @@ type tsdbMetrics struct {
 	dirSyncFailures *prometheus.Desc // sum(thanos_shipper_dir_sync_failures_total)
 	uploads         *prometheus.Desc // sum(thanos_shipper_uploads_total)
 	uploadFailures  *prometheus.Desc // sum(thanos_shipper_upload_failures_total)
+	corruptedBlocks *prometheus.Desc // sum(thanos_shipper_corrupted_blocks_total)
 
 	// Metrics aggregated from TSDB.
 	tsdbCompactionsTotal               *prometheus.Desc
@@ -390,6 +391,10 @@ func newTSDBMetrics(r prometheus.Registerer) *tsdbMetrics {
 			"cortex_ingester_shipper_upload_failures_total",
 			"Total number of TSDB block upload failures",
 			nil, nil),
+		corruptedBlocks: prometheus.NewDesc(
+			"cortex_ingester_shipper_corrupted_blocks_total",
+			"Total number of TSDB blocks corrupted",
+			nil, nil),
 		tsdbCompactionsTotal: prometheus.NewDesc(
 			"cortex_ingester_tsdb_compactions_total",
 			"Total number of TSDB compactions that were executed.",
@@ -579,6 +584,7 @@ func (sm *tsdbMetrics) Describe(out chan<- *prometheus.Desc) {
 	out <- sm.dirSyncFailures
 	out <- sm.uploads
 	out <- sm.uploadFailures
+	out <- sm.corruptedBlocks
 
 	out <- sm.tsdbCompactionsTotal
 	out <- sm.tsdbCompactionDuration
@@ -636,6 +642,7 @@ func (sm *tsdbMetrics) Collect(out chan<- prometheus.Metric) {
 	data.SendSumOfCounters(out, sm.dirSyncFailures, "thanos_shipper_dir_sync_failures_total")
 	data.SendSumOfCounters(out, sm.uploads, "thanos_shipper_uploads_total")
 	data.SendSumOfCounters(out, sm.uploadFailures, "thanos_shipper_upload_failures_total")
+	data.SendSumOfCounters(out, sm.corruptedBlocks, "thanos_shipper_corrupted_blocks_total")
 
 	data.SendSumOfCounters(out, sm.tsdbCompactionsTotal, "prometheus_tsdb_compactions_total")
 	data.SendSumOfHistograms(out, sm.tsdbCompactionDuration, "prometheus_tsdb_compaction_duration_seconds")

pkg/ingester/metrics_test.go

@@ -187,6 +187,11 @@ func TestTSDBMetrics(t *testing.T) {
 			# 4*(12345 + 85787 + 999)
 			cortex_ingester_shipper_upload_failures_total 396524
 
+            # HELP cortex_ingester_shipper_corrupted_blocks_total Total number of TSDB blocks corrupted
+			# TYPE cortex_ingester_shipper_corrupted_blocks_total counter
+			# 30*(12345 + 85787 + 999)
+			cortex_ingester_shipper_corrupted_blocks_total 2973930
+
 			# HELP cortex_ingester_tsdb_compactions_total Total number of TSDB compactions that were executed.
 			# TYPE cortex_ingester_tsdb_compactions_total counter
 			cortex_ingester_tsdb_compactions_total 693917
@@ -446,6 +451,12 @@ func TestTSDBMetricsWithRemoval(t *testing.T) {
 			# 4*(12345 + 85787 + 999)
 			cortex_ingester_shipper_upload_failures_total 396524
 
+            # HELP cortex_ingester_shipper_corrupted_blocks_total Total number of TSDB blocks corrupted
+			# TYPE cortex_ingester_shipper_corrupted_blocks_total counter
+			# 30*(12345 + 85787 + 999)
+			cortex_ingester_shipper_corrupted_blocks_total 2973930
+
+
 			# HELP cortex_ingester_tsdb_compactions_total Total number of TSDB compactions that were executed.
 			# TYPE cortex_ingester_tsdb_compactions_total counter
 			cortex_ingester_tsdb_compactions_total 693917
@@ -688,6 +699,12 @@ func populateTSDBMetrics(base float64) *prometheus.Registry {
 	})
 	uploadFailures.Add(4 * base)
 
+	corruptedBlocks := promauto.With(r).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_shipper_corrupted_blocks_total",
+		Help: "Total number of corrupted blocks",
+	})
+	corruptedBlocks.Add(30 * base)
+
 	// TSDB Head
 	seriesCreated := promauto.With(r).NewCounter(prometheus.CounterOpts{
 		Name: "prometheus_tsdb_head_series_created_total",