block/compact: rework consistency check, make writers only write

- It's weird that on upload errors, we try to clean everything and only
  then write again. It's an extra operation we don't need since whether
  a block exists or not hinges on the existence of meta.json. We don't
  need to delete old, same files before trying to upload them again.
- Consequently, we need to always use the _upload_ time, not block
  creation time when checking for consistency or when deleting partially
  uploaded blocks. Directories as such don't exist in object storages,
  it's a client-side "illusion", so we need to iterate through the
  partial block's directory to fetch the last modified date.

Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com>

Author: GiedriusS

pkg/block/block.go

@@ -141,39 +141,29 @@ func upload(ctx context.Context, logger log.Logger, bkt objstore.Bucket, bdir st
 		return errors.Wrap(err, "gather meta file stats")
 	}
 
-	if err := meta.Write(&metaEncoded); err != nil {
-		return errors.Wrap(err, "encode meta file")
-	}
-
 	if err := objstore.UploadDir(ctx, logger, bkt, filepath.Join(bdir, ChunksDirname), path.Join(id.String(), ChunksDirname), options...); err != nil {
-		return cleanUp(logger, bkt, id, errors.Wrap(err, "upload chunks"))
+		return errors.Wrap(err, "upload chunks")
 	}
 
 	if err := objstore.UploadFile(ctx, logger, bkt, filepath.Join(bdir, IndexFilename), path.Join(id.String(), IndexFilename)); err != nil {
-		return cleanUp(logger, bkt, id, errors.Wrap(err, "upload index"))
+		return errors.Wrap(err, "upload index")
+	}
+
+	meta.Thanos.UploadTime = time.Now().UTC()
+	if err := meta.Write(&metaEncoded); err != nil {
+		return errors.Wrap(err, "encode meta file")
 	}
 
 	// Meta.json always need to be uploaded as a last item. This will allow to assume block directories without meta file to be pending uploads.
 	if err := bkt.Upload(ctx, path.Join(id.String(), MetaFilename), strings.NewReader(metaEncoded.String())); err != nil {
-		// Don't call cleanUp here. Despite getting error, meta.json may have been uploaded in certain cases,
-		// and even though cleanUp will not see it yet, meta.json may appear in the bucket later.
-		// (Eg. S3 is known to behave this way when it returns 503 "SlowDown" error).
-		// If meta.json is not uploaded, this will produce partial blocks, but such blocks will be cleaned later.
+		// Syncer always checks if meta.json exists in the next iteration and will retry if it does not.
+		// This is to avoid partial uploads.
 		return errors.Wrap(err, "upload meta file")
 	}
 
 	return nil
 }
 
