objstore: Added WithExpectedErrs which allows to control instrumentation (e.g not increment failures for expected not found)

cmd/thanos/main_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/compact/downsample"
-	"github.com/thanos-io/thanos/pkg/objstore/inmem"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/testutil"
 	"github.com/thanos-io/thanos/pkg/testutil/e2eutil"
 )
@@ -36,7 +36,7 @@ func TestCleanupIndexCacheFolder(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.WithNoopInstr(objstore.NewInMemBucket())
 
 	// Upload one compaction lvl = 2 block, one compaction lvl = 1.
 	// We generate index cache files only for lvl > 1 blocks.
@@ -97,7 +97,7 @@ func TestCleanupDownsampleCacheFolder(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.WithNoopInstr(objstore.NewInMemBucket())
 	var id ulid.ULID
 	{
 		id, err = e2eutil.CreateBlock(

pkg/block/block_test.go

@@ -19,7 +19,7 @@ import (
 	"github.com/go-kit/kit/log"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
-	"github.com/thanos-io/thanos/pkg/objstore/inmem"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/testutil"
 	"github.com/thanos-io/thanos/pkg/testutil/e2eutil"
 
@@ -81,7 +81,7 @@ func TestUpload(t *testing.T) {
 	testutil.Ok(t, err)
 	defer func() { testutil.Ok(t, os.RemoveAll(tmpDir)) }()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.NewInMemBucket()
 	b1, err := e2eutil.CreateBlock(ctx, tmpDir, []labels.Labels{
 		{{Name: "a", Value: "1"}},
 		{{Name: "a", Value: "2"}},
@@ -185,7 +185,7 @@ func TestDelete(t *testing.T) {
 	testutil.Ok(t, err)
 	defer func() { testutil.Ok(t, os.RemoveAll(tmpDir)) }()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.NewInMemBucket()
 	{
 		b1, err := e2eutil.CreateBlock(ctx, tmpDir, []labels.Labels{
 			{{Name: "a", Value: "1"}},
@@ -232,7 +232,7 @@ func TestMarkForDeletion(t *testing.T) {
 	testutil.Ok(t, err)
 	defer func() { testutil.Ok(t, os.RemoveAll(tmpDir)) }()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.NewInMemBucket()
 	{
 		blockWithoutDeletionMark, err := e2eutil.CreateBlock(ctx, tmpDir, []labels.Labels{
 			{{Name: "a", Value: "1"}},

pkg/block/fetcher.go

@@ -142,7 +142,7 @@ type MetadataModifier interface {
 type BaseFetcher struct {
 	logger      log.Logger
 	concurrency int
-	bkt         objstore.BucketReader
+	bkt         objstore.InstrumentedBucketReader
 
 	// Optional local directory to cache meta.json files.
 	cacheDir string
@@ -152,7 +152,7 @@ type BaseFetcher struct {
 }
 
 // NewBaseFetcher constructs BaseFetcher.
-func NewBaseFetcher(logger log.Logger, concurrency int, bkt objstore.BucketReader, dir string, reg prometheus.Registerer) (*BaseFetcher, error) {
+func NewBaseFetcher(logger log.Logger, concurrency int, bkt objstore.InstrumentedBucketReader, dir string, reg prometheus.Registerer) (*BaseFetcher, error) {
 	if logger == nil {
 		logger = log.NewNopLogger()
 	}
@@ -180,7 +180,7 @@ func NewBaseFetcher(logger log.Logger, concurrency int, bkt objstore.BucketReade
 }
 
 // NewMetaFetcher returns meta fetcher.
-func NewMetaFetcher(logger log.Logger, concurrency int, bkt objstore.BucketReader, dir string, reg prometheus.Registerer, filters []MetadataFilter, modifiers []MetadataModifier) (*MetaFetcher, error) {
+func NewMetaFetcher(logger log.Logger, concurrency int, bkt objstore.InstrumentedBucketReader, dir string, reg prometheus.Registerer, filters []MetadataFilter, modifiers []MetadataModifier) (*MetaFetcher, error) {
 	b, err := NewBaseFetcher(logger, concurrency, bkt, dir, reg)
 	if err != nil {
 		return nil, err
@@ -236,7 +236,7 @@ func (f *BaseFetcher) loadMeta(ctx context.Context, id ulid.ULID) (*metadata.Met
 		}
 	}
 
-	r, err := f.bkt.Get(ctx, metaFile)
+	r, err := f.bkt.ReaderWithExpectedErrs(f.bkt.IsObjNotFoundErr).Get(ctx, metaFile)
 	if f.bkt.IsObjNotFoundErr(err) {
 		// Meta.json was deleted between bkt.Exists and here.
 		return nil, errors.Wrapf(ErrorSyncMetaNotFound, "%v", err)
@@ -740,12 +740,12 @@ func (f *ConsistencyDelayMetaFilter) Filter(_ context.Context, metas map[ulid.UL
 type IgnoreDeletionMarkFilter struct {
 	logger          log.Logger
 	delay           time.Duration
-	bkt             objstore.BucketReader
+	bkt             objstore.InstrumentedBucketReader
 	deletionMarkMap map[ulid.ULID]*metadata.DeletionMark
 }
 
 // NewIgnoreDeletionMarkFilter creates IgnoreDeletionMarkFilter.
-func NewIgnoreDeletionMarkFilter(logger log.Logger, bkt objstore.BucketReader, delay time.Duration) *IgnoreDeletionMarkFilter {
+func NewIgnoreDeletionMarkFilter(logger log.Logger, bkt objstore.InstrumentedBucketReader, delay time.Duration) *IgnoreDeletionMarkFilter {
 	return &IgnoreDeletionMarkFilter{
 		logger: logger,
 		bkt:    bkt,

pkg/block/fetcher_test.go

@@ -75,7 +75,7 @@ func TestMetaFetcher_Fetch(t *testing.T) {
 
 		var ulidToDelete ulid.ULID
 		r := prometheus.NewRegistry()
-		baseFetcher, err := NewBaseFetcher(log.NewNopLogger(), 20, bkt, dir, r)
+		baseFetcher, err := NewBaseFetcher(log.NewNopLogger(), 20, objstore.WithNoopInstr(bkt), dir, r)
 		testutil.Ok(t, err)
 
 		fetcher := baseFetcher.NewMetaFetcher(r, []MetadataFilter{
@@ -1065,7 +1065,7 @@ func TestIgnoreDeletionMarkFilter_Filter(t *testing.T) {
 		now := time.Now()
 		f := &IgnoreDeletionMarkFilter{
 			logger: log.NewNopLogger(),
-			bkt:    bkt,
+			bkt:    objstore.WithNoopInstr(bkt),
 			delay:  48 * time.Hour,
 		}
 

pkg/block/indexheader/json_reader.go

@@ -199,7 +199,7 @@ type JSONReader struct {
 }
 
 // NewJSONReader loads or builds new index-cache.json if not present on disk or object storage.
-func NewJSONReader(ctx context.Context, logger log.Logger, bkt objstore.BucketReader, dir string, id ulid.ULID) (*JSONReader, error) {
+func NewJSONReader(ctx context.Context, logger log.Logger, bkt objstore.InstrumentedBucketReader, dir string, id ulid.ULID) (*JSONReader, error) {
 	cachefn := filepath.Join(dir, id.String(), block.IndexCacheFilename)
 	jr, err := newFileJSONReader(logger, cachefn)
 	if err == nil {
@@ -216,7 +216,7 @@ func NewJSONReader(ctx context.Context, logger log.Logger, bkt objstore.BucketRe
 	}
 
 	// Try to download index cache file from object store.
-	if err = objstore.DownloadFile(ctx, logger, bkt, filepath.Join(id.String(), block.IndexCacheFilename), cachefn); err == nil {
+	if err = objstore.DownloadFile(ctx, logger, bkt.ReaderWithExpectedErrs(bkt.IsObjNotFoundErr), filepath.Join(id.String(), block.IndexCacheFilename), cachefn); err == nil {
 		return newFileJSONReader(logger, cachefn)
 	}
 

pkg/block/metadata/deletionmark.go

@@ -45,10 +45,10 @@ type DeletionMark struct {
 }
 
 // ReadDeletionMark reads the given deletion mark file from <dir>/deletion-mark.json in bucket.
-func ReadDeletionMark(ctx context.Context, bkt objstore.BucketReader, logger log.Logger, dir string) (*DeletionMark, error) {
+func ReadDeletionMark(ctx context.Context, bkt objstore.InstrumentedBucketReader, logger log.Logger, dir string) (*DeletionMark, error) {
 	deletionMarkFile := path.Join(dir, DeletionMarkFilename)
 
-	r, err := bkt.Get(ctx, deletionMarkFile)
+	r, err := bkt.ReaderWithExpectedErrs(bkt.IsObjNotFoundErr).Get(ctx, deletionMarkFile)
 	if err != nil {
 		if bkt.IsObjNotFoundErr(err) {
 			return nil, ErrorDeletionMarkNotFound

pkg/block/metadata/deletionmark_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/fortytw2/leaktest"
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
-	"github.com/thanos-io/thanos/pkg/objstore/inmem"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/testutil"
 )
 
@@ -29,7 +29,7 @@ func TestReadDeletionMark(t *testing.T) {
 	testutil.Ok(t, err)
 	defer func() { testutil.Ok(t, os.RemoveAll(tmpDir)) }()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.WithNoopInstr(objstore.NewInMemBucket())
 	{
 		blockWithoutDeletionMark := ulid.MustNew(uint64(1), nil)
 		_, err := ReadDeletionMark(ctx, bkt, nil, path.Join(tmpDir, blockWithoutDeletionMark.String()))

pkg/compact/clean_test.go

@@ -18,15 +18,15 @@ import (
 	promtest "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/block/metadata"
-	"github.com/thanos-io/thanos/pkg/objstore/inmem"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/testutil"
 )
 
 func TestBestEffortCleanAbortedPartialUploads(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
-	bkt := inmem.NewBucket()
+	bkt := objstore.WithNoopInstr(objstore.NewInMemBucket())
 	logger := log.NewNopLogger()
 
 	metaFetcher, err := block.NewMetaFetcher(nil, 32, bkt, "", nil, nil, nil)

pkg/compact/compact_e2e_test.go

@@ -90,7 +90,7 @@ func TestSyncer_GarbageCollect_e2e(t *testing.T) {
 		}
 
 		duplicateBlocksFilter := block.NewDeduplicateFilter()
-		metaFetcher, err := block.NewMetaFetcher(nil, 32, bkt, "", nil, []block.MetadataFilter{
+		metaFetcher, err := block.NewMetaFetcher(nil, 32, objstore.WithNoopInstr(bkt), "", nil, []block.MetadataFilter{
 			duplicateBlocksFilter,
 		}, nil)
 		testutil.Ok(t, err)
@@ -176,9 +176,9 @@ func TestGroup_Compact_e2e(t *testing.T) {
 
 		reg := prometheus.NewRegistry()
 
-		ignoreDeletionMarkFilter := block.NewIgnoreDeletionMarkFilter(logger, bkt, 48*time.Hour)
+		ignoreDeletionMarkFilter := block.NewIgnoreDeletionMarkFilter(logger, objstore.WithNoopInstr(bkt), 48*time.Hour)
 		duplicateBlocksFilter := block.NewDeduplicateFilter()
-		metaFetcher, err := block.NewMetaFetcher(nil, 32, bkt, "", nil, []block.MetadataFilter{
+		metaFetcher, err := block.NewMetaFetcher(nil, 32, objstore.WithNoopInstr(bkt), "", nil, []block.MetadataFilter{
 			ignoreDeletionMarkFilter,
 			duplicateBlocksFilter,
 		}, nil)

pkg/compact/retention_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/objstore"
-	"github.com/thanos-io/thanos/pkg/objstore/inmem"
 	"github.com/thanos-io/thanos/pkg/testutil"
 )
 
@@ -240,7 +239,7 @@ func TestApplyRetentionPolicyByResolution(t *testing.T) {
 		},
 	} {
 		t.Run(tt.name, func(t *testing.T) {
-			bkt := inmem.NewBucket()
+			bkt := objstore.WithNoopInstr(objstore.NewInMemBucket())
 			for _, b := range tt.blocks {
 				uploadMockBlock(t, bkt, b.id, b.minTime, b.maxTime, int64(b.resolution))
 			}

pkg/objstore/client/factory.go

@@ -42,7 +42,7 @@ type BucketConfig struct {
 
 // NewBucket initializes and returns new object storage clients.
 // NOTE: confContentYaml can contain secrets.
-func NewBucket(logger log.Logger, confContentYaml []byte, reg prometheus.Registerer, component string) (objstore.Bucket, error) {
+func NewBucket(logger log.Logger, confContentYaml []byte, reg prometheus.Registerer, component string) (objstore.InstrumentedBucket, error) {
 	level.Info(logger).Log("msg", "loading bucket configuration")
 	bucketConf := &BucketConfig{}
 	if err := yaml.UnmarshalStrict(confContentYaml, bucketConf); err != nil {

pkg/objstore/cos/cos.go

@@ -52,6 +52,7 @@ func (conf *Config) validate() error {
 	return nil
 }
 
+// NewBucket returns a new Bucket using the provided cos configuration.
 func NewBucket(logger log.Logger, conf []byte, component string) (*Bucket, error) {
 	if logger == nil {
 		logger = log.NewNopLogger()