fix replicate duplicate metrics

Signed-off-by: yeya24 <[email protected]>

Author: yeya24

CHANGELOG.md

@@ -15,6 +15,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 
 - [#2238](https://github.com/thanos-io/thanos/pull/2238) Ruler: Fixed Issue #2204 bug in alert queue signalling filled up queue and alerts were dropped
 - [#2231](https://github.com/thanos-io/thanos/pull/2231) Bucket Web - Sort chunks by thanos.downsample.resolution for better grouping
+- [#2254](https://github.com/thanos-io/thanos/pull/2254) Bucket: Fix metrics registered multiple times in bucket replicate
 
 ### Added
 

pkg/replicate/replicater.go renamed to pkg/replicate/replicator.go

@@ -20,6 +20,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
+	thanosblock "github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/component"
 	"github.com/thanos-io/thanos/pkg/extflag"
@@ -149,6 +150,11 @@ func RunReplicate(
 		Help: "The Duration of replication runs split by success and error.",
 	}, []string{"result"})
 
+	fetcher, err := thanosblock.NewMetaFetcher(logger, 32, fromBkt, "", reg)
+	if err != nil {
+		return errors.Wrapf(err, "create meta fetcher with bucket %v", fromBkt)
+	}
+
 	blockFilter := NewBlockFilter(
 		logger,
 		labelSelector,
@@ -170,7 +176,7 @@ func RunReplicate(
 		logger := log.With(logger, "replication-run-id", ulid.String())
 		level.Info(logger).Log("msg", "running replication attempt")
 
-		if err := newReplicationScheme(logger, metrics, blockFilter, fromBkt, toBkt, reg).execute(ctx); err != nil {
+		if err := newReplicationScheme(logger, metrics, blockFilter, fetcher, fromBkt, toBkt, reg).execute(ctx); err != nil {
 			return errors.Wrap(err, "replication execute")
 		}
 

pkg/replicate/scheme.go

@@ -105,6 +105,7 @@ type replicationScheme struct {
 	toBkt   objstore.Bucket
 
 	blockFilter blockFilterFunc
+	fetcher     thanosblock.MetadataFetcher
 
 	logger  log.Logger
 	metrics *replicationMetrics
@@ -152,14 +153,23 @@ func newReplicationMetrics(reg prometheus.Registerer) *replicationMetrics {
 	return m
 }
 
-func newReplicationScheme(logger log.Logger, metrics *replicationMetrics, blockFilter blockFilterFunc, from objstore.BucketReader, to objstore.Bucket, reg prometheus.Registerer) *replicationScheme {
+func newReplicationScheme(
+	logger log.Logger,
+	metrics *replicationMetrics,
+	blockFilter blockFilterFunc,
+	fetcher thanosblock.MetadataFetcher,
+	from objstore.BucketReader,
+	to objstore.Bucket,
+	reg prometheus.Registerer,
+) *replicationScheme {
 	if logger == nil {
 		logger = log.NewNopLogger()
 	}
 
 	return &replicationScheme{
 		logger:      logger,
 		blockFilter: blockFilter,
+		fetcher:     fetcher,
 		fromBkt:     from,
 		toBkt:       to,
 		metrics:     metrics,
@@ -350,12 +360,7 @@ func (rs *replicationScheme) ensureObjectReplicated(ctx context.Context, objectN
 // partial, this is just a temporary failure, as the block is still being
 // uploaded to the origin bucket.
 func loadMeta(ctx context.Context, rs *replicationScheme, id ulid.ULID) (*metadata.Meta, bool, error) {
-	fetcher, err := thanosblock.NewMetaFetcher(rs.logger, 32, rs.fromBkt, "", rs.reg)
-	if err != nil {
-		return nil, false, errors.Wrapf(err, "create meta fetcher with buecket %v", rs.fromBkt)
-	}
-
-	metas, _, err := fetcher.Fetch(ctx)
+	metas, _, err := rs.fetcher.Fetch(ctx)
 	if err != nil {
 		switch errors.Cause(err) {
 		default:

pkg/replicate/scheme_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/oklog/ulid"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/tsdb"
+	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/objstore"
@@ -302,9 +303,7 @@ func TestReplicationSchemeAll(t *testing.T) {
 		c.prepare(ctx, t, originBucket, targetBucket)
 
 		matcher, err := labels.NewMatcher(labels.MatchEqual, "test-labelname", "test-labelvalue")
-		if err != nil {
-			t.Fatal("Failed to create a matcher.")
-		}
+		testutil.Ok(t, err)
 
 		selector := labels.Selector{
 			matcher,
@@ -314,11 +313,14 @@ func TestReplicationSchemeAll(t *testing.T) {
 		}
 
 		filter := NewBlockFilter(logger, selector, compact.ResolutionLevelRaw, 1).Filter
+		fetcher, err := block.NewMetaFetcher(logger, 32, originBucket, "", nil)
+		testutil.Ok(t, err)
 
 		r := newReplicationScheme(
 			logger,
 			newReplicationMetrics(nil),
 			filter,
+			fetcher,
 			originBucket,
 			targetBucket,
 			nil,