-func cleanUp(logger log.Logger, bkt objstore.Bucket, id ulid.ULID, err error) error {
-	// Cleanup the dir with an uncancelable context.
-	cleanErr := Delete(context.Background(), logger, bkt, id)
-	if cleanErr != nil {
-		return errors.Wrapf(err, "failed to clean block after upload issue. Partial block in system. Err: %s", cleanErr.Error())
-	}
-	return err
-}
-
 // MarkForDeletion creates a file which stores information about when the block was marked for deletion.
 func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID, details string, markedForDeletion prometheus.Counter) error {
 	deletionMarkFile := path.Join(id.String(), metadata.DeletionMarkFilename)

pkg/block/fetcher.go

@@ -1001,10 +1001,18 @@ func NewConsistencyDelayMetaFilterWithoutMetrics(logger log.Logger, consistencyD
 // Filter filters out blocks that filters blocks that have are created before a specified consistency delay.
 func (f *ConsistencyDelayMetaFilter) Filter(_ context.Context, metas map[ulid.ULID]*metadata.Meta, synced GaugeVec, modified GaugeVec) error {
 	for id, meta := range metas {
+		var metaUploadTime = meta.Thanos.UploadTime
+
+		var tooFresh bool
+		if !metaUploadTime.IsZero() {
+			tooFresh = time.Since(metaUploadTime) < f.consistencyDelay
+		} else {
+			tooFresh = ulid.Now()-id.Time() < uint64(f.consistencyDelay/time.Millisecond)
+		}
+
 		// TODO(khyatisoneji): Remove the checks about Thanos Source
 		//  by implementing delete delay to fetch metas.
-		// TODO(bwplotka): Check consistency delay based on file upload / modification time instead of ULID.
-		if ulid.Now()-id.Time() < uint64(f.consistencyDelay/time.Millisecond) &&
+		if tooFresh &&
 			meta.Thanos.Source != metadata.BucketRepairSource &&
 			meta.Thanos.Source != metadata.CompactorSource &&
 			meta.Thanos.Source != metadata.CompactorRepairSource {

pkg/block/fetcher_test.go

@@ -1069,6 +1069,7 @@ func TestConsistencyDelayMetaFilter_Filter_0(t *testing.T) {
 		u.ULID(now.Add(-20 * time.Hour)): {Thanos: metadata.Thanos{Source: metadata.SidecarSource}},
 		u.ULID(now.Add(-20 * time.Hour)): {Thanos: metadata.Thanos{Source: metadata.ReceiveSource}},
 		u.ULID(now.Add(-20 * time.Hour)): {Thanos: metadata.Thanos{Source: metadata.RulerSource}},
+		u.ULID(now):                      {Thanos: metadata.Thanos{UploadTime: time.Now().Add(-20 * time.Hour), Source: metadata.RulerSource}},
 		u.ULID(now.Add(-20 * time.Hour)): {Thanos: metadata.Thanos{Source: metadata.BucketRepairSource}},
 		u.ULID(now.Add(-20 * time.Hour)): {Thanos: metadata.Thanos{Source: metadata.CompactorSource}},
 		u.ULID(now.Add(-20 * time.Hour)): {Thanos: metadata.Thanos{Source: metadata.CompactorRepairSource}},

pkg/block/metadata/meta.go

@@ -14,6 +14,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/oklog/ulid/v2"
@@ -103,6 +104,10 @@ type Thanos struct {
 
 	// Extensions are used for plugin any arbitrary additional information for block. Optional.
 	Extensions any `json:"extensions,omitempty"`
+
+	// UploadTime is used to track when the meta.json file was uploaded to the object storage
+	// without an extra Attributes call. Used for consistency filter.
+	UploadTime time.Time `json:"upload_time,omitempty"`
 }
 
 type IndexStats struct {

pkg/compact/clean.go

@@ -5,13 +5,15 @@ package compact
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid/v2"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/timestamp"
 	"github.com/thanos-io/objstore"
 
 	"github.com/thanos-io/thanos/pkg/block"
@@ -23,6 +25,34 @@ const (
 	PartialUploadThresholdAge = 2 * 24 * time.Hour
 )
 
+// getOldestModifiedTime returns the oldest modified time of a block in the bucket.
+// If it is not possible to get the last modified timestamp then it falls back to the time
+// encoded in the block's ULID.
+func getOldestModifiedTime(ctx context.Context, blockID ulid.ULID, bkt objstore.Bucket) (time.Time, error) {
+	var lastModifiedTime time.Time
+
+	err := bkt.IterWithAttributes(ctx, bkt.Name(), func(attrs objstore.IterObjectAttributes) error {
+		lm, ok := attrs.LastModified()
+		if !ok {
+			return nil
+		}
+		if lm.After(lastModifiedTime) {
+			lastModifiedTime = lm
+		}
+		return nil
+	}, objstore.WithUpdatedAt(), objstore.WithRecursiveIter())
+
+	if err != nil {
+		return timestamp.Time(int64(blockID.Time())), err
+	}
+
+	if lastModifiedTime.IsZero() {
+		return timestamp.Time(int64(blockID.Time())), fmt.Errorf("no last modified time found for block %s, using block creation time instead", blockID.String())
+	}
+
+	return lastModifiedTime, nil
+}
+
 func BestEffortCleanAbortedPartialUploads(
 	ctx context.Context,
 	logger log.Logger,
@@ -34,23 +64,17 @@ func BestEffortCleanAbortedPartialUploads(
 ) {
 	level.Info(logger).Log("msg", "started cleaning of aborted partial uploads")
 
-	// Delete partial blocks that are older than partialUploadThresholdAge.
-	// TODO(bwplotka): This is can cause data loss if blocks are:
-	// * being uploaded longer than partialUploadThresholdAge
-	// * being uploaded and started after their partialUploadThresholdAge
-	// can be assumed in this case. Keep partialUploadThresholdAge long for now.
-	// Mitigate this by adding ModifiedTime to bkt and check that instead of ULID (block creation time).
 	for id := range partial {
-		if ulid.Now()-id.Time() <= uint64(PartialUploadThresholdAge/time.Millisecond) {
-			// Minimum delay has not expired, ignore for now.
+		lastModifiedTime, err := getOldestModifiedTime(ctx, id, bkt)
+		if err != nil {
+			level.Warn(logger).Log("msg", "failed to get last modified time for block; falling back to block creation time", "block", id, "err", err)
+		}
+		if time.Since(lastModifiedTime) <= PartialUploadThresholdAge {
 			continue
 		}
 
 		deleteAttempts.Inc()
-		level.Info(logger).Log("msg", "found partially uploaded block; marking for deletion", "block", id)
-		// We don't gather any information about deletion marks for partial blocks, so let's simply remove it. We waited
-		// long PartialUploadThresholdAge already.
-		// TODO(bwplotka): Fix some edge cases: https://github.com/thanos-io/thanos/issues/2470 .
+		level.Info(logger).Log("msg", "found partially uploaded block; deleting", "block", id)
 		if err := block.Delete(ctx, logger, bkt, id); err != nil {
 			blockCleanupFailures.Inc()
 			level.Warn(logger).Log("msg", "failed to delete aborted partial upload; will retry in next iteration", "block", id, "thresholdAge", PartialUploadThresholdAge, "err", err)

pkg/compact/clean_test.go

@@ -87,3 +87,12 @@ func TestBestEffortCleanAbortedPartialUploads(t *testing.T) {
 	testutil.Ok(t, err)
 	testutil.Equals(t, true, exists)
 }
+
+func TestGetLastModifiedTime(t *testing.T) {
+	now := time.Now().UTC()
+	u := ulid.MustNewDefault(now)
+	tm, err := getOldestModifiedTime(context.Background(), u, objstore.NewInMemBucket())
+	testutil.NotOk(t, err)
+	// NOTE(GiedriusS): ULIDs use millisecond precision.
+	testutil.Equals(t, now.Truncate(time.Second), tm.Truncate(time.Second))
+}

pkg/shipper/shipper_e2e_test.go

@@ -154,6 +154,7 @@ func TestShipper_SyncBlocks_e2e(t *testing.T) {
 			}
 
 			buf := bytes.Buffer{}
+			meta.Thanos.UploadTime = time.Time{}
 			testutil.Ok(t, meta.Write(&buf))
 
 			// We will delete the fifth block and do not expect it to be re-uploaded later.