objstore: Added WithExpectedErrs to Reader which allows to control in…

…strumentation (e.g not increment failures for expected not found). This allows to not wake up oncall in the middle of night, becuase of expeced, properly handled case (: Also: Has to move inmem to objstore for testing. Signed-off-by: Bartlomiej Plotka <[email protected]>
thanos-io · Apr 3, 2020 · a04818c · a04818c
1 parent 84495fa
commit a04818c
Show file tree

Hide file tree

Showing 29 changed files with 498 additions and 345 deletions.
diff --git a/cmd/thanos/main_test.go b/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(

diff --git a/pkg/block/block_test.go b/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"}},

diff --git a/pkg/block/fetcher.go b/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,

diff --git a/pkg/block/fetcher_test.go b/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,
 		}
 

diff --git a/pkg/block/indexheader/json_reader.go b/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)
 	}
 

diff --git a/pkg/block/metadata/deletionmark.go b/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

diff --git a/pkg/block/metadata/deletionmark_test.go b/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()))

diff --git a/pkg/compact/clean_test.go b/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)

diff --git a/pkg/compact/compact_e2e_test.go b/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)

diff --git a/pkg/compact/retention_test.go b/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))
 			}

diff --git a/pkg/objstore/azure/azure.go b/pkg/objstore/azure/azure.go
@@ -64,7 +64,7 @@ func (conf *Config) validate() error {
 	return nil
 }
 
-// NewBucket returns a new Bucket using the provided Azure config.
+// NewInMemBucket returns a new Bucket using the provided Azure config.
 func NewBucket(logger log.Logger, azureConfig []byte, component string) (*Bucket, error) {
 	level.Debug(logger).Log("msg", "creating new Azure bucket connection", "component", component)
 

diff --git a/pkg/objstore/client/factory.go b/pkg/objstore/client/factory.go
@@ -40,9 +40,9 @@ type BucketConfig struct {
 	Config interface{} `yaml:"config"`
 }
 
-// NewBucket initializes and returns new object storage clients.
+// NewInMemBucket 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 {

diff --git a/pkg/objstore/filesystem/filesystem.go b/pkg/objstore/filesystem/filesystem.go
@@ -43,7 +43,7 @@ func NewBucketFromConfig(conf []byte) (*Bucket, error) {
 	return NewBucket(c.Directory)
 }
 
-// NewBucket returns a new filesystem.Bucket.
+// NewInMemBucket returns a new filesystem.Bucket.
 func NewBucket(rootDir string) (*Bucket, error) {
 	absDir, err := filepath.Abs(rootDir)
 	if err != nil {

diff --git a/pkg/objstore/gcs/gcs.go b/pkg/objstore/gcs/gcs.go
@@ -41,7 +41,7 @@ type Bucket struct {
 	closer io.Closer
 }
 
-// NewBucket returns a new Bucket against the given bucket handle.
+// NewInMemBucket returns a new Bucket against the given bucket handle.
 func NewBucket(ctx context.Context, logger log.Logger, conf []byte, component string) (*Bucket, error) {
 	var gc Config
 	if err := yaml.Unmarshal(conf, &gc); err != nil {