Unified fetching/syncing block metadata everywhere #1394

lx223 · 2019-08-09T23:04:40Z

[] CHANGELOG entry if change is relevant to the end user.

Changes

add a new block meta fetcher that fetches block metadata stored in a bucket
replace some of the existing block md fetching code with the new fetcher
change the signature of DownloadMeta to use the more restrictive BucketReader

Verification

Unit test

lx223 · 2019-08-11T20:49:32Z

The CI test fails because of go mod error. It doesn't seem related to my change. Is it a known issue?

bwplotka · 2019-08-15T18:21:02Z

Fixes #1335

@lx223 can you rebase, we fixed this master go.mod error (:

bwplotka

Looking good! Thanks!

Can we use exactly the same thing in bucket.go (essentially in store gateway):

thanos/pkg/store/bucket.go

Line 287 in 865d5ec

func (s *BucketStore) SyncBlocks(ctx context.Context) error {

bwplotka · 2019-08-15T18:23:23Z

pkg/block/mdfetcher.go

+
+// Fetch wil return the mapping from block ID to its metadata that are currently stored in the bucket passed in.
+// If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.
+func (f *MetadataFetcher) Fetch(ctx context.Context) (map[ulid.ULID]*metadata.Meta, error) {


Why fetcher and not e.g Syncer? Especially if we maintain cachedBlockMDs for efficiency?

Let's mention if it's goroutine safe (: It looks like it is, the question is ... should it be?

I think it would be better if you would not make Fetch/Sync goroutine safe. First of all, I don't think it is needed. Secondly, with this, we can be more efficient in terms of memory. We can allow to reuse exactly the same map[ulid.ULID]*metadata.Meta (pass as an argument and return updated one). This might save some allocations as there can be millions of objects potentially in the bucket.

What do you think? With this I am not sure if the struct is needed, maybe just Function is enough? Don't have a strong opinion.

Be aware of this #1408 that changes this code as well (somehow we need to sync both PRs) cc @povilasv

I am happy to rename to syncer if you feel strongly about it. I am alright with both. I chose fetch as I felt it is more about fetching the metadata from remote source. Keeping a cache is just an internal optimisation and might become an issue and get removed if the bucket contains too many objects.

Good point about about the memory allocation. I have a different perspective. I don't think it would add that much to memory as it is using metadata pointers right now. And if we were to reuse the same map, there would still have to be a map to keep what metadata are in the bucket in order to remove those cached but somehow-removed-from-remote-source ones. So the saving won't be that much. Besides, I think streaming metadata might be a better solution to scale with bigger number of objects in the bucket. What do you think?

No strong feeling to the naming. It really depends on what exactly interface we are going to apply as you are exploring streaming suggestion now (: So then ForeachMeta might make sense more?

I think streaming metadata might be a better solution to scale with bigger number of objects in the bucket. What do you think?

Definitely, however, this means we would need to revert to any of Iter concepts like

passing a function to invoke for each meta

Iter interface:

// Iterator is a simple iterator that can only get the next value. type Iterator interface { At() metadata.Meta) Err() error Next() bool }

I think we don't have to use stream for now unless the OOM issue is already big and urgent. I think we can keep this signature which meshes pretty well with the current usage and we can use this as the foundation to identify and make changes when we are ready to explore other options like streaming.

Yea, I think I am happy with this. We already keep cached block MD anyway.

fetcher might be fine then

pkg/block/mdfetcher.go

bwplotka · 2019-08-15T18:33:19Z

pkg/block/mdfetcher.go

+}
+
+// Fetch wil return the mapping from block ID to its metadata that are currently stored in the bucket passed in.
+// If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.


If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.

s/metadta/metadata

Nice idea to delegate the no-meta-json blocks on caller side. I think it makes sense.

However, maybe two maps make sense here then?

(metas map[ulid.ULID]*metadata.Meta, noMeta []ulid.ULID, err error)

This will be easier to use on caller side I think.

sorry, not following about point 1. what is s/metadta/metadata?

I thought about whether to distinguish those two cases. For the places I have replaced, I haven't found a case that would require these two being distinguished. Did I miss a case that would require this?

sorry, not following about point 1. what is s/metadta/metadata?

https://www.tutorialspoint.com/unix/unix-regular-expressions.htm

(:

Hm, what about the case where we remove partially uploaded blocks? (noMeta + delay):

thanos/pkg/compact/compact.go

Line 197 in 865d5ec

if removedOrIgnored := c.removeIfMetaMalformed(workCtx, id); removedOrIgnored {

I think that's when noMeta is useful.

I might be wrong here but I think this is the only case that this required and it is covered in the caller's logic - thanos/pkg/compact/compact.go. But it does sound like something useful in future. Tempted to add when we are seeing more such usages.

bwplotka · 2019-08-15T18:38:25Z

pkg/block/mdfetcher.go

+	remoteBlockMDs := make(map[ulid.ULID]*metadata.Meta, len(f.cachedBlockMDs))
+	wg := &sync.WaitGroup{}
+
+	for i := 0; i < f.concurrencyLimit; i++ {


Will review body of this function when we resolve other comments first (:

Looks good from first glance, looks like you covered all the edge cases, nice! Thanks for tests as well!

bwplotka · 2019-08-15T18:40:22Z

Also we need CHANGELOG for this I think as we not handle partial uploads carefully on downsampling/retention apply

bwplotka · 2019-09-13T09:50:13Z

Hi 👋

Can we finish this? (: If you don't have time, can I grab it and finish?

lx223 · 2019-09-15T00:28:45Z

@bwplotka sorry about that!

I'll change pkg/store/bucket.go and add a changelog tomorrow.

bwplotka · 2019-09-16T09:31:47Z

@lx223 Are you on CNCF slack? Join us so we can discuss quicker offline!

I would like to see this change ASAP as it will help with: #1331

bwplotka

Nice. However, let's make sure to use the same syncer everywhere. I located a few places below. I literally searched for usage of Iter that relates to fetching meta.json. We desperately need to use exactly the same logic everywhere (:

I checked boxed things that I can see we changed in this PR, we might want to add missing bits. (:

Thanos bucket

thanos/cmd/thanos/bucket.go

Line 234 in 865d5ec

return bkt.Iter(ctx, "", func(name string) error {
thanos/cmd/thanos/bucket.go

Line 385 in 865d5ec

func download(ctx context.Context, logger log.Logger, bkt objstore.Bucket) (blocks []metadata.Meta, err error) {
thanos/cmd/thanos/bucket.go

Line 279 in 865d5ec

if err = bkt.Iter(ctx, "", func(name string) error {

Store GW:

thanos/pkg/store/bucket.go

Line 316 in 1389096

err := s.bucket.Iter(ctx, "", func(name string) error {

Compactor:

thanos/cmd/thanos/compact.go

Line 350 in c83332f

if err := bkt.Iter(ctx, "", func(name string) error {
thanos/pkg/compact/compact.go

Line 214 in 865d5ec

err := c.bkt.Iter(ctx, "", func(name string) error {
thanos/cmd/thanos/downsample.go

Line 142 in 0a30a2e

err := bkt.Iter(ctx, "", func(name string) error {
thanos/pkg/compact/retention.go

Line 18 in badeb8f

if err := bkt.Iter(ctx, "", func(name string) error {

Verifier

thanos/pkg/verifier/index_issue.go

Line 31 in 2c5f2cd

err := bkt.Iter(ctx, "", func(name string) error {
thanos/pkg/verifier/overlapped_blocks.go

Line 50 in 865d5ec

err := bkt.Iter(ctx, "", func(name string) error {

Sidecar

thanos/pkg/shipper/shipper.go

Line 200 in c35d47b

if err := c.bucket.Iter(ctx, "", func(path string) error {

bwplotka · 2019-09-16T11:38:16Z

pkg/block/mdfetcher.go

+}
+
+// Fetch wil return the mapping from block ID to its metadata that are currently stored in the bucket passed in.
+// If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.


Hm, what about the case where we remove partially uploaded blocks? (noMeta + delay):

thanos/pkg/compact/compact.go

Line 197 in 865d5ec

if removedOrIgnored := c.removeIfMetaMalformed(workCtx, id); removedOrIgnored {

I think that's when noMeta is useful.

bwplotka · 2019-09-16T11:44:09Z

pkg/block/mdfetcher.go

+
+// Fetch wil return the mapping from block ID to its metadata that are currently stored in the bucket passed in.
+// If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.
+func (f *MetadataFetcher) Fetch(ctx context.Context) (map[ulid.ULID]*metadata.Meta, error) {


No strong feeling to the naming. It really depends on what exactly interface we are going to apply as you are exploring streaming suggestion now (: So then ForeachMeta might make sense more?

I think streaming metadata might be a better solution to scale with bigger number of objects in the bucket. What do you think?

Definitely, however, this means we would need to revert to any of Iter concepts like

passing a function to invoke for each meta

Iter interface:

// Iterator is a simple iterator that can only get the next value. type Iterator interface { At() metadata.Meta) Err() error Next() bool }

bwplotka · 2019-09-16T11:46:16Z

Would love to see this work to be done as this would enormously help with dynamic system where components are restarted in different moments causing partial uploads to be very common (: Let me know if I can help with this and I would love to prioritize this. @lx223

I listed all the places were we can unify our sync/fetch logic.

bwplotka · 2019-09-16T15:08:18Z

Related to #1525

lx223 · 2019-09-21T21:32:40Z

@bwplotka just joined CNCF slack and have made changes to all those places you listed. PTAL

Signed-off-by: Lan Xiao <lx223@users.noreply.github.com>

lx223 · 2019-09-21T21:42:24Z

cmd/thanos/bucket.go

-			id, ok := block.IsBlockDir(name)
-			if !ok {
-				return nil
+		mdFetcher := block.NewMetadataFetcher(logger, defaultBlockMDFetchConcurrency, bkt)


didn't switch to the download function because the original logic suggests it requires block id even for partially uploaded blocks.

bwplotka · 2019-10-31T12:44:59Z

Should we get back to this @lx223 ? (:

lx223 · 2019-11-03T17:31:06Z

@bwplotka sure :D WDYT about the current code? there is some rebasing and test passing to do. But, it probably won't change the main logic.

bwplotka

Yes! I think we covered all of the usages.
I left some comments but to me, it's looking good.

Let's mention we are fixing #1335 issue.

Wonder if we can instrument this fetcher. Maybe fetch runs? or something (:

bwplotka · 2019-11-03T20:18:43Z

cmd/thanos/bucket.go

-			id, ok := block.IsBlockDir(name)
-			if !ok {
-				return nil
+		mdFetcher := block.NewMetadataFetcher(logger, defaultBlockMDFetchConcurrency, bkt)


bwplotka · 2019-11-03T20:19:56Z

pkg/block/block.go

@@ -163,7 +163,7 @@ func deleteDir(ctx context.Context, logger log.Logger, bkt objstore.Bucket, dir

 // DownloadMeta downloads only meta file from bucket by block ID.
 // TODO(bwplotka): Differentiate between network error & partial upload.
-func DownloadMeta(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID) (metadata.Meta, error) {
+func DownloadMeta(ctx context.Context, logger log.Logger, bkt objstore.BucketReader, id ulid.ULID) (metadata.Meta, error) {


bwplotka · 2019-11-03T20:28:44Z

pkg/block/mdfetcher.go

+		return existingMD, nil
+	}
+
+	md, err := DownloadMeta(ctx, f.logger, f.bkt, blockID)


Wonder if we can leverage the file system as a cache, instead of downloading and keep in memory? Let's maybe add TODO at least (: It saves some bandwidth (?) on startup.

bwplotka · 2019-11-03T20:29:59Z

pkg/block/mdfetcher.go

+
+// Fetch wil return the mapping from block ID to its metadata that are currently stored in the bucket passed in.
+// If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.
+func (f *MetadataFetcher) Fetch(ctx context.Context) (map[ulid.ULID]*metadata.Meta, error) {


Yea, I think I am happy with this. We already keep cached block MD anyway.

bwplotka · 2019-11-03T20:30:10Z

pkg/block/mdfetcher.go

+
+// Fetch wil return the mapping from block ID to its metadata that are currently stored in the bucket passed in.
+// If the corresponding value for a block ID key is nil, no valid metadta is retrieved successfully.
+func (f *MetadataFetcher) Fetch(ctx context.Context) (map[ulid.ULID]*metadata.Meta, error) {


fetcher might be fine then

bwplotka · 2019-11-03T20:30:26Z

cmd/thanos/bucket.go

+	mdFetcher := block.NewMetadataFetcher(logger, defaultBlockMDFetchConcurrency, bkt)
+	mds, err := mdFetcher.Fetch(ctx)
+	if err != nil {
+		level.Error(logger).Log("err", err, "msg", "Failed to downloaded block metadata")


double error handling, please choose one

bwplotka · 2019-11-03T20:35:31Z

pkg/block/mdfetcher.go

+			for blockID := range blockIDs {
+				md, err := f.fetchBlockMDIfNotCached(ctx, wg, blockID)
+				if err != nil {
+					level.Debug(f.logger).Log("msg", "failed to fetch block metadata", "block", blockID, "error", err)


do really want to we keep going here? or continue the loop?
If an nil, err is returned, then we'll end up locking and saving nil in the map unnecessarily.

I think we should collect and/or exit on errors. Currently, if there was a transient network error we'll cache nil (negative result) and then never re-try downloading that metadata.

bwplotka · 2019-11-03T20:46:32Z

pkg/store/bucket.go


-	if err != nil {
-		return errors.Wrap(err, "iter")
+		if err := s.addBlock(ctx, blockID); err != nil {


This is important as we download meta.json twice. Once to memory once to disk... Can we improve this?

bwplotka · 2019-11-03T20:58:13Z

pkg/block/mdfetcher.go

+	}
+	wg.Wait()
+
+	f.mu.Lock()


Why this lock? To make fetcher thread-safe? I think we need single flight here TBH with some expiry even (: Can you add TODO(bwplotka): Add single flight with expiry if needed

Also, I am not sure if we can really reuse this lock IMO... I would rather have:
either

two locks, one for concurrency here and second for potential concurrent use of Fetch

remove this lock. I don't think we do currently fetch concurrently so we could remove this and comment that it's not thread safe

I would say let's add TODO

squat · 2019-11-05T15:20:06Z

pkg/block/mdfetcher.go

+			for blockID := range blockIDs {
+				md, err := f.fetchBlockMDIfNotCached(ctx, wg, blockID)
+				if err != nil {
+					level.Debug(f.logger).Log("msg", "failed to fetch block metadata", "block", blockID, "error", err)


do really want to we keep going here? or continue the loop?
If an nil, err is returned, then we'll end up locking and saving nil in the map unnecessarily.

squat · 2019-11-05T15:23:19Z

pkg/block/mdfetcher.go

+	for i := 0; i < f.concurrencyLimit; i++ {
+		go func() {
+			for blockID := range blockIDs {
+				md, err := f.fetchBlockMDIfNotCached(ctx, wg, blockID)


nit, I think it's a bit funny and hard to reason about when the concurrency crosses funcs like this. Maybe just call wg.Done() at the end of the loop block?

squat · 2019-11-05T15:30:20Z

pkg/compact/retention.go

-		m, err := block.DownloadMeta(ctx, logger, bkt, id)
-		if err != nil {
-			return errors.Wrap(err, "download metadata")
+	mdFetcher := block.NewMetadataFetcher(logger, 4, bkt)


Where does this concurrency level come from and why is it different from the others? If 4 is the "right" number, then maybe it should be hardcoded in the block package? I think we need to handle these arguments consistently.

squat · 2019-11-05T15:31:14Z

pkg/shipper/shipper.go

@@ -193,31 +194,27 @@ func newLazyOverlapChecker(logger log.Logger, bucket objstore.Bucket, labels fun
 		labels: labels,

 		lookupMetas: map[ulid.ULID]struct{}{},
+		mdFetcher:   block.NewMetadataFetcher(logger, 3, bucket),


same here, why 3?

squat · 2019-11-05T15:32:42Z

pkg/shipper/shipper.go

-		if err != nil {
-			return err
+	for _, md := range mds {
+		if md == nil {


It looks like every time we want to iterate over the fechted metas we first have to check if the md is nil; why not handle this in Fetch and avoid adding nil metas to the returned slice?

squat · 2019-11-05T15:35:21Z

cmd/thanos/bucket.go

@@ -36,6 +36,10 @@ import (
 	kingpin "gopkg.in/alecthomas/kingpin.v2"
 )

+const (
+	defaultBlockMDFetchConcurrency = 3


default... makes it sound like it can be overridden, but this seems to be a hardcoded value that many different packaged define. Should it instead be defined in the block package one time?

squat · 2019-11-05T16:04:16Z

pkg/block/mdfetcher.go

+				}
+
+				f.mu.Lock()
+				remoteBlockMDs[blockID] = md


I think there is a race currently: we call wg.Done before inserting the returned values in the map, meaning wg.Done() may return early and we may return an incomplete map to the caller.

bwplotka · 2019-11-10T21:36:02Z

@lx223 let us know if you have time to finish this. This fixes quite annoying issue, plus @squat already touching this path for compactor syncer.

If not we are happy to continue this for you (:

lx223 · 2019-11-10T22:08:52Z

@bwplotka sry, might have enough time to finish it fast. Feel free to take it. :)

bwplotka · 2019-11-28T10:09:01Z

Cool, closing this for now then, updating the issue.

lx223 force-pushed the iss-1335-partial-block branch from 01947f2 to 417b9e6 Compare August 9, 2019 23:05

lx223 changed the title ~~Use a shared metadata fetcher to fetch remote block metadata~~ [Compact] Use a shared metadata fetcher to fetch remote block metadata Aug 9, 2019

lx223 force-pushed the iss-1335-partial-block branch 5 times, most recently from ca22f84 to 6bd7cd4 Compare August 11, 2019 20:37

lx223 marked this pull request as ready for review August 11, 2019 20:46

lx223 mentioned this pull request Aug 11, 2019

Handle (ignore) partial uploads on downsampling and apply retention cycles #1335

Closed

bwplotka requested review from bwplotka, povilasv and GiedriusS August 15, 2019 18:18

bwplotka reviewed Aug 15, 2019

View reviewed changes

bwplotka added state: changes-requested feature request/improvement labels Aug 27, 2019

bwplotka mentioned this pull request Aug 27, 2019

Store: Add Time & duration based partitioning #1408

Merged

lx223 force-pushed the iss-1335-partial-block branch 2 times, most recently from d8079e9 to 3680a61 Compare September 14, 2019 23:58

lx223 force-pushed the iss-1335-partial-block branch 2 times, most recently from f67e795 to 3d0b139 Compare September 15, 2019 10:49

bwplotka reviewed Sep 16, 2019

View reviewed changes

bwplotka changed the title ~~[Compact] Use a shared metadata fetcher to fetch remote block metadata~~ Unified fetching/syncing block metadata everywhere Sep 16, 2019

bwplotka mentioned this pull request Sep 16, 2019

Fixed partial delete issues on compactor; Added Upload/Delete tests. #1525

Merged

bwplotka mentioned this pull request Sep 16, 2019

Umbrella issue for RW coordination for object storage. #1528

Closed

9 tasks

lx223 force-pushed the iss-1335-partial-block branch 7 times, most recently from 9f6c3d6 to 1160e83 Compare September 21, 2019 21:31

Use a shared metadata fetcher to fetch remote block metadata

a8cb262

Signed-off-by: Lan Xiao <lx223@users.noreply.github.com>

lx223 force-pushed the iss-1335-partial-block branch from 1160e83 to a8cb262 Compare September 21, 2019 21:40

lx223 commented Sep 21, 2019

View reviewed changes

bwplotka reviewed Nov 3, 2019

View reviewed changes

bwplotka mentioned this pull request Nov 5, 2019

cmd/thanos/compact: add bucket UI #1714

Merged

1 task

squat reviewed Nov 5, 2019

View reviewed changes

bwplotka mentioned this pull request Nov 16, 2019

Compact: Offline deduplication #1276

Closed

bwplotka closed this Nov 28, 2019

Unified fetching/syncing block metadata everywhere #1394

Unified fetching/syncing block metadata everywhere #1394

Conversation

lx223 commented Aug 9, 2019 • edited Loading

Changes

Verification

lx223 commented Aug 11, 2019

bwplotka commented Aug 15, 2019

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka commented Aug 15, 2019

bwplotka commented Sep 13, 2019

lx223 commented Sep 15, 2019

bwplotka commented Sep 16, 2019

bwplotka left a comment

Choose a reason for hiding this comment

Thanos bucket

Store GW:

Compactor:

Verifier

Sidecar

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka commented Sep 16, 2019

bwplotka commented Sep 16, 2019 • edited Loading

lx223 commented Sep 21, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka commented Oct 31, 2019

lx223 commented Nov 3, 2019 • edited Loading

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

squat Nov 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

squat Nov 5, 2019 • edited Loading

Choose a reason for hiding this comment

squat Nov 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka commented Nov 10, 2019

lx223 commented Nov 10, 2019

bwplotka commented Nov 28, 2019

lx223 commented Aug 9, 2019 •

edited

Loading

bwplotka commented Sep 16, 2019 •

edited

Loading

lx223 commented Nov 3, 2019 •

edited

Loading

squat Nov 5, 2019 •

edited

Loading

squat Nov 5, 2019 •

edited

Loading

squat Nov 5, 2019 •

edited

Loading