Compactor: remove malformed blocks after delay

pkg/block/block.go

@@ -145,3 +145,12 @@ func IsBlockDir(path string) (id ulid.ULID, ok bool) {
 	id, err := ulid.Parse(filepath.Base(path))
 	return id, err == nil
 }
+
+// MetaExists checks whether the meta file exists for the provided block
+func MetaExists(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID) (bool, error) {
+	exists, err := bkt.Exists(ctx, path.Join(id.String(), MetaFilename))
+	if err != nil {
+		return false, errors.Wrapf(err, "meta.json bkt get for %s", id.String())
+	}
+	return exists, nil
+}

pkg/compact/compact.go

@@ -31,6 +31,8 @@ const (
 	ResolutionLevelRaw = ResolutionLevel(downsample.ResLevel0)
 	ResolutionLevel5m  = ResolutionLevel(downsample.ResLevel1)
 	ResolutionLevel1h  = ResolutionLevel(downsample.ResLevel2)
+
+	consistencyDelay = time.Duration(30 * time.Minute)
 )
 
 var blockTooFreshSentinelError = errors.New("Block too fresh")
@@ -149,7 +151,8 @@ func NewSyncer(logger log.Logger, reg prometheus.Registerer, bkt objstore.Bucket
 }
 
 // SyncMetas synchronizes all meta files from blocks in the bucket into
-// the memory.
+// the memory.  It removes any partial blocks older than consistencyDelay
+// from the bucket.
 func (c *Syncer) SyncMetas(ctx context.Context) error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
@@ -194,6 +197,9 @@ func (c *Syncer) syncMetas(ctx context.Context) error {
 					continue
 				}
 				if err != nil {
+					if removed := c.removeIfMalformed(workCtx, id); removed {
+						continue
+					}
 					errChan <- err
 					return
 				}
@@ -271,6 +277,30 @@ func (c *Syncer) downloadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta
 	return &meta, nil
 }
 
+func (c *Syncer) removeIfMalformed(ctx context.Context, id ulid.ULID) bool {
+	exists, err := block.MetaExists(ctx, c.logger, c.bkt, id)
+	if err != nil {
+		level.Warn(c.logger).Log("msg", "failed to check if block is malformed", "block", id)
+		return false
+	}
+	if exists {
+		// Meta exists, block is not malformed.
+		return false
+	}
+
+	if ulid.Now()-id.Time() <= uint64(consistencyDelay/time.Millisecond) {
+		// Consistency delay has not expired, so can't say block is malformed yet.
+		return false
+	}
+
+	if err := block.Delete(ctx, c.bkt, id); err != nil {
+		level.Warn(c.logger).Log("msg", "failed to delete malformed block", "block", id)
+	}
+	level.Info(c.logger).Log("msg", "deleted malformed block", "block", id)
+
+	return true
+}
+
 // GroupKey returns a unique identifier for the group the block belongs to. It considers
 // the downsampling resolution and the block's labels.
 func GroupKey(meta metadata.Meta) string {