Merge pull request #7194 from xBazilio/retry-downsample-errors

CHANGELOG.md

@@ -19,6 +19,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#7122](https://github.com/thanos-io/thanos/pull/7122) Store Gateway: Fix lazy expanded postings estimate base cardinality using posting group with remove keys.
 
 ### Added
+- [#7194](https://github.com/thanos-io/thanos/pull/7194) Downsample: retry objstore related errors
 - [#7105](https://github.com/thanos-io/thanos/pull/7105) Rule: add flag `--query.enable-x-functions` to allow usage of extended promql functions (xrate, xincrease, xdelta) in loaded rules
 - [#6867](https://github.com/thanos-io/thanos/pull/6867) Query UI: Tenant input box added to the Query UI, in order to be able to specify which tenant the query should use.
 - [#7175](https://github.com/thanos-io/thanos/pull/7175): Query: Add `--query.mode=distributed` which enables the new distributed mode of the Thanos query engine.

cmd/thanos/downsample.go

@@ -21,6 +21,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
+	"github.com/thanos-io/thanos/pkg/compact"
 
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/objstore/client"
@@ -358,7 +359,7 @@ func processDownsampling(
 
 	err := block.Download(ctx, logger, bkt, m.ULID, bdir, objstore.WithFetchConcurrency(blockFilesConcurrency))
 	if err != nil {
-		return errors.Wrapf(err, "download block %s", m.ULID)
+		return compact.NewRetryError(errors.Wrapf(err, "download block %s", m.ULID))
 	}
 	level.Info(logger).Log("msg", "downloaded block", "id", m.ULID, "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
 
@@ -419,7 +420,7 @@ func processDownsampling(
 
 	err = block.Upload(ctx, logger, bkt, resdir, hashFunc)
 	if err != nil {
-		return errors.Wrapf(err, "upload downsampled block %s", id)
+		return compact.NewRetryError(errors.Wrapf(err, "upload downsampled block %s", id))
 	}
 
 	level.Info(logger).Log("msg", "uploaded block", "id", id, "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())

pkg/compact/compact.go

@@ -967,6 +967,10 @@ type RetryError struct {
 	err error
 }
 
+func NewRetryError(err error) error {
+	return retry(err)
+}
+
 func retry(err error) error {
 	if IsHaltError(err) {
 		return err